Merge pull request 'feat: add backend catalog and review APIs' (#4) from feat/backend-knowledge-management into dev

.gitignore

@@ -8,3 +8,6 @@ dist/
 .vscode/
 .idea/
 *.log
+.ruff_cache/
+.coverage
+htmlcov/

README.md

@@ -1,9 +1,9 @@
-# 那拉提文旅智能问答：前后端基础壳子
+# 那拉提文旅智能问答
 
-基础工程骨架：Vue 3 + TypeScript + Vite 前端；FastAPI + Pydantic + SQLAlchemy 后端；PostgreSQL + pgvector 数据库配置。
+Vue 3 + TypeScript + Vite 前端；FastAPI + Pydantic + SQLAlchemy 后端；PostgreSQL + pgvector。当前后端提供知识管理、来源、文档、人工审核和历史记录，问答仍为占位接口。
 
 ## 启动数据库
-需要 Docker Desktop，在项目根目录运行：
+需要可用的 Docker Engine（Windows 可使用 Docker Desktop），在项目根目录运行：
 ```bash
 docker compose up -d db
 ```
@@ -15,11 +15,25 @@ cd backend
 py -3.12 -m venv .venv
 .\.venv\Scripts\Activate.ps1
 pip install -r requirements.txt
-Copy-Item .env.example .env
+if (!(Test-Path .env)) { Copy-Item .env.example .env }
+alembic upgrade head
 uvicorn app.main:app --reload
 ```
 API 文档：http://127.0.0.1:8000/docs
 
+Linux / macOS 可使用以下等价命令（需要 Python 3.11+；`.env` 已存在时不要覆盖）：
+```bash
+cd backend
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+test -f .env || cp .env.example .env
+alembic upgrade head
+uvicorn app.main:app --reload --host 127.0.0.1
+```
+
+配置始终从 `backend/.env` 读取，环境变量优先。必须先迁移再启动；应用启动只检查数据库版本，不再执行 `create_all`。旧版 `knowledge_items` 会被非破坏性接管；首次迁移前仍须备份。迁移会保留原条目及导入快照，不推测缺失的采集时间。旧版 `verified` 条目因缺少完整审核记录，会转为 `pending`，原状态保存在导入日志中。
+
 ## 启动前端（另开终端）
 ```powershell
 cd frontend
@@ -30,13 +44,57 @@ Vite 通常使用 http://localhost:5173。
 
 ## 已有接口
 - GET `/api/health`
+- GET `/api/ready`：检查数据库连接
 - POST `/api/chat`：占位接口，尚未接入检索和 LLM
 - GET `/api/knowledge`
 - POST `/api/knowledge`
+- GET `/api/knowledge/page`、GET/PATCH/DELETE `/api/knowledge/{id}`
+- POST `/api/knowledge/{id}/review`、GET `/api/knowledge/{id}/history`
+- GET/POST `/api/sources`、GET/PATCH/DELETE `/api/sources/{id}`、GET `/api/sources/{id}/history`
+- GET/POST `/api/documents`、GET/PATCH/DELETE `/api/documents/{id}`、GET `/api/documents/{id}/history`
+
+### 阶段 2 接口约定
+
+- 所有日期均为带时区的 ISO 8601。输入拒绝未知字段，标题/正文不能全为空白，网址只接受 HTTP(S)；可选网址空字符串按未填写处理。
+- `GET /api/knowledge` 保留数组响应，支持 `q`、`category`、`verification_status`、`limit`（默认 100，最大 200）和 `offset`。新页面应使用 `/api/knowledge/page`，避免把首批条目数量当作总数。
+- 分页响应为 `{items, total, page, page_size}`；`page` 从 1 开始，`page_size` 默认 20、最大 100。来源、文档、知识分页和各类历史接口均支持分页。
+- 来源分页支持 `q`；文档分页支持 `q`、`source_id`、`category`；知识分页支持 `q`、`category`、`verification_status`、`source_id`、`document_id`。`q` 为字面文本子串搜索，尚非论文中的关键词检索基线。
+- 创建知识沿用 `title/category/content/source_url`，新增可选 `source_id/document_id/collected_at`；状态固定为 `pending`。关联文档时自动解析其来源，填写的来源或网址若冲突会报错。直接来源链接的旧调用仍受支持。
+- 来源创建字段：`name`、可选 `source_url`、`notes`。文档创建字段：`source_id/title/content/collected_at`，可选 `category/notes`。文档来源归属固定，换来源需新建文档。
+- PATCH 请求必须携带读取时获得的 `expected_version`，并至少提交一个修改字段；DELETE 使用查询参数 `expected_version`。版本过期返回 409，防止覆盖他人修改。
+- 审核请求字段见下方示例。`verification_status` 只允许 `pending/verified/rejected`；核验人和说明必填。通过审核前必须有来源网址及真实采集时间。这里只记录人工结论，不会自动核实网页。
+- 修改条目、关联来源或关联文档都会将相关知识转为 `pending`，清除当前审核字段并记录失效原因。历史快照保留当时的条目、来源和文档。
+- 删除采用软删除；被有效文档或知识引用的来源、被有效知识引用的文档不能删除（409）。已删除记录不出现在列表，但历史仍可查看。
+- 错误响应使用 `detail`：参数错误 422、记录不存在 404、版本/引用冲突 409、数据库暂不可用 503。
+
+审核请求示例（仅为占位测试数据）：
+```json
+{
+  "expected_version": 1,
+  "verification_status": "verified",
+  "reviewer": "test-reviewer",
+  "verification_notes": "示例审核说明，实际使用须填写真实核验依据"
+}
+```
+
+响应新增 `version/updated_at/is_deleted` 及关联、采集、审核字段；完整类型以 `/openapi.json` 和 `/docs` 为准。前端现有录入与数组列表可继续使用；来源、文档、审核及完整分页页面仍需按接口联调。
 
 前端包括问答、知识库管理、来源与更新、实验评估四个页面。实验评估页面没有虚构数据；所有页面明确标记当前功能状态。
 
-注意：当前没有实现 Embedding、RAG、网络搜索、后台更新任务、用户认证或生产级数据库迁移。下一步建议先验证前后端联通，再按模块实现。`.env` 不要提交到 Git。
+注意：当前没有实现 Embedding、RAG、网络搜索、后台更新任务或用户认证。审核人是本机开发环境输入字段，不是认证身份。服务保持仅本机访问；正式部署能力留到后续阶段。`.env` 不要提交到 Git。
+
+### 后端测试和迁移回滚
+
+在 `backend` 目录安装 `requirements-dev.txt` 后执行：
+```bash
+ruff check .
+ruff format --check .
+pytest -q
+```
+
+测试需要真实 PostgreSQL + pgvector：设置 `TEST_DATABASE_URL`（`postgresql+psycopg://...`），或另行 `pip install pgserver` 使用可选的本机临时实例。`pgserver` 仅用于测试，不是应用运行依赖；外部测试数据库账号需有建 Schema 权限且已启用 vector 扩展。每个测试使用独立 Schema，不会清空业务表。测试使用的占位正文和审核人不代表真实资料。
+
+迁移命令为 `alembic upgrade head`，可重复执行。需要撤回阶段 2 时，先备份，再执行 `alembic downgrade 0001_baseline`；它保留基础知识字段，但会删除新增来源、文档、审核字段与历史记录，软删除的基础条目也会重新可见。恢复完整数据应使用备份。不要在有业务数据的数据库执行 `downgrade base`。
 
 ## 团队协作文档
 

backend/alembic.ini

@@ -0,0 +1,4 @@
+[alembic]
+script_location = %(here)s/migrations
+prepend_sys_path = %(here)s
+path_separator = os

backend/app/api/routes/chat.py

@@ -1,7 +1,15 @@
 from fastapi import APIRouter
+
 from app.schemas.chat import ChatRequest, ChatResponse
+
 router = APIRouter(tags=["chat"])
+
+
 @router.post("/chat", response_model=ChatResponse)
 def chat(payload: ChatRequest) -> ChatResponse:
     # TODO: retrieval -> context assembly -> LLM -> evidence/source checks
-    return ChatResponse(answer=f"已收到问题：{payload.question}。当前为基础壳子，尚未接入知识检索或大语言模型。", status="stub", sources=[])
+    return ChatResponse(
+        answer=f"已收到问题：{payload.question}。当前为基础壳子，尚未接入知识检索或大语言模型。",
+        status="stub",
+        sources=[],
+    )

backend/app/api/routes/documents.py

@@ -0,0 +1,99 @@
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from sqlalchemy import or_, select
+from sqlalchemy.orm import Session
+
+from app.db.session import get_db
+from app.models import Document, KnowledgeItem, Source
+from app.schemas.catalog import AuditRead, DocumentCreate, DocumentRead, DocumentUpdate, Page
+from app.services.catalog import (
+    ensure_version,
+    get_item,
+    history,
+    invalidate_knowledge,
+    page_result,
+    record_event,
+    update_values,
+)
+
+router = APIRouter(prefix="/documents", tags=["documents"])
+
+
+@router.get("", response_model=Page[DocumentRead])
+def list_documents(
+    q: str = Query("", max_length=200),
+    source_id: int | None = Query(None, gt=0),
+    category: str | None = Query(None, max_length=80),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    db: Session = Depends(get_db),
+):
+    statement = select(Document).where(Document.is_deleted.is_(False))
+    if source_id is not None:
+        statement = statement.where(Document.source_id == source_id)
+    if category is not None:
+        statement = statement.where(Document.category == category)
+    if q:
+        statement = statement.where(
+            or_(
+                Document.title.icontains(q, autoescape=True),
+                Document.content.icontains(q, autoescape=True),
+            )
+        )
+    return page_result(db, statement.order_by(Document.id.desc()), page, page_size)
+
+
+@router.post("", response_model=DocumentRead, status_code=201)
+def create_document(payload: DocumentCreate, db: Session = Depends(get_db)):
+    get_item(db, Source, payload.source_id, lock=True)
+    item = Document(**payload.model_dump())
+    db.add(item)
+    record_event(db, item, "create")
+    db.commit()
+    return item
+
+
+@router.get("/{item_id}", response_model=DocumentRead)
+def read_document(item_id: int, db: Session = Depends(get_db)):
+    return get_item(db, Document, item_id)
+
+
+@router.patch("/{item_id}", response_model=DocumentRead)
+def update_document(item_id: int, payload: DocumentUpdate, db: Session = Depends(get_db)):
+    item = get_item(db, Document, item_id, lock=True)
+    ensure_version(item, payload.expected_version)
+    update_values(item, payload)
+    record_event(db, item, "update")
+    invalidate_knowledge(db, KnowledgeItem.document_id == item.id)
+    db.commit()
+    return item
+
+
+@router.delete("/{item_id}", status_code=204)
+def delete_document(
+    item_id: int, expected_version: int = Query(..., ge=1), db: Session = Depends(get_db)
+):
+    item = get_item(db, Document, item_id, lock=True)
+    ensure_version(item, expected_version)
+    if (
+        db.scalar(
+            select(KnowledgeItem.id)
+            .where(KnowledgeItem.document_id == item_id, KnowledgeItem.is_deleted.is_(False))
+            .limit(1)
+        )
+        is not None
+    ):
+        raise HTTPException(409, "文档仍被知识条目引用，不能删除")
+    item.is_deleted = True
+    record_event(db, item, "delete")
+    db.commit()
+    return Response(status_code=204)
+
+
+@router.get("/{item_id}/history", response_model=Page[AuditRead])
+def document_history(
+    item_id: int,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    db: Session = Depends(get_db),
+):
+    return history(db, Document, item_id, page, page_size)

backend/app/api/routes/health.py

@@ -1,5 +1,18 @@
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.db.session import get_db
+
 router = APIRouter(tags=["health"])
+
+
 @router.get("/health")
 def health():
     return {"status": "ok", "service": "nalati-rag-api"}
+
+
+@router.get("/ready")
+def ready(db: Session = Depends(get_db)):
+    db.execute(text("SELECT 1"))
+    return {"status": "ok", "database": "connected"}

backend/app/api/routes/knowledge.py

@@ -1,15 +1,151 @@
-from fastapi import APIRouter, Depends
-from sqlalchemy import select
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from sqlalchemy import or_, select
 from sqlalchemy.orm import Session
+
 from app.db.session import get_db
-from app.models.knowledge import KnowledgeItem
-from app.schemas.knowledge import KnowledgeCreate, KnowledgeRead
-router = APIRouter(tags=["knowledge"])
-@router.get("/knowledge", response_model=list[KnowledgeRead])
-def list_knowledge(db: Session = Depends(get_db)):
-    return list(db.scalars(select(KnowledgeItem).order_by(KnowledgeItem.id.desc())).all())
-@router.post("/knowledge", response_model=KnowledgeRead, status_code=201)
+from app.models import KnowledgeItem
+from app.models.catalog import utc_now
+from app.schemas.catalog import AuditRead, Page, Status
+from app.schemas.knowledge import KnowledgeCreate, KnowledgeRead, KnowledgeReview, KnowledgeUpdate
+from app.services.catalog import (
+    ensure_version,
+    get_item,
+    history,
+    page_result,
+    record_event,
+    reset_review,
+    resolve_knowledge_links,
+    update_values,
+)
+
+router = APIRouter(prefix="/knowledge", tags=["knowledge"])
+
+
+def filtered_query(q="", category=None, verification_status=None, source_id=None, document_id=None):
+    statement = select(KnowledgeItem).where(KnowledgeItem.is_deleted.is_(False))
+    for column, value in (
+        (KnowledgeItem.category, category),
+        (KnowledgeItem.verification_status, verification_status),
+        (KnowledgeItem.source_id, source_id),
+        (KnowledgeItem.document_id, document_id),
+    ):
+        if value is not None:
+            statement = statement.where(column == value)
+    if q:
+        statement = statement.where(
+            or_(
+                KnowledgeItem.title.icontains(q, autoescape=True),
+                KnowledgeItem.content.icontains(q, autoescape=True),
+            )
+        )
+    return statement.order_by(KnowledgeItem.id.desc())
+
+
+@router.get("", response_model=list[KnowledgeRead])
+def list_knowledge(
+    q: str = Query("", max_length=200),
+    category: str | None = Query(None, max_length=80),
+    verification_status: Status | None = None,
+    limit: int = Query(100, ge=1, le=200),
+    offset: int = Query(0, ge=0),
+    db: Session = Depends(get_db),
+):
+    """保留数组响应；完整分页与关联筛选使用 /knowledge/page。"""
+    return list(
+        db.scalars(filtered_query(q, category, verification_status).offset(offset).limit(limit))
+    )
+
+
+@router.get("/page", response_model=Page[KnowledgeRead])
+def paged_knowledge(
+    q: str = Query("", max_length=200),
+    category: str | None = Query(None, max_length=80),
+    verification_status: Status | None = None,
+    source_id: int | None = Query(None, gt=0),
+    document_id: int | None = Query(None, gt=0),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    db: Session = Depends(get_db),
+):
+    return page_result(
+        db,
+        filtered_query(q, category, verification_status, source_id, document_id),
+        page,
+        page_size,
+    )
+
+
+@router.post("", response_model=KnowledgeRead, status_code=201)
 def create_knowledge(payload: KnowledgeCreate, db: Session = Depends(get_db)):
-    item = KnowledgeItem(title=payload.title, category=payload.category, content=payload.content, source_url=str(payload.source_url) if payload.source_url else None, verification_status="pending")
-    db.add(item); db.commit(); db.refresh(item)
+    item = KnowledgeItem(
+        **payload.model_dump(exclude={"source_url"}),
+        source_url=str(payload.source_url) if payload.source_url else None,
+        verification_status="pending",
+    )
+    resolve_knowledge_links(db, item)
+    db.add(item)
+    record_event(db, item, "create")
+    db.commit()
     return item
+
+
+@router.get("/{item_id}", response_model=KnowledgeRead)
+def read_knowledge(item_id: int, db: Session = Depends(get_db)):
+    return get_item(db, KnowledgeItem, item_id)
+
+
+@router.patch("/{item_id}", response_model=KnowledgeRead)
+def update_knowledge(item_id: int, payload: KnowledgeUpdate, db: Session = Depends(get_db)):
+    item = get_item(db, KnowledgeItem, item_id)
+    ensure_version(item, payload.expected_version)
+    update_values(item, payload)
+    if payload.document_id is not None and "source_id" not in payload.model_fields_set:
+        item.source_id = None
+    if {
+        "source_id",
+        "document_id",
+    } & payload.model_fields_set and "source_url" not in payload.model_fields_set:
+        item.source_url = None
+    resolve_knowledge_links(db, item)
+    reset_review(item)
+    record_event(db, item, "update")
+    db.commit()
+    return item
+
+
+@router.post("/{item_id}/review", response_model=KnowledgeRead)
+def review_knowledge(item_id: int, payload: KnowledgeReview, db: Session = Depends(get_db)):
+    item = get_item(db, KnowledgeItem, item_id)
+    ensure_version(item, payload.expected_version)
+    resolve_knowledge_links(db, item)
+    if payload.verification_status == "verified" and (not item.source_url or not item.collected_at):
+        raise HTTPException(422, "核验通过前必须补全来源网址和真实采集时间")
+    item.verification_status = payload.verification_status
+    item.reviewer = payload.reviewer
+    item.reviewed_at = utc_now()
+    item.verification_notes = payload.verification_notes
+    record_event(db, item, "review", actor=payload.reviewer, notes=payload.verification_notes)
+    db.commit()
+    return item
+
+
+@router.delete("/{item_id}", status_code=204)
+def delete_knowledge(
+    item_id: int, expected_version: int = Query(..., ge=1), db: Session = Depends(get_db)
+):
+    item = get_item(db, KnowledgeItem, item_id)
+    ensure_version(item, expected_version)
+    item.is_deleted = True
+    record_event(db, item, "delete")
+    db.commit()
+    return Response(status_code=204)
+
+
+@router.get("/{item_id}/history", response_model=Page[AuditRead])
+def knowledge_history(
+    item_id: int,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    db: Session = Depends(get_db),
+):
+    return history(db, KnowledgeItem, item_id, page, page_size)

backend/app/api/routes/sources.py

@@ -0,0 +1,95 @@
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from sqlalchemy import or_, select
+from sqlalchemy.orm import Session
+
+from app.db.session import get_db
+from app.models import Document, KnowledgeItem, Source
+from app.schemas.catalog import AuditRead, Page, SourceCreate, SourceRead, SourceUpdate
+from app.services.catalog import (
+    ensure_version,
+    get_item,
+    history,
+    invalidate_knowledge,
+    page_result,
+    record_event,
+    update_values,
+)
+
+router = APIRouter(prefix="/sources", tags=["sources"])
+
+
+@router.get("", response_model=Page[SourceRead])
+def list_sources(
+    q: str = Query("", max_length=200),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    db: Session = Depends(get_db),
+):
+    statement = select(Source).where(Source.is_deleted.is_(False))
+    if q:
+        statement = statement.where(
+            or_(
+                Source.name.icontains(q, autoescape=True),
+                Source.notes.icontains(q, autoescape=True),
+            )
+        )
+    return page_result(db, statement.order_by(Source.id.desc()), page, page_size)
+
+
+@router.post("", response_model=SourceRead, status_code=201)
+def create_source(payload: SourceCreate, db: Session = Depends(get_db)):
+    item = Source(**payload.model_dump(mode="json"))
+    db.add(item)
+    record_event(db, item, "create")
+    db.commit()
+    return item
+
+
+@router.get("/{item_id}", response_model=SourceRead)
+def read_source(item_id: int, db: Session = Depends(get_db)):
+    return get_item(db, Source, item_id)
+
+
+@router.patch("/{item_id}", response_model=SourceRead)
+def update_source(item_id: int, payload: SourceUpdate, db: Session = Depends(get_db)):
+    item = get_item(db, Source, item_id, lock=True)
+    ensure_version(item, payload.expected_version)
+    update_values(item, payload)
+    record_event(db, item, "update")
+    invalidate_knowledge(
+        db, KnowledgeItem.source_id == item.id, source_url=item.source_url, update_url=True
+    )
+    db.commit()
+    return item
+
+
+@router.delete("/{item_id}", status_code=204)
+def delete_source(
+    item_id: int, expected_version: int = Query(..., ge=1), db: Session = Depends(get_db)
+):
+    item = get_item(db, Source, item_id, lock=True)
+    ensure_version(item, expected_version)
+    for model in (Document, KnowledgeItem):
+        if (
+            db.scalar(
+                select(model.id)
+                .where(model.source_id == item_id, model.is_deleted.is_(False))
+                .limit(1)
+            )
+            is not None
+        ):
+            raise HTTPException(409, "来源仍被文档或知识条目引用，不能删除")
+    item.is_deleted = True
+    record_event(db, item, "delete")
+    db.commit()
+    return Response(status_code=204)
+
+
+@router.get("/{item_id}/history", response_model=Page[AuditRead])
+def source_history(
+    item_id: int,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    db: Session = Depends(get_db),
+):
+    return history(db, Source, item_id, page, page_size)

backend/app/core/config.py

@@ -1,14 +1,21 @@
+from pathlib import Path
+
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+
 class Settings(BaseSettings):
     app_name: str = "Nalati Tourism QA API"
     app_env: str = "development"
     api_prefix: str = "/api"
     database_url: str = "postgresql+psycopg://nalati:change-this-password@127.0.0.1:5432/nalati_rag"
     cors_origins: str = "http://localhost:5173,http://127.0.0.1:5173"
-    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
+    model_config = SettingsConfigDict(
+        env_file=Path(__file__).resolve().parents[2] / ".env", extra="ignore"
+    )
+
     @property
     def cors_origin_list(self) -> list[str]:
         return [x.strip() for x in self.cors_origins.split(",") if x.strip()]
 
+
 settings = Settings()

backend/app/db/session.py

@@ -1,16 +1,25 @@
 from sqlalchemy import create_engine
 from sqlalchemy.orm import DeclarativeBase, sessionmaker
+
 from app.core.config import settings
 
+
 class Base(DeclarativeBase):
     pass
 
-engine = create_engine(settings.database_url, pool_pre_ping=True)
+
+engine = create_engine(
+    settings.database_url, pool_pre_ping=True, connect_args={"connect_timeout": 5}
+)
 SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
 
+
 def get_db():
     db = SessionLocal()
     try:
         yield db
+    except Exception:
+        db.rollback()
+        raise
     finally:
         db.close()

backend/app/main.py

@@ -1,25 +1,64 @@
 from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
 from sqlalchemy import text
+from sqlalchemy.exc import IntegrityError, OperationalError, ProgrammingError
+from sqlalchemy.orm.exc import StaleDataError
+
+from app.api.routes import chat, documents, health, knowledge, sources
 from app.core.config import settings
-from app.db.session import Base, engine
-from app.models import KnowledgeItem  # noqa: F401
-from app.api.routes import health, chat, knowledge
+from app.db.session import engine
 
-@asynccontextmanager
-async def lifespan(app: FastAPI):
-    # Convenience for local bootstrap. Use Alembic migrations before production.
-    with engine.begin() as connection:
-        connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
-    Base.metadata.create_all(bind=engine)
-    yield
 
-app = FastAPI(title=settings.app_name, version="0.1.0", lifespan=lifespan)
-app.add_middleware(CORSMiddleware, allow_origins=settings.cors_origin_list, allow_credentials=True, allow_methods=["*"], allow_headers=["*"])
-app.include_router(health.router, prefix=settings.api_prefix)
-app.include_router(chat.router, prefix=settings.api_prefix)
-app.include_router(knowledge.router, prefix=settings.api_prefix)
-@app.get("/")
-def root():
-    return {"message": "Nalati Tourism QA API", "docs": "/docs"}
+def create_app(db_engine=engine) -> FastAPI:
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        try:
+            with db_engine.connect() as connection:
+                revision = connection.scalar(text("SELECT version_num FROM alembic_version"))
+                if revision != "0002_catalog":
+                    raise RuntimeError(
+                        "数据库版本不匹配，请先在 backend 目录执行 alembic upgrade head"
+                    )
+        except ProgrammingError as exc:
+            raise RuntimeError(
+                "数据库尚未初始化，请先在 backend 目录执行 alembic upgrade head"
+            ) from exc
+        try:
+            yield
+        finally:
+            db_engine.dispose()
+
+    app = FastAPI(title=settings.app_name, version="0.2.0", lifespan=lifespan)
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=settings.cors_origin_list,
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+    for router in (health.router, chat.router, knowledge.router, sources.router, documents.router):
+        app.include_router(router, prefix=settings.api_prefix)
+
+    @app.exception_handler(StaleDataError)
+    async def stale_data_handler(request, exc):
+        return JSONResponse(status_code=409, content={"detail": "记录已更新，请刷新后重试"})
+
+    @app.exception_handler(IntegrityError)
+    async def integrity_handler(request, exc):
+        return JSONResponse(status_code=409, content={"detail": "数据关联或约束冲突"})
+
+    @app.exception_handler(OperationalError)
+    async def database_unavailable_handler(request, exc):
+        return JSONResponse(status_code=503, content={"detail": "数据库暂不可用"})
+
+    @app.get("/")
+    def root():
+        return {"message": "Nalati Tourism QA API", "docs": "/docs"}
+
+    return app
+
+
+app = create_app()

backend/app/models/__init__.py

@@ -1 +1,4 @@
+from app.models.catalog import AuditEvent, Document, Source
 from app.models.knowledge import KnowledgeItem
+
+__all__ = ["AuditEvent", "Document", "KnowledgeItem", "Source"]

backend/app/models/catalog.py

@@ -0,0 +1,54 @@
+from datetime import datetime, timezone
+
+from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy.orm import Mapped, declared_attr, mapped_column
+
+from app.db.session import Base
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+class Versioned:
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=utc_now, onupdate=utc_now
+    )
+    is_deleted: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+
+    @declared_attr.directive
+    def __mapper_args__(cls):
+        return {"version_id_col": cls.version}
+
+
+class Source(Versioned, Base):
+    __tablename__ = "sources"
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String(250), nullable=False)
+    source_url: Mapped[str | None] = mapped_column(String(1000))
+    notes: Mapped[str] = mapped_column(Text, nullable=False, default="")
+
+
+class Document(Versioned, Base):
+    __tablename__ = "documents"
+    id: Mapped[int] = mapped_column(primary_key=True)
+    source_id: Mapped[int] = mapped_column(ForeignKey("sources.id"), index=True)
+    title: Mapped[str] = mapped_column(String(250), nullable=False)
+    category: Mapped[str] = mapped_column(String(80), nullable=False)
+    content: Mapped[str] = mapped_column(Text, nullable=False)
+    collected_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    notes: Mapped[str] = mapped_column(Text, nullable=False, default="")
+
+
+class AuditEvent(Base):
+    __tablename__ = "audit_events"
+    id: Mapped[int] = mapped_column(primary_key=True)
+    entity_type: Mapped[str] = mapped_column(String(30), nullable=False, index=True)
+    entity_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
+    action: Mapped[str] = mapped_column(String(30), nullable=False)
+    actor: Mapped[str | None] = mapped_column(String(100))
+    notes: Mapped[str] = mapped_column(Text, nullable=False, default="")
+    snapshot: Mapped[dict] = mapped_column(JSON, nullable=False)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

backend/app/models/knowledge.py

@@ -1,14 +1,29 @@
 from datetime import datetime
-from sqlalchemy import DateTime, Integer, String, Text, func
-from sqlalchemy.orm import Mapped, mapped_column
-from app.db.session import Base
 
-class KnowledgeItem(Base):
+from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.session import Base
+from app.models.catalog import Versioned
+
+
+class KnowledgeItem(Versioned, Base):
     __tablename__ = "knowledge_items"
+    __table_args__ = (
+        CheckConstraint(
+            "verification_status IN ('pending', 'verified', 'rejected')",
+            name="ck_knowledge_verification_status",
+        ),
+    )
     id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
     title: Mapped[str] = mapped_column(String(250), nullable=False, index=True)
     category: Mapped[str] = mapped_column(String(80), nullable=False, default="未分类")
     content: Mapped[str] = mapped_column(Text, nullable=False)
     source_url: Mapped[str | None] = mapped_column(String(1000), nullable=True)
     verification_status: Mapped[str] = mapped_column(String(30), nullable=False, default="pending")
-    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
+    source_id: Mapped[int | None] = mapped_column(ForeignKey("sources.id"), index=True)
+    document_id: Mapped[int | None] = mapped_column(ForeignKey("documents.id"), index=True)
+    collected_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    reviewer: Mapped[str | None] = mapped_column(String(100))
+    reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    verification_notes: Mapped[str | None] = mapped_column(Text)

backend/app/schemas/catalog.py

@@ -0,0 +1,124 @@
+from datetime import datetime
+from typing import Annotated, Generic, Literal, TypeVar
+
+from pydantic import (
+    AwareDatetime,
+    BaseModel,
+    ConfigDict,
+    Field,
+    HttpUrl,
+    field_validator,
+    model_validator,
+)
+
+Title = Annotated[str, Field(min_length=1, max_length=250)]
+Category = Annotated[str, Field(min_length=1, max_length=80)]
+Body = Annotated[str, Field(min_length=1, max_length=100_000)]
+Notes = Annotated[str, Field(max_length=10_000)]
+Status = Literal["pending", "verified", "rejected"]
+PositiveId = Annotated[int, Field(gt=0)]
+
+
+class Input(BaseModel):
+    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
+
+    @field_validator("source_url", mode="before", check_fields=False)
+    @classmethod
+    def empty_url_is_absent(cls, value):
+        return None if isinstance(value, str) and not value.strip() else value
+
+
+class Patch(Input):
+    expected_version: PositiveId
+
+    @model_validator(mode="after")
+    def require_changes(self):
+        if not self.model_fields_set - {"expected_version"}:
+            raise ValueError("至少提供一个需要更新的字段")
+        return self
+
+    @field_validator("*", mode="before")
+    @classmethod
+    def disallow_required_null(cls, value, info):
+        if (
+            info.field_name
+            in {"name", "title", "category", "content", "notes", "collected_at", "expected_version"}
+            and value is None
+        ):
+            raise ValueError("该字段不能为 null")
+        return value
+
+
+class Read(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    version: int
+    created_at: datetime
+    updated_at: datetime
+    is_deleted: bool
+
+
+class SourceCreate(Input):
+    name: Title
+    source_url: HttpUrl | None = Field(default=None, max_length=1000)
+    notes: Notes = ""
+
+
+class SourceUpdate(Patch):
+    name: Title | None = None
+    source_url: HttpUrl | None = Field(default=None, max_length=1000)
+    notes: Notes | None = None
+
+
+class SourceRead(Read):
+    name: str
+    source_url: str | None
+    notes: str
+
+
+class DocumentCreate(Input):
+    source_id: PositiveId
+    title: Title
+    category: Category = "未分类"
+    content: Body
+    collected_at: AwareDatetime
+    notes: Notes = ""
+
+
+class DocumentUpdate(Patch):
+    title: Title | None = None
+    category: Category | None = None
+    content: Body | None = None
+    collected_at: AwareDatetime | None = None
+    notes: Notes | None = None
+
+
+class DocumentRead(Read):
+    source_id: int
+    title: str
+    category: str
+    content: str
+    collected_at: datetime
+    notes: str
+
+
+T = TypeVar("T")
+
+
+class Page(BaseModel, Generic[T]):
+    items: list[T]
+    total: int
+    page: int
+    page_size: int
+
+
+class AuditRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    entity_type: str
+    entity_id: int
+    action: str
+    actor: str | None
+    notes: str
+    snapshot: dict
+    created_at: datetime

backend/app/schemas/chat.py

@@ -1,12 +1,15 @@
 from pydantic import BaseModel, Field
 
+
 class ChatRequest(BaseModel):
     question: str = Field(min_length=1, max_length=2000)
 
+
 class SourcePreview(BaseModel):
     title: str
     source_url: str | None = None
 
+
 class ChatResponse(BaseModel):
     answer: str
     status: str

backend/app/schemas/knowledge.py

@@ -1,18 +1,46 @@
 from datetime import datetime
-from pydantic import BaseModel, ConfigDict, Field, HttpUrl
 
-class KnowledgeCreate(BaseModel):
-    title: str = Field(min_length=1, max_length=250)
-    category: str = Field(default="未分类", max_length=80)
-    content: str = Field(min_length=1)
-    source_url: HttpUrl | None = None
+from pydantic import AwareDatetime, Field, HttpUrl
 
-class KnowledgeRead(BaseModel):
-    model_config = ConfigDict(from_attributes=True)
-    id: int
+from app.schemas.catalog import Body, Category, Input, Patch, PositiveId, Read, Status, Title
+
+
+class KnowledgeCreate(Input):
+    title: Title
+    category: Category = "未分类"
+    content: Body
+    source_url: HttpUrl | None = Field(default=None, max_length=1000)
+    source_id: PositiveId | None = None
+    document_id: PositiveId | None = None
+    collected_at: AwareDatetime | None = None
+
+
+class KnowledgeUpdate(Patch):
+    title: Title | None = None
+    category: Category | None = None
+    content: Body | None = None
+    source_url: HttpUrl | None = Field(default=None, max_length=1000)
+    source_id: PositiveId | None = None
+    document_id: PositiveId | None = None
+    collected_at: AwareDatetime | None = None
+
+
+class KnowledgeReview(Input):
+    expected_version: PositiveId
+    verification_status: Status
+    reviewer: str = Field(min_length=1, max_length=100)
+    verification_notes: str = Field(min_length=1, max_length=10_000)
+
+
+class KnowledgeRead(Read):
     title: str
     category: str
     content: str
     source_url: str | None
-    verification_status: str
-    created_at: datetime
+    verification_status: Status
+    source_id: int | None
+    document_id: int | None
+    collected_at: datetime | None
+    reviewer: str | None
+    reviewed_at: datetime | None
+    verification_notes: str | None

backend/app/services/catalog.py

@@ -0,0 +1,120 @@
+from fastapi import HTTPException
+from fastapi.encoders import jsonable_encoder
+from sqlalchemy import func, inspect, select
+from sqlalchemy.orm import Session
+
+from app.models import AuditEvent, Document, KnowledgeItem, Source
+from app.models.catalog import utc_now
+
+
+def get_item(db: Session, model, item_id: int, *, include_deleted=False, lock=False):
+    if lock:
+        item = db.scalar(
+            select(model)
+            .where(model.id == item_id)
+            .with_for_update()
+            .execution_options(populate_existing=True)
+        )
+    else:
+        item = db.get(model, item_id)
+    if item is None or (item.is_deleted and not include_deleted):
+        raise HTTPException(404, "记录不存在")
+    return item
+
+
+def ensure_version(item, expected_version: int):
+    if item.version != expected_version:
+        raise HTTPException(409, "记录已更新，请刷新后重试")
+
+
+def values(item) -> dict:
+    return jsonable_encoder(
+        {column.key: getattr(item, column.key) for column in inspect(type(item)).columns}
+    )
+
+
+def record_event(db: Session, item, action: str, *, actor=None, notes=""):
+    db.flush()
+    snapshot = {"item": values(item)}
+    if isinstance(item, KnowledgeItem):
+        source = db.get(Source, item.source_id) if item.source_id else None
+        document = db.get(Document, item.document_id) if item.document_id else None
+        snapshot["source"] = values(source) if source else None
+        snapshot["document"] = values(document) if document else None
+    db.add(
+        AuditEvent(
+            entity_type=item.__tablename__,
+            entity_id=item.id,
+            action=action,
+            actor=actor,
+            notes=notes,
+            snapshot=snapshot,
+        )
+    )
+
+
+def update_values(item, payload):
+    data = payload.model_dump(mode="json", exclude_unset=True, exclude={"expected_version"})
+    data.update(
+        {
+            key: value
+            for key, value in payload.model_dump(exclude_unset=True).items()
+            if key.endswith("_at")
+        }
+    )
+    for name, value in data.items():
+        setattr(item, name, value)
+    item.updated_at = utc_now()
+
+
+def reset_review(item: KnowledgeItem):
+    item.verification_status = "pending"
+    item.reviewer = None
+    item.reviewed_at = None
+    item.verification_notes = None
+    item.updated_at = utc_now()
+
+
+def invalidate_knowledge(db: Session, condition, *, source_url=None, update_url=False):
+    for item in db.scalars(
+        select(KnowledgeItem).where(condition, KnowledgeItem.is_deleted.is_(False))
+    ):
+        reset_review(item)
+        if update_url:
+            item.source_url = source_url
+        record_event(db, item, "invalidate", notes="来源或文档已变更，需要重新核验")
+
+
+def resolve_knowledge_links(db: Session, item: KnowledgeItem):
+    # 先锁依据记录再保存条目，避免引用与父记录修改/删除同时发生。
+    with db.no_autoflush:
+        if item.document_id is not None:
+            document = get_item(db, Document, item.document_id, lock=True)
+            if item.source_id is not None and item.source_id != document.source_id:
+                raise HTTPException(422, "文档与来源不一致")
+            item.source_id = document.source_id
+        if item.source_id is not None:
+            source = get_item(db, Source, item.source_id, lock=True)
+            if item.source_url and item.source_url != source.source_url:
+                raise HTTPException(422, "来源网址与登记来源不一致")
+            item.source_url = source.source_url
+
+
+def page_result(db: Session, statement, page: int, page_size: int):
+    total = db.scalar(select(func.count()).select_from(statement.order_by(None).subquery()))
+    return {
+        "items": list(db.scalars(statement.offset((page - 1) * page_size).limit(page_size))),
+        "total": total,
+        "page": page,
+        "page_size": page_size,
+    }
+
+
+def history(db: Session, model, item_id: int, page: int, page_size: int):
+    get_item(db, model, item_id, include_deleted=True)
+    statement = (
+        select(AuditEvent)
+        .where(AuditEvent.entity_type == model.__tablename__, AuditEvent.entity_id == item_id)
+        .order_by(AuditEvent.id.desc())
+    )
+    return page_result(db, statement, page, page_size)

backend/migrations/env.py

@@ -0,0 +1,26 @@
+from alembic import context
+from sqlalchemy import create_engine, pool
+
+from app import models  # noqa: F401
+from app.core.config import settings
+from app.db.session import Base
+
+config = context.config
+
+
+def run_migrations(connection):
+    context.configure(connection=connection, target_metadata=Base.metadata, compare_type=True)
+    with context.begin_transaction():
+        context.run_migrations()
+
+
+if context.is_offline_mode():
+    raise RuntimeError("迁移需要连接数据库，以检查并保留已有表；请使用 alembic upgrade head")
+elif config.attributes.get("connection") is not None:
+    run_migrations(config.attributes["connection"])
+else:
+    engine = create_engine(
+        settings.database_url, poolclass=pool.NullPool, connect_args={"connect_timeout": 5}
+    )
+    with engine.connect() as connection:
+        run_migrations(connection)

backend/migrations/script.py.mako

@@ -0,0 +1,17 @@
+"""${message}"""
+from alembic import op
+import sqlalchemy as sa
+${imports if imports else ""}
+
+revision = ${repr(up_revision)}
+down_revision = ${repr(down_revision)}
+branch_labels = ${repr(branch_labels)}
+depends_on = ${repr(depends_on)}
+
+
+def upgrade():
+    ${upgrades if upgrades else "pass"}
+
+
+def downgrade():
+    ${downgrades if downgrades else "pass"}

backend/migrations/versions/0001_baseline.py

@@ -0,0 +1,48 @@
+"""Register the original knowledge table without replacing existing data."""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0001_baseline"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    bind = op.get_bind()
+    op.execute("CREATE EXTENSION IF NOT EXISTS vector")
+    inspector = sa.inspect(bind)
+    if inspector.has_table("knowledge_items"):
+        columns = {column["name"] for column in inspector.get_columns("knowledge_items")}
+        expected = {
+            "id",
+            "title",
+            "category",
+            "content",
+            "source_url",
+            "verification_status",
+            "created_at",
+        }
+        if columns != expected:
+            raise RuntimeError("已有 knowledge_items 结构与基础版本不同，请先核对结构并备份")
+        return
+    op.create_table(
+        "knowledge_items",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("title", sa.String(250), nullable=False),
+        sa.Column("category", sa.String(80), nullable=False),
+        sa.Column("content", sa.Text(), nullable=False),
+        sa.Column("source_url", sa.String(1000)),
+        sa.Column("verification_status", sa.String(30), nullable=False),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+    )
+    op.create_index("ix_knowledge_items_id", "knowledge_items", ["id"])
+    op.create_index("ix_knowledge_items_title", "knowledge_items", ["title"])
+
+
+def downgrade():
+    op.drop_table("knowledge_items")
+    # 扩展可能由其他项目共享，不在回滚时删除。

backend/migrations/versions/0002_catalog.py

@@ -0,0 +1,110 @@
+"""Add source, document, review and audit management."""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0002_catalog"
+down_revision = "0001_baseline"
+branch_labels = None
+depends_on = None
+
+
+def version_columns():
+    return [
+        sa.Column("version", sa.Integer(), server_default="1", nullable=False),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.Column("is_deleted", sa.Boolean(), server_default=sa.false(), nullable=False),
+    ]
+
+
+def upgrade():
+    op.create_table(
+        "sources",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("name", sa.String(250), nullable=False),
+        sa.Column("source_url", sa.String(1000)),
+        sa.Column("notes", sa.Text(), nullable=False),
+        *version_columns(),
+    )
+    op.create_table(
+        "documents",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("source_id", sa.Integer(), sa.ForeignKey("sources.id"), nullable=False),
+        sa.Column("title", sa.String(250), nullable=False),
+        sa.Column("category", sa.String(80), nullable=False),
+        sa.Column("content", sa.Text(), nullable=False),
+        sa.Column("collected_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("notes", sa.Text(), nullable=False),
+        *version_columns(),
+    )
+    op.create_index("ix_documents_source_id", "documents", ["source_id"])
+    op.create_table(
+        "audit_events",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column("entity_type", sa.String(30), nullable=False),
+        sa.Column("entity_id", sa.Integer(), nullable=False),
+        sa.Column("action", sa.String(30), nullable=False),
+        sa.Column("actor", sa.String(100)),
+        sa.Column("notes", sa.Text(), nullable=False),
+        sa.Column("snapshot", sa.JSON(), nullable=False),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+    )
+    op.create_index("ix_audit_events_entity_type", "audit_events", ["entity_type"])
+    op.create_index("ix_audit_events_entity_id", "audit_events", ["entity_id"])
+    for column in version_columns():
+        if column.name != "created_at":
+            op.add_column("knowledge_items", column)
+    for name, kind in [
+        ("source_id", sa.Integer()),
+        ("document_id", sa.Integer()),
+        ("collected_at", sa.DateTime(timezone=True)),
+        ("reviewer", sa.String(100)),
+        ("reviewed_at", sa.DateTime(timezone=True)),
+        ("verification_notes", sa.Text()),
+    ]:
+        op.add_column("knowledge_items", sa.Column(name, kind))
+    for name, parent in [("source_id", "sources"), ("document_id", "documents")]:
+        op.create_foreign_key(f"fk_knowledge_{name}", "knowledge_items", parent, [name], ["id"])
+        op.create_index(f"ix_knowledge_items_{name}", "knowledge_items", [name])
+    op.create_check_constraint(
+        "ck_knowledge_verification_status",
+        "knowledge_items",
+        "verification_status IN ('pending', 'verified', 'rejected')",
+    )
+    op.execute("""INSERT INTO audit_events (entity_type, entity_id, action, notes, snapshot)
+        SELECT 'knowledge_items', id, 'import', '保留基础版本数据及原状态',
+               json_build_object('item', row_to_json(k)) FROM knowledge_items k""")
+    op.execute("""UPDATE knowledge_items SET verification_status = 'pending', version = version + 1
+        WHERE verification_status = 'verified'""")
+    op.execute("""INSERT INTO audit_events (entity_type, entity_id, action, notes, snapshot)
+        SELECT 'knowledge_items', id, 'invalidate', '旧版本没有完整审核记录，需要重新核验',
+               json_build_object('item', row_to_json(k)) FROM knowledge_items k WHERE version = 2""")
+
+
+def downgrade():
+    op.drop_constraint("ck_knowledge_verification_status", "knowledge_items", type_="check")
+    for name in ["source_id", "document_id"]:
+        op.drop_index(f"ix_knowledge_items_{name}", table_name="knowledge_items")
+        op.drop_constraint(f"fk_knowledge_{name}", "knowledge_items", type_="foreignkey")
+    for name in [
+        "source_id",
+        "document_id",
+        "collected_at",
+        "reviewer",
+        "reviewed_at",
+        "verification_notes",
+        "version",
+        "updated_at",
+        "is_deleted",
+    ]:
+        op.drop_column("knowledge_items", name)
+    op.drop_table("audit_events")
+    op.drop_table("documents")
+    op.drop_table("sources")

backend/pyproject.toml

@@ -0,0 +1,10 @@
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["."]
+
+[tool.ruff]
+target-version = "py311"
+line-length = 100
+
+[tool.ruff.lint]
+select = ["E4", "E7", "E9", "F", "I"]

backend/requirements-dev.txt

@@ -0,0 +1,6 @@
+-r requirements.txt
+pytest>=8,<10
+httpx>=0.27,<1
+# FastAPI/Starlette TestClient uses the httpx2 package in this runtime.
+httpx2>=2.13,<3
+ruff>=0.9,<1

backend/requirements.txt

@@ -5,3 +5,4 @@ pydantic-settings>=2.4,<3.0
 SQLAlchemy>=2.0,<3.0
 psycopg[binary]>=3.2,<4.0
 pgvector>=0.3,<1.0
+alembic>=1.13,<2.0

backend/tests/conftest.py

@@ -0,0 +1,87 @@
+"""Real PostgreSQL tests, isolated in a fresh schema for every test."""
+
+import os
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from alembic import command
+from alembic.config import Config
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine, text
+from sqlalchemy.engine import make_url
+from sqlalchemy.orm import sessionmaker
+
+from app.db.session import get_db
+from app.main import create_app
+
+BACKEND = Path(__file__).resolve().parents[1]
+
+
+def migrate(engine, revision="head", *, downgrade=False):
+    config = Config(str(BACKEND / "alembic.ini"))
+    with engine.begin() as connection:
+        config.attributes["connection"] = connection
+        (command.downgrade if downgrade else command.upgrade)(config, revision)
+
+
+@pytest.fixture(scope="session")
+def postgres_url(tmp_path_factory):
+    url = os.environ.get("TEST_DATABASE_URL")
+    server = None
+    if not url:
+        try:
+            import pgserver
+        except ImportError:
+            pytest.fail("请设置 TEST_DATABASE_URL，或安装可选的本机测试依赖 pgserver")
+        server = pgserver.get_server(tmp_path_factory.mktemp("postgres"), cleanup_mode="stop")
+        url = server.get_uri().replace("postgresql://", "postgresql+psycopg://", 1)
+    if not url.startswith("postgresql+psycopg://"):
+        pytest.fail("TEST_DATABASE_URL 必须使用 postgresql+psycopg://")
+    bootstrap = create_engine(url)
+    with bootstrap.begin() as connection:
+        connection.execute(text("CREATE EXTENSION IF NOT EXISTS vector WITH SCHEMA public"))
+    bootstrap.dispose()
+    yield url
+    if server:
+        server.cleanup()
+
+
+@pytest.fixture
+def database(postgres_url):
+    schema = "test_" + uuid4().hex
+    admin = create_engine(postgres_url)
+    with admin.begin() as connection:
+        connection.execute(text(f'CREATE SCHEMA "{schema}"'))
+    # 不包含 public，防止误操作已有业务表。
+    url = make_url(postgres_url).update_query_dict({"options": f"-csearch_path={schema}"})
+    engine = create_engine(url)
+    yield engine
+    engine.dispose()
+    with admin.begin() as connection:
+        connection.execute(text(f'DROP SCHEMA "{schema}" CASCADE'))
+    admin.dispose()
+
+
+@pytest.fixture
+def migrated_database(database):
+    migrate(database)
+    return database
+
+
+@pytest.fixture
+def client(migrated_database):
+    app = create_app(migrated_database)
+    factory = sessionmaker(bind=migrated_database)
+
+    def test_session():
+        with factory() as session:
+            try:
+                yield session
+            except Exception:
+                session.rollback()
+                raise
+
+    app.dependency_overrides[get_db] = test_session
+    with TestClient(app) as client:
+        yield client

backend/tests/test_catalog.py

@@ -0,0 +1,269 @@
+from datetime import datetime
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from sqlalchemy.orm.exc import StaleDataError
+
+from app.models import KnowledgeItem
+
+COLLECTED = "2026-10-15T09:00:00+08:00"
+
+
+def create(client, path, data):
+    response = client.post("/api/" + path, json=data)
+    assert response.status_code == 201, response.text
+    return response.json()
+
+
+def source(client):
+    return create(client, "sources", {"name": "示例来源", "source_url": "https://example.com/"})
+
+
+def document(client, source_id):
+    return create(
+        client,
+        "documents",
+        {
+            "source_id": source_id,
+            "title": "示例文档",
+            "content": "仅用于自动化测试的占位内容",
+            "collected_at": COLLECTED,
+        },
+    )
+
+
+def knowledge(client, **overrides):
+    return create(
+        client,
+        "knowledge",
+        {
+            "title": "示例知识",
+            "content": "测试资料，不代表文旅事实",
+            "source_url": "https://example.com/",
+            "collected_at": COLLECTED,
+            **overrides,
+        },
+    )
+
+
+def review(client, item, status="verified"):
+    return client.post(
+        f"/api/knowledge/{item['id']}/review",
+        json={
+            "expected_version": item["version"],
+            "verification_status": status,
+            "reviewer": "test-reviewer",
+            "verification_notes": "仅验证审核流程，非真实核验结论",
+        },
+    )
+
+
+def test_health_docs_and_stub(client):
+    assert client.get("/api/health").json()["status"] == "ok"
+    assert client.get("/api/ready").json()["database"] == "connected"
+    assert client.get("/docs").status_code == 200
+    assert "/api/knowledge/{item_id}/review" in client.get("/openapi.json").json()["paths"]
+    assert client.post("/api/chat", json={"question": "测试"}).json()["status"] == "stub"
+
+
+def test_legacy_create_and_missing_evidence(client):
+    item = create(client, "knowledge", {"title": "旧版调用", "content": "示例"})
+    assert item["verification_status"] == "pending"
+    assert item["collected_at"] is None
+    assert isinstance(client.get("/api/knowledge").json(), list)
+    assert review(client, item).status_code == 422
+    assert client.get(f"/api/knowledge/{item['id']}/history").json()["total"] == 1
+
+
+def test_optional_empty_url_matches_existing_frontend(client):
+    item = create(client, "knowledge", {"title": "示例", "content": "示例", "source_url": ""})
+    assert item["source_url"] is None
+
+
+def test_reassign_document_resolves_new_source(client):
+    first = source(client)
+    first_doc = document(client, first["id"])
+    second = source(client)
+    second_doc = document(client, second["id"])
+    item = knowledge(client, document_id=first_doc["id"])
+    response = client.patch(
+        f"/api/knowledge/{item['id']}",
+        json={
+            "expected_version": item["version"],
+            "document_id": second_doc["id"],
+        },
+    )
+    assert response.status_code == 200
+    assert response.json()["source_id"] == second["id"]
+
+
+def test_browser_preflight_and_document_source_lists(client):
+    response = client.options(
+        "/api/knowledge",
+        headers={
+            "Origin": "http://localhost:5173",
+            "Access-Control-Request-Method": "POST",
+            "Access-Control-Request-Headers": "content-type",
+        },
+    )
+    assert response.status_code == 200
+    assert response.headers["access-control-allow-origin"] == "http://localhost:5173"
+    src = source(client)
+    doc = document(client, src["id"])
+    assert client.get("/api/sources?q=示例").json()["items"][0]["id"] == src["id"]
+    result = client.get(f"/api/documents?source_id={src['id']}&q=占位").json()
+    assert result["total"] == 1 and result["items"][0]["id"] == doc["id"]
+
+
+def test_review_edit_and_history(client):
+    item = knowledge(client)
+    verified = review(client, item)
+    assert verified.status_code == 200
+    verified = verified.json()
+    assert verified["version"] == 2
+    assert datetime.fromisoformat(verified["reviewed_at"]).tzinfo is not None
+    assert review(client, item).status_code == 409
+    response = client.patch(
+        f"/api/knowledge/{item['id']}",
+        json={
+            "expected_version": verified["version"],
+            "content": "修订后的示例正文",
+        },
+    )
+    assert response.status_code == 200
+    updated = response.json()
+    assert updated["verification_status"] == "pending"
+    assert updated["reviewer"] is None and updated["reviewed_at"] is None
+    events = client.get(f"/api/knowledge/{item['id']}/history").json()
+    assert [event["action"] for event in events["items"]] == ["update", "review", "create"]
+    assert events["items"][1]["snapshot"]["item"]["verification_status"] == "verified"
+
+
+def test_source_document_changes_invalidate_evidence(client):
+    src = source(client)
+    doc = document(client, src["id"])
+    item = knowledge(client, source_id=src["id"], document_id=doc["id"])
+    assert review(client, item).status_code == 200
+    response = client.patch(
+        f"/api/documents/{doc['id']}",
+        json={
+            "expected_version": doc["version"],
+            "content": "修订后的文档",
+        },
+    )
+    assert response.status_code == 200
+    current = client.get(f"/api/knowledge/{item['id']}").json()
+    assert current["verification_status"] == "pending"
+    assert review(client, current).status_code == 200
+    response = client.patch(
+        f"/api/sources/{src['id']}",
+        json={
+            "expected_version": src["version"],
+            "source_url": "https://example.com/new",
+        },
+    )
+    assert response.status_code == 200
+    current = client.get(f"/api/knowledge/{item['id']}").json()
+    assert current["verification_status"] == "pending"
+    assert current["source_url"] == "https://example.com/new"
+    events = client.get(f"/api/knowledge/{item['id']}/history").json()["items"]
+    reviewed = next(e for e in events if e["action"] == "review")
+    assert reviewed["snapshot"]["source"]["source_url"] == "https://example.com/"
+    assert reviewed["snapshot"]["document"]["content"] == "修订后的文档"
+
+
+def test_delete_preserves_history_and_guards_references(client):
+    src = source(client)
+    doc = document(client, src["id"])
+    item = knowledge(client, document_id=doc["id"])
+    assert item["source_id"] == src["id"]
+    assert client.delete(f"/api/sources/{src['id']}?expected_version=1").status_code == 409
+    assert client.delete(f"/api/documents/{doc['id']}?expected_version=1").status_code == 409
+    assert client.delete(f"/api/knowledge/{item['id']}?expected_version=2").status_code == 409
+    assert client.delete(f"/api/knowledge/{item['id']}?expected_version=1").status_code == 204
+    assert client.get(f"/api/knowledge/{item['id']}").status_code == 404
+    assert client.get("/api/knowledge").json() == []
+    assert (
+        client.get(f"/api/knowledge/{item['id']}/history").json()["items"][0]["action"] == "delete"
+    )
+    assert client.delete(f"/api/documents/{doc['id']}?expected_version=1").status_code == 204
+    assert client.delete(f"/api/sources/{src['id']}?expected_version=1").status_code == 204
+    assert client.get("/api/sources").json()["total"] == 0
+
+
+def test_filter_pagination_and_literal_search(client):
+    knowledge(client, title="示例%资料", category="交通出行")
+    other = knowledge(client, title="其他资料", category="民宿服务")
+    assert review(client, other, "rejected").status_code == 200
+    page = client.get("/api/knowledge/page?page_size=1").json()
+    assert page["total"] == 2 and len(page["items"]) == 1
+    assert page["items"][0]["id"] == other["id"]
+    assert client.get("/api/knowledge/page?page=3&page_size=1").json()["items"] == []
+    assert client.get("/api/knowledge/page?q=%25").json()["total"] == 1
+    assert len(client.get("/api/knowledge?category=交通出行").json()) == 1
+    assert client.get("/api/knowledge/page?verification_status=verified").json()["total"] == 0
+    assert client.get("/api/knowledge/page?verification_status=rejected").json()["total"] == 1
+
+
+@pytest.mark.parametrize(
+    "patch",
+    [
+        {"title": "   "},
+        {"title": None},
+        {"content": ""},
+        {"category": ""},
+        {"collected_at": "2026-10-15T09:00:00"},
+        {"source_url": "javascript:alert(1)"},
+        {"verification_status": "verified"},
+        {},
+    ],
+)
+def test_invalid_update_cannot_mutate_or_audit(client, patch):
+    item = knowledge(client)
+    response = client.patch(f"/api/knowledge/{item['id']}", json={"expected_version": 1, **patch})
+    assert response.status_code == 422
+    assert client.get(f"/api/knowledge/{item['id']}").json()["version"] == 1
+    assert client.get(f"/api/knowledge/{item['id']}/history").json()["total"] == 1
+
+
+def test_invalid_relationship_and_query(client):
+    src = source(client)
+    doc = document(client, src["id"])
+    second = source(client)
+    response = client.post(
+        "/api/knowledge",
+        json={
+            "title": "示例",
+            "content": "示例",
+            "source_id": second["id"],
+            "document_id": doc["id"],
+        },
+    )
+    assert response.status_code == 422
+    assert (
+        client.post(
+            "/api/knowledge", json={"title": "示例", "content": "示例", "document_id": 9999}
+        ).status_code
+        == 404
+    )
+    for path in ["knowledge", "sources", "documents"]:
+        assert client.get(f"/api/{path}/9999").status_code == 404
+        assert client.get(f"/api/{path}/9999/history").status_code == 404
+    assert client.get("/api/knowledge/page?page_size=101").status_code == 422
+    assert client.get("/api/knowledge/page?verification_status=unknown").status_code == 422
+    assert client.get("/api/knowledge/page?page=0").status_code == 422
+
+
+def test_optimistic_lock_prevents_concurrent_overwrite(client, migrated_database):
+    item = knowledge(client)
+    with Session(migrated_database) as first, Session(migrated_database) as second:
+        first_item = first.scalar(select(KnowledgeItem).where(KnowledgeItem.id == item["id"]))
+        second_item = second.get(KnowledgeItem, item["id"])
+        first_item.content = "先提交的内容"
+        first.commit()
+        second_item.content = "过期编辑"
+        with pytest.raises(StaleDataError):
+            second.commit()
+        second.rollback()
+    assert client.get(f"/api/knowledge/{item['id']}").json()["content"] == "先提交的内容"

backend/tests/test_migrations.py

@@ -0,0 +1,41 @@
+import pytest
+from alembic.autogenerate import compare_metadata
+from alembic.migration import MigrationContext
+from conftest import migrate
+from fastapi.testclient import TestClient
+from sqlalchemy import text
+
+from app.db.session import Base
+from app.main import create_app
+
+
+def test_startup_requires_migration(database):
+    with pytest.raises(RuntimeError, match="alembic upgrade head"):
+        with TestClient(create_app(database)):
+            pass
+
+
+def test_upgrade_preserves_original_rows_and_repeated_upgrade(database):
+    migrate(database, "0001_baseline")
+    with database.begin() as connection:
+        connection.execute(
+            text("""INSERT INTO knowledge_items
+            (title, category, content, source_url, verification_status)
+            VALUES ('示例旧数据', '未分类', '原始正文', 'https://example.com/', 'verified')""")
+        )
+        # 模拟 create_all 建表后没有 Alembic 版本记录的旧工程。
+        connection.execute(text("DELETE FROM alembic_version"))
+    migrate(database)
+    migrate(database)
+    with database.connect() as connection:
+        row = connection.execute(text("SELECT * FROM knowledge_items")).mappings().one()
+        assert row["title"] == "示例旧数据" and row["content"] == "原始正文"
+        assert row["collected_at"] is None
+        assert row["verification_status"] == "pending"
+        assert connection.scalar(text("SELECT count(*) FROM audit_events")) == 2
+        assert connection.scalar(text("SELECT version_num FROM alembic_version")) == "0002_catalog"
+        assert compare_metadata(MigrationContext.configure(connection), Base.metadata) == []
+    migrate(database, "0001_baseline", downgrade=True)
+    with database.connect() as connection:
+        assert connection.scalar(text("SELECT content FROM knowledge_items")) == "原始正文"
+    migrate(database)

backend/tests/test_runtime.py

@@ -0,0 +1,78 @@
+"""Run the default application lifecycle over HTTP and verify persistence."""
+
+import os
+import socket
+import subprocess
+import sys
+import time
+from contextlib import contextmanager
+from pathlib import Path
+
+import httpx
+
+
+@contextmanager
+def server(engine):
+    with socket.socket() as socket_:
+        socket_.bind(("127.0.0.1", 0))
+        port = socket_.getsockname()[1]
+    environment = {**os.environ, "DATABASE_URL": engine.url.render_as_string(hide_password=False)}
+    backend = Path(__file__).resolve().parents[1]
+    process = subprocess.Popen(
+        [
+            sys.executable,
+            "-m",
+            "uvicorn",
+            "app.main:app",
+            "--host",
+            "127.0.0.1",
+            "--port",
+            str(port),
+            "--log-level",
+            "error",
+        ],
+        cwd=backend,
+        env=environment,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    try:
+        with httpx.Client(base_url=f"http://127.0.0.1:{port}", timeout=2) as client:
+            deadline = time.monotonic() + 15
+            while time.monotonic() < deadline:
+                if process.poll() is not None:
+                    raise AssertionError("默认后端进程启动失败")
+                try:
+                    if client.get("/api/ready").status_code == 200:
+                        break
+                except httpx.TransportError:
+                    pass
+                time.sleep(0.05)
+            else:
+                raise AssertionError("默认后端未在时限内就绪")
+            yield client
+    finally:
+        process.terminate()
+        try:
+            process.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            process.kill()
+            process.wait(timeout=5)
+
+
+def test_default_startup_and_persistence_across_restart(migrated_database):
+    with server(migrated_database) as client:
+        assert client.get("/docs").status_code == 200
+        response = client.post(
+            "/api/knowledge",
+            json={
+                "title": "持久化验收占位",
+                "content": "仅为自动化测试数据",
+                "source_url": "",
+            },
+        )
+        assert response.status_code == 201
+        item = response.json()
+    with server(migrated_database) as client:
+        assert client.get(f"/api/knowledge/{item['id']}").json()["content"] == "仅为自动化测试数据"
+        assert client.get("/api/knowledge").json()[0]["id"] == item["id"]

docs/ARCHITECTURE.md

@@ -14,7 +14,17 @@ FastAPI + Pydantic
           |-- 来源获取 -> 人工核验 -> 文本切分 -> 生成向量 -> 发布版本
 ```
 
-当前工程仅实现健康检查、知识条目新增/列表及问答占位接口；pgvector 扩展已在启动时启用，但未实现向量字段、索引和 RAG。前端来源管理与实验页为静态原型。
+当前后端实现健康/数据库就绪检查、来源与文档管理、知识增删改查、分页筛选、人工审核及历史快照；问答仍为占位接口。数据库由 Alembic 显式迁移并启用 pgvector，尚无向量字段、向量索引和 RAG。前端来源管理与实验页仍为静态原型，审核与分页页面需继续联调。
+
+## 阶段 2 数据边界
+
+- `sources` 登记来源名称、HTTP(S) 网址和备注；`documents` 登记所属来源、正文、分类与采集时间。
+- `knowledge_items` 保留基础版本全部字段，新增可选来源/文档引用、采集时间、人工审核字段、版本和软删除状态。历史条目不强行补造来源或时间。
+- `audit_events` 在同一事务中保存创建、修改、审核、失效、删除和迁移导入事件；知识快照包含当时关联的来源与文档。
+- `version` 用于内容并发控制，PATCH/审核/DELETE 必须匹配读取版本；ORM 版本检查处理并发写入冲突，引用操作对来源/文档加行锁，避免与修改和删除交错。
+- 修改知识或关联依据会撤销当前审核结论。未来正式检索必须同时限定 `verification_status = 'verified'` 和 `is_deleted = false`；当前管理列表不承担检索或索引发布功能。
+- 文档版本通过递增版本号及完整历史快照留存；`chunks`、Embedding 模型版本与索引发布版本在检索阶段单独实现，不能把记录版本当作已实现索引切换。
+- 旧库首次升级自动接管基础表，保留条目与导入快照。没有完整审核记录的旧 `verified` 状态回到待审核。启动不自动迁移，缺失或不匹配的版本会明确报错。
 
 ## 安全边界
 当前无认证与权限控制，仅供本机开发；正式部署前须补充登录鉴权、接口限流、数据库迁移、密钥管理、日志脱敏、CORS 收敛和 HTTPS。

docs/DECISIONS.md

@@ -14,3 +14,11 @@
 
 ## ADR-005 更新策略
 状态：设计确定、尚未实现。新文档版本后台生成向量，完成后原子发布；失败保留旧索引。
+
+## ADR-006 阶段 2 知识管理
+
+状态：分支实现，待合并请求审阅。采用来源、文档、知识条目和追加式事件日志；记录版本用于并发控制，快照用于追溯。所有删除均为软删除，被有效记录引用的父记录禁止删除。依据或正文修改即回待审核；核验通过必须有来源链接、采集时间、核验人和说明。保留旧版知识创建请求及列表数组结构，新页面使用独立分页接口。当前为本机人工审核流程，认证身份、向量切分和索引发布仍属后续阶段。
+
+## ADR-007 显式迁移与旧数据保留
+
+状态：分支实现，待合并请求审阅。以 Alembic 取代应用启动时的 `create_all`。基础迁移接管旧知识表，新增字段和表通过增量迁移完成；不推测旧数据采集时间。旧审核结论缺少完整记录时保留原快照并转待审核。迁移回滚需先备份，新增历史与审核元数据无法通过降级完整保留。

docs/TASKS.md

@@ -22,3 +22,25 @@
 - 已使用临时 Uvicorn 服务（关闭数据库生命周期初始化）验证 `GET /api/health` 返回 `200` 和预期状态数据，验证后服务已关闭。
 - 本机 Docker CLI 已安装，但 Docker Desktop 引擎无法启动，且 WSL 尚未处于可用状态。因此尚未执行 PostgreSQL 容器启动、知识条目新增/读取、默认生命周期启动和浏览器端联调。
 - 后续操作：重启 Windows 后启动 Docker Desktop；确认 `docker version` 能显示 Server 信息，再在项目根目录运行 `docker compose up -d db`，随后继续 T02、T03 和 T01 的剩余验收。
+
+## 2026-10-15 后端阶段 1–2 分支验收
+
+分支：`feat/backend-knowledge-management`。以下为分支验收结果，须经既定审阅流程合入后再关闭任务；不覆盖前述其他环境的历史记录。
+
+| 范围 | 已完成的验证 | 剩余事项 |
+|---|---|---|
+| T02 | 默认生命周期真实启动 Uvicorn；HTTP `/docs`、健康检查、数据库就绪检查通过 | 合并请求审阅、团队环境复核 |
+| T03 | 真实 PostgreSQL 16 测试实例完成迁移、知识录入和读取；后端进程重启后数据仍可读取 | 当前主机 Docker 守护进程未运行，启动需管理员权限；Compose 容器与浏览器端联调待复核 |
+| T05（阶段 2 范围） | 来源、文档、知识关联、版本、人工审核、软删除和历史快照；验证旧基础表迁移保留数据 | `chunks` 和索引版本留到检索阶段；字段与前端、资料负责人共同确认 |
+| T10（部分） | Alembic 迁移、并发版本检查、PostgreSQL 自动化测试、启动说明 | 登录鉴权、限流、生产部署留到后续工程阶段 |
+
+验证环境：Python 3.11；使用可选 `pgserver` 启动独立 PostgreSQL 16.2 测试实例并启用 pgvector。该方式用于验证 SQL 和接口，不等同于 Docker 容器验收。
+
+实际命令与结果（在 `backend/` 执行）：
+
+- `ruff check .`：通过。
+- `ruff format --check .`：通过。
+- `pytest -q`：22 passed；依赖库产生一条 TestClient 弃用提示，不影响本轮用例。
+- 覆盖：新库迁移、已有基础表接管、重复升级、阶段 2 降级后基础正文保留、ORM 与迁移结构一致、审核前置条件、正文/来源/文档变更后重新审核、历史快照、软删除引用保护、筛选与分页、并发写入冲突、跨域预检、真实 HTTP 与服务重启持久化。
+
+所有测试正文、来源和审核人均为占位数据，临时数据库不入库。当前问答仍是占位接口，未开展论文对照实验。Issue、合并请求和最终批准结论须补入 Gitea 后按团队规则归档。