cas / fw-pms-ai

"""
任务相关 API 路由
"""

import logging
from typing import Optional, List, Dict, Any
import json
from datetime import datetime
from decimal import Decimal

from fastapi import APIRouter, Query, HTTPException
from pydantic import BaseModel

from ...services.db import get_connection

logger = logging.getLogger(__name__)
router = APIRouter()


class TaskResponse(BaseModel):
    """任务响应模型"""
    id: int
    task_no: str
    group_id: int
    dealer_grouping_id: int
    dealer_grouping_name: Optional[str] = None
    brand_grouping_id: Optional[int] = None
    plan_amount: float = 0
    actual_amount: float = 0
    part_count: int = 0
    base_ratio: Optional[float] = None
    status: int = 0
    status_text: str = ""
    error_message: Optional[str] = None
    llm_provider: Optional[str] = None
    llm_model: Optional[str] = None
    llm_total_tokens: int = 0
    statistics_date: Optional[str] = None
    start_time: Optional[str] = None
    end_time: Optional[str] = None
    duration_seconds: Optional[int] = None
    create_time: Optional[str] = None


class TaskListResponse(BaseModel):
    """任务列表响应"""
    total: int
    page: int
    page_size: int
    items: List[TaskResponse]


class DetailResponse(BaseModel):
    """配件建议明细响应"""
    id: int
    task_no: str
    shop_id: int
    shop_name: Optional[str] = None
    part_code: str
    part_name: Optional[str] = None
    unit: Optional[str] = None
    cost_price: float = 0
    current_ratio: Optional[float] = None
    base_ratio: Optional[float] = None
    post_plan_ratio: Optional[float] = None
    valid_storage_cnt: float = 0
    avg_sales_cnt: float = 0
    suggest_cnt: int = 0
    suggest_amount: float = 0
    suggestion_reason: Optional[str] = None
    priority: int = 2
    llm_confidence: Optional[float] = None
    statistics_date: Optional[str] = None


class DetailListResponse(BaseModel):
    """配件建议明细列表响应"""
    total: int
    page: int
    page_size: int
    items: List[DetailResponse]


def format_datetime(dt) -> Optional[str]:
    """格式化日期时间"""
    if dt is None:
        return None
    if isinstance(dt, datetime):
        return dt.strftime("%Y-%m-%d %H:%M:%S")
    return str(dt)


def get_status_text(status: int) -> str:
    """获取状态文本"""
    status_map = {0: "运行中", 1: "成功", 2: "失败"}
    return status_map.get(status, "未知")


@router.get("/tasks", response_model=TaskListResponse)
async def list_tasks(
    page: int = Query(1, ge=1, description="页码"),
    page_size: int = Query(20, ge=1, le=100, description="每页数量"),
    status: Optional[int] = Query(None, description="状态筛选: 0-运行中 1-成功 2-失败"),
    dealer_grouping_id: Optional[int] = Query(None, description="商家组合ID"),
    statistics_date: Optional[str] = Query(None, description="统计日期"),
):
    """获取任务列表"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    
    try:
        # 构建查询条件
        where_clauses = []
        params = []
        
        if status is not None:
            where_clauses.append("status = %s")
            params.append(status)
        
        if dealer_grouping_id is not None:
            where_clauses.append("dealer_grouping_id = %s")
            params.append(dealer_grouping_id)
        
        if statistics_date:
            where_clauses.append("statistics_date = %s")
            params.append(statistics_date)
        
        where_sql = " AND ".join(where_clauses) if where_clauses else "1=1"
        
        # 查询总数
        count_sql = f"SELECT COUNT(*) as total FROM ai_replenishment_task WHERE {where_sql}"
        cursor.execute(count_sql, params)
        total = cursor.fetchone()["total"]
        
        # 查询分页数据
        offset = (page - 1) * page_size
        data_sql = f"""
            SELECT t.*, 
                COALESCE(
                    NULLIF(t.llm_total_tokens, 0),
                    (SELECT SUM(l.llm_tokens) FROM ai_task_execution_log l WHERE l.task_no = t.task_no)
                ) as calculated_tokens
            FROM ai_replenishment_task t
            WHERE {where_sql}
            ORDER BY t.create_time DESC
            LIMIT %s OFFSET %s
        """
        cursor.execute(data_sql, params + [page_size, offset])
        rows = cursor.fetchall()
        
        items = []
        for row in rows:
            # 计算执行时长
            duration = None
            if row.get("start_time") and row.get("end_time"):
                duration = int((row["end_time"] - row["start_time"]).total_seconds())
            
            items.append(TaskResponse(
                id=row["id"],
                task_no=row["task_no"],
                group_id=row["group_id"],
                dealer_grouping_id=row["dealer_grouping_id"],
                dealer_grouping_name=row.get("dealer_grouping_name"),
                brand_grouping_id=row.get("brand_grouping_id"),
                plan_amount=float(row.get("plan_amount") or 0),
                actual_amount=float(row.get("actual_amount") or 0),
                part_count=row.get("part_count") or 0,
                base_ratio=float(row["base_ratio"]) if row.get("base_ratio") else None,
                status=row.get("status") or 0,
                status_text=get_status_text(row.get("status") or 0),
                error_message=row.get("error_message"),
                llm_provider=row.get("llm_provider"),
                llm_model=row.get("llm_model"),
                llm_total_tokens=int(row.get("calculated_tokens") or 0),
                statistics_date=row.get("statistics_date"),
                start_time=format_datetime(row.get("start_time")),
                end_time=format_datetime(row.get("end_time")),
                duration_seconds=duration,
                create_time=format_datetime(row.get("create_time")),
            ))
        
        return TaskListResponse(
            total=total,
            page=page,
            page_size=page_size,
            items=items,
        )
    
    finally:
        cursor.close()
        conn.close()


@router.get("/tasks/{task_no}", response_model=TaskResponse)
async def get_task(task_no: str):
    """获取任务详情"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    
    try:
        cursor.execute(
            """
            SELECT t.*, 
                COALESCE(
                    NULLIF(t.llm_total_tokens, 0),
                    (SELECT SUM(l.llm_tokens) FROM ai_task_execution_log l WHERE l.task_no = t.task_no)
                ) as calculated_tokens
            FROM ai_replenishment_task t
            WHERE t.task_no = %s
            """,
            (task_no,)
        )
        row = cursor.fetchone()
        
        if not row:
            raise HTTPException(status_code=404, detail="任务不存在")
        
        duration = None
        if row.get("start_time") and row.get("end_time"):
            duration = int((row["end_time"] - row["start_time"]).total_seconds())
        
        return TaskResponse(
            id=row["id"],
            task_no=row["task_no"],
            group_id=row["group_id"],
            dealer_grouping_id=row["dealer_grouping_id"],
            dealer_grouping_name=row.get("dealer_grouping_name"),
            brand_grouping_id=row.get("brand_grouping_id"),
            plan_amount=float(row.get("plan_amount") or 0),
            actual_amount=float(row.get("actual_amount") or 0),
            part_count=row.get("part_count") or 0,
            base_ratio=float(row["base_ratio"]) if row.get("base_ratio") else None,
            status=row.get("status") or 0,
            status_text=get_status_text(row.get("status") or 0),
            error_message=row.get("error_message"),
            llm_provider=row.get("llm_provider"),
            llm_model=row.get("llm_model"),
            llm_total_tokens=int(row.get("calculated_tokens") or 0),
            statistics_date=row.get("statistics_date"),
            start_time=format_datetime(row.get("start_time")),
            end_time=format_datetime(row.get("end_time")),
            duration_seconds=duration,
            create_time=format_datetime(row.get("create_time")),
        )
    
    finally:
        cursor.close()
        conn.close()


@router.get("/tasks/{task_no}/details", response_model=DetailListResponse)
async def get_task_details(
    task_no: str,
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    sort_by: str = Query("suggest_amount", description="排序字段"),
    sort_order: str = Query("desc", description="排序方向: asc/desc"),
    part_code: Optional[str] = Query(None, description="配件编码搜索"),
):
    """获取任务的配件建议明细"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    
    try:
        # 验证排序字段
        allowed_sort_fields = [
            "suggest_amount", "suggest_cnt", "cost_price", 
            "avg_sales_cnt", "current_ratio", "part_code"
        ]
        if sort_by not in allowed_sort_fields:
            sort_by = "suggest_amount"
        
        sort_direction = "DESC" if sort_order.lower() == "desc" else "ASC"
        
        # 构建查询条件
        where_sql = "task_no = %s"
        params = [task_no]
        
        if part_code:
            where_sql += " AND part_code LIKE %s"
            params.append(f"%{part_code}%")
        
        # 查询总数
        cursor.execute(
            f"SELECT COUNT(*) as total FROM ai_replenishment_detail WHERE {where_sql}",
            params
        )
        total = cursor.fetchone()["total"]
        
        # 查询分页数据
        offset = (page - 1) * page_size
        cursor.execute(
            f"""
            SELECT * FROM ai_replenishment_detail 
            WHERE {where_sql}
            ORDER BY {sort_by} {sort_direction}
            LIMIT %s OFFSET %s
            """,
            params + [page_size, offset]
        )
        rows = cursor.fetchall()
        
        items = []
        for row in rows:
            items.append(DetailResponse(
                id=row["id"],
                task_no=row["task_no"],
                shop_id=row["shop_id"],
                shop_name=row.get("shop_name"),
                part_code=row["part_code"],
                part_name=row.get("part_name"),
                unit=row.get("unit"),
                cost_price=float(row.get("cost_price") or 0),
                current_ratio=float(row["current_ratio"]) if row.get("current_ratio") else None,
                base_ratio=float(row["base_ratio"]) if row.get("base_ratio") else None,
                post_plan_ratio=float(row["post_plan_ratio"]) if row.get("post_plan_ratio") else None,
                valid_storage_cnt=float(row.get("valid_storage_cnt") or 0),
                avg_sales_cnt=float(row.get("avg_sales_cnt") or 0),
                suggest_cnt=row.get("suggest_cnt") or 0,
                suggest_amount=float(row.get("suggest_amount") or 0),
                suggestion_reason=row.get("suggestion_reason"),
                priority=row.get("priority") or 2,
                llm_confidence=float(row["llm_confidence"]) if row.get("llm_confidence") else None,
                statistics_date=row.get("statistics_date"),
            ))
        
        return DetailListResponse(
            total=total,
            page=page,
            page_size=page_size,
            items=items,
        )
    
    finally:
        cursor.close()
        conn.close()


class ExecutionLogResponse(BaseModel):
    """执行日志响应"""
    id: int
    task_no: str
    step_name: str
    step_order: int
    status: int
    status_text: str = ""
    input_data: Optional[str] = None
    output_data: Optional[str] = None
    error_message: Optional[str] = None
    retry_count: int = 0
    llm_tokens: int = 0
    execution_time_ms: int = 0
    start_time: Optional[str] = None
    end_time: Optional[str] = None
    create_time: Optional[str] = None


class ExecutionLogListResponse(BaseModel):
    """执行日志列表响应"""
    total: int
    items: List[ExecutionLogResponse]


def get_log_status_text(status: int) -> str:
    """获取日志状态文本"""
    status_map = {0: "运行中", 1: "成功", 2: "失败", 3: "跳过"}
    return status_map.get(status, "未知")


def get_step_name_display(step_name: str) -> str:
    """获取步骤名称显示"""
    step_map = {
        "fetch_part_ratio": "获取配件数据",
        "sql_agent": "AI分析建议",
        "allocate_budget": "分配预算",
        "generate_report": "生成报告",
    }
    return step_map.get(step_name, step_name)


@router.get("/tasks/{task_no}/logs", response_model=ExecutionLogListResponse)
async def get_task_logs(task_no: str):
    """获取任务执行日志"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    
    try:
        cursor.execute(
            """
            SELECT * FROM ai_task_execution_log 
            WHERE task_no = %s 
            ORDER BY step_order ASC
            """,
            (task_no,)
        )
        rows = cursor.fetchall()
        
        items = []
        for row in rows:
            items.append(ExecutionLogResponse(
                id=row["id"],
                task_no=row["task_no"],
                step_name=row["step_name"],
                step_order=row.get("step_order") or 0,
                status=row.get("status") or 0,
                status_text=get_log_status_text(row.get("status") or 0),
                input_data=row.get("input_data"),
                output_data=row.get("output_data"),
                error_message=row.get("error_message"),
                retry_count=row.get("retry_count") or 0,
                llm_tokens=row.get("llm_tokens") or 0,
                execution_time_ms=row.get("execution_time_ms") or 0,
                start_time=format_datetime(row.get("start_time")),
                end_time=format_datetime(row.get("end_time")),
                create_time=format_datetime(row.get("create_time")),
            ))
        
        return ExecutionLogListResponse(
            total=len(items),
            items=items,
        )
    
    finally:
        cursor.close()
        conn.close()


class PartSummaryResponse(BaseModel):
    """配件汇总响应"""
    id: int
    task_no: str
    part_code: str
    part_name: Optional[str] = None
    unit: Optional[str] = None
    cost_price: float = 0
    total_storage_cnt: float = 0
    total_avg_sales_cnt: float = 0
    group_current_ratio: Optional[float] = None
    total_suggest_cnt: int = 0
    total_suggest_amount: float = 0
    shop_count: int = 0
    need_replenishment_shop_count: int = 0
    part_decision_reason: Optional[str] = None
    priority: int = 2
    llm_confidence: Optional[float] = None
    statistics_date: Optional[str] = None
    group_post_plan_ratio: Optional[float] = None


class PartSummaryListResponse(BaseModel):
    """配件汇总列表响应"""
    total: int
    page: int
    page_size: int
    items: List[PartSummaryResponse]


@router.get("/tasks/{task_no}/part-summaries", response_model=PartSummaryListResponse)
async def get_task_part_summaries(
    task_no: str,
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    sort_by: str = Query("total_suggest_amount", description="排序字段"),
    sort_order: str = Query("desc", description="排序方向: asc/desc"),
    part_code: Optional[str] = Query(None, description="配件编码筛选"),
    priority: Optional[int] = Query(None, description="优先级筛选"),
):
    """获取任务的配件汇总列表"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    
    try:
        # 验证排序字段
        allowed_sort_fields = [
            "total_suggest_amount", "total_suggest_cnt", "cost_price",
            "total_avg_sales_cnt", "group_current_ratio", "part_code",
            "priority", "need_replenishment_shop_count",
            "total_storage_cnt", "shop_count", "group_post_plan_ratio"
        ]
        if sort_by not in allowed_sort_fields:
            sort_by = "total_suggest_amount"
        
        sort_direction = "DESC" if sort_order.lower() == "desc" else "ASC"
        
        # 构建查询条件
        where_clauses = ["task_no = %s"]
        params = [task_no]
        
        if part_code:
            where_clauses.append("part_code LIKE %s")
            params.append(f"%{part_code}%")
        
        if priority is not None:
            where_clauses.append("priority = %s")
            params.append(priority)
            
        where_sql = " AND ".join(where_clauses)
        
        # 查询总数
        cursor.execute(
            f"SELECT COUNT(*) as total FROM ai_replenishment_part_summary WHERE {where_sql}",
            params
        )
        total = cursor.fetchone()["total"]
        
        # 查询分页数据
        offset = (page - 1) * page_size
        
        # 动态计算计划后库销比: (库存 + 建议) / 月均销
        # 注意: total_avg_sales_cnt 可能为 0, 需要处理除以零的情况
        query_sql = f"""
            SELECT *,
                (
                    (COALESCE(total_storage_cnt, 0) + COALESCE(total_suggest_cnt, 0)) / 
                    NULLIF(total_avg_sales_cnt, 0)
                ) as group_post_plan_ratio
            FROM ai_replenishment_part_summary 
            WHERE {where_sql}
            ORDER BY {sort_by} {sort_direction}
            LIMIT %s OFFSET %s
        """
        
        cursor.execute(query_sql, params + [page_size, offset])
        rows = cursor.fetchall()
        
        items = []
        for row in rows:
            items.append(PartSummaryResponse(
                id=row["id"],
                task_no=row["task_no"],
                part_code=row["part_code"],
                part_name=row.get("part_name"),
                unit=row.get("unit"),
                cost_price=float(row.get("cost_price") or 0),
                total_storage_cnt=float(row.get("total_storage_cnt") or 0),
                total_avg_sales_cnt=float(row.get("total_avg_sales_cnt") or 0),
                group_current_ratio=float(row["group_current_ratio"]) if row.get("group_current_ratio") else None,
                group_post_plan_ratio=float(row["group_post_plan_ratio"]) if row.get("group_post_plan_ratio") is not None else None,
                total_suggest_cnt=row.get("total_suggest_cnt") or 0,
                total_suggest_amount=float(row.get("total_suggest_amount") or 0),
                shop_count=row.get("shop_count") or 0,
                need_replenishment_shop_count=row.get("need_replenishment_shop_count") or 0,
                part_decision_reason=row.get("part_decision_reason"),
                priority=row.get("priority") or 2,
                llm_confidence=float(row["llm_confidence"]) if row.get("llm_confidence") else None,
                statistics_date=row.get("statistics_date"),
            ))
        
        return PartSummaryListResponse(
            total=total,
            page=page,
            page_size=page_size,
            items=items,
        )
    
    finally:
        cursor.close()
        conn.close()


@router.get("/tasks/{task_no}/parts/{part_code}/shops")
async def get_part_shop_details(
    task_no: str,
    part_code: str,
):
    """获取指定配件的门店明细"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    
    try:
        cursor.execute(
            """
            SELECT * FROM ai_replenishment_detail 
            WHERE task_no = %s AND part_code = %s
            ORDER BY suggest_amount DESC
            """,
            (task_no, part_code)
        )
        rows = cursor.fetchall()
        
        items = []
        for row in rows:
            items.append(DetailResponse(
                id=row["id"],
                task_no=row["task_no"],
                shop_id=row["shop_id"],
                shop_name=row.get("shop_name"),
                part_code=row["part_code"],
                part_name=row.get("part_name"),
                unit=row.get("unit"),
                cost_price=float(row.get("cost_price") or 0),
                current_ratio=float(row["current_ratio"]) if row.get("current_ratio") else None,
                base_ratio=float(row["base_ratio"]) if row.get("base_ratio") else None,
                post_plan_ratio=float(row["post_plan_ratio"]) if row.get("post_plan_ratio") else None,
                valid_storage_cnt=float(row.get("valid_storage_cnt") or 0),
                avg_sales_cnt=float(row.get("avg_sales_cnt") or 0),
                suggest_cnt=row.get("suggest_cnt") or 0,
                suggest_amount=float(row.get("suggest_amount") or 0),
                suggestion_reason=row.get("suggestion_reason"),
                priority=row.get("priority") or 2,
                llm_confidence=float(row["llm_confidence"]) if row.get("llm_confidence") else None,
                statistics_date=row.get("statistics_date"),
            ))
        
        return {
            "total": len(items),
            "items": items,
        }
    
    finally:
        cursor.close()
        conn.close()


class AnalysisReportResponse(BaseModel):
    """分析报告响应"""
    id: int
    task_no: str
    group_id: int
    dealer_grouping_id: int
    dealer_grouping_name: Optional[str] = None
    brand_grouping_id: Optional[int] = None
    report_type: str
    
    # JSON 字段，使用 Any 或 Dict 来接收解析后的对象
    replenishment_insights: Optional[Dict[str, Any]] = None
    urgency_assessment: Optional[Dict[str, Any]] = None
    strategy_recommendations: Optional[Dict[str, Any]] = None
    execution_guide: Optional[Dict[str, Any]] = None
    expected_outcomes: Optional[Dict[str, Any]] = None
    
    total_suggest_cnt: int = 0
    total_suggest_amount: float = 0
    shortage_risk_cnt: int = 0
    excess_risk_cnt: int = 0
    stagnant_cnt: int = 0
    low_freq_cnt: int = 0
    
    llm_provider: Optional[str] = None
    llm_model: Optional[str] = None
    llm_tokens: int = 0
    execution_time_ms: int = 0
    statistics_date: Optional[str] = None
    create_time: Optional[str] = None


@router.get("/tasks/{task_no}/analysis-report", response_model=Optional[AnalysisReportResponse])
async def get_analysis_report(task_no: str):
    """获取任务的分析报告"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    
    try:
        cursor.execute(
            """
            SELECT * FROM ai_analysis_report 
            WHERE task_no = %s
            ORDER BY create_time DESC
            LIMIT 1
            """,
            (task_no,)
        )
        row = cursor.fetchone()
        
        if not row:
            return None
            
        # 辅助函数：解析 JSON 字符串
        def parse_json(value):
            if not value:
                return None
            if isinstance(value, str):
                try:
                    return json.loads(value)
                except json.JSONDecodeError:
                    return None
            return value

        return AnalysisReportResponse(
            id=row["id"],
            task_no=row["task_no"],
            group_id=row["group_id"],
            dealer_grouping_id=row["dealer_grouping_id"],
            dealer_grouping_name=row.get("dealer_grouping_name"),
            brand_grouping_id=row.get("brand_grouping_id"),
            report_type=row.get("report_type", "replenishment"),
            
            replenishment_insights=parse_json(row.get("replenishment_insights")),
            urgency_assessment=parse_json(row.get("urgency_assessment")),
            strategy_recommendations=parse_json(row.get("strategy_recommendations")),
            execution_guide=parse_json(row.get("execution_guide")),
            expected_outcomes=parse_json(row.get("expected_outcomes")),
            
            total_suggest_cnt=row.get("total_suggest_cnt") or 0,
            total_suggest_amount=float(row.get("total_suggest_amount") or 0),
            shortage_risk_cnt=row.get("shortage_risk_cnt") or 0,
            excess_risk_cnt=row.get("excess_risk_cnt") or 0,
            stagnant_cnt=row.get("stagnant_cnt") or 0,
            low_freq_cnt=row.get("low_freq_cnt") or 0,
            
            llm_provider=row.get("llm_provider"),
            llm_model=row.get("llm_model"),
            llm_tokens=row.get("llm_tokens") or 0,
            execution_time_ms=row.get("execution_time_ms") or 0,
            statistics_date=row.get("statistics_date"),
            create_time=format_datetime(row.get("create_time")),
        )
    
    finally:
        cursor.close()
        conn.close()