"""External-task multi-agent scheduled analysis API (任务分析定时).

Two audiences share this router:

- **对方系统（external）**：创建/删除任务（推送式接入）、时间线详情（他们详情页
  拉取渲染）、问答历史查询、**定时数据交付**（6 块 JSON + 时间轴期刻度）。鉴权统一
  走现有 Bearer token（对方后端/前端用我方
  登录接口换 token），不建独立 API-Key 机制。
- **我方管理页（internal）**：任务列表 / 绑定 CRUD / 运行记录与席位过程消息 /
  手动触发 / 数据块标题与排序编辑。与任务分析前端页（official-task-analysis）一一对应。

数据交付（第三期）：智能体在定时执行里经 ``POST /data/submit`` 入库（内部接口，同样
走 Bearer，技能脚本缺省经公共 token 端点现签）；提交落在「提交时刻的当前最新一期」，
首次提交无期自动开期。合并规则：顶层 key 合并、同 key 后写覆盖。
"""

from __future__ import annotations

import logging
from typing import Any, Literal

from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field

from deerflow.persistence.task_analysis import TaskAnalysisStore
from deerflow.runtime.task_analysis.data_delivery import (
    block_defs_payload,
    period_detail_payload,
    periods_payload,
    submit_block_data,
)
from deerflow.runtime.task_analysis.live_messages import (
    live_rows_to_process_messages,
    merge_process_messages,
)
from deerflow.runtime.task_analysis.service import TaskAnalysisService, next_run_at

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/task-analysis", tags=["task-analysis"])


def _store(request: Request) -> TaskAnalysisStore:
    store = getattr(request.app.state, "task_analysis_store", None)
    if store is None:
        raise HTTPException(status_code=503, detail="Task analysis store not available")
    return store


def _service(request: Request) -> TaskAnalysisService:
    service = getattr(request.app.state, "task_analysis_service", None)
    if service is None:
        raise HTTPException(status_code=503, detail="Task analysis service not running")
    return service


def _user_id(request: Request) -> str:
    user = getattr(request.state, "user", None)
    if user is not None:
        return str(user.id)
    from deerflow.runtime.user_context import get_effective_user_id

    return get_effective_user_id()


async def _task_or_404(store: TaskAnalysisStore, *, task_id: str | None = None, source: str | None = None) -> dict[str, Any]:
    task = (
        await store.get_task_by_source(source) if source else await store.get_task(task_id or "")
    )
    if task is None:
        raise HTTPException(status_code=404, detail="任务不存在")
    return task


# ── 对外接口（对方系统调用；Bearer token 鉴权与内部接口一致） ────────────────


class ExternalTaskCreateRequest(BaseModel):
    source_task_id: str = Field(min_length=1, max_length=128)
    name: str = Field(min_length=1, max_length=512)
    description: str | None = Field(default=None, max_length=8192)


@router.post("/external/tasks")
async def external_create_task(request: Request, body: ExternalTaskCreateRequest) -> dict:
    """对方创建完任务后调用；响应带 default_chain_id（我方记录的最近一次绑定链）。"""
    store = _store(request)
    try:
        task = await store.create_task(
            {"source_task_id": body.source_task_id, "name": body.name, "description": body.description}
        )
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc
    return {"task_id": task["id"], "default_chain_id": await store.last_bound_chain_id()}


@router.delete("/external/tasks/{source_task_id}")
async def external_delete_task(request: Request, source_task_id: str) -> dict:
    """对方删除任务时同步删除：任务 + 绑定 + 执行记录 + 过程消息 + 时间线版本。"""
    store = _store(request)
    task = await _task_or_404(store, source=source_task_id)
    await store.delete_task_cascade(task["id"])
    return {"deleted": True, "task_id": task["id"]}


@router.get("/external/tasks/{source_task_id}/detail")
async def external_task_detail(
    request: Request, source_task_id: str, limit: int = 20, offset: int = 0
) -> dict:
    """时间线详情：对方详情页按任务号拉取（版本倒序，含各块内容）。"""
    store = _store(request)
    task = await _task_or_404(store, source=source_task_id)
    binding = await store.get_binding(task["id"])
    versions = await store.list_versions(task["id"], limit=limit, offset=offset)
    return {
        "task": {"task_id": task["id"], "source_task_id": task["source_task_id"], "name": task["name"]},
        "binding": {
            "chain_id": binding.get("chain_id"),
            "cron_expr": binding.get("cron_expr"),
            "schedule_type": binding.get("schedule_type") or "cron",
            "enabled": binding.get("enabled"),
        }
        if binding
        else None,
        "timeline": versions,
    }


@router.get("/external/runs/{run_id}/messages")
async def external_run_messages(
    request: Request, run_id: str, seat: str | None = None, limit: int = 500
) -> dict:
    """问答历史 / 执行过程查询（seat 过滤可只看某席位或 __coordinator__ 总控）。"""
    store = _store(request)
    run = await store.get_run(run_id)
    if run is None:
        raise HTTPException(status_code=404, detail="运行记录不存在")
    messages = await _messages_for_run(request, run, seat=seat, limit=limit)
    return {"run": {k: run[k] for k in ("id", "trigger", "status", "started_at", "finished_at")}, "messages": messages}


# ── 定时数据交付（6 块 JSON + 时间轴；第三期） ───────────────────────────────


class DataSubmitRequest(BaseModel):
    # 外部任务号（对方系统的任务 id，非 DeerFlow 内部 task_id）。
    task_id: str = Field(min_length=1, max_length=128)
    block_key: str = Field(min_length=1, max_length=32)
    # JSON 对象即可，不做 schema 校验（schema 会变）；非对象由 pydantic 拒绝（422）。
    data: dict[str, Any]


@router.post("/data/submit")
async def submit_data_block(request: Request, body: DataSubmitRequest) -> dict:
    """智能体数据入库：落「提交时刻的当前最新一期」并按流水合并（顶层 key 合并、同 key 后写覆盖）。

    供 task-data-submit 技能（席位执行时 bash 调用）使用；首次提交无期自动开期
    （trigger=external），保证写入永远成功。
    """
    store = _store(request)
    source = body.task_id.strip()
    task = await store.get_task_by_source(source)
    if task is None:
        raise HTTPException(status_code=404, detail="任务不存在")
    try:
        result = await submit_block_data(
            store,
            task_id=task["id"],
            block_key=body.block_key,
            data=body.data,
            submitted_by=_user_id(request),
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return {"source_task_id": task["source_task_id"], "task_id": task["id"], **result}


@router.get("/external/tasks/{source_task_id}/blocks")
async def external_task_blocks(request: Request, source_task_id: str) -> dict:
    """数据块定义列表（块 key 固定 block_1~block_6，标题/排序可经管理端修改）。"""
    task = await _task_or_404(_store(request), source=source_task_id)
    return await block_defs_payload(_store(request), task)


@router.get("/external/tasks/{source_task_id}/periods")
async def external_task_periods(request: Request, source_task_id: str, limit: int = 500) -> dict:
    """时间轴刻度列表（期号升序；含是否当前期与各块更新时间摘要，进行中的刻度实时可见）。"""
    task = await _task_or_404(_store(request), source=source_task_id)
    return await periods_payload(_store(request), task, limit=max(1, min(limit, 2000)))


@router.get("/external/tasks/{source_task_id}/periods/{period}")
async def external_task_period_detail(request: Request, source_task_id: str, period: str) -> dict:
    """某期 6 块完整 JSON + 各块 updated_at；``period`` 传期号数字或 ``latest``。"""
    store = _store(request)
    task = await _task_or_404(store, source=source_task_id)
    if period != "latest" and not period.isdigit():
        raise HTTPException(status_code=400, detail="period 必须是期号数字或 latest")
    payload = await period_detail_payload(store, task, period)
    if payload is None:
        raise HTTPException(status_code=404, detail="期不存在")
    return payload


# ── 内部管理接口（任务分析页） ───────────────────────────────────────────────


class BindingUpsertRequest(BaseModel):
    # 业务链；缺省时使用「默认业务链」配置（default-chain 接口维护），都没有则 400。
    chain_id: str | None = Field(default=None, max_length=64)
    cron: str = Field(default="0 9 * * *", max_length=64)
    timezone: str = Field(default="Asia/Shanghai", max_length=64)
    # 执行方式：cron = 按频率定时；once = 只执行一次（创建即立刻执行，之后不再自动）。
    # None = 更新时保持原值（新建缺省 cron），避免局部更新（如启停）把 once 改回 cron。
    schedule_type: Literal["cron", "once"] | None = None
    enabled: bool = True
    model: str | None = Field(default=None, max_length=128)
    coordinator_prompt: str | None = Field(default=None, max_length=8192)


class TaskCreateRequest(BaseModel):
    name: str = Field(min_length=1, max_length=512)
    source_task_id: str | None = Field(default=None, max_length=128)
    description: str | None = Field(default=None, max_length=8192)
    binding: BindingUpsertRequest | None = None


@router.get("/tasks")
async def list_tasks(request: Request) -> dict:
    store = _store(request)
    return {"tasks": await store.list_tasks()}


@router.post("/tasks")
async def create_task(request: Request, body: TaskCreateRequest) -> dict:
    """页面一步式创建：任务信息 + 可选绑定（业务链 + 频率 + 启用 + 总控提示）。"""
    store = _store(request)
    user_id = _user_id(request)
    source = body.source_task_id or f"TS-{_next_source_suffix()}"
    try:
        task = await store.create_task(
            {"source_task_id": source, "name": body.name, "description": body.description}
        )
    except ValueError as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc

    binding = None
    if body.binding is not None:
        binding = await _upsert_binding(request, store, task["id"], body.binding, user_id)
    # 只执行一次：创建即立刻执行（enabled 关闭则不执行，等手动「立即执行」）。
    run = None
    if binding is not None and binding.get("schedule_type") == "once" and binding.get("enabled"):
        try:
            run = await _service(request).launch_binding(binding["id"], manual=False)
        except Exception:
            logger.exception("task-analysis once binding %s immediate launch failed", binding.get("id"))
    return {"task": task, "binding": binding, "run": run}


@router.delete("/tasks/{task_id}")
async def delete_task(request: Request, task_id: str) -> dict:
    store = _store(request)
    task = await _task_or_404(store, task_id=task_id)
    await store.delete_task_cascade(task["id"])
    return {"deleted": True}


@router.put("/tasks/{task_id}/binding")
async def upsert_binding(request: Request, task_id: str, body: BindingUpsertRequest) -> dict:
    store = _store(request)
    task = await _task_or_404(store, task_id=task_id)
    return {"binding": await _upsert_binding(request, store, task["id"], body, _user_id(request))}


@router.delete("/tasks/{task_id}/binding")
async def delete_binding(request: Request, task_id: str) -> dict:
    store = _store(request)
    task = await _task_or_404(store, task_id=task_id)
    return {"deleted": await store.delete_binding(task["id"])}


@router.post("/tasks/{task_id}/trigger")
async def trigger_run(request: Request, task_id: str) -> dict:
    """立即执行一次（手动触发的定时语义执行，非总控问答）。"""
    task = await _task_or_404(_store(request), task_id=task_id)
    binding = await _store(request).get_binding(task["id"])
    if binding is None:
        raise HTTPException(status_code=400, detail="任务未绑定分析方案")
    run = await _service(request).launch_binding(binding["id"], manual=True)
    if run is None:
        raise HTTPException(status_code=409, detail="执行被跳过（业务链缺失或已停用）")
    return {"run": run}


class BlockDefUpdate(BaseModel):
    block_key: str = Field(min_length=1, max_length=32)
    title: str | None = Field(default=None, max_length=128)
    sort_order: int | None = Field(default=None, ge=0, le=9999)


class BlockDefsUpdateRequest(BaseModel):
    blocks: list[BlockDefUpdate] = Field(min_length=1, max_length=32)


@router.put("/tasks/{task_id}/blocks")
async def update_task_blocks(request: Request, task_id: str, body: BlockDefsUpdateRequest) -> dict:
    """修改数据块标题/排序（管理端；块集合固定 6 条，不能增删）。"""
    store = _store(request)
    task = await _task_or_404(store, task_id=task_id)
    try:
        defs = await store.update_block_defs(
            task["id"], [block.model_dump() for block in body.blocks]
        )
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return {"blocks": defs}


@router.get("/tasks/{task_id}/runs")
async def list_runs(request: Request, task_id: str, limit: int = 50) -> list[dict]:
    store = _store(request)
    task = await _task_or_404(store, task_id=task_id)
    return await store.list_runs(task["id"], limit=limit)


@router.get("/runs/{run_id}")
async def get_run(request: Request, run_id: str) -> dict:
    run = await _store(request).get_run(run_id)
    if run is None:
        raise HTTPException(status_code=404, detail="运行记录不存在")
    return run


@router.get("/runs/{run_id}/messages")
async def get_run_messages(
    request: Request, run_id: str, seat: str | None = None, limit: int = 500
) -> dict:
    store = _store(request)
    run = await store.get_run(run_id)
    if run is None:
        raise HTTPException(status_code=404, detail="运行记录不存在")
    return {"messages": await _messages_for_run(request, run, seat=seat, limit=limit)}


# ── helpers ──────────────────────────────────────────────────────────────


async def _messages_for_run(
    request: Request,
    run: dict[str, Any],
    *,
    seat: str | None,
    limit: int,
) -> list[dict[str, Any]]:
    """Stored orchestration notes + live agent events (same source as scheduled tasks)."""
    store = _store(request)
    stored = await store.list_run_messages(run["id"], seat=seat, limit=limit)
    seat_ids = [seat] if seat else list((run.get("seat_results") or {}).keys())
    if not seat_ids:
        return stored
    merged: list[dict[str, Any]] = []
    for seat_id in seat_ids:
        seat_stored = [row for row in stored if row.get("seat_agent_id") == seat_id] if not seat else stored
        if seat and seat_id != seat:
            continue
        live_rows = await _live_agent_event_rows(request, run, seat_id)
        live = live_rows_to_process_messages(
            live_rows,
            seat=seat_id,
            stage_index=int((run.get("seat_results") or {}).get(seat_id, {}).get("stage_index") or 0),
        )
        merged.extend(merge_process_messages(stored=seat_stored, live=live, limit=limit))
    if seat:
        return merged[:limit]
    # Unfiltered: keep original stored order when there is no live overlay.
    if not any((row.get("metadata") or {}).get("source") == "live" for row in merged):
        return stored[:limit]
    return merged[:limit]


async def _live_agent_event_rows(
    request: Request, run: dict[str, Any], seat: str
) -> list[dict[str, Any]]:
    info = ((run.get("seat_results") or {}).get(seat) or {})
    thread_id = info.get("thread_id")
    agent_run_id = info.get("agent_run_id")
    if not thread_id:
        return []
    rows: list[dict[str, Any]] = []
    try:
        from app.gateway.deps import get_run_event_store

        event_store = get_run_event_store(request)
        if agent_run_id:
            rows = await event_store.list_messages_by_run(
                thread_id, agent_run_id, limit=500, user_id=None
            )
        if not rows:
            rows = await event_store.list_messages(thread_id, limit=500)
    except Exception:
        logger.debug("task-analysis live event-store read failed for seat %s", seat, exc_info=True)
        rows = []
    if rows:
        return rows
    return await _checkpoint_event_rows(request, thread_id)


async def _checkpoint_event_rows(request: Request, thread_id: str) -> list[dict[str, Any]]:
    """Same fallback scheduled tasks use: project the live LangGraph checkpoint."""
    try:
        from app.gateway.deps import get_checkpointer
        from deerflow.runtime.serialization import serialize_channel_values

        checkpointer = get_checkpointer(request)
        tup = await checkpointer.aget_tuple(
            {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
        )
    except Exception:
        logger.debug("task-analysis checkpoint read failed for %s", thread_id, exc_info=True)
        return []
    if tup is None:
        return []
    checkpoint = getattr(tup, "checkpoint", {}) or {}
    try:
        values = serialize_channel_values(checkpoint.get("channel_values") or {})
    except Exception:
        return []
    messages = values.get("messages") or []
    rows: list[dict[str, Any]] = []
    for index, message in enumerate(messages):
        if not isinstance(message, dict):
            continue
        raw_type = message.get("type") or message.get("role") or ""
        if raw_type in {"human", "user"}:
            event_type = "llm.human.input"
        elif raw_type in {"ai", "assistant"}:
            event_type = "llm.ai.response"
        elif raw_type == "tool":
            event_type = "llm.tool.result"
        else:
            continue
        rows.append(
            {
                "seq": index + 1,
                "event_type": event_type,
                "content": message,
                "metadata": {"source": "checkpoint"},
            }
        )
    return rows


async def _upsert_binding(
    request: Request,
    store: TaskAnalysisStore,
    external_task_id: str,
    body: BindingUpsertRequest,
    user_id: str,
) -> dict[str, Any]:
    # 业务链缺省：使用模块设置里的默认业务链。
    chain_id = body.chain_id or await store.get_setting("default_chain_id")
    if not chain_id:
        raise HTTPException(
            status_code=400,
            detail="未指定业务链，且未配置默认业务链（可在任务分析页或 PUT /default-chain 配置）",
        )
    # 绑定时固化业务链快照（seats/stages/goals/总控提示），执行前由 service 再刷新。
    chain_snapshot = await _load_chain_snapshot(request, chain_id, user_id)
    # 执行方式：请求未带时沿用现值（新建缺省 cron）；once 永不排期（next_run_at 恒空）。
    existing = await store.get_binding(external_task_id)
    schedule_type = body.schedule_type or (existing or {}).get("schedule_type") or "cron"
    next_at = (
        None if schedule_type == "once" else next_run_at(body.cron, body.timezone).isoformat()
    )
    return await store.upsert_binding(
        external_task_id,
        {
            "chain_id": chain_id,
            "chain_snapshot": chain_snapshot,
            "cron_expr": body.cron,
            "timezone": body.timezone,
            "schedule_type": schedule_type,
            "enabled": body.enabled,
            "next_run_at": next_at,
            "model": body.model,
            "coordinator_prompt": body.coordinator_prompt,
            "created_by": user_id,
        },
    )


async def _load_chain_snapshot(request: Request, chain_id: str, user_id: str) -> dict[str, Any]:
    from deerflow.runtime.task_analysis.service import build_chain_snapshot

    chain_store = getattr(request.app.state, "roundtable_chain_store", None)
    if chain_store is None:
        return {}
    try:
        chain = await chain_store.get_chain(chain_id, user_id)
    except Exception:
        logger.exception("task-analysis load chain %s for snapshot failed", chain_id)
        return {}
    if chain is None:
        raise HTTPException(status_code=400, detail=f"业务链不存在或不可见：{chain_id}")
    return build_chain_snapshot(chain)


def _next_source_suffix() -> str:
    import time

    return f"LOCAL-{int(time.time() * 1000) % 100000000:08d}"


# ── 默认业务链 / 任务完成状态 ──────────────────────────────────────────────


class DefaultChainUpdateRequest(BaseModel):
    chain_id: str | None = Field(default=None, max_length=64)


def _is_admin(request: Request) -> bool:
    """宽松管理员判定：无鉴权部署视为单管理空间（对齐 business_mapping）。"""
    user = getattr(request.state, "user", None)
    return user is None or getattr(user, "system_role", None) == "admin"


@router.get("/default-chain")
async def get_default_chain(request: Request) -> dict:
    """当前默认业务链（绑定不传 chain_id 时使用）。"""
    store = _store(request)
    chain_id = await store.get_setting("default_chain_id")
    title = None
    if chain_id:
        snapshot = await _load_chain_snapshot(request, chain_id, _user_id(request))
        title = snapshot.get("title")
    return {"chain_id": chain_id, "chain_title": title}


@router.put("/default-chain")
async def set_default_chain(request: Request, body: DefaultChainUpdateRequest) -> dict:
    """配置默认业务链（仅管理员；chain_id 传空清除）。"""
    if not _is_admin(request):
        raise HTTPException(status_code=403, detail="仅管理员可配置默认业务链")
    store = _store(request)
    if body.chain_id:
        # 配置前校验链存在（快照能加载即存在；不存在会 400）。
        await _load_chain_snapshot(request, body.chain_id, _user_id(request))
        await store.set_setting("default_chain_id", body.chain_id)
    else:
        await store.set_setting("default_chain_id", "")
    return await get_default_chain(request)


@router.get("/external/tasks/{source_task_id}/status")
async def get_external_task_status(request: Request, source_task_id: str) -> dict:
    """任务完成状态：一次都没执行成功过 = pending（未完成）；成功过 = completed（已完成）。"""
    store = _store(request)
    task = await _task_or_404(store, source=source_task_id)
    runs = await store.list_runs(task["id"], limit=200)
    succeeded = [r for r in runs if r.get("status") == "succeeded"]
    last_finished = max(
        (r.get("finished_at") for r in succeeded if r.get("finished_at")), default=None
    )
    return {
        "source_task_id": source_task_id,
        "task_id": task["id"],
        "status": "completed" if succeeded else "pending",
        "succeeded_runs": len(succeeded),
        "total_runs": len(runs),
        "last_finished_at": last_finished,
    }