DataOps-platform/app/api/knowledge_base/routes.py

from __future__ import annotations

import uuid

from flask import current_app, g, jsonify, request
from sqlalchemy import text

from app import db
from app.api.knowledge_base import bp
from app.core.knowledge.access import build_access_context
from app.core.knowledge.admin import (
    change_set_detail,
    knowledge_status,
    list_change_sets,
    retry_change_set,
    rollback_change_set,
)
from app.core.knowledge.audit import run_canonical_audit
from app.core.knowledge.qa import AnswerSynthesizer, DeepSeekAnswerModel
from app.core.knowledge.retrieval.pipeline import KnowledgeRetrievalPipeline
from app.core.knowledge.retrieval.sql import (
    SqlLexicalRetriever,
    SqlVectorRetriever,
    UnavailableVectorRetriever,
)
from app.models.result import failed, success


def _bounded_int(value, *, field: str, default: int, lower: int, upper: int) -> int:
    try:
        parsed = int(default if value is None else value)
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{field} 必须为整数") from exc
    return min(max(parsed, lower), upper)


def _pipeline() -> KnowledgeRetrievalPipeline:
    configured = current_app.extensions.get("knowledge_retrieval_pipeline")
    if configured is not None:
        return configured
    api_key = current_app.config.get("QWEN_EMBEDDING_API_KEY", "")
    base_url = current_app.config.get("QWEN_EMBEDDING_BASE_URL", "")
    if api_key and base_url:
        from app.services.embedding.qwen import QwenEmbeddingClient

        embedder = QwenEmbeddingClient(
            api_key=api_key,
            base_url=base_url,
            model=current_app.config["QWEN_EMBEDDING_MODEL"],
            dimension=current_app.config["QWEN_EMBEDDING_DIMENSION"],
        )
        vector = SqlVectorRetriever(db.session, embedder)
    else:
        vector = UnavailableVectorRetriever()
    return KnowledgeRetrievalPipeline(
        lexical=SqlLexicalRetriever(db.session),
        vector=vector,
    )


def _serialize_evidence(item) -> dict:
    return {
        "chunk_id": item.chunk_id,
        "content": item.content,
        "score": item.score,
        "retriever": item.retriever,
        "object_uid": item.object_uid,
        "object_type": item.object_type,
        "object_version": item.object_version,
        "business_domain_uid": item.business_domain_uid,
        "point_keys": list(item.point_keys),
        "point_revisions": list(item.point_revisions),
        "index_generation": item.generation,
        "source_updated_at": item.source_updated_at,
        "freshness_status": item.freshness_status,
        "section_path": item.section_path,
    }


def _answer_synthesizer() -> AnswerSynthesizer:
    configured = current_app.extensions.get("knowledge_answer_synthesizer")
    return configured or AnswerSynthesizer(DeepSeekAnswerModel())


def _request_context(payload):
    requested_domains = payload.get("business_domain_uids")
    if requested_domains is not None and not isinstance(requested_domains, list):
        raise ValueError("business_domain_uids 必须为数组")
    correlation_id = request.headers.get("X-Correlation-ID") or str(uuid.uuid4())
    context = build_access_context(
        db.session,
        identity=g.current_user,
        requested_business_domains=requested_domains,
        correlation_id=correlation_id,
    )
    return context, correlation_id


@bp.post("/search")
def search():
    payload = request.get_json(silent=True) or {}
    query = str(payload.get("query") or "").strip()
    if not query:
        return jsonify(failed("query 不能为空", code=400)), 400
    mode = str(payload.get("mode") or "auto")
    if mode not in {"auto", "exact", "semantic", "relationship", "global"}:
        return jsonify(failed("不支持的检索模式", code=400)), 400
    try:
        limit = _bounded_int(
            payload.get("limit"), field="limit", default=20, lower=1, upper=100
        )
        context, correlation_id = _request_context(payload)
    except ValueError as exc:
        return jsonify(failed(str(exc), code=400)), 400
    result = _pipeline().search(
        query,
        context=context,
        mode=mode,
        limit=limit,
    )
    return jsonify(
        success(
            {
                "query": query,
                "mode": result.mode,
                "evidence": [_serialize_evidence(item) for item in result.evidence],
                "degraded_components": list(result.degraded_components),
                "correlation_id": correlation_id,
            }
        )
    )


@bp.post("/ask")
def ask():
    payload = request.get_json(silent=True) or {}
    query = str(payload.get("query") or "").strip()
    if not query:
        return jsonify(failed("query 不能为空", code=400)), 400
    mode = str(payload.get("mode") or "auto")
    if mode not in {"auto", "exact", "semantic", "relationship", "global"}:
        return jsonify(failed("不支持的检索模式", code=400)), 400
    try:
        context, correlation_id = _request_context(payload)
    except ValueError as exc:
        return jsonify(failed(str(exc), code=400)), 400
    retrieval = _pipeline().search(query, context=context, mode=mode, limit=30)
    answer = _answer_synthesizer().answer(query, retrieval.evidence)
    return jsonify(
        success(
            {
                "query_id": correlation_id,
                "mode": retrieval.mode,
                "answer": answer.answer,
                "answer_status": answer.status,
                "degraded_components": list(retrieval.degraded_components),
                "citations": [citation.__dict__ for citation in answer.citations],
                "freshness_status": answer.freshness_status,
            }
        )
    )


def _source_document(source_uid: str, version: int | None = None):
    context, _correlation_id = _request_context({})
    version_clause = (
        "AND d.object_version = :version"
        if version is not None
        else "AND d.status = 'active'"
    )
    row = (
        db.session.execute(
            text(
                f"""
            SELECT d.object_uid::text, d.object_type, d.object_version, d.object_name,
                   d.business_domain_uid::text, d.content, d.source_updated_at,
                   d.active_generation, d.status
            FROM public.governance_documents d
            WHERE d.object_uid = CAST(:uid AS uuid) {version_clause}
            """
            ),
            {"uid": source_uid, "version": version},
        )
        .mappings()
        .one_or_none()
    )
    if row is None or not context.permits_domain(row["business_domain_uid"]):
        return None
    return dict(row)


@bp.get("/sources/<source_uid>")
def source(source_uid: str):
    row = _source_document(source_uid)
    if row is None:
        return jsonify(failed("来源不存在或无权访问", code=404)), 404
    return jsonify(success(row))


@bp.get("/sources/<source_uid>/versions/<int:version>")
def source_version(source_uid: str, version: int):
    row = _source_document(source_uid, version)
    if row is None:
        return jsonify(failed("来源版本不存在或无权访问", code=404)), 404
    return jsonify(success(row))


@bp.get("/capabilities")
def capabilities():
    return jsonify(
        success(
            {
                "standard_retrieval": True,
                "lightrag_enabled": bool(
                    current_app.config.get("KNOWLEDGE_LIGHTRAG_ENABLED", False)
                ),
                "lightrag_shadow_only": bool(
                    current_app.config.get("KNOWLEDGE_LIGHTRAG_SHADOW_ONLY", True)
                ),
                "answer_generation": bool(current_app.config.get("DEEPSEEK_API_KEY")),
            }
        )
    )


@bp.get("/admin/sync")
def admin_sync():
    return jsonify(success(knowledge_status(db.session)))


@bp.get("/admin/change-sets")
def admin_change_sets():
    try:
        limit = _bounded_int(
            request.args.get("limit"), field="limit", default=50, lower=1, upper=200
        )
    except ValueError as exc:
        return jsonify(failed(str(exc), code=400)), 400
    return jsonify(success(list_change_sets(db.session, limit=limit)))


@bp.get("/admin/change-sets/<change_set_id>")
def admin_change_set(change_set_id: str):
    detail = change_set_detail(db.session, change_set_id)
    if detail is None:
        return jsonify(failed("change set 不存在", code=404)), 404
    return jsonify(success(detail))


@bp.post("/admin/change-sets/<change_set_id>/retry")
def admin_retry_change_set(change_set_id: str):
    if not retry_change_set(db.session, change_set_id):
        db.session.rollback()
        return jsonify(failed("当前状态不允许重试", code=409)), 409
    db.session.commit()
    return jsonify(success({"change_set_id": change_set_id, "status": "pending"}))


@bp.post("/admin/change-sets/<change_set_id>/rollback")
def admin_rollback_change_set(change_set_id: str):
    if not rollback_change_set(db.session, change_set_id):
        db.session.rollback()
        return jsonify(failed("没有可安全回退的上一版本", code=409)), 409
    db.session.commit()
    return jsonify(success({"change_set_id": change_set_id, "status": "rolled_back"}))


@bp.post("/admin/audit")
def admin_audit():
    payload = request.get_json(silent=True) or {}
    repair = payload.get("repair") is True
    findings = run_canonical_audit(db.session)
    repaired = 0
    if repair:
        repaired = db.session.execute(
            text(
                "DELETE FROM public.knowledge_cache_dependencies "
                "WHERE expires_at <= CURRENT_TIMESTAMP"
            )
        ).rowcount
        db.session.commit()
    return jsonify(
        success(
            {
                "mode": "repair" if repair else "report",
                "findings": [finding.__dict__ for finding in findings],
                "repaired_cache_dependencies": repaired,
            }
        )
    )


@bp.post("/admin/retry-projection")
def admin_retry_projection():
    projection_id = str(
        (request.get_json(silent=True) or {}).get("projection_id") or ""
    )
    if not projection_id:
        return jsonify(failed("projection_id 不能为空", code=400)), 400
    updated = db.session.execute(
        text(
            """
            UPDATE public.knowledge_index_projections
            SET status = 'pending', last_error = NULL, external_track_id = NULL,
                available_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
            WHERE id = CAST(:id AS uuid) AND status IN ('failed','unverified')
            """
        ),
        {"id": projection_id},
    ).rowcount
    if updated != 1:
        db.session.rollback()
        return jsonify(failed("当前投影状态不允许重试", code=409)), 409
    db.session.commit()
    return jsonify(success({"projection_id": projection_id, "status": "pending"}))


@bp.get("/admin/evaluations")
def admin_evaluations():
    rows = db.session.execute(
        text(
            """
            SELECT run.id::text, evaluation_set.name, run.status, run.configuration,
                   run.started_at, run.finished_at,
                   COUNT(result.case_id) AS case_count,
                   COUNT(result.case_id) FILTER (WHERE result.passed) AS passed_count
            FROM public.knowledge_evaluation_runs run
            JOIN public.knowledge_evaluation_sets evaluation_set
              ON evaluation_set.id = run.evaluation_set_id
            LEFT JOIN public.knowledge_evaluation_results result ON result.run_id = run.id
            GROUP BY run.id, evaluation_set.name
            ORDER BY run.started_at DESC
            LIMIT 50
            """
        )
    ).mappings()
    return jsonify(success([dict(row) for row in rows]))