按原型图重构核心页面并完善生成链路

2026-07-02 16:09:34 +08:00
parent 5a43cc70d5
commit a8716aa3c6
10 changed files with 1228 additions and 210 deletions
@@ -1,5 +1,4 @@
 import json
-import time
 import os
 import uuid
 from datetime import datetime
@@ -7,6 +6,7 @@ from datetime import datetime
 from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sse_starlette.sse import EventSourceResponse

 from config import settings
 from database import get_db
@@ -17,6 +17,13 @@ from models.paragraph import Paragraph
 from models.template import Template
 from schemas.schemas import GenerateFullRequest, GenerateTestRequest, Response
 from services.ai_service import call_ai
+from services.generation_runtime import (
+    build_mock_content,
+    generation_progress,
+    request_cancel,
+    run_generation,
+    update_progress,
+)
 from services.minio_client import upload_bytes

 router = APIRouter()
@@ -36,47 +43,6 @@ def _serialize_document(document: Document) -> dict:
        "updated_at": document.updated_at,
    }

-
-def _build_mock_content(paragraph: Paragraph) -> dict:
-    if paragraph.output_format == "table":
-        return {
-            "content": [
-                {
-                    "type": "table",
-                    "title": paragraph.title,
-                    "headers": ["字段", "内容"],
-                    "rows": [
-                        ["段落标题", paragraph.title],
-                        ["生成说明", paragraph.prompt_text or "根据模板内容生成"],
-                    ],
-                }
-            ]
-        }
-
-    blocks = [
-        {
-            "type": "text",
-            "text": f"这是“{paragraph.title}”的示例生成内容，可用于前端联调与流程验证。"
-        }
-    ]
-    if paragraph.content:
-        blocks.append({"type": "text", "text": f"模板上下文：{paragraph.content[:200]}"})
-    if paragraph.need_prompt and paragraph.prompt_text:
-        blocks.append({"type": "text", "text": f"预设提示词：{paragraph.prompt_text[:200]}"})
-    return {"content": blocks}
-
-
-async def _get_effective_model(db: AsyncSession, paragraph: Paragraph) -> AiModel | None:
-    if paragraph.model_id:
-        model = await db.get(AiModel, paragraph.model_id)
-        if model is not None and model.status == "enabled":
-            return model
-    result = await db.execute(
-        select(AiModel).where(AiModel.status == "enabled").order_by(AiModel.id.asc()).limit(1)
-    )
-    return result.scalars().first()
-
-
@router.post("/test")
 async def generate_test(body: GenerateTestRequest, db: AsyncSession = Depends(get_db)):
    paragraph = await db.get(Paragraph, body.paragraph_id)
@@ -90,7 +56,7 @@ async def generate_test(body: GenerateTestRequest, db: AsyncSession = Depends(ge
        model = await db.get(AiModel, paragraph.model_id)

    if model is None or model.status != "enabled":
-        content = _build_mock_content(paragraph)
+        content = build_mock_content(paragraph)
        message = "当前未找到可用模型，返回本地模拟生成结果。"
    else:
        result = await call_ai(paragraph, model)
@@ -162,54 +128,32 @@ async def generate_full(body: GenerateFullRequest, db: AsyncSession = Depends(ge
    )
    db.add(document)
    await db.flush()
-
-    done_count = 0
-    failed_count = 0
-    for paragraph in paragraphs:
-        if paragraph.edit_mode == "manual":
-            content = {"content": [{"type": "text", "text": paragraph.content or "该段落为人工编辑模式。"}]}
-            status = "success"
-            duration = 0
-            error_message = ""
-        else:
-            start = time.perf_counter()
-            model = await _get_effective_model(db, paragraph)
-            try:
-                if model is None:
-                    content = _build_mock_content(paragraph)
-                else:
-                    result = await call_ai(paragraph, model)
-                    content = result.content
-                status = "success"
-                error_message = ""
-            except Exception as error:
-                content = _build_mock_content(paragraph)
-                status = "failed"
-                error_message = str(error)
-                failed_count += 1
-            duration = round(time.perf_counter() - start, 4)
-
-        log = GenerationLog(
-            document_id=document.id,
-            paragraph_id=paragraph.id,
-            model_id=paragraph.model_id,
-            status=status,
-            content=json.dumps(content, ensure_ascii=False),
-            duration=duration,
-            error_msg=error_message,
-        )
-        db.add(log)
-        done_count += 1
-
-    document.para_count_done = done_count
-    document.status = "completed" if failed_count == 0 else "failed"
-    document.error = "" if failed_count == 0 else f"{failed_count} 个段落生成失败，已回退为模拟结果。"
-    document.file_path = f"mock://document/{document.id}"
    await db.commit()
    await db.refresh(document)
+    update_progress(document.id, status="pending", percent=0, done=0, total=len(paragraphs), message="任务已创建")
+    asyncio.create_task(run_generation(document.id, template.id))
    return Response(data=_serialize_document(document))


+@router.get("/progress/{document_id}")
+async def generate_progress(document_id: int):
+    async def event_generator():
+        while True:
+            state = generation_progress.get(
+                document_id,
+                {"status": "pending", "percent": 0, "message": "等待中", "done": 0, "total": 0},
+            )
+            yield {
+                "event": "progress",
+                "data": json.dumps(state, ensure_ascii=False),
+            }
+            if state.get("status") in {"completed", "failed", "cancelled"}:
+                break
+            await asyncio.sleep(1)
+
+    return EventSourceResponse(event_generator())
+
+
@router.get("/documents")
 async def list_documents(
    page: int = Query(1, ge=1),
@@ -263,9 +207,10 @@ async def cancel_document(document_id: int, db: AsyncSession = Depends(get_db)):
    if document is None:
        raise HTTPException(status_code=404, detail="生成记录不存在")

-    document.status = "cancelled"
-    await db.commit()
-    await db.refresh(document)
+    if document.status in {"completed", "failed", "cancelled"}:
+        return Response(data=_serialize_document(document))
+
+    request_cancel(document_id)
    return Response(data=_serialize_document(document))


@@ -2,6 +2,7 @@ import asyncio
 import json
 import re
 from dataclasses import dataclass
+from urllib.parse import urlparse

 import httpx

@@ -60,6 +61,10 @@ def _build_prompt(paragraph: Paragraph) -> tuple[str, str]:

 def _normalize_openai_endpoint(api_endpoint: str) -> str:
    endpoint = api_endpoint.rstrip("/")
+    parsed = urlparse(endpoint if "://" in endpoint else f"https://{endpoint}")
+    host = parsed.netloc or parsed.path.split("/")[0]
+    if host == "api.deepseek.com":
+        return "https://api.deepseek.com/chat/completions"
    if endpoint.endswith("/chat/completions"):
        return endpoint
    if endpoint.endswith("/v1"):
@@ -88,7 +93,7 @@ async def _post_with_retry(
            response = await client.post(url, headers=headers, json=payload)
            if response.status_code in (429, 500, 502, 503, 504):
                raise httpx.HTTPStatusError(
-                    f"上游模型响应异常: {response.status_code}",
+                    f"上游模型响应异常: {response.status_code} - {response.text[:500]}",
                    request=response.request,
                    response=response,
                )
@@ -99,7 +104,10 @@ async def _post_with_retry(
            if attempt == settings.AI_MAX_RETRIES - 1:
                break
            await asyncio.sleep(2 ** attempt)
-    raise RuntimeError(f"模型调用失败：{last_error}")
+    error_message = str(last_error)
+    if isinstance(last_error, httpx.HTTPStatusError) and last_error.response is not None:
+        error_message = f"{error_message}\n响应内容: {last_error.response.text[:1000]}"
+    raise RuntimeError(f"模型调用失败：{error_message}")


 async def _call_openai_compatible(model: AiModel, system_prompt: str, user_prompt: str) -> AiCallResult:
@@ -116,7 +124,7 @@ async def _call_openai_compatible(model: AiModel, system_prompt: str, user_promp
        "temperature": 0.3,
    }
    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
-    async with httpx.AsyncClient(timeout=settings.AI_REQUEST_TIMEOUT) as client:
+    async with httpx.AsyncClient(timeout=settings.AI_REQUEST_TIMEOUT, trust_env=False) as client:
        response = await _post_with_retry(client, _normalize_openai_endpoint(model.api_endpoint), headers, payload)
    body = response.json()
    text = body["choices"][0]["message"]["content"]
@@ -139,7 +147,7 @@ async def _call_anthropic(model: AiModel, system_prompt: str, user_prompt: str)
        "anthropic-version": "2023-06-01",
        "content-type": "application/json",
    }
-    async with httpx.AsyncClient(timeout=settings.AI_REQUEST_TIMEOUT) as client:
+    async with httpx.AsyncClient(timeout=settings.AI_REQUEST_TIMEOUT, trust_env=False) as client:
        response = await _post_with_retry(client, _normalize_anthropic_endpoint(model.api_endpoint), headers, payload)
    body = response.json()
    text = ""
@@ -0,0 +1,174 @@
+import asyncio
+import json
+import time
+from datetime import datetime
+
+from sqlalchemy import select
+
+from database import async_session
+from models.ai_model import AiModel
+from models.document import Document
+from models.generation_log import GenerationLog
+from models.paragraph import Paragraph
+from models.template import Template
+from services.ai_service import call_ai
+
+generation_progress: dict[int, dict] = {}
+generation_cancel_flags: dict[int, bool] = {}
+
+
+def build_mock_content(paragraph: Paragraph) -> dict:
+    if paragraph.output_format == "table":
+        return {
+            "content": [
+                {
+                    "type": "table",
+                    "title": paragraph.title,
+                    "headers": ["字段", "内容"],
+                    "rows": [
+                        ["段落标题", paragraph.title],
+                        ["生成说明", paragraph.prompt_text or "根据模板内容生成"],
+                    ],
+                }
+            ]
+        }
+
+    blocks = [
+        {
+            "type": "text",
+            "text": f"这是“{paragraph.title}”的示例生成内容，可用于前端联调与流程验证。"
+        }
+    ]
+    if paragraph.content:
+        blocks.append({"type": "text", "text": f"模板上下文：{paragraph.content[:200]}"})
+    if paragraph.need_prompt and paragraph.prompt_text:
+        blocks.append({"type": "text", "text": f"预设提示词：{paragraph.prompt_text[:200]}"})
+    return {"content": blocks}
+
+
+async def get_effective_model(paragraph: Paragraph) -> AiModel | None:
+    async with async_session() as db:
+        if paragraph.model_id:
+            model = await db.get(AiModel, paragraph.model_id)
+            if model is not None and model.status == "enabled":
+                return model
+        result = await db.execute(
+            select(AiModel).where(AiModel.status == "enabled").order_by(AiModel.id.asc()).limit(1)
+        )
+        return result.scalars().first()
+
+
+def update_progress(document_id: int, **kwargs):
+    state = generation_progress.setdefault(
+        document_id,
+        {"percent": 0, "status": "pending", "message": "等待中", "done": 0, "total": 0},
+    )
+    state.update(kwargs)
+
+
+def request_cancel(document_id: int):
+    generation_cancel_flags[document_id] = True
+    update_progress(document_id, status="cancelling", message="正在取消...")
+
+
+def is_cancel_requested(document_id: int) -> bool:
+    return generation_cancel_flags.get(document_id, False)
+
+
+async def run_generation(document_id: int, template_id: int):
+    async with async_session() as db:
+        document = await db.get(Document, document_id)
+        template = await db.get(Template, template_id)
+        if document is None or template is None:
+            update_progress(document_id, status="failed", message="生成任务初始化失败")
+            return
+
+        result = await db.execute(
+            select(Paragraph)
+            .where(Paragraph.template_id == template_id)
+            .order_by(Paragraph.sort_index.asc(), Paragraph.id.asc())
+        )
+        paragraphs = result.scalars().all()
+        total = len(paragraphs)
+        update_progress(document_id, status="generating", total=total, done=0, percent=0, message="开始生成...")
+
+        done_count = 0
+        failed_count = 0
+        try:
+            for index, paragraph in enumerate(paragraphs, start=1):
+                if is_cancel_requested(document_id):
+                    document.status = "cancelled"
+                    document.error = "用户已取消生成"
+                    await db.commit()
+                    update_progress(document_id, status="cancelled", percent=min(99, int(done_count / max(total, 1) * 100)), message="已取消生成", done=done_count)
+                    return
+
+                if paragraph.edit_mode == "manual":
+                    content = {"content": [{"type": "text", "text": paragraph.content or "该段落为人工编辑模式。"}]}
+                    status = "success"
+                    duration = 0
+                    error_message = ""
+                    model_id = paragraph.model_id
+                else:
+                    start = time.perf_counter()
+                    model = await get_effective_model(paragraph)
+                    model_id = model.id if model is not None else paragraph.model_id
+                    try:
+                        if model is None:
+                            content = build_mock_content(paragraph)
+                        else:
+                            result_data = await call_ai(paragraph, model)
+                            content = result_data.content
+                        status = "success"
+                        error_message = ""
+                    except Exception as error:
+                        content = build_mock_content(paragraph)
+                        status = "failed"
+                        error_message = str(error)
+                        failed_count += 1
+                    duration = round(time.perf_counter() - start, 4)
+
+                log = GenerationLog(
+                    document_id=document.id,
+                    paragraph_id=paragraph.id,
+                    model_id=model_id,
+                    status=status,
+                    content=json.dumps(content, ensure_ascii=False),
+                    duration=duration,
+                    error_msg=error_message,
+                )
+                db.add(log)
+                done_count += 1
+                document.para_count_done = done_count
+                percent = int(done_count / max(total, 1) * 100)
+                update_progress(
+                    document_id,
+                    status="generating",
+                    percent=percent,
+                    done=done_count,
+                    total=total,
+                    current_paragraph=paragraph.title,
+                    message=f"正在生成：{paragraph.title}",
+                )
+                await db.commit()
+
+            document.status = "completed" if failed_count == 0 else "failed"
+            document.error = "" if failed_count == 0 else f"{failed_count} 个段落生成失败，已回退为模拟结果。"
+            document.file_path = f"mock://document/{document.id}"
+            document.updated_at = datetime.now()
+            await db.commit()
+            update_progress(
+                document_id,
+                status=document.status,
+                percent=100,
+                done=done_count,
+                total=total,
+                message="生成完成" if failed_count == 0 else document.error,
+            )
+        except Exception as error:
+            document.status = "failed"
+            document.error = str(error)
+            await db.commit()
+            update_progress(document_id, status="failed", message=str(error), done=done_count, total=total)
+        finally:
+            generation_cancel_flags.pop(document_id, None)