diff --git a/backend/app/api/knowledge_articles.py b/backend/app/api/knowledge_articles.py index fa783f5d..10a5c4d8 100644 --- a/backend/app/api/knowledge_articles.py +++ b/backend/app/api/knowledge_articles.py @@ -10,6 +10,7 @@ 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"))], @@ -51,7 +52,21 @@ def list_articles( ) ) if related_page: - q = q.filter(KnowledgeArticle.related_page.contains(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() diff --git a/backend/app/utils/route_key.py b/backend/app/utils/route_key.py new file mode 100644 index 00000000..82f21108 --- /dev/null +++ b/backend/app/utils/route_key.py @@ -0,0 +1,128 @@ +# -*- coding: utf-8 -*- +"""路由键归一与候选键派生(KBFF-v11c,2026-10-12) + +背景(隐患): + 知识文章列表接口原用 ``KnowledgeArticle.related_page.contains(related_page)`` + ⇒ SQL ``LIKE '%%'`` **子串包含**。prop 取短键/前缀即超集命中: + prop=``/maps`` 会同时命中 ``/maps/canvas`` 与 ``/maps-review``;prop=``/kpi`` 命中 ``/kpis``。 + +口径(本模块 = related_page 匹配口径的**唯一实现处**): + 匹配 = 「归一化候选键集合的**精确等值**」∪「模板子路由(``<键>/:<占位符>``)」。 + 调用方只传原始路由串,**不得**自行截断/拼接/大小写变换 —— 否则会出现第二套口径。 + +归一化规则(``normalize_route``): + ① 去首尾空白;丢弃 ``?query`` 与 ``#fragment``; + ② 反斜杠 → ``/``;压缩连续 ``/``;补前导 ``/``; + ③ 去末尾 ``/``(根 ``/`` 保留); + ④ 统一小写(CMA 前端路由均小写定义,容忍手工录入大小写差异)。 + +候选键(``candidate_route_keys``)—— 带参路由兼容: + 入参可能是**具体实例路径**(如 ``/maps/canvas/42``),而表内存的是**页面身份** + (``/maps/canvas``)或**模板形式**(``/maps/canvas/:id``,见模型注释)。 + 故候选键 = 归一化完整路径 ∪ 父路径(末段为变量时)∪ 各自模板形式, + 匹配走 ``related_page IN (候选键)`` —— 全程等值,无任意子串包含。 + + 变量段判据:纯数字 / 长十六进制(≥8) / UUID。 + 占位符命名不统一时(他方可能写 ``:articleId``),由调用方叠加 + ``route_template_child_pattern(key)`` 给出的前缀模式兜住,见该函数。 +""" +from __future__ import annotations + +import re +from typing import List, Optional, Tuple + +#: 变量段占位符(与 app/models/knowledge_article.py 注释 ``如 /maps/canvas/:id`` 同约定) +PARAM_PLACEHOLDER = ":id" + +_UUID_RE = re.compile( + r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" +) +_HEX_RE = re.compile(r"^[0-9a-f]{8,}$") +_NUM_RE = re.compile(r"^\d+$") +_MULTI_SLASH_RE = re.compile(r"/{2,}") + + +def normalize_route(path: Optional[str]) -> str: + """把任意路由串归一为比较键。不可归一时返回空串(调用方须按“无命中”处理)。""" + if path is None: + return "" + s = str(path).strip() + if not s: + return "" + # 丢弃 query / fragment(保留原有 path 语义) + s = s.split("?", 1)[0].split("#", 1)[0] + s = s.replace("\\", "/").strip() + if not s: + return "" + if not s.startswith("/"): + s = "/" + s + s = _MULTI_SLASH_RE.sub("/", s) + s = s.lower() + if len(s) > 1: + s = s.rstrip("/") + return s or "/" + + +def is_variable_segment(seg: str) -> bool: + """段是否形如变量实例(数字 / 长十六进制 / UUID)。""" + if not seg: + return False + if _UUID_RE.match(seg): + return True + if _NUM_RE.match(seg): + return True + return bool(_HEX_RE.match(seg)) + + +def is_dynamic_segment(seg: str) -> bool: + """段是否「动态」:变量实例(``42``)或占位符本身(``:id`` / ``:articleId``)。 + + 用于从动态末段回推页面身份(父路径),使「传具体实例」与「传模板串」两种调用姿势 + 都能命中表内的页面身份键。 + """ + return seg.startswith(":") or is_variable_segment(seg) + + +def _split(path: str) -> List[str]: + return [p for p in path.split("/") if p] + + +def template_route(path: str) -> str: + """把变量段替换为占位符,得到模板形式(``/maps/canvas/42`` → ``/maps/canvas/:id``)。""" + segs = _split(path) + if not segs: + return path + return "/" + "/".join(PARAM_PLACEHOLDER if is_variable_segment(s) else s for s in segs) + + +def candidate_route_keys(path: Optional[str]) -> Tuple[str, ...]: + """归一化候选键集合(已排序去重)。空元组 = 不可归一 ⇒ 无命中。""" + key = normalize_route(path) + if not key: + return () + keys = {key} + segs = _split(key) + if segs: + keys.add(template_route(key)) + # 末段是动态段(变量实例 / 占位符)⇒ 补父路径(及父路径的模板形式) + if is_dynamic_segment(segs[-1]) and len(segs) > 1: + parent = "/" + "/".join(segs[:-1]) + keys.add(parent) + keys.add(template_route(parent)) + return tuple(sorted(keys)) + + +def escape_like(value: str) -> str: + """转义 LIKE 元字符(配合 ``escape='\\\\'`` 使用)。""" + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def route_template_child_pattern(key: str) -> str: + """模板子路由前缀模式:``<键>/:<任意占位符>``(含更深层级)。 + + ``/maps/canvas`` → ``/maps/canvas/:%`` —— 只命中“把该页声明为模板”的存储值 + (``/maps/canvas/:id`` / ``/maps/canvas/:articleId``), + 不会命中兄弟页(``/maps-review`` 之类),也不命中更深的静态页 + (``/maps/canvas/detail`` 不含前导 ``:``)。 + """ + return escape_like(key) + "/:%" diff --git a/backend/tests/test_kbff_v11c_related_page_exact_match_20261012.py b/backend/tests/test_kbff_v11c_related_page_exact_match_20261012.py new file mode 100644 index 00000000..8d0cc3ff --- /dev/null +++ b/backend/tests/test_kbff_v11c_related_page_exact_match_20261012.py @@ -0,0 +1,255 @@ +# -*- coding: utf-8 -*- +"""KBFF-v11c · `/knowledge-articles?related_page=` 过滤精确化回归(2026-10-12) + +派单:`KBFF-v11c-relatedpage-exact-match-filter`(A 侧 wecom-yanxue) +口径实现:`app/utils/route_key.py`;改造点:`app/api/knowledge_articles.py` +纪律:本文件只新增测试,不改 app/**、conftest.py、任何既有 test_*.py。 + +覆盖: + §1 单位:normalize_route / candidate_route_keys / route_template_child_pattern(表驱动) + §2 负对照:证明改造前 contains 语义确会超集命中(同库同 seed 直构旧表达式) + §3 生产 seed 7 值逐值回归(值取自 backend/scripts/seed_knowledge_articles.py) + —— 每个值各自只中自己那 1 行;短键 `/maps` 0 行、`/kpi` 0 行 + §4 带参路由 / 尾斜杠 / query-fragment / 大小写 / 不可归一 兼容面 + §5 模板子路由(存储值写占位符,任意占位符命名均可命中) + §6 既有语义保持:账套分区 + category 组合过滤不受影响 +""" +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models import Entity +from app.models.knowledge_article import KnowledgeArticle +from app.utils.route_key import ( + candidate_route_keys, + normalize_route, + route_template_child_pattern, + template_route, +) +from tests.conftest import auth_header, create_test_user + +KA = "/api/cma/knowledge-articles" + +#: 生产 seed 的 7 个 related_page 值(backend/scripts/seed_knowledge_articles.py:41,61,83,113,139,166,189) +SEED = [ + (1, "/maps/canvas"), + (2, "/kpis"), + (3, "/budget"), + (4, "/deviations"), + (5, "/predict"), + (6, "/maps-review"), + (7, "/alerts"), +] + + +# ── 夹具 ──────────────────────────────────────────────────────────── + +def _ensure_entity(db: Session, eid: int, name: str = None, short: str = None) -> Entity: + e = db.query(Entity).filter(Entity.id == eid).first() + if e: + return e + e = Entity(id=eid, name=name or f"实体{eid}", short_name=short or f"E{eid}", status="active") + db.add(e) + db.commit() + db.refresh(e) + return e + + +def _login(client: TestClient, entity_id: int = 1) -> str: + r = client.post("/api/cma/auth/login", json={ + "username": "testadmin", "password": "test-fixture-pass", "entity_id": entity_id, + }) + assert r.status_code == 200, r.text + j = r.json() + return j.get("token") or j.get("access_token") + + +def _seed_rows(db: Session, rows) -> None: + """rows: [(id, related_page)],全局共享(entity_id=None)便于账套无关复算。""" + for i, rp in rows: + db.add(KnowledgeArticle(title=f"seed-{i}", content="c", category="term", + entity_id=None, related_page=rp, sort_order=i)) + db.commit() + + +# ── §1 单位 ───────────────────────────────────────────────────────── + +@pytest.mark.parametrize("raw,expected", [ + ("/maps/canvas", "/maps/canvas"), + ("/maps/canvas/", "/maps/canvas"), + ("/maps//canvas/", "/maps/canvas"), + ("maps/canvas", "/maps/canvas"), + ("/MAPS/Canvas", "/maps/canvas"), + ("/maps/canvas?id=42", "/maps/canvas"), + ("/maps/canvas#top", "/maps/canvas"), + (" /budget ", "/budget"), + ("/", "/"), + ("", ""), + (" ", ""), + (None, ""), + ("?only=query", ""), +]) +def test_u1_normalize_route(raw, expected): + assert normalize_route(raw) == expected + + +@pytest.mark.parametrize("raw,expect_members,expect_absent", [ + ("/maps/canvas", {"/maps/canvas"}, {"/maps", "/maps/canvas/:id"}), + ("/maps/canvas/42", {"/maps/canvas/42", "/maps/canvas", "/maps/canvas/:id"}, + set()), + ("/maps/canvas/:id", {"/maps/canvas/:id", "/maps/canvas"}, {"/maps"}), + ("/maps", {"/maps"}, {"/maps/canvas"}), + ("/kpi", {"/kpi"}, {"/kpis"}), + ("", set(), set()), + (" ", set(), set()), +]) +def test_u2_candidate_route_keys(raw, expect_members, expect_absent): + keys = candidate_route_keys(raw) + assert set(keys) >= expect_members, f"{raw} -> {keys}" + assert not (set(keys) & expect_absent), f"{raw} 不该派生 {expect_absent}:{keys}" + assert list(keys) == sorted(keys), "候选键须排序去重,便于同库复算与取证" + + +@pytest.mark.parametrize("path,expected", [ + ("/maps/canvas/42", "/maps/canvas/:id"), + ("/maps/canvas", "/maps/canvas"), + ("/kpis", "/kpis"), +]) +def test_u3_template_route(path, expected): + assert template_route(path) == expected + + +def test_u4_template_child_pattern_shape(): + assert route_template_child_pattern("/maps/canvas") == "/maps/canvas/:%" + # LIKE 元字符须转义(否则用户串可注入通配) + assert route_template_child_pattern("/a_b%c") == "/a\\_b\\%c/:%" + + +# ── §2 负对照(改造前语义) ────────────────────────────────────────── + +def test_n1_contains_semantics_superset_control(client: TestClient, db: Session): + """负对照:旧 `.contains()` 语义下 `/maps` 超集命中 2 行、`/kpi` 命中 `/kpis`。 + 证明本改造确有行为变化(不是“改了个等价写法”)。""" + create_test_user(db) + _seed_rows(db, SEED) + tok = _login(client) + assert client.get(f"{KA}?related_page=/maps/canvas", headers=auth_header(tok)).status_code == 200 + + def old_query(v): + return (db.query(KnowledgeArticle) + .filter(KnowledgeArticle.related_page.contains(v)) + .order_by(KnowledgeArticle.id.asc()).all()) + + assert [r.related_page for r in old_query("/maps")] == ["/maps/canvas", "/maps-review"] + assert [r.related_page for r in old_query("/kpi")] == ["/kpis"] + # 生成 SQL 亦确认为子串包含 + sql = str(db.query(KnowledgeArticle) + .filter(KnowledgeArticle.related_page.contains("/maps")) + .statement.compile(compile_kwargs={"literal_binds": True})) + assert "LIKE" in sql and "%" in sql + + +# ── §3 生产 seed 7 值逐值回归 ──────────────────────────────────────── + +def test_f1_seven_seed_values_exact(client: TestClient, db: Session): + create_test_user(db) + _seed_rows(db, SEED) + tok = _login(client) + + for sid, rp in SEED: + r = client.get(f"{KA}?related_page={rp}", headers=auth_header(tok)) + assert r.status_code == 200, r.text + got = [(i["id"], i["related_page"]) for i in r.json()["data"]] + assert got == [(sid, rp)], f"{rp} 期望精确 1 行({sid}),实得 {got}" + + +def test_f2_short_key_and_prefix_zero(client: TestClient, db: Session): + """短键/前缀必须 0 行:/maps(否则连带 /maps/canvas + /maps-review)、/kpi(否则 /kpis)。""" + create_test_user(db) + _seed_rows(db, SEED) + tok = _login(client) + for prop in ["/maps", "/kpi", "/map", "/budg", "/alerts-extra", "/maps-revie"]: + r = client.get(f"{KA}?related_page={prop}", headers=auth_header(tok)) + assert r.status_code == 200, r.text + assert r.json()["data"] == [], f"{prop} 应 0 行,实得 {r.json()['data']}" + + +# ── §4 兼容面 ─────────────────────────────────────────────────────── + +@pytest.mark.parametrize("prop,expect_id", [ + ("/maps/canvas/42", 1), # 带参实例 → 父页 + 模板派生 + ("/maps/canvas/:id", 1), # 模板形式直传 + ("/maps/canvas/", 1), # 尾斜杠 + ("/MAPS/CANVAS", 1), # 大小写 + ("/maps/canvas?id=42", 1), # query + ("/budget%23top", 3), # fragment(编码传参,验证落库前的截断) + ("%20%20/kpis%20%20", 2), # 首尾空白 + ("/alerts", 7), +]) +def test_f3_compat_forms(client: TestClient, db: Session, prop, expect_id): + create_test_user(db) + _seed_rows(db, SEED) + tok = _login(client) + r = client.get(f"{KA}?related_page={prop}", headers=auth_header(tok)) + assert r.status_code == 200, r.text + assert {i["id"] for i in r.json()["data"]} == {expect_id}, f"{prop} -> {r.json()['data']}" + + +def test_f4_unmatchable_prop_returns_empty_not_all(client: TestClient, db: Session): + """不可归一入参(纯 query/空白)⇒ 空集,**不得**退化为“不过滤=全量”。""" + create_test_user(db) + _seed_rows(db, SEED) + tok = _login(client) + for prop in ["%20", "?only=query", "%23frag"]: + r = client.get(f"{KA}?related_page={prop}", headers=auth_header(tok)) + assert r.status_code == 200, r.text + assert r.json()["data"] == [], f"{prop} 应空集,实得 {len(r.json()['data'])} 行" + + +# ── §5 模板子路由 ─────────────────────────────────────────────────── + +@pytest.mark.parametrize("stored,prop,hit", [ + ("/maps/canvas/:articleId", "/maps/canvas", True), + ("/maps/canvas/:id/step", "/maps/canvas", True), + ("/maps/canvas", "/maps/canvas", True), + ("/maps-review", "/maps", False), + ("/maps/canvas/detail", "/maps/canvas/detail", True), +]) +def test_f5_template_child(db: Session, client: TestClient, stored, prop, hit): + create_test_user(db) + _seed_rows(db, [(1, stored)]) + tok = _login(client) + r = client.get(f"{KA}?related_page={prop}", headers=auth_header(tok)) + assert r.status_code == 200, r.text + got = r.json()["data"] + if hit: + assert [i["related_page"] for i in got] == [stored], f"{prop} 应命中 {stored},实得 {got}" + else: + assert got == [], f"{prop} 不应命中 {stored}" + + +# ── §6 既有语义保持 ──────────────────────────────────────────────── + +def test_f6_partition_and_category_still_hold(client: TestClient, db: Session): + """账套分区(本企业 ∪ 全局)与 category 组合过滤在精确口径下不变。""" + _ensure_entity(db, 2, "陕西博海", "博海") + rows = [ + KnowledgeArticle(title="K1", content="c", entity_id=1, related_page="/kpis", category="term"), + KnowledgeArticle(title="K2", content="c", entity_id=1, related_page="/kpis", category="practice"), + KnowledgeArticle(title="K3", content="c", entity_id=1, related_page="/budget", category="term"), + KnowledgeArticle(title="K4", content="c", entity_id=None, related_page=None, category="term"), + KnowledgeArticle(title="K5", content="c", entity_id=2, related_page="/kpis", category="term"), + ] + db.add_all(rows) + db.commit() + for a in rows: + db.refresh(a) + create_test_user(db) + tok = _login(client) + + rp = client.get(f"{KA}?related_page=/kpis", headers=auth_header(tok)).json()["data"] + assert {i["id"] for i in rp} == {rows[0].id, rows[1].id} + rc = client.get(f"{KA}?category=term", headers=auth_header(tok)).json()["data"] + assert {i["id"] for i in rc} == {rows[0].id, rows[2].id, rows[3].id} + rb = client.get(f"{KA}?related_page=/kpis&category=term", headers=auth_header(tok)).json()["data"] + assert {i["id"] for i in rb} == {rows[0].id}