border-red-post-nalati/backend/app/api/routes/knowledge.py

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 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.auth import require_admin
from app.services.catalog import (
    ensure_version,
    get_item,
    history,
    page_result,
    record_event,
    reset_review,
    resolve_knowledge_links,
    update_values,
)
from app.services.indexing import catalog_write_guard

router = APIRouter(
    prefix="/knowledge",
    tags=["knowledge"],
    dependencies=[Depends(require_admin), Depends(catalog_write_guard)],
)


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(
        **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)