feat(api): 统一响应格式并重构 API 端点

在 api.v1.__init__ 中添加共享的 ok() 响应封装函数，统一响应结构
移除 documents、export、export_records 模块中重复的 _ok() 辅助函数
更新所有 API 端点，统一使用集中定义的 ok() 函数
将 export.py 中的导入语句移至文件顶部，提升代码组织性
移除 export 端点中对 DocumentNotFoundError 多余的 try-catch 处理
合并 export 处理器中的数据库会话与 Service 初始化逻辑
将 documents.py 中的 math 导入移至模块顶层，去掉内联导入
简化导出响应的构造方式，直接链式调用 model_dump()
确保所有 v1 端点的 API 响应格式保持一致

app/api/v1/__init__.py

@@ -0,0 +1,2 @@
+def ok(data: dict) -> dict:
+    return {"code": 0, "message": "Success", "data": data}

app/api/v1/documents.py

@@ -1,8 +1,10 @@
+import math
 from typing import Literal, Optional
 
 from fastapi import APIRouter, Depends, Query
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.api.v1 import ok
 from app.core.dependencies import get_db
 from app.schemas.document import (
     CreateDocumentRequest,
@@ -16,10 +18,6 @@ from app.services.document_service import DocumentService
 router = APIRouter(prefix="/documents", tags=["Documents"])
 
 
-def _ok(data: dict) -> dict:
-    return {"code": 0, "message": "Success", "data": data}
-
-
 # ------------------------------------------------------------------ #
 # POST /documents  创建文档
 # ------------------------------------------------------------------ #
@@ -30,7 +28,7 @@ async def create_document(
 ) -> dict:
     svc = DocumentService(db)
     doc = await svc.create_document(body)
-    return _ok(
+    return ok(
         {
             "documentId": doc.id,
             "title": doc.title,
@@ -60,7 +58,6 @@ async def list_documents(
         sort_by=sort_by,
         sort_order=sort_order,
     )
-    import math
 
     items = [
         DocumentListItem.model_validate(d).model_dump(by_alias=True) for d in docs
@@ -72,7 +69,7 @@ async def list_documents(
         total_pages=math.ceil(total / page_size) if page_size else 1,
     ).model_dump(by_alias=True)
 
-    return _ok({"documents": items, "pagination": pagination})
+    return ok({"documents": items, "pagination": pagination})
 
 
 # ------------------------------------------------------------------ #
@@ -85,7 +82,7 @@ async def get_document(
 ) -> dict:
     svc = DocumentService(db)
     doc = await svc.get_document(document_id)
-    return _ok(DocumentResponse.model_validate(doc).model_dump(by_alias=True))
+    return ok(DocumentResponse.model_validate(doc).model_dump(by_alias=True))
 
 
 # ------------------------------------------------------------------ #
@@ -99,7 +96,7 @@ async def update_document(
 ) -> dict:
     svc = DocumentService(db)
     doc = await svc.update_document(document_id, body)
-    return _ok(
+    return ok(
         {
             "documentId": doc.id,
             "updatedAt": int(doc.updated_at.timestamp() * 1000),

app/api/v1/export.py

@@ -1,16 +1,16 @@
 from fastapi import APIRouter
 
-from app.core.exceptions import DocumentNotFoundError, ExportError
+from app.api.v1 import ok
+from app.core.exceptions import ExportError
 from app.schemas.export import ExportDocRequest, ExportDocResponse
 from app.services.export_service import export_doc
+from app.core.database import AsyncSessionLocal
+from app.schemas.document import UpdateDocumentRequest
+from app.services.document_service import DocumentService
 
 router = APIRouter(tags=["Export"])
 
 
-def _ok(data: dict) -> dict:
-    return {"code": 0, "message": "Success", "data": data}
-
-
 # ------------------------------------------------------------------ #
 # POST /export/doc  — 导出 .doc，写入记录，返回永久下载链接
 # ------------------------------------------------------------------ #
@@ -20,18 +20,12 @@ async def export_document(body: ExportDocRequest) -> dict:
     # 若传入 documentId，可选同步更新草稿（失败不阻塞导出）
     if body.document_id:
         try:
-            from app.core.database import AsyncSessionLocal
-            from app.schemas.document import UpdateDocumentRequest
-            from app.services.document_service import DocumentService
-
             async with AsyncSessionLocal() as db:
                 svc = DocumentService(db)
                 await svc.update_document(
                     body.document_id,
                     UpdateDocumentRequest(content=body.content),
                 )
-        except DocumentNotFoundError:
-            pass
         except Exception:
             pass
 
@@ -43,11 +37,10 @@ async def export_document(body: ExportDocRequest) -> dict:
         document_id=body.document_id,
     )
 
-    resp = ExportDocResponse(
+    return ok(ExportDocResponse(
         record_id=result["record_id"],
         download_url=result["download_url"],
         file_name=result["file_name"],
         style_id=result["style_id"],
         warning=result["warning"],
-    )
-    return _ok(resp.model_dump(by_alias=True))
+    ).model_dump(by_alias=True))

app/api/v1/export_records.py

@@ -1,22 +1,20 @@
 """export_records.py — 导出记录管理路由及管理端存储查询。"""
 
 import math
-
+from pathlib import Path
 from fastapi import APIRouter
 from fastapi.responses import FileResponse
 
+from app.api.v1 import ok
 from app.core.database import AsyncSessionLocal
 from app.core.exceptions import RecordNotFoundError
+from app.schemas.export import ExportRecordSchema
 from app.services.export_record_service import ExportRecordService
 from app.services.storage_monitor import get_storage_info
 
 router = APIRouter(tags=["Export Records"])
 
 
-def _ok(data: dict) -> dict:
-    return {"code": 0, "data": data}
-
-
 # ------------------------------------------------------------------ #
 # GET /export/records  — 分页查询导出记录
 # ------------------------------------------------------------------ #
@@ -41,19 +39,10 @@ async def list_export_records(
         )
 
     total_pages = math.ceil(total / page_size) if total > 0 else 1
-    return _ok(
+    return ok(
         {
             "records": [
-                {
-                    "recordId": r.id,
-                    "userId": r.user_id,
-                    "fileName": r.file_name,
-                    "fileSize": r.file_size,
-                    "downloadUrl": r.download_url,
-                    "documentId": r.document_id,
-                    "styleId": r.style_id,
-                    "createdAt": int(r.created_at.timestamp() * 1000),
-                }
+                ExportRecordSchema.from_record(r).model_dump(by_alias=True)
                 for r in records
             ],
             "pagination": {
@@ -72,8 +61,6 @@ async def list_export_records(
 
 @router.get("/export/records/{recordId}/download", summary="重新下载导出文件")
 async def download_export_record(recordId: str, userId: str) -> FileResponse:
-    from pathlib import Path
-
     async with AsyncSessionLocal() as db:
         svc = ExportRecordService(db)
         record = await svc.get_record(recordId, userId)
@@ -109,7 +96,7 @@ async def delete_export_record(recordId: str, userId: str) -> dict:
 @router.get("/admin/storage", summary="管理端：存储使用情况")
 async def admin_storage() -> dict:
     info = get_storage_info()
-    return _ok(
+    return ok(
         {
             "diskTotalBytes": info["disk_total_bytes"],
             "diskUsedBytes": info["disk_used_bytes"],

app/core/dependencies.py

@@ -7,10 +7,4 @@ from app.core.database import AsyncSessionLocal
 
 async def get_db() -> AsyncGenerator[AsyncSession, None]:
     async with AsyncSessionLocal() as session:
-        try:
-            yield session
-        except Exception:
-            await session.rollback()
-            raise
-        finally:
-            await session.close()
+        yield session

app/core/exceptions.py

@@ -24,39 +24,23 @@ class RecordNotFoundError(Exception):
         super().__init__(f"Export record '{record_id}' not found")
 
 
+# (异常类, HTTP 状态码, 响应 code 字段)
+_HANDLERS: list[tuple[type[Exception], int, int]] = [
+    (DocumentNotFoundError, 404,  404),
+    (ContentTooLargeError,  413,  413),
+    (ExportError,           500,  500),
+    (RecordNotFoundError,   404, 4041),
+]
+
+
 def register_exception_handlers(app: FastAPI) -> None:
-    @app.exception_handler(DocumentNotFoundError)
-    async def document_not_found_handler(
-        request: Request, exc: DocumentNotFoundError
-    ) -> JSONResponse:
-        return JSONResponse(
-            status_code=404,
-            content={"code": 404, "message": str(exc), "data": None},
-        )
-
-    @app.exception_handler(ContentTooLargeError)
-    async def content_too_large_handler(
-        request: Request, exc: ContentTooLargeError
-    ) -> JSONResponse:
-        return JSONResponse(
-            status_code=413,
-            content={"code": 413, "message": str(exc), "data": None},
-        )
-
-    @app.exception_handler(ExportError)
-    async def export_error_handler(
-        request: Request, exc: ExportError
-    ) -> JSONResponse:
-        return JSONResponse(
-            status_code=500,
-            content={"code": 500, "message": str(exc), "data": None},
-        )
-
-    @app.exception_handler(RecordNotFoundError)
-    async def record_not_found_handler(
-        request: Request, exc: RecordNotFoundError
-    ) -> JSONResponse:
-        return JSONResponse(
-            status_code=404,
-            content={"code": 4041, "message": str(exc), "data": None},
-        )
+    def _make_handler(http_status: int, code: int):
+        async def handler(request: Request, exc: Exception) -> JSONResponse:
+            return JSONResponse(
+                status_code=http_status,
+                content={"code": code, "message": str(exc), "data": None},
+            )
+        return handler
+
+    for exc_class, http_status, code in _HANDLERS:
+        app.add_exception_handler(exc_class, _make_handler(http_status, code))

app/schemas/export.py

@@ -1,7 +1,12 @@
-from typing import Literal, Optional
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Literal, Optional
 
 from pydantic import BaseModel, Field
 
+if TYPE_CHECKING:
+    from app.models.export_record import ExportRecord
+
 
 # ------------------------------------------------------------------ #
 # 导出请求 / 响应
@@ -44,6 +49,19 @@ class ExportRecordSchema(BaseModel):
 
     model_config = {"populate_by_name": True}
 
+    @classmethod
+    def from_record(cls, r: "ExportRecord") -> "ExportRecordSchema":
+        return cls(
+            record_id=r.id,
+            user_id=r.user_id,
+            file_name=r.file_name,
+            file_size=r.file_size,
+            download_url=r.download_url,
+            document_id=r.document_id,
+            style_id=r.style_id,
+            created_at=int(r.created_at.timestamp() * 1000),
+        )
+
 
 class PaginationSchema(BaseModel):
     page: int

app/services/document_service.py

@@ -1,7 +1,7 @@
 import re
 from datetime import datetime, timezone
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.exceptions import ContentTooLargeError, DocumentNotFoundError
@@ -23,8 +23,9 @@ class DocumentService:
         data: CreateDocumentRequest,
         user_id: str | None = None,
     ) -> Document:
-        if len(data.content.encode("utf-8")) > CONTENT_MAX_BYTES:
-            raise ContentTooLargeError(len(data.content.encode("utf-8")))
+        size = len(data.content.encode("utf-8"))
+        if size > CONTENT_MAX_BYTES:
+            raise ContentTooLargeError(size)
 
         doc = Document(
             title=data.title,
@@ -71,11 +72,9 @@ class DocumentService:
         else:
             query = query.order_by(sort_col.desc())
 
-        # 总数
-        count_result = await self.db.execute(
-            query.with_only_columns(Document.id)
-        )
-        total = len(count_result.all())
+        # 总数：SELECT COUNT(*) 而非拉全部 id 再 len()
+        count_q = select(func.count()).select_from(query.subquery())
+        total: int = (await self.db.execute(count_q)).scalar_one()
 
         # 分页
         offset = (page - 1) * page_size
@@ -95,8 +94,9 @@ class DocumentService:
             doc.title = data.title
 
         if data.content is not None:
-            if len(data.content.encode("utf-8")) > CONTENT_MAX_BYTES:
-                raise ContentTooLargeError(len(data.content.encode("utf-8")))
+            size = len(data.content.encode("utf-8"))
+            if size > CONTENT_MAX_BYTES:
+                raise ContentTooLargeError(size)
             doc.content = data.content
 
         if data.blocks is not None:
@@ -144,12 +144,6 @@ class DocumentService:
             heading_info.append((line_idx, level, idx))
             level_counter[level] = idx + 1
 
-        # 构建查找字典 (level, index) → line_idx
-        heading_map: dict[tuple[int, int], int] = {
-            (level, index): line_idx
-            for line_idx, level, index in heading_info
-        }
-
         # 将 lines 拆成块
         # 块边界 = 各标题行的 line_idx
         split_points = sorted({line_idx for line_idx, _, _ in heading_info})
@@ -157,22 +151,20 @@ class DocumentService:
 
         # 前置正文（首个标题之前的内容）
         pre_content_end = split_points[0] if split_points else len(lines)
-        chunks: list[str] = []
-        chunks.append("\n".join(lines[:pre_content_end]))
+        chunks: list[str] = ["\n".join(lines[:pre_content_end])]
 
-        # 各标题块
-        block_keys: list[tuple[int, int] | None] = [None]  # 对应 pre_content
+        # 各标题块，同时构建 (level, index) → chunk_idx 映射，O(1) 定位
+        key_to_chunk: dict[tuple[int, int], int] = {}
         for i, sp in enumerate(split_points[:-1]):
             end = split_points[i + 1]
             chunks.append("\n".join(lines[sp:end]))
             _, level, index = heading_info[i]
-            block_keys.append((level, index))
+            key_to_chunk[(level, index)] = len(chunks) - 1
 
         # 执行替换
         for block_update in blocks:
             key = (block_update.level, block_update.index)
-            if key in heading_map:
-                chunk_idx = block_keys.index(key)
-                chunks[chunk_idx] = block_update.content
+            if key in key_to_chunk:
+                chunks[key_to_chunk[key]] = block_update.content
 
         return "\n".join(chunks).strip() + "\n"

app/services/export_service.py

@@ -6,6 +6,7 @@ import secrets
 import unicodedata
 from pathlib import Path
 from typing import Optional
+from datetime import date
 
 import mistune
 from docx import Document
@@ -16,6 +17,9 @@ from lxml import etree
 
 from app.config import settings
 from app.core.exceptions import ExportError
+from app.core.database import AsyncSessionLocal
+from app.services.export_record_service import ExportRecordService
+from app.services.storage_monitor import check_quota
 
 
 # ------------------------------------------------------------------ #
@@ -58,8 +62,7 @@ def build_style_map(style_data: dict) -> dict[str, dict]:
 def dict_to_element(d: dict) -> etree._Element:
     """将 element_to_dict() 产生的字典还原为 lxml Element（styles.py 的逆操作）。"""
     tag = d["@tag"]
-    attrib = {k: v for k, v in d.get("@attrib", {}).items()}
-    elem = etree.Element(tag, attrib=attrib)
+    elem = etree.Element(tag, attrib=dict(d.get("@attrib", {})))
 
     if d.get("#text"):
         elem.text = d["#text"]
@@ -126,14 +129,7 @@ class DocxRenderer(mistune.BaseRenderer):
     """将 mistune AST token 流渲染到 python-docx Document 对象。"""
 
     # Word 内置标题样式的英文名（用于查找 style_map 时的候选 key）
-    _HEADING_ALIASES = {
-        1: ["Heading 1", "heading 1"],
-        2: ["Heading 2", "heading 2"],
-        3: ["Heading 3", "heading 3"],
-        4: ["Heading 4", "heading 4"],
-        5: ["Heading 5", "heading 5"],
-        6: ["Heading 6", "heading 6"],
-    }
+    _HEADING_ALIASES = {i: [f"Heading {i}", f"heading {i}"] for i in range(1, 7)}
 
     def __init__(self, style_map: dict, style_data: dict) -> None:
         """初始化渲染器，注入完整样式定义并缓存常用样式 ID。"""
@@ -415,7 +411,6 @@ async def export_doc(
     document_id: Optional[str] = None,
 ) -> dict:
     """导出入口：加载样式、生成 .docx，按用户/日期分区写入，写入 export_records 记录，返回 { record_id, download_url, file_name, style_id, warning }。"""
-    from datetime import date
 
     # 1. 加载样式文件，构建映射
     style_data = load_style_file(style_id)
@@ -445,16 +440,7 @@ async def export_doc(
 
     file_size = file_path.stat().st_size
 
-    # 4. 生成永久下载链接（无过期时间）
-    download_url = (
-        f"{settings.base_url.rstrip('/')}/api/v1/export/records"
-        # 占位，record_id 写入 DB 后拼接
-    )
-
-    # 5. 写入 export_records 数据库记录
-    from app.core.database import AsyncSessionLocal
-    from app.services.export_record_service import ExportRecordService
-
+    # 4. 写入 export_records 数据库记录
     async with AsyncSessionLocal() as db:
         svc = ExportRecordService(db)
         record = await svc.create_record(
@@ -462,7 +448,7 @@ async def export_doc(
             file_name=final_name,
             file_path=str(file_path),
             file_size=file_size,
-            download_url="",          # 先占位，下面用 record_id 补全
+            download_url="",
             document_id=document_id,
             style_id=actual_style_id,
         )
@@ -476,7 +462,6 @@ async def export_doc(
         await db.commit()
 
     # 6. 导出后检查磁盘配额
-    from app.services.storage_monitor import check_quota
     warning = check_quota(user_id)
 
     return {

app/services/storage_monitor.py

@@ -72,31 +72,36 @@ def get_storage_info() -> dict:
 # 配额检查（导出后调用 / 定时任务共用）
 # ------------------------------------------------------------------ #
 
+def _log_quota_warnings(info: dict, prefix: str) -> None:
+    """将超限情况写入日志，供 check_quota 和定时任务共用。"""
+    if info["quota_exceeded"]:
+        logger.warning(
+            "[%s] tmp/ 总占用 %d bytes 超过配额 %d bytes",
+            prefix,
+            info["tmp_total_bytes"],
+            info["quota_bytes"],
+        )
+    for u in info["users"]:
+        if u["quota_exceeded"]:
+            logger.warning(
+                "[%s] 用户 %s 占用 %d bytes 超过均分配额 %d bytes",
+                prefix,
+                u["user_id"],
+                u["used_bytes"],
+                info["per_user_quota_bytes"],
+            )
+
+
 def check_quota(user_id: str) -> str | None:
     """
     检查存储配额，返回用户侧 warning 文本；无超限时返回 None。
     同时将管理员级别超限情况写入日志。
     """
     info = get_storage_info()
+    _log_quota_warnings(info, prefix="存储告警")
 
-    # 全局超限 → 记录管理员日志
-    if info["quota_exceeded"]:
-        logger.warning(
-            "[存储告警] tmp/ 总占用 %d bytes 超过配额 %d bytes（磁盘 %.0f%%）",
-            info["tmp_total_bytes"],
-            info["quota_bytes"],
-            settings.disk_quota_ratio * 100,
-        )
-
-    # 检查当前用户是否超个人配额
     user_entry = next((u for u in info["users"] if u["user_id"] == user_id), None)
     if user_entry and user_entry["quota_exceeded"]:
-        logger.warning(
-            "[存储告警] 用户 %s 占用 %d bytes 超过均分配额 %d bytes",
-            user_id,
-            user_entry["used_bytes"],
-            info["per_user_quota_bytes"],
-        )
         return "您的存储空间已超出限额，请删除旧文件释放空间"
 
     return None
@@ -111,21 +116,7 @@ async def _periodic_check(interval_seconds: int = 1800) -> None:
     while True:
         await asyncio.sleep(interval_seconds)
         try:
-            info = get_storage_info()
-            if info["quota_exceeded"]:
-                logger.warning(
-                    "[定时检查] tmp/ 总占用 %d bytes 超过配额 %d bytes",
-                    info["tmp_total_bytes"],
-                    info["quota_bytes"],
-                )
-            for u in info["users"]:
-                if u["quota_exceeded"]:
-                    logger.warning(
-                        "[定时检查] 用户 %s 占用 %d bytes 超过均分配额 %d bytes",
-                        u["user_id"],
-                        u["used_bytes"],
-                        info["per_user_quota_bytes"],
-                    )
+            _log_quota_warnings(get_storage_info(), prefix="定时检查")
         except Exception:
             logger.exception("[定时检查] 磁盘检查异常")