diff --git a/.gitignore b/.gitignore index 2b2678b..0a5b602 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,6 @@ dist/ .vscode/ .idea/ *.log +.ruff_cache/ +.coverage +htmlcov/ diff --git a/README.md b/README.md index 69f3588..c2e1227 100644 --- a/README.md +++ b/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`。 ## 团队协作文档 diff --git a/backend/alembic.ini b/backend/alembic.ini new file mode 100644 index 0000000..8d81203 --- /dev/null +++ b/backend/alembic.ini @@ -0,0 +1,4 @@ +[alembic] +script_location = %(here)s/migrations +prepend_sys_path = %(here)s +path_separator = os diff --git a/backend/app/api/routes/chat.py b/backend/app/api/routes/chat.py index c601685..1d495c8 100644 --- a/backend/app/api/routes/chat.py +++ b/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=[], + ) diff --git a/backend/app/api/routes/documents.py b/backend/app/api/routes/documents.py new file mode 100644 index 0000000..a525cfd --- /dev/null +++ b/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) diff --git a/backend/app/api/routes/health.py b/backend/app/api/routes/health.py index cbfebf6..9099983 100644 --- a/backend/app/api/routes/health.py +++ b/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"} diff --git a/backend/app/api/routes/knowledge.py b/backend/app/api/routes/knowledge.py index e97bb4a..a967c7e 100644 --- a/backend/app/api/routes/knowledge.py +++ b/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) diff --git a/backend/app/api/routes/sources.py b/backend/app/api/routes/sources.py new file mode 100644 index 0000000..ba95e1b --- /dev/null +++ b/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) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index 8aef48f..068adb0 100644 --- a/backend/app/core/config.py +++ b/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() diff --git a/backend/app/db/session.py b/backend/app/db/session.py index 12aea8d..8140f74 100644 --- a/backend/app/db/session.py +++ b/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() diff --git a/backend/app/main.py b/backend/app/main.py index 0e8f89e..7f1739c 100644 --- a/backend/app/main.py +++ b/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() diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index 03a2db7..771117b 100644 --- a/backend/app/models/__init__.py +++ b/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"] diff --git a/backend/app/models/catalog.py b/backend/app/models/catalog.py new file mode 100644 index 0000000..569cdc1 --- /dev/null +++ b/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()) diff --git a/backend/app/models/knowledge.py b/backend/app/models/knowledge.py index d825dec..48ab814 100644 --- a/backend/app/models/knowledge.py +++ b/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) diff --git a/backend/app/schemas/catalog.py b/backend/app/schemas/catalog.py new file mode 100644 index 0000000..36f4a73 --- /dev/null +++ b/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 diff --git a/backend/app/schemas/chat.py b/backend/app/schemas/chat.py index 40ccb80..3880934 100644 --- a/backend/app/schemas/chat.py +++ b/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 diff --git a/backend/app/schemas/knowledge.py b/backend/app/schemas/knowledge.py index 69a3260..b6a6217 100644 --- a/backend/app/schemas/knowledge.py +++ b/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 diff --git a/backend/app/services/__init__.py b/backend/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/services/catalog.py b/backend/app/services/catalog.py new file mode 100644 index 0000000..293e26d --- /dev/null +++ b/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) diff --git a/backend/migrations/env.py b/backend/migrations/env.py new file mode 100644 index 0000000..409fe1e --- /dev/null +++ b/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) diff --git a/backend/migrations/script.py.mako b/backend/migrations/script.py.mako new file mode 100644 index 0000000..5449f62 --- /dev/null +++ b/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"} diff --git a/backend/migrations/versions/0001_baseline.py b/backend/migrations/versions/0001_baseline.py new file mode 100644 index 0000000..98f89bd --- /dev/null +++ b/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") + # 扩展可能由其他项目共享，不在回滚时删除。 diff --git a/backend/migrations/versions/0002_catalog.py b/backend/migrations/versions/0002_catalog.py new file mode 100644 index 0000000..97ca417 --- /dev/null +++ b/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") diff --git a/backend/pyproject.toml b/backend/pyproject.toml new file mode 100644 index 0000000..f8307c5 --- /dev/null +++ b/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"] diff --git a/backend/requirements-dev.txt b/backend/requirements-dev.txt new file mode 100644 index 0000000..296cea3 --- /dev/null +++ b/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 diff --git a/backend/requirements.txt b/backend/requirements.txt index eb450d8..1be4a2c 100644 --- a/backend/requirements.txt +++ b/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 diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 0000000..3b7fe5a --- /dev/null +++ b/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 diff --git a/backend/tests/test_catalog.py b/backend/tests/test_catalog.py new file mode 100644 index 0000000..dc400dc --- /dev/null +++ b/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"] == "先提交的内容" diff --git a/backend/tests/test_migrations.py b/backend/tests/test_migrations.py new file mode 100644 index 0000000..0747b06 --- /dev/null +++ b/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) diff --git a/backend/tests/test_runtime.py b/backend/tests/test_runtime.py new file mode 100644 index 0000000..eb3536f --- /dev/null +++ b/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"] diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 5fb5d97..d698576 100644 --- a/docs/ARCHITECTURE.md +++ b/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。 diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 92395b5..5e955fd 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -14,3 +14,11 @@ ## ADR-005 更新策略 状态：设计确定、尚未实现。新文档版本后台生成向量，完成后原子发布；失败保留旧索引。 + +## ADR-006 阶段 2 知识管理 + +状态：分支实现，待合并请求审阅。采用来源、文档、知识条目和追加式事件日志；记录版本用于并发控制，快照用于追溯。所有删除均为软删除，被有效记录引用的父记录禁止删除。依据或正文修改即回待审核；核验通过必须有来源链接、采集时间、核验人和说明。保留旧版知识创建请求及列表数组结构，新页面使用独立分页接口。当前为本机人工审核流程，认证身份、向量切分和索引发布仍属后续阶段。 + +## ADR-007 显式迁移与旧数据保留 + +状态：分支实现，待合并请求审阅。以 Alembic 取代应用启动时的 `create_all`。基础迁移接管旧知识表，新增字段和表通过增量迁移完成；不推测旧数据采集时间。旧审核结论缺少完整记录时保留原快照并转待审核。迁移回滚需先备份，新增历史与审核元数据无法通过降级完整保留。 diff --git a/docs/TASKS.md b/docs/TASKS.md index fc24215..263f347 100644 --- a/docs/TASKS.md +++ b/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 后按团队规则归档。