batch_scheduler.py

"""
バッチスケジューラークラス
2時、3時、4時の時刻別バッチ処理機能を実装し、
指定時刻の未処理リクエストを処理する機能を提供します。
"""

import asyncio
import os
from datetime import datetime, timedelta
from typing import Dict, Any, List, Optional
import logging
import traceback

from letter_request_manager import RequestManager
from letter_generator import LetterGenerator
from async_storage_manager import AsyncStorageManager
from async_rate_limiter import AsyncRateLimitManager
from letter_user_manager import UserManager

# ログ設定
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class BatchSchedulerError(Exception):
    """バッチスケジューラー関連のエラー"""
    pass


class BatchScheduler:
    """非同期バッチ処理スケジューラー"""
    
    def __init__(self, storage_manager: Optional[AsyncStorageManager] = None):
        """
        バッチスケジューラーを初期化
        
        Args:
            storage_manager: ストレージマネージャー（指定しない場合は新規作成）
        """
        # ストレージマネージャーの初期化
        if storage_manager is None:
            storage_path = os.getenv("STORAGE_PATH", "/mnt/data/letters.json")
            self.storage_manager = AsyncStorageManager(storage_path)
        else:
            self.storage_manager = storage_manager
        
        # 依存コンポーネントの初期化
        self.rate_limiter = AsyncRateLimitManager(self.storage_manager)
        self.request_manager = RequestManager(self.storage_manager, self.rate_limiter)
        self.letter_generator = LetterGenerator()
        self.user_manager = UserManager(self.storage_manager)
        
        # 設定値
        self.available_hours = [2, 3, 4]  # 2時、3時、4時
        self.max_concurrent_generations = int(os.getenv("MAX_CONCURRENT_GENERATIONS", "3"))
        self.generation_timeout = int(os.getenv("GENERATION_TIMEOUT", "300"))  # 5分
        self.retry_failed_requests = os.getenv("RETRY_FAILED_REQUESTS", "true").lower() == "true"
        
        logger.info(f"BatchScheduler初期化完了 - 対象時刻: {self.available_hours}")
    
    async def run_hourly_batch(self, hour: int) -> Dict[str, Any]:
        """
        指定時刻のバッチ処理を実行
        
        Args:
            hour: 実行時刻（2, 3, 4のいずれか）
            
        Returns:
            Dict: バッチ処理の結果
        """
        start_time = datetime.now()
        batch_id = f"{start_time.strftime('%Y%m%d_%H%M%S')}_{hour}"
        
        logger.info(f"=== {hour}時のバッチ処理開始 (ID: {batch_id}) ===")
        
        # 時刻の検証
        if hour not in self.available_hours:
            error_msg = f"無効な時刻が指定されました: {hour}時 (有効: {self.available_hours})"
            logger.error(error_msg)
            return {
                "success": False,
                "error": error_msg,
                "batch_id": batch_id,
                "hour": hour,
                "start_time": start_time.isoformat(),
                "end_time": datetime.now().isoformat()
            }
        
        try:
            # バッチ実行の記録開始
            await self._record_batch_start(batch_id, hour, start_time)
            
            # 指定時刻の未処理リクエストを処理
            result = await self.process_pending_requests_for_hour(hour)
            
            # バッチ実行の記録完了
            end_time = datetime.now()
            await self._record_batch_completion(batch_id, hour, start_time, end_time, result)
            
            execution_time = (end_time - start_time).total_seconds()
            
            logger.info(f"=== {hour}時のバッチ処理完了 (所要時間: {execution_time:.2f}秒) ===")
            
            return {
                "success": True,
                "batch_id": batch_id,
                "hour": hour,
                "start_time": start_time.isoformat(),
                "end_time": end_time.isoformat(),
                "execution_time": execution_time,
                "processed_count": result.get("processed_count", 0),
                "success_count": result.get("success_count", 0),
                "failed_count": result.get("failed_count", 0),
                "errors": result.get("errors", [])
            }
            
        except Exception as e:
            end_time = datetime.now()
            execution_time = (end_time - start_time).total_seconds()
            error_msg = f"バッチ処理中にエラーが発生しました: {str(e)}"
            
            logger.error(f"{error_msg}\n{traceback.format_exc()}")
            
            # エラーの記録
            await self._record_batch_error(batch_id, hour, start_time, end_time, error_msg)
            
            return {
                "success": False,
                "error": error_msg,
                "batch_id": batch_id,
                "hour": hour,
                "start_time": start_time.isoformat(),
                "end_time": end_time.isoformat(),
                "execution_time": execution_time
            }
    
    async def process_pending_requests_for_hour(self, hour: int) -> Dict[str, Any]:
        """
        指定時刻の未処理リクエストを処理
        
        Args:
            hour: 処理対象の時刻
            
        Returns:
            Dict: 処理結果の詳細
        """
        try:
            # 未処理リクエストを取得
            pending_requests = await self.request_manager.get_pending_requests_by_hour(hour)
            
            if not pending_requests:
                logger.info(f"{hour}時の未処理リクエストはありません")
                return {
                    "processed_count": 0,
                    "success_count": 0,
                    "failed_count": 0,
                    "errors": []
                }
            
            logger.info(f"{hour}時の未処理リクエスト数: {len(pending_requests)}")
            
            # 並行処理用のセマフォ
            semaphore = asyncio.Semaphore(self.max_concurrent_generations)
            
            # 各リクエストを並行処理
            tasks = []
            for request in pending_requests:
                task = self._process_single_request(semaphore, request)
                tasks.append(task)
            
            # 全てのタスクを実行
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # 結果を集計
            success_count = 0
            failed_count = 0
            errors = []
            
            for i, result in enumerate(results):
                if isinstance(result, Exception):
                    failed_count += 1
                    error_msg = f"リクエスト処理例外: {str(result)}"
                    errors.append({
                        "request_index": i,
                        "user_id": pending_requests[i].get("user_id", "unknown"),
                        "error": error_msg
                    })
                    logger.error(error_msg)
                elif result.get("success", False):
                    success_count += 1
                else:
                    failed_count += 1
                    errors.append({
                        "request_index": i,
                        "user_id": pending_requests[i].get("user_id", "unknown"),
                        "error": result.get("error", "不明なエラー")
                    })
            
            logger.info(f"処理完了 - 成功: {success_count}, 失敗: {failed_count}")
            
            return {
                "processed_count": len(pending_requests),
                "success_count": success_count,
                "failed_count": failed_count,
                "errors": errors
            }
            
        except Exception as e:
            error_msg = f"未処理リクエスト処理中にエラーが発生: {str(e)}"
            logger.error(f"{error_msg}\n{traceback.format_exc()}")
            return {
                "processed_count": 0,
                "success_count": 0,
                "failed_count": 0,
                "errors": [{"error": error_msg}]
            }
    
    async def _process_single_request(self, semaphore: asyncio.Semaphore, request: Dict[str, Any]) -> Dict[str, Any]:
        """
        単一のリクエストを処理
        
        Args:
            semaphore: 並行処理制御用のセマフォ
            request: 処理対象のリクエスト
            
        Returns:
            Dict: 処理結果
        """
        async with semaphore:
            user_id = request["user_id"]
            theme = request["theme"]
            date = request["date"]
            
            try:
                logger.info(f"手紙生成開始 - ユーザー: {user_id}, テーマ: {theme[:50]}...")
                
                # タイムアウト付きで手紙生成を実行
                user_history = await self.user_manager.get_user_profile(user_id)
                
                generation_task = self.letter_generator.generate_letter(user_id, theme, user_history)
                letter_result = await asyncio.wait_for(generation_task, timeout=self.generation_timeout)
                
                # 生成された手紙をストレージに保存
                await self._save_generated_letter(user_id, date, theme, letter_result)
                
                # リクエストを完了としてマーク
                await self.request_manager.mark_request_processed(user_id, date, "completed")
                
                # ユーザー履歴を更新
                await self.user_manager.update_user_history(user_id, {
                    "date": date,
                    "theme": theme,
                    "status": "completed",
                    "generated_at": datetime.now().isoformat()
                })
                
                logger.info(f"手紙生成完了 - ユーザー: {user_id}")
                
                return {
                    "success": True,
                    "user_id": user_id,
                    "theme": theme,
                    "generation_time": letter_result["metadata"].get("generation_time", 0)
                }
                
            except asyncio.TimeoutError:
                error_msg = f"手紙生成がタイムアウトしました（{self.generation_timeout}秒）"
                logger.error(f"{error_msg} - ユーザー: {user_id}")
                
                await self.request_manager.mark_request_failed(user_id, date, error_msg)
                
                return {
                    "success": False,
                    "user_id": user_id,
                    "error": error_msg
                }
                
            except Exception as e:
                error_msg = f"手紙生成中にエラーが発生: {str(e)}"
                logger.error(f"{error_msg} - ユーザー: {user_id}\n{traceback.format_exc()}")
                
                await self.request_manager.mark_request_failed(user_id, date, error_msg)
                
                return {
                    "success": False,
                    "user_id": user_id,
                    "error": error_msg
                }
    
    async def _save_generated_letter(self, user_id: str, date: str, theme: str, letter_result: Dict[str, Any]) -> None:
        """
        生成された手紙をストレージに保存
        
        Args:
            user_id: ユーザーID
            date: 日付
            theme: テーマ
            letter_result: 生成結果
        """
        try:
            user_data = await self.storage_manager.get_user_data(user_id)
            
            # 手紙データを作成
            letter_data = {
                "theme": theme,
                "content": letter_result["content"],
                "status": "completed",
                "generated_at": datetime.now().isoformat(),
                "metadata": letter_result["metadata"]
            }
            
            # ユーザーデータに追加
            user_data["letters"][date] = letter_data
            user_data["profile"]["total_letters"] = user_data["profile"].get("total_letters", 0) + 1
            user_data["profile"]["last_request"] = date
            
            # ストレージに保存
            await self.storage_manager.update_user_data(user_id, user_data)
            
            logger.info(f"手紙をストレージに保存しました - ユーザー: {user_id}, 日付: {date}")
            
        except Exception as e:
            logger.error(f"手紙保存エラー - ユーザー: {user_id}, 日付: {date}: {str(e)}")
            raise
    
    async def _record_batch_start(self, batch_id: str, hour: int, start_time: datetime) -> None:
        """バッチ実行開始を記録"""
        try:
            system_info = await self.storage_manager.get_system_info()
            
            if "batch_runs" not in system_info:
                system_info["batch_runs"] = {}
            
            system_info["batch_runs"][batch_id] = {
                "hour": hour,
                "start_time": start_time.isoformat(),
                "status": "running"
            }
            
            await self.storage_manager.update_system_info(system_info)
            
        except Exception as e:
            logger.error(f"バッチ開始記録エラー: {str(e)}")
    
    async def _record_batch_completion(self, batch_id: str, hour: int, start_time: datetime, 
                                     end_time: datetime, result: Dict[str, Any]) -> None:
        """バッチ実行完了を記録"""
        try:
            system_info = await self.storage_manager.get_system_info()
            
            if batch_id in system_info.get("batch_runs", {}):
                system_info["batch_runs"][batch_id].update({
                    "end_time": end_time.isoformat(),
                    "status": "completed",
                    "execution_time": (end_time - start_time).total_seconds(),
                    "processed_count": result.get("processed_count", 0),
                    "success_count": result.get("success_count", 0),
                    "failed_count": result.get("failed_count", 0),
                    "error_count": len(result.get("errors", []))
                })
                
                await self.storage_manager.update_system_info(system_info)
            
        except Exception as e:
            logger.error(f"バッチ完了記録エラー: {str(e)}")
    
    async def _record_batch_error(self, batch_id: str, hour: int, start_time: datetime, 
                                end_time: datetime, error_msg: str) -> None:
        """バッチ実行エラーを記録"""
        try:
            system_info = await self.storage_manager.get_system_info()
            
            if batch_id in system_info.get("batch_runs", {}):
                system_info["batch_runs"][batch_id].update({
                    "end_time": end_time.isoformat(),
                    "status": "failed",
                    "execution_time": (end_time - start_time).total_seconds(),
                    "error": error_msg
                })
                
                await self.storage_manager.update_system_info(system_info)
            
        except Exception as e:
            logger.error(f"バッチエラー記録エラー: {str(e)}")
    
    def schedule_all_hours(self) -> None:
        """
        全ての対象時刻でのスケジュール設定
        注意: この関数は実際のスケジューリングライブラリと組み合わせて使用する
        """
        logger.info("バッチスケジュールを設定します")
        
        for hour in self.available_hours:
            logger.info(f"  - {hour}:00 に手紙生成バッチを設定")
        
        # 実際のスケジューリングは外部ライブラリ（APScheduler等）で実装
        # ここでは設定情報のログ出力のみ
        logger.info("スケジュール設定完了（実際の実行は外部スケジューラーが必要）")
    
    async def cleanup_old_data(self, days: int = 90) -> Dict[str, Any]:
        """
        古いデータの自動削除
        
        Args:
            days: 保持日数
            
        Returns:
            Dict: 削除結果
        """
        try:
            logger.info(f"{days}日以前の古いデータを削除します")
            
            # ストレージマネージャーの削除機能を使用
            deleted_count = await self.storage_manager.cleanup_old_data(days)
            
            # リクエストマネージャーの削除機能も使用
            deleted_requests = await self.request_manager.cleanup_old_requests(days)
            
            # バックアップの作成
            backup_path = await self.storage_manager.backup_data()
            
            result = {
                "success": True,
                "deleted_letters": deleted_count,
                "deleted_requests": deleted_requests,
                "backup_created": backup_path,
                "cleanup_date": datetime.now().isoformat()
            }
            
            logger.info(f"古いデータ削除完了: {result}")
            return result
            
        except Exception as e:
            error_msg = f"古いデータ削除中にエラーが発生: {str(e)}"
            logger.error(f"{error_msg}\n{traceback.format_exc()}")
            
            return {
                "success": False,
                "error": error_msg,
                "cleanup_date": datetime.now().isoformat()
            }
    
    async def get_batch_statistics(self, days: int = 7) -> Dict[str, Any]:
        """
        バッチ処理の統計情報を取得
        
        Args:
            days: 統計対象日数
            
        Returns:
            Dict: 統計情報
        """
        try:
            system_info = await self.storage_manager.get_system_info()
            batch_runs = system_info.get("batch_runs", {})
            
            # 指定日数以内のバッチ実行を抽出
            cutoff_date = datetime.now() - timedelta(days=days)
            recent_batches = []
            
            for batch_id, batch_info in batch_runs.items():
                try:
                    start_time = datetime.fromisoformat(batch_info["start_time"])
                    if start_time >= cutoff_date:
                        recent_batches.append(batch_info)
                except (ValueError, KeyError):
                    continue
            
            # 統計を計算
            total_runs = len(recent_batches)
            successful_runs = len([b for b in recent_batches if b.get("status") == "completed"])
            failed_runs = len([b for b in recent_batches if b.get("status") == "failed"])
            
            total_processed = sum(b.get("processed_count", 0) for b in recent_batches)
            total_success = sum(b.get("success_count", 0) for b in recent_batches)
            total_failed = sum(b.get("failed_count", 0) for b in recent_batches)
            
            avg_execution_time = 0
            if recent_batches:
                execution_times = [b.get("execution_time", 0) for b in recent_batches if b.get("execution_time")]
                if execution_times:
                    avg_execution_time = sum(execution_times) / len(execution_times)
            
            # 時刻別統計
            hourly_stats = {hour: {"runs": 0, "processed": 0, "success": 0} for hour in self.available_hours}
            
            for batch in recent_batches:
                hour = batch.get("hour")
                if hour in hourly_stats:
                    hourly_stats[hour]["runs"] += 1
                    hourly_stats[hour]["processed"] += batch.get("processed_count", 0)
                    hourly_stats[hour]["success"] += batch.get("success_count", 0)
            
            return {
                "period_days": days,
                "total_runs": total_runs,
                "successful_runs": successful_runs,
                "failed_runs": failed_runs,
                "success_rate": (successful_runs / total_runs * 100) if total_runs > 0 else 0,
                "total_processed": total_processed,
                "total_success": total_success,
                "total_failed": total_failed,
                "processing_success_rate": (total_success / total_processed * 100) if total_processed > 0 else 0,
                "avg_execution_time": avg_execution_time,
                "hourly_stats": hourly_stats,
                "generated_at": datetime.now().isoformat()
            }
            
        except Exception as e:
            logger.error(f"統計情報取得エラー: {str(e)}")
            return {"error": str(e)}
    
    async def test_batch_processing(self, test_hour: int = 2) -> Dict[str, Any]:
        """
        バッチ処理のテスト実行
        
        Args:
            test_hour: テスト対象の時刻
            
        Returns:
            Dict: テスト結果
        """
        try:
            logger.info(f"バッチ処理テストを開始します（{test_hour}時）")
            
            # API接続テスト
            api_status = await self.letter_generator.check_api_connections()
            if not api_status.get("overall", False):
                return {
                    "success": False,
                    "error": "API接続テストに失敗しました",
                    "api_status": api_status
                }
            
            # テスト用のバッチ処理を実行
            result = await self.run_hourly_batch(test_hour)
            
            return {
                "success": True,
                "message": "バッチ処理テスト完了",
                "api_status": api_status,
                "batch_result": result
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": f"バッチ処理テスト中にエラーが発生: {str(e)}"
            }


# テスト用の関数
async def test_batch_scheduler():
    """BatchSchedulerのテスト"""
    import tempfile
    import uuid
    
    # 一時ディレクトリでテスト
    with tempfile.TemporaryDirectory() as temp_dir:
        test_file = os.path.join(temp_dir, "test_letters.json")
        storage = AsyncStorageManager(test_file)
        scheduler = BatchScheduler(storage)
        
        print("=== BatchSchedulerテスト開始 ===")
        
        # テスト用のリクエストを作成
        user_id = str(uuid.uuid4())
        success, message = await scheduler.request_manager.submit_request(user_id, "テストテーマ", 2)
        print(f"✓ テストリクエスト作成: {success} - {message}")
        
        # バッチ処理テスト
        result = await scheduler.run_hourly_batch(2)
        print(f"✓ バッチ処理テスト: {result['success']}")
        
        # 統計情報テスト
        stats = await scheduler.get_batch_statistics()
        print(f"✓ 統計情報取得テスト: {stats}")
        
        # 古いデータ削除テスト
        cleanup_result = await scheduler.cleanup_old_data(0)  # 全て削除
        print(f"✓ データ削除テスト: {cleanup_result['success']}")
        
        print("=== 全てのテストが完了しました！ ===")


if __name__ == "__main__":
    asyncio.run(test_batch_scheduler())