merge: T6 知识库实体与图谱(T-P2-01/05)

This commit is contained in:
xsl
2026-08-14 20:09:59 +08:00
10 changed files with 1499 additions and 10 deletions
+19 -6
View File
@@ -1,14 +1,27 @@
# `packages/kb/` — 美学知识库(占位)
# `packages/kb/` — 美学知识库
**职责**(架构 §4.1):知识实体/关系/规则/素材的读写与发布。
**不负责**:检索算法(调 `packages/retrieval`)。
## 待实现
## 本批已落地(T6,T-P2-01/05)
- 表:`kb_entity` `kb_relation` `kb_rule` `kb_chunk`(vector 1024) `kb_asset`
`kb_color` `kb_gap`
- FR-KB-01~12:本体 Schema、自建色板、内容录入、规则库、白底图流水线、
草稿→审核→发布→归档流程、知识缺口埋点
- `schemas.py`:12 类 `EntityType` Literal、`EntityStatus`、CRUD 与图谱
响应模型(雪花 ID 一律字符串出参,防 JS 精度丢失)
- `repo.py`:`kb_entity`/`kb_relation`/`kb_color` 数据访问;
`traverse` 为递归 CTE 1~3 跳有界遍历(防环:`path` 数组 + `NOT @>`)
- `service.py`:状态机(draft→reviewing→published→archived,撤稿支线
draft/reviewing→archived,非法跃迁 40900)、内容修改 version+1 且
published 只读(要改必须撤稿归档后重录新草稿)、关系写入(自环/重复/
端点缺失校验)、色板 role/hex(`#RRGGBB`)校验
- 路由 `apps/api/routers/kb.py`:`GET/POST /kb/items`、
`GET/PATCH /kb/items/{id}`、`GET /kb/items/{id}/graph?max_hops=1..3`
## 待实现(后续批次)
- `kb_rule` 决策表 CRUD(T7 规则引擎)
- `kb_chunk` 向量+全文混合检索(T8)
- `kb_asset` 白底图流水线、`kb_gap` 知识缺口埋点(T-P2-06+)
- 关系/色板的运营 HTTP 入口(后台角色 T-P2-09 就绪后挂路由)
## 为什么它是数据不是代码
+5 -1
View File
@@ -1 +1,5 @@
"""kb 域:知识实体/关系/规则/素材的读写与发布。占位。"""
"""kb 域:知识实体/关系/规则/素材的读写与发布。
本批已落地实体底座(schemas / repo / service,见本目录 README);
规则(kb_rule)归 packages/rules + T7,检索(kb_chunk)归 T8。
"""
+168
View File
@@ -0,0 +1,168 @@
"""kb 域数据访问层。**只有本模块允许触碰 kb_entity / kb_relation / kb_color。**
``traverse`` 用递归 CTE 而不是在 Python 里逐跳发查询:TD-09 的结论是
知识图谱的查询都是「从某实体出发的浅层有界遍历」(1~3 跳),邻接表 +
CTE 在 PG 内一次往返就能完成,不值得为此引入图数据库的第二套存储与运维。
防环在 CTE 里做(``path`` 数组 + ``@>`` 包含判断),保证有环数据
(A→B→A)也不会让递归发散。
"""
from __future__ import annotations
from collections.abc import Iterable
from dataclasses import dataclass
from sqlalchemy import ColumnElement, func, select, text
from sqlalchemy.ext.asyncio import AsyncSession
from packages.shared.models import KbColor, KbEntity, KbRelation
#: 递归遍历 SQL(1~3 跳)。``path`` 记录已经过的实体 id,``rels`` 与逐跳
#: rel_type 平行——每行结果本身就是一条完整可解释路径,不需要在 Python
#: 侧再拼接。深度上限由参数 ``:max_hops`` 控制(service 层已校验 1~3)。
#: ``::varchar[]`` 显转型不可省:锚点项推得 varchar(64)[]、递归项推得
#: varchar[],PG 要求两段类型完全一致,不转型直接 DatatypeMismatchError。
_TRAVERSE_SQL = text(
"""
WITH RECURSIVE walk(from_id, to_id, rel_type, hop, path, rels) AS (
SELECT r.from_id, r.to_id, r.rel_type, 1,
ARRAY[r.from_id, r.to_id]::bigint[], ARRAY[r.rel_type]::varchar[]
FROM kb_relation r
WHERE r.from_id = :start_id
UNION ALL
SELECT r.from_id, r.to_id, r.rel_type, w.hop + 1,
w.path || r.to_id, w.rels || ARRAY[r.rel_type]::varchar[]
FROM kb_relation r
JOIN walk w ON r.from_id = w.to_id
WHERE w.hop < :max_hops
AND NOT w.path @> ARRAY[r.to_id]
)
SELECT hop, path, rels FROM walk ORDER BY hop, path, rels
"""
)
@dataclass(frozen=True, slots=True)
class RawPath:
"""CTE 直接吐出的一条路径(防环已由 SQL 保证,len(rel_types)==len(node_ids)-1)。"""
node_ids: tuple[int, ...]
rel_types: tuple[str, ...]
# --- kb_entity ---------------------------------------------------------------
async def get_entity(session: AsyncSession, entity_id: int) -> KbEntity | None:
return await session.get(KbEntity, entity_id)
async def get_entity_for_update(session: AsyncSession, entity_id: int) -> KbEntity | None:
result = await session.execute(
select(KbEntity).where(KbEntity.entity_id == entity_id).with_for_update()
)
return result.scalar_one_or_none()
async def get_entities_by_ids(session: AsyncSession, ids: Iterable[int]) -> dict[int, KbEntity]:
id_list = list(ids)
if not id_list:
return {}
result = await session.execute(select(KbEntity).where(KbEntity.entity_id.in_(id_list)))
return {entity.entity_id: entity for entity in result.scalars()}
async def find_entity_by_name(
session: AsyncSession, *, entity_type: str, name: str
) -> KbEntity | None:
# UNIQUE(entity_type, name) 保证至多一行
result = await session.execute(
select(KbEntity).where(KbEntity.entity_type == entity_type, KbEntity.name == name)
)
return result.scalar_one_or_none()
def add_entity(session: AsyncSession, entity: KbEntity) -> KbEntity:
session.add(entity)
return entity
async def search_entities(
session: AsyncSession,
*,
entity_type: str | None = None,
status: str | None = None,
name_ilike: str | None = None,
limit: int = 20,
offset: int = 0,
) -> tuple[list[KbEntity], int]:
"""过滤 + 分页。返回 ``(当页行, 满足过滤条件的总数)``。
``name_ilike`` 不转义 ``%``/``_``:这是后台运营的模糊搜索框,
通配符被当通配符用符合直觉,也不是注入面(参数化绑定)。
"""
conds: list[ColumnElement[bool]] = []
if entity_type is not None:
conds.append(KbEntity.entity_type == entity_type)
if status is not None:
conds.append(KbEntity.status == status)
if name_ilike:
conds.append(KbEntity.name.ilike(f"%{name_ilike}%"))
total = int(await session.scalar(select(func.count()).select_from(KbEntity).where(*conds)) or 0)
result = await session.execute(
select(KbEntity)
.where(*conds)
.order_by(KbEntity.created_at.desc(), KbEntity.entity_id.desc())
.limit(limit)
.offset(offset)
)
return list(result.scalars().all()), total
# --- kb_relation -------------------------------------------------------------
async def find_relation(
session: AsyncSession, *, from_id: int, rel_type: str, to_id: int
) -> KbRelation | None:
result = await session.execute(
select(KbRelation).where(
KbRelation.from_id == from_id,
KbRelation.rel_type == rel_type,
KbRelation.to_id == to_id,
)
)
return result.scalar_one_or_none()
def add_relation(session: AsyncSession, relation: KbRelation) -> KbRelation:
session.add(relation)
return relation
async def traverse(session: AsyncSession, *, from_id: int, max_hops: int) -> list[RawPath]:
"""从 ``from_id`` 出发沿 from→to 方向做 ``max_hops`` 跳的有界遍历。
深度合法性(1~3)由 service 层校验,这里信任调用方。返回按
(跳数, 路径, 关系序列)排序的全部路径——同一终点多条路径都保留,
「为什么推荐」恰恰需要把每条依据都摆出来。
"""
result = await session.execute(_TRAVERSE_SQL, {"start_id": from_id, "max_hops": max_hops})
return [RawPath(node_ids=tuple(row.path), rel_types=tuple(row.rels)) for row in result]
# --- kb_color ----------------------------------------------------------------
async def list_colors(session: AsyncSession, *, season_code: str | None = None) -> list[KbColor]:
stmt = select(KbColor).order_by(KbColor.season_code, KbColor.color_id)
if season_code is not None:
stmt = stmt.where(KbColor.season_code == season_code)
result = await session.execute(stmt)
return list(result.scalars().all())
def add_color(session: AsyncSession, color: KbColor) -> KbColor:
session.add(color)
return color
+166
View File
@@ -0,0 +1,166 @@
"""kb 域请求/响应模型(架构文档 §6.2,FR-KB-01/02/05).
``apps/api/schemas.py`` 是全局共享契约文件,本轮并行开发禁改;kb 接口的
请求/响应模型统一放本包,由路由层装配进 ``ok(...)`` 信封(NFR-MAINT-01
「契约先行」:客户端类型从 OpenAPI 生成)。
雪花 ID 一律以字符串出现在载荷里:JS Number 只有 2^53 精度,直接给数字
会被四舍五入成另一个实体。
"""
from __future__ import annotations
from datetime import datetime
from decimal import Decimal
from typing import Any, Literal, get_args
from pydantic import BaseModel, Field
#: 12 类实体(知识库文档 §3.1 本体)。写死成 Literal 而不是普通 str:
#: 新增类目是需要评审的低频动作,宁可发版,也不给运营侧随手写错值留口子。
EntityType = Literal[
"item",
"color",
"style",
"body_shape",
"face_hair",
"occasion",
"material",
"accessory",
"makeup",
"rule",
"outfit",
"trend",
]
ENTITY_TYPES: tuple[str, ...] = get_args(EntityType)
#: 生命周期四态。合法跃迁矩阵见 :mod:`packages.kb.service`(非法跃迁 40900)。
EntityStatus = Literal["draft", "reviewing", "published", "archived"]
ENTITY_STATUSES: tuple[str, ...] = get_args(EntityStatus)
#: kb_color.role:基础色 / 点缀色 / 回避色(TD-18 自建色板)。
ColorRole = Literal["base", "accent", "avoid"]
#: 图谱遍历深度上限(TD-09:浅层有界遍历,1~3 跳,不上图数据库)。
#: 放在这里是因为路由 Query 与 service 校验必须引用同一个数字,两处硬编码必然漂移。
MAX_GRAPH_HOPS = 3
class EntityCreate(BaseModel):
"""``POST /kb/items`` 载荷。新实体固定从 draft/v1 起步,不收客户端指定状态。"""
entity_type: EntityType = Field(description="12 类本体之一,见 EntityType")
name: str = Field(min_length=1, max_length=128, description="同类目内唯一(UNIQUE 守护)")
attrs: dict[str, Any] = Field(
default_factory=dict, description="类目差异化字段,由 attrs JSONB 承载(架构 §6.2)"
)
source: str | None = Field(default=None, max_length=255, description="知识来源,版权审计用")
license: str | None = Field(default=None, max_length=64, description="内容许可证标识")
class EntityUpdate(BaseModel):
"""``PATCH /kb/items/{id}`` 载荷:内容字段与状态可单独或同时提交。
- 只出现内容字段(name/attrs/source/license)→ 内容修改,version+1,
仅 draft/reviewing 状态允许(published 只读,见 service)。
- 只出现 status → 纯状态跃迁,走状态机校验,version 不变。
- 与当前状态相同的 status → 幂等重放,视为 no-op 成功(客户端重试友好)。
"""
name: str | None = Field(default=None, min_length=1, max_length=128)
attrs: dict[str, Any] | None = None
source: str | None = Field(default=None, max_length=255)
license: str | None = Field(default=None, max_length=64)
status: EntityStatus | None = None
def content_fields(self) -> dict[str, Any]:
"""本次请求显式提供的内容字段(PATCH 语义:未提供的字段不动)。
``exclude_unset`` 保证显式传 ``null``(清空 source/license)与
「未传」可区分——这是 PATCH 覆盖语义的最低要求。
"""
return self.model_dump(exclude_unset=True, exclude={"status"})
class EntityOut(BaseModel):
"""实体对外视图(``data`` 载荷,单条或列表元素)。"""
entity_id: str = Field(description="雪花 ID,字符串形式")
entity_type: str
name: str
attrs: dict[str, Any]
status: str
version: int = Field(description="内容版本号,每次内容修改 +1")
source: str | None
license: str | None
updated_by: str | None
created_at: datetime
updated_at: datetime
class RelationCreate(BaseModel):
"""``kb_relation`` 三元组写入载荷。
本批次未暴露 HTTP 路由(关系录入属运营后台 T-P2-09 的范围),但写入
校验(自环/重复/端点存在)在 service 层先行落地,集成测试直接驱动。
"""
from_entity_id: int
to_entity_id: int
rel_type: str = Field(min_length=1, max_length=64, description="关系类型,如 suits/fits")
weight: Decimal = Field(default=Decimal("1"), ge=0, le=Decimal("9.999"), description="权重")
confidence: Decimal = Field(
default=Decimal("1"), ge=0, le=Decimal("9.999"), description="知识置信度"
)
evidence: str | None = Field(default=None, max_length=255, description="证据/出处")
class GraphNode(BaseModel):
"""路径上的实体摘要:图谱路径要能「读给人听」,只有 ID 不可解释。"""
entity_id: str
entity_type: str
name: str
status: str
class GraphHop(BaseModel):
"""一跳:从哪个实体经什么关系到哪个实体(T-P2-05 可解释路径的最小单位)。"""
from_entity_id: str
rel_type: str
to_entity_id: str
class GraphPath(BaseModel):
"""一条从起点出发的完整路径:节点序列 + 每跳 rel_type。
``len(rel_types) == len(nodes) - 1``;同一终点可能有多条路径,
遍历结果不去重合并——「为什么推荐」恰恰需要把每条依据都摆出来。
"""
nodes: list[GraphNode]
hops: list[GraphHop]
class GraphQueryResult(BaseModel):
"""``GET /kb/items/{id}/graph`` 载荷。"""
from_entity_id: str
max_hops: int = Field(description="本次遍历深度(1~3)")
paths: list[GraphPath] = Field(description="按跳数升序的全部可解释路径")
class ColorOut(BaseModel):
"""色板条目(TD-18 自建色板)。Lab 供后续 ΔE 数值匹配(T8 检索),此处只读展示。"""
color_id: str
season_code: str = Field(description="季型代码,如 spring_warm")
role: str = Field(description="base=基础色 / accent=点缀色 / avoid=回避色")
lab_l: float
lab_a: float
lab_b: float
hex: str = Field(description="#RRGGBB")
name_cn: str
+392
View File
@@ -0,0 +1,392 @@
"""kb 域 service:实体生命周期、关系写入、图谱遍历、自建色板(FR-KB-01/02/05).
状态机与格式校验拆成模块级纯函数(不碰 DB,单测直接覆盖矩阵);
``KbService`` 只负责组装、落库与跨表编排。所有写操作只 ``flush`` 不
``commit``——事务边界交给 ``session_dep`` 的统一提交/回滚,与 task /
billing 域一致。
"""
from __future__ import annotations
import re
from datetime import UTC, datetime
from decimal import Decimal
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from packages.domain.ids import next_id
from packages.kb import repo
from packages.kb.schemas import (
MAX_GRAPH_HOPS,
ColorOut,
EntityCreate,
EntityUpdate,
GraphHop,
GraphNode,
GraphPath,
GraphQueryResult,
RelationCreate,
)
from packages.shared.errors import AppError, ErrorCode, NotFoundError
from packages.shared.logging import get_logger
from packages.shared.models import KbColor, KbEntity, KbRelation
logger = get_logger(__name__)
#: 合法跃迁矩阵(知识库文档「发布回滚」语义):
#: draft → reviewing → published → archived 主线;
#: draft/reviewing → archived 撤稿支线;
#: archived 终态。其余(含 published→draft、reviewing→draft、自跃迁)
#: 一律非法——想改已发布内容,就撤稿归档后以新草稿重录,让新版本重走评审。
LEGAL_TRANSITIONS: frozenset[tuple[str, str]] = frozenset(
{
("draft", "reviewing"),
("reviewing", "published"),
("published", "archived"),
("draft", "archived"),
("reviewing", "archived"),
}
)
#: 内容可编辑的状态。published 内容只读(版本语义的基石),archived 终态。
CONTENT_EDITABLE_STATUSES: frozenset[str] = frozenset({"draft", "reviewing"})
_HEX_COLOR_PATTERN = re.compile(r"^#[0-9A-Fa-f]{6}$")
_COLOR_ROLES: frozenset[str] = frozenset({"base", "accent", "avoid"})
# --- 纯校验函数(单测直接覆盖,不触碰 DB) ----------------------------------
def can_transition(current: str, target: str) -> bool:
return (current, target) in LEGAL_TRANSITIONS
def assert_transition(current: str, target: str) -> None:
"""非法跃迁 → ``STATE_CONFLICT``(409)。矩阵见 :data:`LEGAL_TRANSITIONS`。"""
if not can_transition(current, target):
raise AppError(
ErrorCode.STATE_CONFLICT,
message="实体状态不允许该跃迁",
data={"current": current, "target": target},
)
def ensure_content_editable(status: str) -> None:
"""published/archived 的内容不可直接修改(version 递增的前提)。"""
if status not in CONTENT_EDITABLE_STATUSES:
raise AppError(
ErrorCode.STATE_CONFLICT,
message=f"{status} 状态的实体内容不可修改",
hint="已发布内容要改:先撤稿归档,再以新草稿重录,新版本重走评审",
data={"status": status},
)
def ensure_valid_hex(hex_code: str) -> None:
"""色板 hex 必须 ``#RRGGBB``(大小写不敏感,入库前不归一化,展示原样保留)。"""
if _HEX_COLOR_PATTERN.fullmatch(hex_code) is None:
raise AppError(
ErrorCode.BAD_REQUEST,
message="色板 hex 格式非法,应为 #RRGGBB",
data={"hex": hex_code},
)
def ensure_valid_color_role(role: str) -> None:
if role not in _COLOR_ROLES:
raise AppError(
ErrorCode.BAD_REQUEST,
message="色板 role 非法,应为 base/accent/avoid",
data={"role": role},
)
def ensure_valid_hops(max_hops: int) -> None:
"""遍历深度 1~3(TD-09 有界遍历;上限同时也是防环的保险丝)。"""
if not 1 <= max_hops <= MAX_GRAPH_HOPS:
raise AppError(
ErrorCode.BAD_REQUEST,
message=f"max_hops 超出允许范围 1~{MAX_GRAPH_HOPS}",
data={"max_hops": max_hops},
)
# --- service -----------------------------------------------------------------
class KbService:
def __init__(self, session: AsyncSession) -> None:
self._session = session
# -- 实体 -----------------------------------------------------------------
async def create_entity(
self, *, payload: EntityCreate, updated_by: str | None = None
) -> KbEntity:
"""新建实体:固定 draft / version=1 起步。同名同类目拒绝(400)。"""
existing = await repo.find_entity_by_name(
self._session, entity_type=payload.entity_type, name=payload.name
)
if existing is not None:
raise AppError(
ErrorCode.BAD_REQUEST,
message="同名同类目的实体已存在",
data={"entity_type": payload.entity_type, "name": payload.name},
)
entity = repo.add_entity(
self._session,
KbEntity(
entity_id=next_id(),
entity_type=payload.entity_type,
name=payload.name,
attrs=payload.attrs,
status="draft",
version=1,
source=payload.source,
license=payload.license,
updated_by=updated_by,
),
)
try:
await self._session.flush()
except IntegrityError as exc:
# 并发同名是唯一预期内的约束冲突(UNIQUE(entity_type,name) 兜底)
raise AppError(
ErrorCode.BAD_REQUEST,
message="同名同类目的实体已存在",
data={"entity_type": payload.entity_type, "name": payload.name},
) from exc
logger.info(
"kb_entity_created",
entity_id=entity.entity_id,
entity_type=entity.entity_type,
name=entity.name,
)
return entity
async def get_entity(self, *, entity_id: int) -> KbEntity:
entity = await repo.get_entity(self._session, entity_id)
if entity is None:
raise NotFoundError("实体不存在")
return entity
async def search_entities(
self,
*,
entity_type: str | None = None,
status: str | None = None,
name_ilike: str | None = None,
limit: int = 20,
offset: int = 0,
) -> tuple[list[KbEntity], int]:
return await repo.search_entities(
self._session,
entity_type=entity_type,
status=status,
name_ilike=name_ilike,
limit=limit,
offset=offset,
)
async def update_entity(
self, *, entity_id: int, payload: EntityUpdate, updated_by: str | None = None
) -> KbEntity:
"""内容修改与状态跃迁共用一个 PATCH 入口,规则:
- 内容字段仅 draft/reviewing 可改,每次内容修改 version+1;
- 状态变化必须走合法跃迁矩阵;与当前相同的 status 是幂等 no-op;
- 先全部校验、后统一变更——半个请求不会落库。
"""
entity = await repo.get_entity_for_update(self._session, entity_id)
if entity is None:
raise NotFoundError("实体不存在")
content = payload.content_fields()
if content:
ensure_content_editable(entity.status)
if payload.status is not None and payload.status != entity.status:
assert_transition(entity.status, payload.status)
if content:
for field, value in content.items():
setattr(entity, field, value)
entity.version += 1
if payload.status is not None:
entity.status = payload.status
if content or payload.status is not None:
entity.updated_by = updated_by or entity.updated_by
entity.updated_at = datetime.now(UTC)
try:
await self._session.flush()
except IntegrityError as exc:
# 改名撞上另一条同类目实体
raise AppError(
ErrorCode.BAD_REQUEST,
message="同名同类目的实体已存在",
data={"entity_id": str(entity_id)},
) from exc
logger.info(
"kb_entity_updated",
entity_id=entity_id,
version=entity.version,
status=entity.status,
content_changed=bool(content),
)
return entity
# -- 关系 -----------------------------------------------------------------
async def add_relation(self, *, payload: RelationCreate) -> KbRelation:
"""写入三元组。自环与重复(UNIQUE 三元组)拒绝(400),端点缺失 404。"""
if payload.from_entity_id == payload.to_entity_id:
raise AppError(
ErrorCode.BAD_REQUEST,
message="不允许自环关系",
data={"entity_id": str(payload.from_entity_id)},
)
for end_id in (payload.from_entity_id, payload.to_entity_id):
if await repo.get_entity(self._session, end_id) is None:
raise NotFoundError("关联实体不存在")
duplicate = await repo.find_relation(
self._session,
from_id=payload.from_entity_id,
rel_type=payload.rel_type,
to_id=payload.to_entity_id,
)
if duplicate is not None:
raise AppError(
ErrorCode.BAD_REQUEST,
message="该 from/rel_type/to 三元组已存在",
data={
"from_entity_id": str(payload.from_entity_id),
"rel_type": payload.rel_type,
"to_entity_id": str(payload.to_entity_id),
},
)
relation = repo.add_relation(
self._session,
KbRelation(
relation_id=next_id(),
from_id=payload.from_entity_id,
rel_type=payload.rel_type,
to_id=payload.to_entity_id,
weight=payload.weight,
confidence=payload.confidence,
evidence=payload.evidence,
),
)
await self._session.flush()
logger.info(
"kb_relation_added",
from_id=payload.from_entity_id,
rel_type=payload.rel_type,
to_id=payload.to_entity_id,
)
return relation
# -- 图谱 -----------------------------------------------------------------
async def traverse_graph(
self, *, entity_id: int, max_hops: int = MAX_GRAPH_HOPS
) -> GraphQueryResult:
"""1~3 跳可解释路径(T-P2-05)。
CTE 只回 id 与 rel_type 序列;节点摘要(类型/名称/状态)由第二次
查询补齐——遍历有界,涉及的实体集合很小,两次往返远比在递归 SQL
里 JOIN 实体表好维护。
"""
ensure_valid_hops(max_hops)
await self.get_entity(entity_id=entity_id) # 起点 404 语义保持一致
raw_paths = await repo.traverse(self._session, from_id=entity_id, max_hops=max_hops)
node_ids: set[int] = set()
for raw in raw_paths:
node_ids.update(raw.node_ids)
nodes_map = await repo.get_entities_by_ids(self._session, node_ids)
paths: list[GraphPath] = []
for raw in raw_paths:
nodes = [_node_summary(nodes_map, node_id) for node_id in raw.node_ids]
hops = [
GraphHop(
from_entity_id=str(hop_from),
rel_type=rel_type,
to_entity_id=str(hop_to),
)
# node_ids 比 rel_types 多一个元素(n 个节点 n-1 跳),首尾
# 错位 zip 即得「第 i 跳 = nodes[i] --rels[i]--> nodes[i+1]」
for hop_from, rel_type, hop_to in zip(
raw.node_ids[:-1], raw.rel_types, raw.node_ids[1:], strict=True
)
]
paths.append(GraphPath(nodes=nodes, hops=hops))
return GraphQueryResult(from_entity_id=str(entity_id), max_hops=max_hops, paths=paths)
# -- 色板(TD-18 自建色板) ------------------------------------------------
async def add_color(
self,
*,
season_code: str,
role: str,
lab_l: float,
lab_a: float,
lab_b: float,
hex_code: str,
name_cn: str,
) -> ColorOut:
"""录入色板条目。role/hex 校验在 service 层(运营后台 T-P2-09 直接复用)。"""
ensure_valid_color_role(role)
ensure_valid_hex(hex_code)
color = repo.add_color(
self._session,
KbColor(
color_id=next_id(),
season_code=season_code,
role=role,
lab_l=Decimal(str(lab_l)),
lab_a=Decimal(str(lab_a)),
lab_b=Decimal(str(lab_b)),
hex=hex_code,
name_cn=name_cn,
),
)
await self._session.flush()
logger.info("kb_color_added", season_code=season_code, role=role, hex=hex_code)
return _color_out(color)
async def list_colors(self, *, season_code: str | None = None) -> list[ColorOut]:
rows = await repo.list_colors(self._session, season_code=season_code)
return [_color_out(row) for row in rows]
def _node_summary(nodes_map: dict[int, KbEntity], node_id: int) -> GraphNode:
"""路径节点摘要。关系端点在写入时校验过存在,理论必中;万一实体行
被运维手工删除,退化为占位摘要而不是让整个图查询 500。"""
entity = nodes_map.get(node_id)
if entity is None:
logger.warning("kb_graph_node_missing", entity_id=node_id)
return GraphNode(
entity_id=str(node_id), entity_type="unknown", name="unknown", status="unknown"
)
return GraphNode(
entity_id=str(entity.entity_id),
entity_type=entity.entity_type,
name=entity.name,
status=entity.status,
)
def _color_out(color: KbColor) -> ColorOut:
return ColorOut(
color_id=str(color.color_id),
season_code=color.season_code,
role=color.role,
lab_l=float(color.lab_l),
lab_a=float(color.lab_a),
lab_b=float(color.lab_b),
hex=color.hex,
name_cn=color.name_cn,
)