Neo-ZQYY/apps/backend/app/routers/admin_task_engine.py

# -*- coding: utf-8 -*-
"""P18 任务引擎运营看板 API

提供转移日志查看、待审核任务管理、参数配置等端点。
所有端点需要 JWT 认证；写操作仅限 super_admin。
"""

from __future__ import annotations

import logging
from datetime import date

from fastapi import APIRouter, Depends, HTTPException, Query, status
from psycopg2.extras import RealDictCursor

from app.auth.dependencies import CurrentUser, get_current_user
from app.database import get_connection
from app.schemas.admin_task_engine import (
    CandidateAssistant,
    CandidateListResponse,
    CloseRequest,
    CloseResponse,
    ConfigParam,
    ConfigParamCreate,
    ConfigParamList,
    ConfigParamResponse,
    ConfigParamUpdate,
    PendingReviewItem,
    PendingReviewPage,
    ReassignRequest,
    ReassignResponse,
    TransferLogItem,
    TransferLogPage,
)

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/admin/task-engine", tags=["任务引擎管理"])

# ---- 任务类型中文映射 ----
TASK_TYPE_LABELS = {
    "high_priority_recall": "高优先召回",
    "priority_recall": "优先召回",
    "follow_up_visit": "客户回访",
    "relationship_building": "关系构建",
}


# ---- 权限辅助函数 ----

def _require_super_admin(user: CurrentUser) -> None:
    """写操作权限校验：仅超级管理员可执行。"""
    if "super_admin" not in user.roles:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="仅超级管理员可执行此操作",
        )


def _filter_site_id(user: CurrentUser, query_site_id: int | None) -> int | None:
    """读操作门店过滤：门店管理员强制按自身 site_id 过滤。"""
    if "super_admin" in user.roles:
        return query_site_id
    return user.site_id


# =====================================================================
# 1. 转移日志
# =====================================================================

@router.get("/transfer-log", response_model=TransferLogPage)
async def list_transfer_logs(
    site_id: int | None = Query(None, description="门店 ID"),
    from_date: date | None = Query(None, description="起始日期"),
    to_date: date | None = Query(None, description="截止日期"),
    assistant_id: int | None = Query(None, description="助教 ID"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    user: CurrentUser = Depends(get_current_user),
) -> TransferLogPage:
    """转移日志分页列表。"""
    effective_site_id = _filter_site_id(user, site_id)

    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            conditions = []
            params: list = []

            if effective_site_id is not None:
                conditions.append("t.site_id = %s")
                params.append(effective_site_id)
            if from_date is not None:
                conditions.append("t.created_at >= %s")
                params.append(from_date)
            if to_date is not None:
                conditions.append("t.created_at < %s::date + interval '1 day'")
                params.append(to_date)
            if assistant_id is not None:
                conditions.append("(t.from_assistant_id = %s OR t.to_assistant_id = %s)")
                params.extend([assistant_id, assistant_id])

            where_clause = " AND ".join(conditions) if conditions else "1=1"

            # 总数
            cur.execute(
                f"SELECT count(*) AS cnt FROM biz.coach_task_transfer_log t WHERE {where_clause}",
                params,
            )
            total = cur.fetchone()["cnt"]

            # 分页数据
            offset = (page - 1) * page_size
            cur.execute(
                f"""
                SELECT t.*, s.site_name
                FROM biz.coach_task_transfer_log t
                LEFT JOIN biz.sites s ON s.site_id = t.site_id
                WHERE {where_clause}
                ORDER BY t.created_at DESC
                LIMIT %s OFFSET %s
                """,
                params + [page_size, offset],
            )
            rows = cur.fetchall()

        items = [TransferLogItem(**row) for row in rows]
        return TransferLogPage(items=items, total=total)
    except HTTPException:
        raise
    except Exception as exc:
        logger.exception("查询转移日志失败")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询转移日志失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()


@router.get("/transfer-log/{member_id}/history", response_model=list[TransferLogItem])
async def get_member_transfer_history(
    member_id: int,
    user: CurrentUser = Depends(get_current_user),
) -> list[TransferLogItem]:
    """某客户全部转移历史。"""
    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            # 门店管理员只能看自己门店的记录
            effective_site_id = _filter_site_id(user, None)
            if effective_site_id is not None:
                cur.execute(
                    """
                    SELECT t.*, s.site_name
                    FROM biz.coach_task_transfer_log t
                    LEFT JOIN biz.sites s ON s.site_id = t.site_id
                    WHERE t.member_id = %s AND t.site_id = %s
                    ORDER BY t.created_at DESC
                    """,
                    [member_id, effective_site_id],
                )
            else:
                cur.execute(
                    """
                    SELECT t.*, s.site_name
                    FROM biz.coach_task_transfer_log t
                    LEFT JOIN biz.sites s ON s.site_id = t.site_id
                    WHERE t.member_id = %s
                    ORDER BY t.created_at DESC
                    """,
                    [member_id],
                )
            rows = cur.fetchall()

        return [TransferLogItem(**row) for row in rows]
    except HTTPException:
        raise
    except Exception as exc:
        logger.exception("查询客户转移历史失败: member_id=%s", member_id)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询客户转移历史失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()


# =====================================================================
# 2. 待审核任务
# =====================================================================

@router.get("/pending-review", response_model=PendingReviewPage)
async def list_pending_reviews(
    site_id: int | None = Query(None, description="门店 ID"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    user: CurrentUser = Depends(get_current_user),
) -> PendingReviewPage:
    """待审核任务列表。"""
    effective_site_id = _filter_site_id(user, site_id)

    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            conditions = ["ct.status = 'pending_review'"]
            params: list = []

            if effective_site_id is not None:
                conditions.append("ct.site_id = %s")
                params.append(effective_site_id)

            where_clause = " AND ".join(conditions)

            # 总数
            cur.execute(
                f"SELECT count(*) AS cnt FROM biz.coach_tasks ct WHERE {where_clause}",
                params,
            )
            total = cur.fetchone()["cnt"]

            # 分页数据
            offset = (page - 1) * page_size
            cur.execute(
                f"""
                SELECT ct.*, s.site_name
                FROM biz.coach_tasks ct
                LEFT JOIN biz.sites s ON s.site_id = ct.site_id
                WHERE {where_clause}
                ORDER BY ct.created_at DESC
                LIMIT %s OFFSET %s
                """,
                params + [page_size, offset],
            )
            rows = cur.fetchall()

        items = []
        for row in rows:
            row["task_type_label"] = TASK_TYPE_LABELS.get(row.get("task_type", ""), "")
            items.append(PendingReviewItem(**row))

        return PendingReviewPage(items=items, total=total)
    except HTTPException:
        raise
    except Exception as exc:
        logger.exception("查询待审核任务失败")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询待审核任务失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()


@router.post("/pending-review/{task_id}/reassign", response_model=ReassignResponse)
async def reassign_task(
    task_id: int,
    body: ReassignRequest,
    user: CurrentUser = Depends(get_current_user),
) -> ReassignResponse:
    """重新分配待审核任务（仅超级管理员）。

    逻辑：原任务 status → 'transferred'，新建 active 任务，写 transfer_log。
    """
    _require_super_admin(user)

    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            # 查询原任务
            cur.execute(
                "SELECT * FROM biz.coach_tasks WHERE id = %s FOR UPDATE",
                [task_id],
            )
            task = cur.fetchone()
            if task is None:
                raise HTTPException(status_code=404, detail="任务不存在")
            if task["status"] != "pending_review":
                raise HTTPException(status_code=400, detail="任务状态不是待审核，无法重新分配")

            # 原任务标记为 transferred
            cur.execute(
                "UPDATE biz.coach_tasks SET status = 'transferred', updated_at = now() WHERE id = %s",
                [task_id],
            )

            # 新建 active 任务
            cur.execute(
                """
                INSERT INTO biz.coach_tasks
                    (site_id, member_id, assistant_id, task_type, priority_score, status, created_at, updated_at)
                VALUES (%s, %s, %s, %s, %s, 'active', now(), now())
                RETURNING id
                """,
                [task["site_id"], task["member_id"], body.to_assistant_id,
                 task["task_type"], task.get("priority_score")],
            )
            new_task_id = cur.fetchone()["id"]

            # 写转移日志
            cur.execute(
                """
                INSERT INTO biz.coach_task_transfer_log
                    (site_id, member_id, from_assistant_id, to_assistant_id,
                     transfer_reason, transfer_score, created_at)
                VALUES (%s, %s, %s, %s, %s, %s, now())
                """,
                [task["site_id"], task["member_id"], task["assistant_id"],
                 body.to_assistant_id, "manual_reassign", task.get("priority_score")],
            )

        conn.commit()

        # 触发 AI 任务分配链（App4 → App5）
        try:
            from app.services.trigger_scheduler import fire_event
            fire_event(
                "ai_task_assigned",
                {
                    "site_id": task["site_id"],
                    "member_id": task["member_id"],
                    "assistant_id": body.to_assistant_id,
                },
            )
        except Exception:
            logger.exception(
                "触发 ai_task_assigned 事件失败: task_id=%s new_task_id=%s",
                task_id, new_task_id,
            )

        return ReassignResponse(success=True, new_task_id=new_task_id)
    except HTTPException:
        conn.rollback()
        raise
    except Exception as exc:
        conn.rollback()
        logger.exception("重新分配任务失败: task_id=%s", task_id)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"重新分配任务失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()


@router.post("/pending-review/{task_id}/close", response_model=CloseResponse)
async def close_task(
    task_id: int,
    body: CloseRequest,
    user: CurrentUser = Depends(get_current_user),
) -> CloseResponse:
    """关闭待审核任务（仅超级管理员）。

    逻辑：任务 status → 'inactive'，记录 abandon_reason。
    """
    _require_super_admin(user)

    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                "SELECT id, status FROM biz.coach_tasks WHERE id = %s FOR UPDATE",
                [task_id],
            )
            task = cur.fetchone()
            if task is None:
                raise HTTPException(status_code=404, detail="任务不存在")
            if task["status"] != "pending_review":
                raise HTTPException(status_code=400, detail="任务状态不是待审核，无法关闭")

            cur.execute(
                """
                UPDATE biz.coach_tasks
                SET status = 'inactive', abandon_reason = %s, updated_at = now()
                WHERE id = %s
                """,
                [body.reason, task_id],
            )

        conn.commit()
        return CloseResponse(success=True)
    except HTTPException:
        conn.rollback()
        raise
    except Exception as exc:
        conn.rollback()
        logger.exception("关闭任务失败: task_id=%s", task_id)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"关闭任务失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()


# =====================================================================
# 3. 参数管理
# =====================================================================

# 权重参数 key 列表（联合校验用）
_WEIGHT_KEYS = {"w_rs", "w_ms", "w_ml"}


@router.get("/config", response_model=ConfigParamList)
async def list_config_params(
    site_id: int | None = Query(None, description="门店 ID（不传则返回全部）"),
    user: CurrentUser = Depends(get_current_user),
) -> ConfigParamList:
    """参数列表。"""
    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            conditions: list[str] = []
            params: list = []

            if site_id is not None:
                # 返回指定门店覆盖 + 全局默认
                conditions.append("(p.site_id = %s OR p.site_id IS NULL)")
                params.append(site_id)

            where_clause = " AND ".join(conditions) if conditions else "1=1"

            cur.execute(
                f"""
                SELECT p.*, s.site_name
                FROM biz.cfg_task_generator_params p
                LEFT JOIN biz.sites s ON s.site_id = p.site_id
                WHERE {where_clause}
                ORDER BY p.site_id NULLS FIRST, p.param_key
                """,
                params,
            )
            rows = cur.fetchall()

        return ConfigParamList(params=[ConfigParam(**row) for row in rows])
    except HTTPException:
        raise
    except Exception as exc:
        logger.exception("查询参数配置失败")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"查询参数配置失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()


@router.put("/config/{param_id}", response_model=ConfigParamResponse)
async def update_config_param(
    param_id: int,
    body: ConfigParamUpdate,
    user: CurrentUser = Depends(get_current_user),
) -> ConfigParamResponse:
    """更新参数值（仅超级管理员）。

    权重参数（w_rs / w_ms / w_ml）更新后会校验三者之和是否为 1.0。
    """
    _require_super_admin(user)

    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            # 查询当前参数
            cur.execute(
                "SELECT * FROM biz.cfg_task_generator_params WHERE id = %s FOR UPDATE",
                [param_id],
            )
            param = cur.fetchone()
            if param is None:
                raise HTTPException(status_code=404, detail="参数不存在")

            # 更新
            cur.execute(
                """
                UPDATE biz.cfg_task_generator_params
                SET param_value = %s, updated_at = now()
                WHERE id = %s
                """,
                [body.param_value, param_id],
            )

            # 权重参数联合校验：w_rs + w_ms + w_ml = 1.0
            if param["param_key"] in _WEIGHT_KEYS:
                cur.execute(
                    """
                    SELECT param_key, param_value
                    FROM biz.cfg_task_generator_params
                    WHERE site_id IS NOT DISTINCT FROM %s
                      AND param_key = ANY(%s)
                    """,
                    [param["site_id"], list(_WEIGHT_KEYS)],
                )
                weight_rows = cur.fetchall()
                weight_sum = sum(r["param_value"] for r in weight_rows)
                if abs(weight_sum - 1.0) > 0.001:
                    conn.rollback()
                    raise HTTPException(
                        status_code=400,
                        detail=f"权重参数之和必须为 1.0，当前为 {weight_sum:.4f}",
                    )

        conn.commit()
        return ConfigParamResponse(success=True, id=param_id)
    except HTTPException:
        conn.rollback()
        raise
    except Exception as exc:
        conn.rollback()
        logger.exception("更新参数失败: param_id=%s", param_id)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"更新参数失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()


@router.post("/config", response_model=ConfigParamResponse)
async def create_config_param(
    body: ConfigParamCreate,
    user: CurrentUser = Depends(get_current_user),
) -> ConfigParamResponse:
    """新增门店覆盖参数（仅超级管理员）。"""
    _require_super_admin(user)

    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            # 检查是否已存在同 site_id + param_key 的记录
            cur.execute(
                """
                SELECT id FROM biz.cfg_task_generator_params
                WHERE site_id = %s AND param_key = %s
                """,
                [body.site_id, body.param_key],
            )
            if cur.fetchone() is not None:
                raise HTTPException(
                    status_code=400,
                    detail=f"门店 {body.site_id} 已存在参数 {body.param_key} 的覆盖配置",
                )

            cur.execute(
                """
                INSERT INTO biz.cfg_task_generator_params
                    (site_id, param_key, param_value, updated_at)
                VALUES (%s, %s, %s, now())
                RETURNING id
                """,
                [body.site_id, body.param_key, body.param_value],
            )
            new_id = cur.fetchone()["id"]

        conn.commit()
        return ConfigParamResponse(success=True, id=new_id)
    except HTTPException:
        conn.rollback()
        raise
    except Exception as exc:
        conn.rollback()
        logger.exception("新增参数失败")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"新增参数失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()


@router.delete("/clear-all-tasks")
async def clear_all_tasks(
    user: CurrentUser = Depends(get_current_user),
) -> dict:
    """【测试用】清空所有 coach_tasks 及关联数据（仅超级管理员）。

    用于开发/测试阶段重置任务数据，让 task_generator 重新生成。
    按外键依赖顺序删除：transfer_log → notes → history → tasks。
    """
    _require_super_admin(user)

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            # 按外键依赖顺序：先删引用表，再删主表
            cur.execute("DELETE FROM biz.coach_task_transfer_log")
            transfer_count = cur.rowcount
            cur.execute("DELETE FROM biz.notes WHERE task_id IS NOT NULL")
            notes_count = cur.rowcount
            cur.execute("DELETE FROM biz.coach_task_history")
            history_count = cur.rowcount
            # coach_tasks 有自引用 FK，先清 parent_task_id 和 transferred_from
            cur.execute("UPDATE biz.coach_tasks SET parent_task_id = NULL, transferred_from = NULL")
            cur.execute("DELETE FROM biz.coach_tasks")
            task_count = cur.rowcount
        conn.commit()
        return {
            "success": True,
            "message": f"已清空 {task_count} 条任务 + {history_count} 条历史 + {transfer_count} 条转移日志 + {notes_count} 条备注",
            "deleted_tasks": task_count,
            "deleted_history": history_count,
            "deleted_transfers": transfer_count,
            "deleted_notes": notes_count,
        }
    except Exception as exc:
        conn.rollback()
        logger.exception("清空任务失败")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"清空任务失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()


@router.delete("/config/{param_id}", response_model=ConfigParamResponse)
async def delete_config_param(
    param_id: int,
    user: CurrentUser = Depends(get_current_user),
) -> ConfigParamResponse:
    """删除门店覆盖参数（仅超级管理员）。

    不允许删除 site_id IS NULL 的全局默认参数。
    """
    _require_super_admin(user)

    conn = get_connection()
    try:
        with conn.cursor(cursor_factory=RealDictCursor) as cur:
            cur.execute(
                "SELECT id, site_id FROM biz.cfg_task_generator_params WHERE id = %s",
                [param_id],
            )
            param = cur.fetchone()
            if param is None:
                raise HTTPException(status_code=404, detail="参数不存在")
            if param["site_id"] is None:
                raise HTTPException(status_code=400, detail="不允许删除全局默认参数")

            cur.execute(
                "DELETE FROM biz.cfg_task_generator_params WHERE id = %s",
                [param_id],
            )

        conn.commit()
        return ConfigParamResponse(success=True, id=param_id)
    except HTTPException:
        conn.rollback()
        raise
    except Exception as exc:
        conn.rollback()
        logger.exception("删除参数失败: param_id=%s", param_id)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"删除参数失败: {str(exc)[:200]}",
        )
    finally:
        conn.close()