""" CSVファイルからチーム情報を読み込んでメール送信するDjango管理コマンド Outlook SMTP (rogaining@gifuai.net) での送信に対応 Usage: # ドライラン（実際には送信しない） docker compose exec app python manage.py send_team_emails --csv_file="CPLIST/input/team_mail.csv" --dry_run # 実際のメール送信（送信間隔1秒） docker compose exec app python manage.py send_team_emails --csv_file="CPLIST/input/team_mail.csv" # カスタム送信間隔（3秒間隔） docker compose exec app python manage.py send_team_emails --csv_file="CPLIST/input/team_mail.csv" --delay=3 Author: システム開発チーム Date: 2025-09-05 """ import csv import os import time from django.core.management.base import BaseCommand, CommandError from django.core.mail import send_mail from django.conf import settings from django.template.loader import render_to_string from datetime import datetime class Command(BaseCommand): help = 'CSVファイルからチーム情報を読み込んでOutlook SMTPでメール送信' def add_arguments(self, parser): parser.add_argument( '--csv_file', type=str, required=True, help='CSVファイルのパス (例: CPLIST/input/team_mail.csv)' ) parser.add_argument( '--dry_run', action='store_true', help='ドライラン（実際にはメール送信しない）' ) parser.add_argument( '--delay', type=int, default=1, help='メール送信間隔（秒）デフォルト: 1秒' ) parser.add_argument( '--test_email', type=str, help='テスト用メールアドレス（指定した場合、全てのメールをこのアドレスに送信）' ) def handle(self, *args, **options): csv_file = options['csv_file'] dry_run = options['dry_run'] delay = options['delay'] test_email = options.get('test_email') mode = '[DRY RUN] ' if dry_run else '' self.stdout.write(f'{mode}CSVメール送信開始: csv_file={csv_file}') if test_email: self.stdout.write(f'🧪 テストモード: 全メールを {test_email} に送信') # CSVファイル存在確認 if not os.path.exists(csv_file): raise CommandError(f'CSVファイルが見つかりません: {csv_file}') # SMTP設定確認 self.verify_email_settings() # 統計情報 stats = { 'total_rows': 0, 'emails_sent': 0, 'errors': [] } # CSVファイル読み込み・メール送信 try: with open(csv_file, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row_num, row in enumerate(reader, start=2): stats['total_rows'] += 1 try: # CSVデータ取得 team_data = self.extract_team_data(row) # メール内容生成 email_content = self.generate_email_content(team_data) subject = email_content['subject'] body = email_content['body'] # 送信先決定（テストモードならtest_emailに） recipient = test_email if test_email else team_data['email'] self.stdout.write(f'{mode}行 {row_num}: {team_data["team_name"]} → {recipient}') if dry_run: self.show_email_preview(subject, body) else: # 実際のメール送信 self.send_email(subject, body, recipient, team_data) self.stdout.write(f' ✅ メール送信完了') # 送信間隔 if delay > 0: time.sleep(delay) stats['emails_sent'] += 1 except Exception as e: error_msg = f'行 {row_num}: {str(e)}' stats['errors'].append(error_msg) self.stdout.write(self.style.ERROR(error_msg)) continue except Exception as e: raise CommandError(f'CSVファイル読み込みエラー: {str(e)}') # 結果レポート self.print_stats(stats, dry_run) def verify_email_settings(self): """SMTP設定確認""" required_settings = [ 'EMAIL_HOST', 'EMAIL_PORT', 'EMAIL_HOST_USER', 'EMAIL_HOST_PASSWORD', 'DEFAULT_FROM_EMAIL' ] for setting in required_settings: if not hasattr(settings, setting) or not getattr(settings, setting): raise CommandError(f'メール設定が不完全です: {setting}') self.stdout.write(f'📧 SMTP設定確認完了: {settings.EMAIL_HOST}:{settings.EMAIL_PORT}') self.stdout.write(f'📧 送信者: {settings.DEFAULT_FROM_EMAIL}') def extract_team_data(self, row): """CSVデータからチーム情報を抽出""" team_data = { 'team_name': row.get('チーム名', '').strip(), 'email': row.get('メール', '').strip(), 'category': row.get('部門', '').strip(), 'duration': row.get('時間', '').strip(), 'member_name': row.get('氏名１', '').strip(), 'phone': row.get('電話', '').strip(), } # 必須項目チェック if not team_data['team_name']: raise ValueError('チーム名が必要です') if not team_data['email']: raise ValueError('メールアドレスが必要です') return team_data def generate_email_content(self, team_data): """メール内容生成（外部テンプレートファイル使用）""" # テンプレートファイルから件名を読み込み subject = render_to_string( 'emails/team_registration_subject.txt', team_data ).strip() # テンプレートファイルから本文を読み込み body = render_to_string( 'emails/team_registration_body.txt', team_data ).strip() return { 'subject': subject, 'body': body } def show_email_preview(self, subject, body): """メールプレビュー表示（ドライラン用）""" self.stdout.write(f' 件名: {subject}') self.stdout.write(f' 本文プレビュー: {body[:100]}...') def send_email(self, subject, body, recipient, team_data): """実際のメール送信""" try: send_mail( subject=subject, message=body, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[recipient], fail_silently=False ) except Exception as e: raise ValueError(f'メール送信エラー ({recipient}): {str(e)}') def print_stats(self, stats, dry_run): """統計情報の表示""" mode = '[DRY RUN] ' if dry_run else '' self.stdout.write(self.style.SUCCESS('\n' + '='*50)) self.stdout.write(self.style.SUCCESS(f'{mode}メール送信結果')) self.stdout.write(self.style.SUCCESS('='*50)) self.stdout.write(f'処理行数: {stats["total_rows"]}') self.stdout.write(f'メール送信数: {stats["emails_sent"]}') if stats['errors']: self.stdout.write(f'エラー数: {len(stats["errors"])}') for error in stats['errors']: self.stdout.write(f' {error}') if not dry_run: self.stdout.write(self.style.SUCCESS('\nメール送信完了！')) else: self.stdout.write(self.style.WARNING('\n※ ドライランのため、実際のメール送信は行われていません')) def send_email_to_team(self, row, template_content, template_name, row_num, stats): """実際のメール送信""" team_name = row.get('チーム名', '').strip() email = row.get('メール', '').strip() category = row.get('部門', '').strip() duration = row.get('時間', '').strip() leader_name = row.get('氏名１', '').strip() phone = row.get('電話番号', '').strip() if not email: raise ValueError('メールアドレスが必要です') if not team_name: raise ValueError('チーム名が必要です') # テンプレート処理 context_data = { 'event_name': '岐阜ロゲイニング2025', 'team_name': team_name, 'category': category, 'duration': duration, 'leader_name': leader_name, 'email': email, 'phone': phone, 'password': row.get('パスワード', '').strip() } # テンプレートファイルから件名・本文を生成 subject = render_to_string('emails/team_registration_subject.txt', context_data).strip() body = render_to_string('emails/team_registration_body.txt', context_data).strip() # メール送信 try: send_mail( subject=subject, message=body, from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[email], fail_silently=False, ) self.stdout.write(f'行 {row_num}: メール送信完了 {team_name} ({email})') stats['emails_sent'] += 1 except Exception as e: raise ValueError(f'メール送信エラー: {str(e)}') def print_stats(self, stats, dry_run): """統計情報の表示""" mode = '[DRY RUN] ' if dry_run else '' self.stdout.write(self.style.SUCCESS('\n' + '='*50)) self.stdout.write(self.style.SUCCESS(f'{mode}メール送信結果')) self.stdout.write(self.style.SUCCESS('='*50)) self.stdout.write(f'処理行数: {stats["total_rows"]}') self.stdout.write(f'メール送信数: {stats["emails_sent"]}') if stats['errors']: self.stdout.write(f'\nエラー数: {len(stats["errors"])}') for error in stats['errors']: self.stdout.write(f' {error}') if dry_run: self.stdout.write(self.style.WARNING('\n※ ドライランのため、実際にはメールは送信されていません')) else: self.stdout.write(self.style.SUCCESS('\nメール送信完了！'))