fix(knowledge): related_page 过滤精确化（子串包含 -> 归一候选键等值）
KBFF-v11c（派单 KBFF-v11c-relatedpage-exact-match-filter · A 侧 wecom-yanxue · 执行人 wecom-project） 本提交=代码面切片：未 push、未部署；生产部署另走发布单 + approval-gate（回滚 = git revert <本 sha>）。 - backend/app/api/knowledge_articles.py：related_page 过滤由 LIKE '%<v>%'（子串包含，短键/前缀超集命中） 收敛为「归一候选键 IN ∪ 模板子路由 <键>/:<占位符>」；不可归一入参 -> 空集（不退化为全量） - backend/app/utils/route_key.py（新增）：匹配口径唯一实现处（归一化 / 候选键派生 / 模板模式） - backend/tests/test_kbff_v11c_related_page_exact_match_20261012.py（新增）：42 用例 —— 生产 seed 7 值逐值回归（各中 1 行）、短键 /maps->0 行 /kpi->0 行、带参路由与尾斜杠/query/大小写兼容、 改造前 contains 语义负对照、账套分区与 category 组合语义保持 (cherry picked from commit 2a5cb182c237ae724d4b506b68eb48810ba27b16)
This commit is contained in:
wecom-project
committed by
Hermes CI Fix
Hermes CI Fix
parent
60b6ef3bfe
commit
5fa83e2532
@@ -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 '%<v>%'」收敛为
|
||||
# 「归一化候选键精确等值 ∪ 模板子路由」——消除短键/前缀超集命中
|
||||
# （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()
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""路由键归一与候选键派生（KBFF-v11c，2026-10-12）
|
||||
|
||||
背景（隐患）：
|
||||
知识文章列表接口原用 ``KnowledgeArticle.related_page.contains(related_page)``
|
||||
⇒ SQL ``LIKE '%<prop>%'`` **子串包含**。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) + "/:%"
|
||||
@@ -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}
|
||||
Reference in New Issue
Block a user