checkin status tool
This commit is contained in:
528
register_event_users.py
Normal file
528
register_event_users.py
Normal file
@ -0,0 +1,528 @@
|
||||
#!/usr/bin/env python
|
||||
"""
|
||||
イベントユーザー登録スクリプト
|
||||
|
||||
外部システムAPI仕様書.mdを前提に、ユーザーデータCSVから、
|
||||
各ユーザーごとにユーザー登録、チーム登録、エントリー登録、イベント参加を行う
|
||||
docker composeで実施するPythonスクリプト
|
||||
|
||||
使用方法:
|
||||
python register_event_users.py --event_code 大垣2509
|
||||
|
||||
ユーザーデータのCSVは以下の項目を持つ：
|
||||
部門別数,時間,部門,チーム名,メール,パスワード,電話番号,氏名１,誕生日１,氏名２,誕生日２,氏名３,誕生日３,氏名４,誕生日４,氏名５,誕生日５,氏名６,誕生日６,氏名７,誕生日７,,
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import csv
|
||||
import requests
|
||||
import argparse
|
||||
import logging
|
||||
from datetime import datetime, date
|
||||
import time
|
||||
import json
|
||||
from typing import Dict, List, Optional, Tuple
|
||||
|
||||
# ログ設定
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(levelname)s - %(message)s',
|
||||
handlers=[
|
||||
logging.FileHandler('register_event_users.log'),
|
||||
logging.StreamHandler()
|
||||
]
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class EventUserRegistration:
|
||||
def __init__(self, event_code: str, base_url: str = "http://localhost:8000", dry_run: bool = False):
|
||||
"""
|
||||
イベントユーザー登録クラス
|
||||
|
||||
Args:
|
||||
event_code: イベントコード（例: 大垣2509）
|
||||
base_url: APIベースURL
|
||||
dry_run: テスト実行フラグ
|
||||
"""
|
||||
self.event_code = event_code
|
||||
self.base_url = base_url.rstrip('/')
|
||||
self.dry_run = dry_run
|
||||
self.session = requests.Session()
|
||||
self.admin_token = None
|
||||
|
||||
# 統計情報
|
||||
self.stats = {
|
||||
'processed_teams': 0,
|
||||
'users_created': 0,
|
||||
'users_updated': 0,
|
||||
'teams_registered': 0,
|
||||
'entries_created': 0,
|
||||
'participations_created': 0,
|
||||
'errors': []
|
||||
}
|
||||
|
||||
logger.info(f"Event User Registration initialized for event: {event_code}")
|
||||
if dry_run:
|
||||
logger.info("DRY RUN MODE - No actual API calls will be made")
|
||||
|
||||
def get_or_create_user(self, email: str, password: str, firstname: str, lastname: str,
|
||||
date_of_birth: str, phone: str) -> Tuple[bool, Optional[str], Optional[str]]:
|
||||
"""
|
||||
メールアドレスをキーに既存ユーザーを取得、存在しなければ新規作成
|
||||
|
||||
Args:
|
||||
email: メールアドレス
|
||||
password: パスワード
|
||||
firstname: 名前
|
||||
lastname: 姓
|
||||
date_of_birth: 生年月日 (YYYY/MM/DD形式)
|
||||
phone: 電話番号
|
||||
|
||||
Returns:
|
||||
Tuple[success, user_id, token]
|
||||
"""
|
||||
if self.dry_run:
|
||||
logger.info(f"[DRY RUN] Would get or create user: {email}")
|
||||
return True, "dummy_user_id", "dummy_token"
|
||||
|
||||
try:
|
||||
# まずログインを試行して既存ユーザーかチェック
|
||||
login_data = {
|
||||
"identifier": email,
|
||||
"password": password
|
||||
}
|
||||
|
||||
response = self.session.post(f"{self.base_url}/login/", json=login_data)
|
||||
|
||||
if response.status_code == 200:
|
||||
# 既存ユーザーの場合、パスワード更新（実際にはパスワード更新APIが必要）
|
||||
result = response.json()
|
||||
token = result.get('token')
|
||||
user_id = result.get('user', {}).get('id')
|
||||
|
||||
logger.info(f"既存ユーザーでログイン成功: {email}")
|
||||
self.stats['users_updated'] += 1
|
||||
return True, str(user_id), token
|
||||
|
||||
elif response.status_code == 401:
|
||||
# ユーザーが存在しないか、パスワードが間違っている場合、新規登録を試行
|
||||
return self._create_new_user(email, password, firstname, lastname, date_of_birth)
|
||||
|
||||
else:
|
||||
logger.error(f"ログイン試行でエラー: {response.status_code} - {response.text}")
|
||||
return False, None, None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"ユーザー認証エラー: {str(e)}")
|
||||
return False, None, None
|
||||
|
||||
def _create_new_user(self, email: str, password: str, firstname: str, lastname: str,
|
||||
date_of_birth: str) -> Tuple[bool, Optional[str], Optional[str]]:
|
||||
"""
|
||||
新規ユーザーを作成
|
||||
"""
|
||||
try:
|
||||
# 生年月日をYYYY-MM-DD形式に変換
|
||||
if '/' in date_of_birth:
|
||||
date_parts = date_of_birth.split('/')
|
||||
if len(date_parts) == 3:
|
||||
birth_date = f"{date_parts[0]}-{date_parts[1].zfill(2)}-{date_parts[2].zfill(2)}"
|
||||
else:
|
||||
birth_date = "1990-01-01" # デフォルト値
|
||||
else:
|
||||
birth_date = date_of_birth
|
||||
|
||||
# 仮ユーザー登録
|
||||
register_data = {
|
||||
"email": email,
|
||||
"password": password,
|
||||
"firstname": firstname,
|
||||
"lastname": lastname,
|
||||
"date_of_birth": birth_date,
|
||||
"female": False, # デフォルト値
|
||||
"is_rogaining": True
|
||||
}
|
||||
|
||||
response = self.session.post(f"{self.base_url}/register/", json=register_data)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
logger.info(f"仮ユーザー登録成功: {email}")
|
||||
|
||||
# 実際のシステムでは、メール認証コードを使って本登録を完了する必要があります
|
||||
# ここでは簡略化のため、直接ログインを試行します
|
||||
time.sleep(1) # 少し待機
|
||||
|
||||
login_data = {
|
||||
"identifier": email,
|
||||
"password": password
|
||||
}
|
||||
|
||||
login_response = self.session.post(f"{self.base_url}/login/", json=login_data)
|
||||
|
||||
if login_response.status_code == 200:
|
||||
result = login_response.json()
|
||||
token = result.get('token')
|
||||
user_id = result.get('user', {}).get('id')
|
||||
|
||||
logger.info(f"新規ユーザーのログイン成功: {email}")
|
||||
self.stats['users_created'] += 1
|
||||
return True, str(user_id), token
|
||||
else:
|
||||
logger.warning(f"新規ユーザーのログインに失敗: {email}")
|
||||
# メール認証が必要な可能性があります
|
||||
self.stats['users_created'] += 1
|
||||
return True, "pending_verification", None
|
||||
|
||||
else:
|
||||
error_msg = response.text
|
||||
logger.error(f"ユーザー登録失敗: {email} - {error_msg}")
|
||||
return False, None, None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"新規ユーザー作成エラー: {str(e)}")
|
||||
return False, None, None
|
||||
|
||||
def register_team_and_members(self, team_data: Dict, zekken_number: int) -> Tuple[bool, Optional[str]]:
|
||||
"""
|
||||
チーム登録とメンバー登録
|
||||
|
||||
Args:
|
||||
team_data: チームデータ（CSVの1行分）
|
||||
zekken_number: ゼッケン番号
|
||||
|
||||
Returns:
|
||||
Tuple[success, team_id]
|
||||
"""
|
||||
if self.dry_run:
|
||||
logger.info(f"[DRY RUN] Would register team: {team_data['チーム名']} with zekken: {zekken_number}")
|
||||
return True, "dummy_team_id"
|
||||
|
||||
try:
|
||||
# チーム登録データを準備
|
||||
register_data = {
|
||||
"zekken_number": zekken_number,
|
||||
"event_code": self.event_code,
|
||||
"team_name": team_data['チーム名'],
|
||||
"class_name": team_data['部門'],
|
||||
"password": team_data['パスワード']
|
||||
}
|
||||
|
||||
# チーム登録API呼び出し
|
||||
response = self.session.post(f"{self.base_url}/register_team", json=register_data)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
result = response.json()
|
||||
if result.get('status') == 'OK':
|
||||
team_id = result.get('team_id')
|
||||
logger.info(f"チーム登録成功: {team_data['チーム名']} (zekken: {zekken_number})")
|
||||
self.stats['teams_registered'] += 1
|
||||
|
||||
# メンバー登録
|
||||
success = self._register_team_members(team_data, team_id)
|
||||
return success, str(team_id)
|
||||
else:
|
||||
logger.error(f"チーム登録エラー: {result.get('message')}")
|
||||
return False, None
|
||||
else:
|
||||
logger.error(f"チーム登録API呼び出し失敗: {response.status_code} - {response.text}")
|
||||
return False, None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"チーム登録エラー: {str(e)}")
|
||||
return False, None
|
||||
|
||||
def _register_team_members(self, team_data: Dict, team_id: str) -> bool:
|
||||
"""
|
||||
チームメンバーを登録（最大7名）
|
||||
|
||||
Args:
|
||||
team_data: チームデータ
|
||||
team_id: チームID
|
||||
|
||||
Returns:
|
||||
成功フラグ
|
||||
"""
|
||||
if self.dry_run:
|
||||
logger.info(f"[DRY RUN] Would register team members for team: {team_id}")
|
||||
return True
|
||||
|
||||
try:
|
||||
success_count = 0
|
||||
|
||||
# メンバー1-7を順番に処理
|
||||
for i in range(1, 8):
|
||||
name_key = f'氏名{i}'
|
||||
birth_key = f'誕生日{i}'
|
||||
|
||||
if name_key in team_data and team_data[name_key].strip():
|
||||
name = team_data[name_key].strip()
|
||||
birth_date = team_data.get(birth_key, '1990/01/01')
|
||||
|
||||
# ダミーメールアドレスを生成
|
||||
dummy_email = f"{name.replace(' ', '')}_{team_id}_{i}@dummy.local"
|
||||
|
||||
# メンバー追加データ
|
||||
member_data = {
|
||||
"email": dummy_email,
|
||||
"firstname": name.split()[0] if ' ' in name else name,
|
||||
"lastname": name.split()[-1] if ' ' in name else "",
|
||||
"date_of_birth": birth_date.replace('/', '-'),
|
||||
"female": False # デフォルト値
|
||||
}
|
||||
|
||||
# メンバー追加API呼び出し
|
||||
response = self.session.post(
|
||||
f"{self.base_url}/teams/{team_id}/members/",
|
||||
json=member_data
|
||||
)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
logger.info(f"メンバー追加成功: {name} -> チーム{team_id}")
|
||||
success_count += 1
|
||||
else:
|
||||
logger.warning(f"メンバー追加失敗: {name} - {response.text}")
|
||||
|
||||
logger.info(f"チーム{team_id}のメンバー登録完了: {success_count}名")
|
||||
return success_count > 0
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"メンバー登録エラー: {str(e)}")
|
||||
return False
|
||||
|
||||
def create_event_entry(self, team_id: str, category_id: int = 1) -> Tuple[bool, Optional[str]]:
|
||||
"""
|
||||
イベントエントリー登録
|
||||
|
||||
Args:
|
||||
team_id: チームID
|
||||
category_id: カテゴリID
|
||||
|
||||
Returns:
|
||||
Tuple[success, entry_id]
|
||||
"""
|
||||
if self.dry_run:
|
||||
logger.info(f"[DRY RUN] Would create event entry for team: {team_id}")
|
||||
return True, "dummy_entry_id"
|
||||
|
||||
try:
|
||||
# エントリーデータ準備
|
||||
entry_data = {
|
||||
"team_id": team_id,
|
||||
"event_code": self.event_code,
|
||||
"category": category_id,
|
||||
"entry_date": datetime.now().strftime("%Y-%m-%d")
|
||||
}
|
||||
|
||||
# エントリー登録API呼び出し
|
||||
response = self.session.post(f"{self.base_url}/entry/", json=entry_data)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
result = response.json()
|
||||
entry_id = result.get('id') or result.get('entry_id')
|
||||
logger.info(f"エントリー登録成功: team_id={team_id}, entry_id={entry_id}")
|
||||
self.stats['entries_created'] += 1
|
||||
return True, str(entry_id)
|
||||
else:
|
||||
logger.error(f"エントリー登録失敗: {response.status_code} - {response.text}")
|
||||
return False, None
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"エントリー登録エラー: {str(e)}")
|
||||
return False, None
|
||||
|
||||
def participate_in_event(self, entry_id: str, zekken_number: int) -> bool:
|
||||
"""
|
||||
イベント参加処理
|
||||
|
||||
Args:
|
||||
entry_id: エントリーID
|
||||
zekken_number: ゼッケン番号
|
||||
|
||||
Returns:
|
||||
成功フラグ
|
||||
"""
|
||||
if self.dry_run:
|
||||
logger.info(f"[DRY RUN] Would participate in event: entry_id={entry_id}, zekken={zekken_number}")
|
||||
return True
|
||||
|
||||
try:
|
||||
# イベント参加データ準備
|
||||
participation_data = {
|
||||
"entry_id": entry_id,
|
||||
"event_code": self.event_code,
|
||||
"zekken_number": zekken_number,
|
||||
"participation_date": datetime.now().strftime("%Y-%m-%d")
|
||||
}
|
||||
|
||||
# イベント参加API呼び出し（実際のAPIエンドポイントに合わせて調整が必要）
|
||||
response = self.session.post(f"{self.base_url}/start_from_rogapp", json=participation_data)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
logger.info(f"イベント参加成功: entry_id={entry_id}, zekken={zekken_number}")
|
||||
self.stats['participations_created'] += 1
|
||||
return True
|
||||
else:
|
||||
logger.warning(f"イベント参加API呼び出し結果: {response.status_code} - {response.text}")
|
||||
# 参加処理は必須ではないため、警告のみでTrueを返す
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"イベント参加エラー: {str(e)}")
|
||||
return True # 参加処理は必須ではないため、エラーでもTrueを返す
|
||||
|
||||
def process_csv_file(self, csv_file_path: str) -> bool:
|
||||
"""
|
||||
CSVファイルを処理してユーザー登録からイベント参加まで実行
|
||||
|
||||
Args:
|
||||
csv_file_path: CSVファイルパス
|
||||
|
||||
Returns:
|
||||
成功フラグ
|
||||
"""
|
||||
try:
|
||||
if not os.path.exists(csv_file_path):
|
||||
logger.error(f"CSVファイルが見つかりません: {csv_file_path}")
|
||||
return False
|
||||
|
||||
with open(csv_file_path, 'r', encoding='utf-8') as file:
|
||||
csv_reader = csv.DictReader(file)
|
||||
|
||||
for row_num, row in enumerate(csv_reader, start=1):
|
||||
try:
|
||||
self._process_team_row(row, row_num)
|
||||
|
||||
# API呼び出し間隔を空ける
|
||||
if not self.dry_run:
|
||||
time.sleep(0.5)
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"行{row_num}の処理でエラー: {str(e)}"
|
||||
logger.error(error_msg)
|
||||
self.stats['errors'].append(error_msg)
|
||||
continue
|
||||
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"CSVファイル処理エラー: {str(e)}")
|
||||
return False
|
||||
|
||||
def _process_team_row(self, row: Dict, row_num: int):
|
||||
"""
|
||||
CSVの1行（1チーム）を処理
|
||||
|
||||
Args:
|
||||
row: CSV行データ
|
||||
row_num: 行番号
|
||||
"""
|
||||
team_name = row.get('チーム名', '').strip()
|
||||
email = row.get('メール', '').strip()
|
||||
password = row.get('password', '').strip()
|
||||
phone = row.get('電話番号', '').strip()
|
||||
|
||||
if not all([team_name, email, password]):
|
||||
logger.warning(f"行{row_num}: 必須項目が不足 - チーム名={team_name}, メール={email}")
|
||||
return
|
||||
|
||||
logger.info(f"行{row_num}の処理開始: チーム={team_name}")
|
||||
|
||||
# ゼッケン番号を生成（行番号ベース、実際の運用では別途管理が必要）
|
||||
zekken_number = row_num
|
||||
|
||||
# 2-1. カスタムユーザー登録
|
||||
# 最初のメンバー（氏名１）をメインユーザーとして使用
|
||||
firstname = row.get('氏名１', team_name).strip()
|
||||
lastname = ""
|
||||
if ' ' in firstname:
|
||||
parts = firstname.split(' ', 1)
|
||||
firstname = parts[0]
|
||||
lastname = parts[1]
|
||||
|
||||
date_of_birth = row.get('誕生日１', '1990/01/01')
|
||||
|
||||
user_success, user_id, token = self.get_or_create_user(
|
||||
email, password, firstname, lastname, date_of_birth, phone
|
||||
)
|
||||
|
||||
if not user_success:
|
||||
logger.error(f"行{row_num}: ユーザー登録/取得失敗")
|
||||
return
|
||||
|
||||
# 2-2. チーム登録、メンバー登録
|
||||
team_success, team_id = self.register_team_and_members(row, zekken_number)
|
||||
|
||||
if not team_success:
|
||||
logger.error(f"行{row_num}: チーム登録失敗")
|
||||
return
|
||||
|
||||
# 2-3. エントリー登録
|
||||
entry_success, entry_id = self.create_event_entry(team_id)
|
||||
|
||||
if not entry_success:
|
||||
logger.error(f"行{row_num}: エントリー登録失敗")
|
||||
return
|
||||
|
||||
# 2-4. イベント参加
|
||||
participation_success = self.participate_in_event(entry_id, zekken_number)
|
||||
|
||||
if participation_success:
|
||||
logger.info(f"行{row_num}: 全処理完了 - チーム={team_name}, zekken={zekken_number}")
|
||||
self.stats['processed_teams'] += 1
|
||||
else:
|
||||
logger.warning(f"行{row_num}: イベント参加処理で警告")
|
||||
|
||||
def print_statistics(self):
|
||||
"""
|
||||
処理統計を出力
|
||||
"""
|
||||
logger.info("=== 処理統計 ===")
|
||||
logger.info(f"処理完了チーム数: {self.stats['processed_teams']}")
|
||||
logger.info(f"作成ユーザー数: {self.stats['users_created']}")
|
||||
logger.info(f"更新ユーザー数: {self.stats['users_updated']}")
|
||||
logger.info(f"登録チーム数: {self.stats['teams_registered']}")
|
||||
logger.info(f"作成エントリー数: {self.stats['entries_created']}")
|
||||
logger.info(f"参加登録数: {self.stats['participations_created']}")
|
||||
logger.info(f"エラー数: {len(self.stats['errors'])}")
|
||||
|
||||
if self.stats['errors']:
|
||||
logger.error("エラー詳細:")
|
||||
for error in self.stats['errors']:
|
||||
logger.error(f" - {error}")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='イベントユーザー登録スクリプト')
|
||||
parser.add_argument('--event_code', required=True, help='イベントコード（例: 大垣2509）')
|
||||
parser.add_argument('--csv_file', default='CPLIST/input/team2025.csv', help='CSVファイルパス')
|
||||
parser.add_argument('--base_url', default='http://localhost:8000', help='APIベースURL')
|
||||
parser.add_argument('--dry_run', action='store_true', help='テスト実行（実際のAPI呼び出しなし）')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
logger.info(f"イベントユーザー登録処理開始: event_code={args.event_code}")
|
||||
|
||||
# 登録処理実行
|
||||
registration = EventUserRegistration(
|
||||
event_code=args.event_code,
|
||||
base_url=args.base_url,
|
||||
dry_run=args.dry_run
|
||||
)
|
||||
|
||||
success = registration.process_csv_file(args.csv_file)
|
||||
|
||||
# 統計出力
|
||||
registration.print_statistics()
|
||||
|
||||
if success:
|
||||
logger.info("処理が正常に完了しました")
|
||||
return 0
|
||||
else:
|
||||
logger.error("処理中にエラーが発生しました")
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user