#!/usr/bin/env python3
"""
task_scheduler.py — 自动化消费任务调度器

职责:
  1. 定时触发数据清洗任务 (每小时)
  2. 定时触发批处理分析任务 (每5分钟)
  3. 定时生成报表 (每日/每周)
  4. 定时导出数据 (每日)
  5. 定时执行质量检查 (每10分钟)

调度配置:
  - clean_task: 每小时执行
  - analytics_task: 每5分钟执行
  - report_daily: 每天 08:00, 20:00
  - report_weekly: 每周一 08:00
  - export_task: 每天 01:00
  - quality_check: 每10分钟
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import threading
import time
from collections import defaultdict
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Optional

# ── 配置 ─────────────────────────────────────────────────────────

APP_DIR = Path(__file__).resolve().parent.parent.parent
TASKS_DIR = APP_DIR / "logs" / "scheduler"
TASK_LOG = TASKS_DIR / "task_history.ndjson"
TASK_STATE_FILE = TASKS_DIR / "task_state.json"

# 默认调度间隔 (秒)
SCHEDULE = {
    "clean_task": 3600,       # 1小时
    "analytics_task": 300,    # 5分钟
    "report_daily": 43200,    # 12小时 (每天两次)
    "report_weekly": 604800,  # 7天
    "export_task": 86400,     # 24小时
    "quality_check": 600,     # 10分钟
    "strategy_sync": 300,     # 5分钟
    "insights_sync": 300,     # 5分钟
}


# ── 数据类 ───────────────────────────────────────────────────────

@dataclass
class Task:
    """任务定义"""
    name: str
    func: Callable
    interval_sec: int
    enabled: bool = True
    timeout_sec: int = 300
    retry_on_fail: bool = True
    max_retries: int = 3


@dataclass
class TaskResult:
    """任务执行结果"""
    task_name: str = ""
    started_at: str = ""
    completed_at: str = ""
    duration_sec: float = 0.0
    success: bool = False
    output: str = ""
    error: str = ""
    retries: int = 0


@dataclass
class SchedulerState:
    """调度器状态"""
    last_runs: dict[str, str] = field(default_factory=dict)
    next_runs: dict[str, str] = field(default_factory=dict)
    failures: dict[str, int] = field(default_factory=dict)
    total_runs: dict[str, int] = field(default_factory=dict)


# ── 任务执行器 ───────────────────────────────────────────────────

class TaskExecutor:
    """任务执行器 (带超时和重试)"""

    def __init__(self, max_retries: int = 3, timeout_sec: int = 300):
        self.max_retries = max_retries
        self.timeout_sec = timeout_sec

    def execute(self, task: Task) -> TaskResult:
        result = TaskResult()
        result.task_name = task.name
        result.started_at = datetime.now(timezone.utc).isoformat()

        for attempt in range(task.max_retries if task.retry_on_fail else 1):
            try:
                if task.timeout_sec > 0:
                    # 使用线程执行以支持超时
                    import concurrent.futures
                    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
                        future = executor.submit(task.func)
                        try:
                            output = future.result(timeout=task.timeout_sec)
                            result.output = str(output)[:2000]  # 限制输出长度
                            result.success = True
                            break
                        except concurrent.futures.TimeoutError:
                            result.error = f"Task timed out after {task.timeout_sec}s"
                            if attempt == task.max_retries - 1:
                                result.success = False
                        except Exception as e:
                            result.error = str(e)
                            result.success = False
                else:
                    result.output = str(task.func())
                    result.success = True
                    break

            except Exception as e:
                result.error = str(e)
                result.success = False
                if attempt < task.max_retries - 1:
                    time.sleep(2 ** attempt)  # 指数退避

            result.retries = attempt + 1

        result.completed_at = datetime.now(timezone.utc).isoformat()
        result.duration_sec = round(
            (datetime.fromisoformat(result.completed_at) -
             datetime.fromisoformat(result.started_at)).total_seconds(),
            3
        )

        return result


# ── 任务定义 ─────────────────────────────────────────────────────

def _clean_task() -> str:
    """数据清洗任务"""
    from src.data_consume.data_cleaner import DataCleaner

    DATA_ROOT = APP_DIR / "data" / "server"
    cleaned = 0

    # 扫描原始数据
    for trades_dir in (DATA_ROOT / "trades").iterdir():
        if not trades_dir.is_dir():
            continue
        tenant_id = trades_dir.name
        events_file = trades_dir / "events.ndjson"

        if not events_file.exists():
            continue

        events = []
        with open(events_file, "r", encoding="utf-8") as f:
            for line in f:
                try:
                    events.append(json.loads(line.strip()))
                except json.JSONDecodeError:
                    continue

        if events:
            cleaner = DataCleaner()
            cleaned_events, _ = cleaner.clean_batch(events)
            from src.data_consume.data_cleaner import write_cleaned_events
            write_cleaned_events(cleaned_events, tenant_id)
            cleaned += len(cleaned_events)

    return f"Cleaned {cleaned} events"


def _analytics_task() -> str:
    """分析引擎任务"""
    from src.data_consume.analytics_engine import get_engine

    engine = get_engine()
    report = engine.run_batch_analysis(tenant_id="default")
    engine.save_report(report, "scheduled")
    return f"Analytics report: {report.total_events} events, PnL ${report.total_pnl:.2f}"


def _report_daily_task() -> str:
    """日报生成任务"""
    from src.data_consume.report_generator import get_generator

    gen = get_generator()
    results = gen.generate_all(tenant_id="default")
    return f"Generated: {', '.join(str(p) for p in results.values())}"


def _export_task() -> str:
    """数据导出任务"""
    from datetime import timedelta
    from src.data_consume.report_generator import get_generator

    gen = get_generator()
    end = datetime.now(timezone.utc)
    start = end - timedelta(days=7)
    csv_path = gen.generate_csv_export(start, end, "default")
    return f"Exported CSV: {csv_path}"


def _quality_check_task() -> str:
    """质量检查任务"""
    from src.data_consume.quality_monitor import generate_quality_report, save_quality_report

    report = generate_quality_report("default")
    save_quality_report(report, "default")
    status = report.get("status", "UNKNOWN")
    score = report.get("overall_score", 0)
    return f"Quality: {status}, Score {score}"


def _strategy_sync_task() -> str:
    """策略同步任务"""
    try:
        from src.strategy_sync import sync_params, check_risk_alerts, process_alerts
        sync_result = sync_params()
        alerts = check_risk_alerts()
        if alerts:
            process_alerts(alerts)
        return f"Strategy sync: applied={sync_result.get('applied')}, alerts={len(alerts)}"
    except Exception as e:
        return f"Strategy sync failed: {e}"


# ── 调度器主类 ───────────────────────────────────────────────────

class TaskScheduler:
    """自动化任务调度器"""

    def __init__(self):
        self.executor = TaskExecutor()
        self.tasks: dict[str, Task] = {}
        self.state = self._load_state()
        self._running = False
        self._thread: threading.Thread | None = None

        self._register_tasks()

    def _register_tasks(self) -> None:
        """注册所有任务"""
        task_defs = [
            Task("clean_task", _clean_task, SCHEDULE["clean_task"], enabled=True),
            Task("analytics_task", _analytics_task, SCHEDULE["analytics_task"], enabled=True),
            Task("report_daily", _report_daily_task, SCHEDULE["report_daily"], enabled=True),
            Task("export_task", _export_task, SCHEDULE["export_task"], enabled=True),
            Task("quality_check", _quality_check_task, SCHEDULE["quality_check"], enabled=True),
            Task("strategy_sync", _strategy_sync_task, SCHEDULE["strategy_sync"], enabled=True),
        ]

        for td in task_defs:
            self.tasks[td.name] = td

    def _load_state(self) -> SchedulerState:
        """加载调度器状态"""
        if TASK_STATE_FILE.exists():
            try:
                data = json.loads(TASK_STATE_FILE.read_text(encoding="utf-8"))
                return SchedulerState(**data)
            except Exception:
                pass
        return SchedulerState()

    def _save_state(self) -> None:
        """保存调度器状态"""
        TASKS_DIR.mkdir(parents=True, exist_ok=True)
        TASK_STATE_FILE.write_text(json.dumps(asdict(self.state), ensure_ascii=False))

    def _log_result(self, result: TaskResult) -> None:
        """记录任务结果"""
        TASKS_DIR.mkdir(parents=True, exist_ok=True)
        with open(TASK_LOG, "a", encoding="utf-8") as f:
            f.write(json.dumps(asdict(result), ensure_ascii=False) + "\n")

        # 更新状态
        self.state.last_runs[result.task_name] = result.completed_at
        self.state.total_runs[result.task_name] = self.state.total_runs.get(result.task_name, 0) + 1

        if not result.success:
            self.state.failures[result.task_name] = self.state.failures.get(result.task_name, 0) + 1
        else:
            self.state.failures[result.task_name] = 0

        self._save_state()

    def run_task(self, task_name: str) -> TaskResult:
        """手动触发单个任务"""
        if task_name not in self.tasks:
            result = TaskResult()
            result.task_name = task_name
            result.error = f"Unknown task: {task_name}"
            return result

        task = self.tasks[task_name]
        print(f"[Scheduler] Running task: {task_name}", flush=True)
        result = self.executor.execute(task)
        self._log_result(result)
        print(f"[Scheduler] {task_name}: {'SUCCESS' if result.success else 'FAILED'} ({result.duration_sec:.2f}s)",
              flush=True)
        return result

    def check_and_run_due_tasks(self) -> list[TaskResult]:
        """检查并运行到期任务"""
        results = []
        now = datetime.now(timezone.utc)

        for name, task in self.tasks.items():
            if not task.enabled:
                continue

            last_run = self.state.last_runs.get(name, "")
            if last_run:
                try:
                    last_dt = datetime.fromisoformat(last_run)
                    next_run = last_dt.timestamp() + task.interval_sec
                    if now.timestamp() < next_run:
                        continue
                except Exception:
                    pass

            # 任务到期，执行
            result = self.run_task(name)
            results.append(result)
            self.state.last_runs[name] = result.completed_at

        return results

    def start(self, daemon: bool = True) -> None:
        """启动调度器"""
        self._running = True
        print(f"[Scheduler] Starting scheduler with {len(self.tasks)} tasks", flush=True)

        if daemon:
            self._thread = threading.Thread(target=self._run_loop, daemon=True)
            self._thread.start()
        else:
            self._run_loop()

    def _run_loop(self) -> None:
        """调度循环"""
        check_interval = 60  # 每分钟检查一次

        while self._running:
            try:
                self.check_and_run_due_tasks()
            except Exception as e:
                print(f"[Scheduler] Error in loop: {e}", flush=True)

            for _ in range(check_interval):
                if not self._running:
                    break
                time.sleep(1)

    def stop(self) -> None:
        """停止调度器"""
        self._running = False
        if self._thread:
            self._thread.join(timeout=5)

    def get_status(self) -> dict[str, Any]:
        """获取调度器状态"""
        now = datetime.now(timezone.utc)
        next_runs = {}

        for name, task in self.tasks.items():
            last_run = self.state.last_runs.get(name, "")
            if last_run:
                try:
                    last_dt = datetime.fromisoformat(last_run)
                    next_ts = last_dt.timestamp() + task.interval_sec
                    next_dt = datetime.fromtimestamp(next_ts, tz=timezone.utc)
                    next_runs[name] = next_dt.isoformat()
                except Exception:
                    next_runs[name] = "unknown"
            else:
                next_runs[name] = "due now"

        return {
            "running": self._running,
            "tasks": {
                name: {
                    "enabled": task.enabled,
                    "interval_sec": task.interval_sec,
                    "last_run": self.state.last_runs.get(name, ""),
                    "next_run": next_runs.get(name, ""),
                    "failures": self.state.failures.get(name, 0),
                    "total_runs": self.state.total_runs.get(name, 0),
                }
                for name, task in self.tasks.items()
            },
        }


# ── 便捷函数 ─────────────────────────────────────────────────────

_scheduler: TaskScheduler | None = None


def get_scheduler() -> TaskScheduler:
    global _scheduler
    if _scheduler is None:
        _scheduler = TaskScheduler()
    return _scheduler


def run_scheduler_daemon() -> None:
    """启动调度器守护进程"""
    sched = get_scheduler()
    sched.start(daemon=True)


# ── 主入口 ───────────────────────────────────────────────────────

if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Task scheduler")
    parser.add_argument("--run", metavar="TASK", help="Run specific task")
    parser.add_argument("--status", action="store_true", help="Show scheduler status")
    parser.add_argument("--daemon", action="store_true", help="Run as daemon")
    args = parser.parse_args()

    sched = get_scheduler()

    if args.run:
        result = sched.run_task(args.run)
        print(json.dumps(asdict(result), ensure_ascii=False, indent=2))
    elif args.status:
        status = sched.get_status()
        print(json.dumps(status, ensure_ascii=False, indent=2))
    elif args.daemon:
        print("Starting scheduler daemon...")
        sched.start(daemon=True)
        while True:
            time.sleep(60)
    else:
        # 运行所有到期任务 (非守护模式)
        sched.check_and_run_due_tasks()