"""知识库文章 API — P1-3 嵌入功能模块用 提供按关联页面查询知识文章的功能。 """ from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from sqlalchemy import or_ from typing import Optional from app.database import get_db from app.auth_middleware import require_role from app.deps import get_entity_id from app.models.knowledge_article import KnowledgeArticle from app.utils.route_key import candidate_route_keys, route_template_child_pattern router = APIRouter(prefix="/api/cma/knowledge-articles", tags=["知识库嵌入"], dependencies=[Depends(require_role("ceo", "finance", "business", "it"))], ) def article_to_dict(a: KnowledgeArticle) -> dict: return { "id": a.id, "entity_id": a.entity_id, "title": a.title, "summary": a.summary, "content": a.content, "category": a.category, "icon": a.icon, "related_page": a.related_page, "sort_order": a.sort_order, } @router.get("") def list_articles( related_page: Optional[str] = Query(None, description="按关联页面路由筛选"), category: Optional[str] = Query(None), db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """查询知识文章(本企业专属 ∪ 全局共享),可按关联页面或分类筛选 隔离语义(2026-09-03 多租户分区): - entity_id IS NULL → 全局共享(CMA 系统教学知识,所有企业可见) - entity_id == 当前账套 → 企业专属(只本企业可见) - 其他企业的专属文章不可见 """ q = db.query(KnowledgeArticle).filter( or_( KnowledgeArticle.entity_id == entity_id, KnowledgeArticle.entity_id.is_(None), ) ) if related_page: # KBFF-v11c(2026-10-12):匹配口径由「子串包含 LIKE '%%'」收敛为 # 「归一化候选键精确等值 ∪ 模板子路由」——消除短键/前缀超集命中 # (prop=/maps 曾连带命中 /maps/canvas + /maps-review;prop=/kpi 命中 /kpis)。 # 口径实现唯一处 = app/utils/route_key.py(调用方不得二次截断/拼接)。 keys = candidate_route_keys(related_page) if not keys: # 不可归一(纯 query/空白等)⇒ 明确无命中,不退化为“不过滤=全量” return {"data": []} conds = [KnowledgeArticle.related_page.in_(keys)] for k in keys: # 模板子路由:存储值形如 <键>/:<占位符>(占位符命名不限),仍属该页 conds.append( KnowledgeArticle.related_page.like(route_template_child_pattern(k), escape="\\") ) q = q.filter(or_(*conds)) if category: q = q.filter(KnowledgeArticle.category == category) articles = q.order_by(KnowledgeArticle.sort_order.asc(), KnowledgeArticle.id.asc()).all() return {"data": [article_to_dict(a) for a in articles]} @router.get("/{article_id}") def get_article( article_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id), ): """获取文章详情(企业专属文章仅本企业可见,防跨企业探测)""" a = ( db.query(KnowledgeArticle) .filter(KnowledgeArticle.id == article_id) .first() ) if not a: raise HTTPException(404, "文章不存在") if a.entity_id is not None and a.entity_id != entity_id: raise HTTPException(404, "文章不存在") return article_to_dict(a)