deerflow-code/offline-backend-20260512/backend/app/gateway/routers/ai_writing.py

"""AI 写作路由 —— 业务表 CRUD + transcript 持久化 + admin cleanup。

写作主流程（``/start /stream /resume /state``）已迁到 LangGraph Server 的
``ai_writing`` assistant（前端走 ``useAIWritingStream`` hook + SDK），路由层
不再承载 graph 推进，只保留以下「不依赖 graph 状态」的辅助接口：

GET    /api/ai-writing/sessions                   列出用户的会话历史
GET    /api/ai-writing/sessions/{id}              获取单个历史会话
PATCH  /api/ai-writing/sessions/{id}              重命名会话标题
DELETE /api/ai-writing/sessions/{id}              删除会话
PUT    /api/ai-writing/sessions/{id}/transcript   保存完整对话时间线
GET    /api/ai-writing/article-types              列出所有文章类型
POST   /api/ai-writing/article-types              新增文章类型
PATCH  /api/ai-writing/article-types/{id}         更新文章类型
DELETE /api/ai-writing/article-types/{id}         删除文章类型
POST   /api/ai-writing/admin/cleanup              （admin）手动清理会话
"""
from __future__ import annotations

import logging
from typing import Optional

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

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/ai-writing", tags=["ai-writing"])

# ── 请求/响应模型 ─────────────────────────────────────────────────────────────

class RenameSessionRequest(BaseModel):
    title: str


class AdminCleanupRequest(BaseModel):
    """手动清理接口的可选覆盖参数。

    全部留空 → 用 config.yaml 的 ai_writing.cleanup 段默认值；显式传值
    → 覆盖配置（一般用于运维想验证一个临时阈值，例如清掉「最近 1 小时
    以前的」会话来模拟整轮调度）。
    """
    retention_days: Optional[int] = Field(default=None, ge=0, description="保留多少天以内的会话；0 = 不保留")
    only_finished: Optional[bool] = Field(default=None, description="True 时只清 done/error 状态")
    batch_size: Optional[int] = Field(default=None, ge=1, le=10000, description="单批最多删多少行")


class AdminCleanupResponse(BaseModel):
    deleted_db: int
    deleted_checkpoints: int
    # 因「仍在跑 / 等用户干预」被保护掉的会话数（threads_meta.status in
    # running/interrupted）。即使 retention_days=0 也不会秒杀用户正在写的稿。
    skipped_active: int = 0
    elapsed_seconds: float
    params: dict


class SaveTranscriptRequest(BaseModel):
    # 前端组装的完整对话时间线（progressEvents + completedInterventions）
    transcript: dict


class ArticleTypeRequest(BaseModel):
    key: str
    label: str
    description: Optional[str] = None
    prompt_hint: Optional[str] = None
    default_word_count: int = 800
    sort_order: int = 0


class UpdateArticleTypeRequest(BaseModel):
    key: Optional[str] = None
    label: Optional[str] = None
    description: Optional[str] = None
    prompt_hint: Optional[str] = None
    default_word_count: Optional[int] = None
    sort_order: Optional[int] = None


class ArticleTypeResponse(BaseModel):
    id: str
    key: str
    label: str
    description: Optional[str] = None
    prompt_hint: Optional[str] = None
    default_word_count: int
    sort_order: int
    created_at: str
    updated_at: str


class ResearchTemplateRequest(BaseModel):
    name: str
    description: Optional[str] = None
    article_type_key: Optional[str] = None
    outline: Optional[str] = None
    material_source: str = "general"
    word_count: int = 800
    audience: Optional[str] = None
    strict_mode: bool = False
    topic: Optional[str] = None
    sort_order: int = 0


class UpdateResearchTemplateRequest(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    article_type_key: Optional[str] = None
    outline: Optional[str] = None
    material_source: Optional[str] = None
    word_count: Optional[int] = None
    audience: Optional[str] = None
    strict_mode: Optional[bool] = None
    topic: Optional[str] = None
    sort_order: Optional[int] = None


class ResearchTemplateResponse(BaseModel):
    id: str
    name: str
    description: Optional[str] = None
    article_type_key: Optional[str] = None
    outline: Optional[str] = None
    material_source: str
    word_count: int
    audience: Optional[str] = None
    strict_mode: bool
    topic: Optional[str] = None
    sort_order: int
    created_at: str
    updated_at: str


class SessionResponse(BaseModel):
    id: str
    user_id: Optional[str] = None
    title: str
    user_intent: Optional[str] = None
    status: str
    draft_title: Optional[str] = None
    draft_markdown: Optional[str] = None
    completed_interventions: Optional[list] = None
    review_result: Optional[dict] = None
    transcript: Optional[dict] = None
    created_at: str
    updated_at: str


# ── 路由 ──────────────────────────────────────────────────────────────────────
@router.get("/sessions", response_model=list[SessionResponse])
async def list_sessions(request: Request):
    """列出当前用户的 AI 写作历史会话。"""
    from app.gateway.deps import get_ai_writing_session_store, get_current_user
    from app.gateway.routers._ai_writing_seed import ensure_ai_writing_functional_agents

    # 调用前自检:打开写作页必经此接口。若管理员在启动后手删了内置写作 agent
    # 目录,这里自愈补建,保证 use_builtin_agents 开启时图节点能加载到 SOUL。
    ensure_ai_writing_functional_agents()

    repo = get_ai_writing_session_store(request)
    if repo is None:
        return []
    user_id = await get_current_user(request)
    return await repo.list(user_id=user_id)


@router.get("/sessions/{session_id}", response_model=SessionResponse)
async def get_session(session_id: str, request: Request):
    """获取单个历史会话。"""
    from app.gateway.deps import get_ai_writing_session_store, get_current_user

    repo = get_ai_writing_session_store(request)
    if repo is None:
        raise HTTPException(status_code=503, detail="Session store not available")
    user_id = await get_current_user(request)
    record = await repo.get(session_id, user_id=user_id)
    if record is None:
        raise HTTPException(status_code=404, detail="Session not found")
    return record


@router.patch("/sessions/{session_id}", response_model=SessionResponse)
async def rename_session(session_id: str, body: RenameSessionRequest, request: Request):
    """重命名会话标题。"""
    from app.gateway.deps import get_ai_writing_session_store, get_current_user

    repo = get_ai_writing_session_store(request)
    if repo is None:
        raise HTTPException(status_code=503, detail="Session store not available")
    user_id = await get_current_user(request)
    updated = await repo.update(session_id, user_id=user_id, title=body.title.strip())
    if updated is None:
        raise HTTPException(status_code=404, detail="Session not found")
    return updated


@router.delete("/sessions/{session_id}")
async def delete_session(session_id: str, request: Request):
    """删除历史会话。"""
    from app.gateway.deps import get_ai_writing_session_store, get_current_user

    repo = get_ai_writing_session_store(request)
    if repo is None:
        raise HTTPException(status_code=503, detail="Session store not available")
    user_id = await get_current_user(request)
    deleted = await repo.delete(session_id, user_id=user_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Session not found")
    return {"success": True}


@router.put("/sessions/{session_id}/transcript")
async def save_transcript(session_id: str, body: SaveTranscriptRequest, request: Request):
    """保存会话的完整对话时间线，供历史会话回显出与实时对话一致的富时间线。

    时间线持久化属于「锦上添花」功能：即便保存失败（如内网部署缺少 transcript
    列、或 MySQL TEXT 列超长），也绝不能阻断写作流程。任何异常都吞掉并返回
    success=False，由前端忽略即可。
    """
    from app.gateway.deps import get_ai_writing_session_store, get_current_user

    repo = get_ai_writing_session_store(request)
    if repo is None:
        return {"success": False, "reason": "store_unavailable"}
    try:
        user_id = await get_current_user(request)
        updated = await repo.update(session_id, user_id=user_id, transcript=body.transcript)
    except Exception:
        logger.exception("ai_writing: failed to save transcript: session=%s", session_id)
        return {"success": False, "reason": "save_failed"}
    if updated is None:
        # 会话不存在或不属于当前用户：同样不抛 500，返回 success=False 即可
        return {"success": False, "reason": "session_not_found"}
    return {"success": True}
@router.get("/article-types", response_model=list[ArticleTypeResponse])
async def list_article_types(request: Request):
    """列出所有文章类型，按 sort_order 升序排列。"""
    from app.gateway.deps import get_article_type_store
    from app.gateway.routers._ai_writing_seed import ensure_ai_writing_functional_agents

    # 防御性兜底,与 list_sessions 同理(写作表单加载文章类型时也会经过这里)。
    ensure_ai_writing_functional_agents()

    store = get_article_type_store(request)
    if store is None:
        return []
    return await store.list()


@router.post("/article-types", response_model=ArticleTypeResponse, status_code=201)
async def create_article_type(body: ArticleTypeRequest, request: Request):
    """新增文章类型。key 必须全局唯一。"""
    from app.gateway.deps import get_article_type_store

    store = get_article_type_store(request)
    if store is None:
        raise HTTPException(status_code=503, detail="Article type store not available")
    existing = await store.get_by_key(body.key)
    if existing is not None:
        raise HTTPException(status_code=409, detail=f"Article type key '{body.key}' already exists")
    return await store.create(
        key=body.key,
        label=body.label,
        description=body.description,
        prompt_hint=body.prompt_hint,
        default_word_count=body.default_word_count,
        sort_order=body.sort_order,
    )


@router.patch("/article-types/{article_type_id}", response_model=ArticleTypeResponse)
async def update_article_type(article_type_id: str, body: UpdateArticleTypeRequest, request: Request):
    """更新文章类型字段（只传需要修改的字段）。"""
    from app.gateway.deps import get_article_type_store

    store = get_article_type_store(request)
    if store is None:
        raise HTTPException(status_code=503, detail="Article type store not available")
    # 如果修改了 key，检查唯一性
    if body.key is not None:
        existing = await store.get_by_key(body.key)
        if existing is not None and existing["id"] != article_type_id:
            raise HTTPException(status_code=409, detail=f"Article type key '{body.key}' already exists")

    _UNSET = object.__new__(object)
    kwargs = {}
    for field in ("key", "label", "description", "prompt_hint", "default_word_count", "sort_order"):
        val = getattr(body, field)
        if val is not None:
            kwargs[field] = val

    updated = await store.update(article_type_id, **kwargs)
    if updated is None:
        raise HTTPException(status_code=404, detail="Article type not found")
    return updated


@router.delete("/article-types/{article_type_id}")
async def delete_article_type(article_type_id: str, request: Request):
    """删除文章类型。"""
    from app.gateway.deps import get_article_type_store

    store = get_article_type_store(request)
    if store is None:
        raise HTTPException(status_code=503, detail="Article type store not available")
    deleted = await store.delete(article_type_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Article type not found")
    return {"success": True}


# ── 研究模板（课题研究左栏一键预填表单，需求 0616 #5）────────────────────────────


@router.get("/research-templates", response_model=list[ResearchTemplateResponse])
async def list_research_templates(request: Request):
    """列出所有研究模板，按 sort_order 升序。

    新机器/空表时自动写入内置范例模板（网摘/周报/要讯），幂等仅在表为空时执行。
    """
    from app.gateway.deps import get_research_template_store

    store = get_research_template_store(request)
    if store is None:
        return []
    try:
        await store.seed_defaults()
    except Exception:
        logger.exception("ai_writing: failed to seed default research templates")
    return await store.list()


@router.post("/research-templates", response_model=ResearchTemplateResponse, status_code=201)
async def create_research_template(body: ResearchTemplateRequest, request: Request):
    """新增研究模板。"""
    from app.gateway.deps import get_research_template_store

    store = get_research_template_store(request)
    if store is None:
        raise HTTPException(status_code=503, detail="Research template store not available")
    return await store.create(
        name=body.name,
        description=body.description,
        article_type_key=body.article_type_key,
        outline=body.outline,
        material_source=body.material_source,
        word_count=body.word_count,
        audience=body.audience,
        strict_mode=body.strict_mode,
        topic=body.topic,
        sort_order=body.sort_order,
    )


@router.patch("/research-templates/{template_id}", response_model=ResearchTemplateResponse)
async def update_research_template(template_id: str, body: UpdateResearchTemplateRequest, request: Request):
    """更新研究模板字段（只传需要修改的字段）。"""
    from app.gateway.deps import get_research_template_store

    store = get_research_template_store(request)
    if store is None:
        raise HTTPException(status_code=503, detail="Research template store not available")

    kwargs = {}
    for field in (
        "name",
        "description",
        "article_type_key",
        "outline",
        "material_source",
        "word_count",
        "audience",
        "strict_mode",
        "topic",
        "sort_order",
    ):
        val = getattr(body, field)
        if val is not None:
            kwargs[field] = val

    updated = await store.update(template_id, **kwargs)
    if updated is None:
        raise HTTPException(status_code=404, detail="Research template not found")
    return updated


@router.delete("/research-templates/{template_id}")
async def delete_research_template(template_id: str, request: Request):
    """删除研究模板。"""
    from app.gateway.deps import get_research_template_store

    store = get_research_template_store(request)
    if store is None:
        raise HTTPException(status_code=503, detail="Research template store not available")
    deleted = await store.delete(template_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="Research template not found")
    return {"success": True}


# ── 管理接口 ──────────────────────────────────────────────────────────────────

async def _require_admin(request: Request) -> None:
    """放行 admin / auth-disabled，其余 403。

    与 scheduled_tasks._require_admin 同套语义：auth_disabled 模式下
    get_optional_user_from_request 返回 None，此时放行（开发环境一致体验）。
    """
    from app.gateway.deps import get_optional_user_from_request

    user = await get_optional_user_from_request(request)
    if user is None:
        return
    if getattr(user, "system_role", None) != "admin":
        raise HTTPException(status_code=403, detail="AI 写作清理管理接口仅限管理员")


def _resolve_cleanup_params(override: AdminCleanupRequest) -> dict:
    """合并 config.yaml 默认值 + 请求显式覆盖。

    显式传 0 / False 也算覆盖（用 ``is not None`` 判定），方便测「保留 0 天 =
    全删」这种极端场景。
    """
    from deerflow.config import get_app_config

    raw = getattr(get_app_config(), "ai_writing", None) or {}
    cfg = (raw.get("cleanup") if isinstance(raw, dict) else None) or {}

    return {
        "retention_days": (
            override.retention_days if override.retention_days is not None
            else int(cfg.get("retention_days", 7))
        ),
        "only_finished": (
            override.only_finished if override.only_finished is not None
            else bool(cfg.get("delete_only_finished", False))
        ),
        "batch_size": (
            override.batch_size if override.batch_size is not None
            else int(cfg.get("batch_size", 500))
        ),
    }


@router.post("/admin/cleanup", response_model=AdminCleanupResponse)
async def admin_cleanup(
    body: AdminCleanupRequest,
    request: Request,
) -> AdminCleanupResponse:
    """**仅管理员** 手动触发一次会话清理（不等凌晨 3 点定时调度）。

    用途：
      - 验证 cleanup 配置改动是否生效
      - 内网紧急清理（磁盘满 / MySQL 行数过多）
      - 调试：传 ``retention_days: 0`` 一刷拉清空，方便重新跑回归

    所有参数留空 → 沿用 config.yaml 的 ``ai_writing.cleanup`` 段默认。显式
    传值会覆盖配置但**不修改 config.yaml**（一次性参数，下次定时调度还是用
    配置文件的值）。
    """
    import time

    from app.gateway.ai_writing_cleanup import run_cleanup_once

    await _require_admin(request)
    params = _resolve_cleanup_params(body)

    # 透传主 LangGraph checkpointer / thread_store —— 新路径下 ai_writing 的
    # thread 状态在主 checkpointer 里，thread_store 用来按 status 过滤活跃 thread。
    # getattr 嵌套防御：测试用 stub Request 时 ``request.app`` 可能不存在。
    _app_state = getattr(getattr(request, "app", None), "state", None)
    main_checkpointer = getattr(_app_state, "checkpointer", None)
    main_thread_store = getattr(_app_state, "thread_store", None)

    started = time.monotonic()
    try:
        result = await run_cleanup_once(
            **params,
            checkpointer=main_checkpointer,
            thread_store=main_thread_store,
        )
    except Exception as e:
        logger.exception("ai_writing admin cleanup failed: params=%s", params)
        # 抛 500 是合理的 —— 管理员触发清理失败需要看到失败原因，跟「业务
        # 接口被失败拖累」是两回事
        raise HTTPException(
            status_code=500,
            detail=f"清理执行失败: {type(e).__name__}: {e}",
        ) from e

    elapsed = time.monotonic() - started
    logger.info(
        "ai_writing admin cleanup done: params=%s deleted_db=%d deleted_cp=%d elapsed=%.2fs",
        params,
        result["deleted_db"],
        result["deleted_checkpoints"],
        elapsed,
    )
    return AdminCleanupResponse(
        deleted_db=result["deleted_db"],
        deleted_checkpoints=result["deleted_checkpoints"],
        skipped_active=result.get("skipped_active", 0),
        elapsed_seconds=round(elapsed, 3),
        params=params,
    )