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

from __future__ import annotations

import uuid
from time import perf_counter

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.device_scope import (
    DeviceSourceScopeError,
    DeviceSourceScopeService,
    SqlAlchemyDeviceSourceScopeRepository,
    source_scope_access,
)
from app.core.knowledge.qa import AnswerSynthesizer, DeepSeekAnswerModel
from app.core.knowledge.query_audit import (
    SqlKnowledgeQueryAuditRepository,
    build_query_audit,
)
from app.core.knowledge.retrieval.device import (
    SqlDeviceKnowledgeRepository,
    SqlDeviceKnowledgeRetriever,
)
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,
        device=SqlDeviceKnowledgeRetriever(_device_repository()),
    )


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 _device_repository():
    configured = current_app.extensions.get("device_knowledge_repository")
    return configured or SqlDeviceKnowledgeRepository(db.session)


def _scope_service():
    configured = current_app.extensions.get("device_source_scope_service")
    return configured or DeviceSourceScopeService(
        SqlAlchemyDeviceSourceScopeRepository(db.session),
        commit=db.session.commit,
        rollback=db.session.rollback,
    )


def _audit_repository():
    configured = current_app.extensions.get(
        "knowledge_query_audit_repository"
    )
    return configured or SqlKnowledgeQueryAuditRepository(db.session)


def _persist_query_audit(
    *,
    query,
    context,
    result,
    cited_points=(),
    started_at,
) -> bool:
    record = build_query_audit(
        query=query,
        context=context,
        mode=result.mode,
        evidence=result.evidence,
        cited_points=tuple(cited_points),
        degraded_components=result.degraded_components,
        latency_ms=int((perf_counter() - started_at) * 1000),
    )
    try:
        _audit_repository().record(record)
        return True
    except Exception:
        current_app.logger.exception("mandatory knowledge query audit failed")
        return False


def _query(payload) -> str:
    query = str(payload.get("query") or "").strip()
    if not query:
        raise ValueError("query 不能为空")
    if len(query) > 300:
        raise ValueError("query 不能超过 300 个字符")
    return query


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 {}
    try:
        query = _query(payload)
    except ValueError as exc:
        return jsonify(failed(str(exc), 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
    started_at = perf_counter()
    result = _pipeline().search(
        query,
        context=context,
        mode=mode,
        limit=limit,
    )
    if not _persist_query_audit(
        query=query,
        context=context,
        result=result,
        started_at=started_at,
    ):
        return jsonify(failed("知识查询审计写入失败", code=503)), 503
    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 {}
    try:
        query = _query(payload)
    except ValueError as exc:
        return jsonify(failed(str(exc), 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
    started_at = perf_counter()
    retrieval = _pipeline().search(query, context=context, mode=mode, limit=30)
    answer = _answer_synthesizer().answer(query, retrieval.evidence)
    cited_points = tuple(
        (citation.point_key, citation.point_revision)
        for citation in answer.citations
    )
    if not _persist_query_audit(
        query=query,
        context=context,
        result=retrieval,
        cited_points=cited_points,
        started_at=started_at,
    ):
        return jsonify(failed("知识查询审计写入失败", code=503)), 503
    evidence_by_chunk = {
        item.chunk_id: _serialize_evidence(item)
        for item in retrieval.evidence
    }
    citations = []
    for citation in answer.citations:
        serialized = dict(citation.__dict__)
        selected = evidence_by_chunk.get(citation.chunk_id, {})
        serialized["content"] = selected.get("content")
        serialized["business_domain_uid"] = selected.get(
            "business_domain_uid"
        )
        citations.append(serialized)
    return jsonify(
        success(
            {
                "query_id": correlation_id,
                "mode": retrieval.mode,
                "answer": answer.answer,
                "answer_status": answer.status,
                "degraded_components": list(retrieval.degraded_components),
                "citations": citations,
                "evidence": list(evidence_by_chunk.values()),
                "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:
        context, _correlation_id = _request_context({})
        device = _device_repository().get_detail(
            source_uid,
            global_access=context.global_access,
            business_domain_uids=tuple(
                sorted(context.business_domain_uids)
            ),
        )
        if device is not None:
            row = {
                "object_uid": device.asset_uid,
                "object_type": "DeviceAsset",
                "object_version": device.current_version,
                "object_name": device.name,
                "asset_type": device.asset_type,
                "business_domain_uid": device.business_domain_uid,
                "location": device.location,
                "organization": device.organization,
                "responsible_person": device.responsible_person,
                "source_codes": list(device.source_codes),
                "related_events": [
                    {
                        "event_type": event_type,
                        "title": title,
                        "source_code": source_code,
                    }
                    for event_type, title, source_code in device.related_events
                ],
                "source_updated_at": (
                    device.updated_at.isoformat()
                    if device.updated_at is not None
                    else None
                ),
            }
    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/device-sources")
def admin_device_sources():
    return jsonify(success(list(_scope_service().list())))


@bp.put("/admin/device-sources/<source_uid>/scope")
def admin_device_source_scope(source_uid: str):
    payload = request.get_json(silent=True) or {}
    try:
        source_record = _scope_service().update(
            source_uid,
            payload,
            actor_is_admin="admin" in g.current_user.get("roles", ()),
        )
    except DeviceSourceScopeError as exc:
        return jsonify(failed(str(exc), code=exc.http_status)), exc.http_status
    access = source_scope_access(source_record.permission_scope)
    return jsonify(
        success(
            {
                "uid": source_record.uid,
                "name": source_record.name,
                "source_type": source_record.source_type,
                "status": source_record.status,
                **access,
                "updated_at": (
                    source_record.updated_at.isoformat()
                    if source_record.updated_at is not None
                    else None
                ),
            }
        )
    )


@bp.get("/admin/query-audits")
def admin_query_audits():
    try:
        limit = _bounded_int(
            request.args.get("limit"),
            field="limit",
            default=100,
            lower=1,
            upper=200,
        )
    except ValueError as exc:
        return jsonify(failed(str(exc), code=400)), 400
    return jsonify(success(list(_audit_repository().list(limit=limit))))


@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]))