feat(entity-dim): 账套维度收口批 — subjects/analysis_results 加列 + 分析结论账套化 + 模板实例化构造点补账套
ci/woodpecker/push/main Pipeline failed

- ① kpi_templates instantiate 构造点显式传会话账套(原缺 entity_id ⇒ 静默落 e1),kpi_code 判重按账套收紧
- ② kpi_subject_map 3 个读取点加护栏注释(不加列、不改语义)
- ③ subjects 分层:读口 union(NULL=全局标准准则科目 / N=账套私有),UPSERT 唯一键改 (entity_id, subject_code)
- ④ analysis_results 账套化:读/写/删三端点按会话账套判定归属(列未落地时 kpi_code JOIN 止血),
     无归属行进 unattributed 登记、删除一律 404
- 模型层按现网 DDL 补齐:KPIDefinition(template_id/is_system + uk_entity_kpi)、AnalysisResult.entity_id、Subject.entity_id
- 新增 migrate_{subjects,analysis_results}_entity_id_20260923.py(--check/--apply/--down)、
  22 条对账用例 test_entity_dimension_convergence_20260923.py

DDL 已由 ops 受控通道落地(subjects 21:31:56 / analysis_results 21:31:59 CST,审批 1e9b4d9a/5ffb1640)
本提交为切片提交,未 push(发布须另取放行)
This commit is contained in:
Hermes CI Fix
2026-09-23 21:39:24 +08:00
parent 0b3a8aa111
commit fb4609be14
14 changed files with 1429 additions and 65 deletions
+159 -28
View File
@@ -1,10 +1,25 @@
"""置信度评分系统 — 财务Bot分析结论管理""" """置信度评分系统 — 财务Bot分析结论管理
账套维度(2026-09-23 账套维度收口批,任务 0-P1-project-20260923-entity-dimension-convergence
§二③ 读侧止血 + §三④ 账套化):
处置前:三个端点均无 `get_entity_id` ⇒ e1 / e2 双 token 拿到**逐字节相同**的返回体
(实测 2 行存量:`id=1 TEST001` / `id=2 kpi_code=NULL`)。
处置后:读/写/删均按会话账套判定归属;`entity_id` 列未落地时读侧退化为
「`kpi_code` JOIN `kpi_definitions.entity_id`」止血口径(DDL 与代码可分两步上线)。
归属三级(**任何一级都不猜账套**):
① `analysis_results.entity_id`(列已落地且非 NULL)
② 该行 `kpi_code` 在 `kpi_definitions` 中**唯一归属单账套**时所指向的 `entity_id`
③ 判不出归属(`kpi_code` 为空 / 库内无该码 / 该码跨账套二义 / 列为 NULL)→ 进
`unattributed` 登记池,**不返回给任何账套**;删除端点对其一律 404。
"""
from fastapi import APIRouter, Depends, HTTPException, Query from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from sqlalchemy import func from sqlalchemy import func, inspect
from app.database import get_db from app.database import get_db
from app.deps import get_entity_id
from app.auth_middleware import require_auth from app.auth_middleware import require_auth
from app.models import AnalysisResult, KPIValue, KPIDefinition from app.models import AnalysisResult, KPIValue, KPIDefinition, OperationLog
from typing import Optional, List from typing import Optional, List
from datetime import datetime from datetime import datetime
@@ -12,6 +27,95 @@ router = APIRouter(prefix="/api/cma/analysis", tags=["置信度评分"],
dependencies=[Depends(require_auth)], dependencies=[Depends(require_auth)],
) )
# `analysis_results.entity_id` 列存在性缓存(键 = engine 对象 id;DDL 前后同一进程内可能变化,
# 故只缓存判定结果、不做进程级常量:迁移后重启服务即刷新)
_ENTITY_COL_CACHE: dict = {}
def _has_entity_column(db) -> bool:
"""`analysis_results.entity_id` 是否已落地(DDL 迁移 2026-09-23 未执行时为 False)。
读侧止血必须先于 DDL 合入 ⇒ 不能无条件引用该列;列不在时按 `kpi_code` JOIN 判归属。
"""
bind = db.get_bind()
key = id(bind)
cached = _ENTITY_COL_CACHE.get(key)
if cached is None:
try:
cols = {c["name"] for c in inspect(bind).get_columns(AnalysisResult.__tablename__)}
cached = "entity_id" in cols
except Exception: # noqa: BLE001 —— 结构探不通时按"列不存在"的保守口径走 join
cached = False
_ENTITY_COL_CACHE[key] = cached
return cached
def _kpi_owners(db, codes: List[Optional[str]]) -> dict:
"""`kpi_code` → 拥有该码的账套集合(同码可跨账套存在,故为集合)"""
codes = [c for c in codes if c]
if not codes:
return {}
owners: dict = {}
rows = (
db.query(KPIDefinition.kpi_code, KPIDefinition.entity_id)
.filter(KPIDefinition.kpi_code.in_(codes))
.all()
)
for code, eid in rows:
if eid is not None:
owners.setdefault(code, set()).add(eid)
return owners
def _row_owners(db, r, owners_map: dict) -> set:
"""单行归属账套集合;空集 = 判不出归属(不猜账套)
列口径:`entity_id` 非 NULL 即为唯一归属。
join 口径(列未落地 / 列为 NULL 的历史行):只有该 `kpi_code` 在 `kpi_definitions`
中**唯一归属单账套**时才认定——同码跨账套(库内 22 个码属两账套)属归属二义,
按「不猜账套」进登记池,宁可不可见也不误判给他账套。
"""
if _has_entity_column(db):
eid = getattr(r, "entity_id", None)
if eid is not None:
return {eid}
if r.kpi_code:
owners = owners_map.get(r.kpi_code, set())
return owners if len(owners) == 1 else set()
return set()
def _to_dict(r) -> dict:
return {
"id": r.id,
"period": r.period,
"结论": r.conclusion,
"置信度": f"{r.confidence}%",
"数据来源": r.data_source,
"计算逻辑": r.calculation_logic,
"可比基准": r.comparable_benchmark,
"局限": r.limitations,
"kpi_code": r.kpi_code,
"kpi_name": r.kpi_name,
"has_actual": bool(r.has_actual),
"has_target": bool(r.has_target),
"has_trend": bool(r.has_trend),
"has_review": bool(r.has_review),
"created_at": r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else None,
}
def _unattributed_dict(r) -> dict:
"""无归属登记项:只回标识与元信息,不回结论正文(登记而非共享)"""
return {
"id": r.id,
"period": r.period,
"kpi_code": r.kpi_code,
"kpi_name": r.kpi_name,
"confidence": r.confidence,
"created_at": r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else None,
}
def _calc_confidence(has_actual: bool, has_target: bool, has_trend: bool, has_review: bool) -> int: def _calc_confidence(has_actual: bool, has_target: bool, has_trend: bool, has_review: bool) -> int:
"""基于数据完整度自动计算置信度""" """基于数据完整度自动计算置信度"""
@@ -31,8 +135,14 @@ async def get_analysis_results(
period: Optional[str] = Query(None, description="期间 YYYY-MM"), period: Optional[str] = Query(None, description="期间 YYYY-MM"),
kpi_code: Optional[str] = Query(None, description="KPI编码"), kpi_code: Optional[str] = Query(None, description="KPI编码"),
db: Session = Depends(get_db), db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
): ):
"""查询分析结论(按期间和/或KPI编码过滤)""" """查询分析结论(按期间和/或KPI编码过滤)— 账套隔离(2026-09-23)
返回体字段:原有字段全保留(`total` 语义收窄为「本账套可见行数」,`results` 仍为列表),
新增 `entity_id` / `unattributed_total` / `unattributed` / `entity_scope`(均为附加字段,前端契约不破坏)。
无归属行(存量 id=1/2)进 `unattributed`:**对任何账套都不进 `results`**,只作登记。
"""
query = db.query(AnalysisResult).order_by(AnalysisResult.created_at.desc()) query = db.query(AnalysisResult).order_by(AnalysisResult.created_at.desc())
if period: if period:
@@ -41,36 +151,34 @@ async def get_analysis_results(
query = query.filter(AnalysisResult.kpi_code == kpi_code) query = query.filter(AnalysisResult.kpi_code == kpi_code)
results = query.all() results = query.all()
owners_map = _kpi_owners(db, [r.kpi_code for r in results])
mine, unattributed = [], []
for r in results:
owners = _row_owners(db, r, owners_map)
if not owners:
unattributed.append(r)
elif entity_id in owners:
mine.append(r)
return { return {
"total": len(results), "total": len(mine),
"period": period, "period": period,
"results": [ "entity_id": entity_id,
{ "entity_scope": "column" if _has_entity_column(db) else "kpi_join",
"id": r.id, "unattributed_total": len(unattributed),
"period": r.period, "unattributed": [_unattributed_dict(r) for r in unattributed],
"结论": r.conclusion, "results": [_to_dict(r) for r in mine],
"置信度": f"{r.confidence}%",
"数据来源": r.data_source,
"计算逻辑": r.calculation_logic,
"可比基准": r.comparable_benchmark,
"局限": r.limitations,
"kpi_code": r.kpi_code,
"kpi_name": r.kpi_name,
"has_actual": bool(r.has_actual),
"has_target": bool(r.has_target),
"has_trend": bool(r.has_trend),
"has_review": bool(r.has_review),
"created_at": r.created_at.strftime("%Y-%m-%d %H:%M:%S") if r.created_at else None,
}
for r in results
],
} }
@router.post("/result") @router.post("/result")
def create_analysis_result(data: dict = None, db: Session = Depends(get_db)): def create_analysis_result(data: dict = None, db: Session = Depends(get_db),
"""提交分析结果(自动计算置信度)""" entity_id: int = Depends(get_entity_id)):
"""提交分析结果(自动计算置信度)— 账套维度:本次写入挂**会话账套**
`entity_id` 列未落地时(DDL 迁移前)先把会话账套写进 `OperationLog.detail.entity_id`,
列落地后直接写列(响应始终回 `entity_id`)。
"""
if not data: if not data:
data = {} data = {}
period = data.get("period", "") period = data.get("period", "")
@@ -88,6 +196,7 @@ def create_analysis_result(data: dict = None, db: Session = Depends(get_db)):
confidence = _calc_confidence(has_actual, has_target, has_trend, has_review) confidence = _calc_confidence(has_actual, has_target, has_trend, has_review)
has_column = _has_entity_column(db)
result = AnalysisResult( result = AnalysisResult(
period=period, period=period,
conclusion=conclusion, conclusion=conclusion,
@@ -103,10 +212,25 @@ def create_analysis_result(data: dict = None, db: Session = Depends(get_db)):
has_trend=1 if has_trend else 0, has_trend=1 if has_trend else 0,
has_review=1 if has_review else 0, has_review=1 if has_review else 0,
) )
if has_column:
result.entity_id = entity_id
db.add(result) db.add(result)
db.commit() db.commit()
db.refresh(result) db.refresh(result)
if not has_column:
# DDL 未落地前的过渡留痕:会话账套只存在于日志,不猜列
import json
db.add(OperationLog(
user_id=None, action="create", target_type="analysis_result",
target_id=result.id,
detail=json.dumps({"entity_id": entity_id, "kpi_code": kpi_code,
"period": period,
"note": "analysis_results.entity_id 列未落地,账套暂存于日志"},
ensure_ascii=False),
))
db.commit()
return { return {
"id": result.id, "id": result.id,
"period": result.period, "period": result.period,
@@ -117,6 +241,8 @@ def create_analysis_result(data: dict = None, db: Session = Depends(get_db)):
"可比基准": result.comparable_benchmark, "可比基准": result.comparable_benchmark,
"局限": result.limitations, "局限": result.limitations,
"confidence_score": result.confidence, "confidence_score": result.confidence,
"entity_id": entity_id,
"entity_persisted": "column" if has_column else "operation_log",
"message": "分析结论已保存", "message": "分析结论已保存",
} }
@@ -125,11 +251,16 @@ def create_analysis_result(data: dict = None, db: Session = Depends(get_db)):
async def delete_analysis_result( async def delete_analysis_result(
result_id: int, result_id: int,
db: Session = Depends(get_db), db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
): ):
"""删除分析结论""" """删除分析结论 — 归属校验:非本账套 或 无归属一律 **404**(不引入 403,避免探测面)"""
result = db.query(AnalysisResult).filter(AnalysisResult.id == result_id).first() result = db.query(AnalysisResult).filter(AnalysisResult.id == result_id).first()
if not result: if not result:
raise HTTPException(404, "分析结论不存在") raise HTTPException(404, "分析结论不存在")
owners = _row_owners(db, result, _kpi_owners(db, [result.kpi_code]))
if entity_id not in owners:
# 他账套 id 与无归属行走同一文案(不暴露"该行存在但不属于你")
raise HTTPException(404, "分析结论不存在")
db.delete(result) db.delete(result)
db.commit() db.commit()
return {"message": "已删除"} return {"message": "已删除"}
+49 -16
View File
@@ -1036,13 +1036,16 @@ def _norm_subj_active(v):
return None, f"is_active 只允许 0/1(实际 {v!r})" return None, f"is_active 只允许 0/1(实际 {v!r})"
def _validate_subject_rows(rows, db): def _validate_subject_rows(rows, db, entity_id: int = None):
"""行级全量校验(任一错误 → 400 且零落库),返回标准化行 [{字段}] """行级全量校验(任一错误 → 400 且零落库),返回标准化行 [{字段}]
规则(任务 dama-cma-subjects-write-endpoint-20260912): 规则(任务 dama-cma-subjects-write-endpoint-20260912):
subject_code 非空且 ≤_SUBJ_CODE_MAXLEN(=20,DB varchar(20) 列宽)+ 批次内去重;subject_name 非空(DB NOT NULL)且 ≤200; subject_code 非空且 ≤_SUBJ_CODE_MAXLEN(=20,DB varchar(20) 列宽)+ 批次内去重;subject_name 非空(DB NOT NULL)且 ≤200;
level ∈ {1,2,3}(缺省 1);parent_code 为空或存在于本表(或本批次内,支持层级同批导入); level ∈ {1,2,3}(缺省 1);parent_code 为空或存在于本表(或本批次内,支持层级同批导入);
category 限枚举;is_active ∈ {0,1}(缺省 1)。 category 限枚举;is_active ∈ {0,1}(缺省 1)。
账套维度(2026-09-23 `subjects` 分层):parent_code 存在性只认
**全局科目(`entity_id IS NULL`)∪ 本会话账套私有科目**——他账套私有父科目不构成合法父级。
""" """
if not isinstance(rows, list) or not rows: if not isinstance(rows, list) or not rows:
raise HTTPException(400, '请求体必须是非空行数组: {"subjects": [{...}]}') raise HTTPException(400, '请求体必须是非空行数组: {"subjects": [{...}]}')
@@ -1134,11 +1137,13 @@ def _validate_subject_rows(rows, db):
if not bad: if not bad:
clean.append(item) clean.append(item)
# parent_code 存在性:库内(本表)∪ 本批次 # parent_code 存在性:库内(本表,按账套分层)∪ 本批次
unknown_parents = sorted(parents - set(seen.keys())) unknown_parents = sorted(parents - set(seen.keys()))
if unknown_parents: if unknown_parents:
found = {c for (c,) in db.query(Subject.subject_code) pq = db.query(Subject.subject_code).filter(Subject.subject_code.in_(unknown_parents))
.filter(Subject.subject_code.in_(unknown_parents)).all()} if entity_id is not None:
pq = pq.filter((Subject.entity_id.is_(None)) | (Subject.entity_id == entity_id))
found = {c for (c,) in pq.all()}
missing = [p for p in unknown_parents if p not in found] missing = [p for p in unknown_parents if p not in found]
if missing: if missing:
errs.append(f"parent_code 不存在于 subjects 表且不在本批次内: {missing}") errs.append(f"parent_code 不存在于 subjects 表且不在本批次内: {missing}")
@@ -1181,50 +1186,76 @@ def bot_subjects_upsert(
bot: dict = Depends(verify_bot_key), bot: dict = Depends(verify_bot_key),
payload: dict = Depends(_read_subjects_body), payload: dict = Depends(_read_subjects_body),
db: Session = Depends(get_db), db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
): ):
"""会计科目写入(subjects 幂等 UPSERT,单事务;绝不 DELETE) """会计科目写入(subjects 幂等 UPSERT,单事务;绝不 DELETE)
body: {"subjects": [{"subject_code","subject_name","parent_code","level","category","is_active"}], body: {"subjects": [{"subject_code","subject_name","parent_code","level","category","is_active"}],
"dry_run": false, "update_existing": false} "dry_run": false, "update_existing": false}
账套(`entity_id`)来源:会话账套(`get_entity_id`:token → query/header `X-Entity-Id`/`entity_id` →
默认 1);**不接受 body 内的 entity_id**(Bot 通道按 header/query 显式传,防越权落错账套)。
- 唯一键口径(2026-09-23 `subjects` 分层):`(entity_id, subject_code)`——同账套内同码才视为「已存在」;
不同账套可各有同码科目(如两账套都有 `1001 库存现金`)
- 全量校验通过才落库;任一非法行 → 400 且库内零变化 - 全量校验通过才落库;任一非法行 → 400 且库内零变化
- 已存在行:update_existing=false → skip(幂等默认);true → 更新 name/parent/level/category - 已存在行(同码**可见**=全局标准科目 ∪ 本账套私有行):update_existing=false → skip(幂等默认);
true → 更新 name/parent/level/category —— **但只更新本账套私有行**;
同码若只以**全局标准科目**(`entity_id IS NULL`)存在,则**永不被账套 UPSERT 改写**(计入 `skipped_global`),
同时也不再插入本账套同码副本(避免标准科目重复)。
- dry_run=true → 只校验与回显,零写入 - dry_run=true → 只校验与回显,零写入
返回: {"ok":true,"inserted":n,"updated":m,"skipped":k,"errors":[]};未授权 401。 返回: {"ok":true,"inserted":n,"updated":m,"skipped":k,"skipped_global":g,"entity_id":e,"errors":[]};未授权 401。
""" """
clean = _validate_subject_rows(payload.get("rows"), db) clean = _validate_subject_rows(payload.get("rows"), db, entity_id=entity_id)
dry_run = payload.get("dry_run", False) dry_run = payload.get("dry_run", False)
update_existing = payload.get("update_existing", False) update_existing = payload.get("update_existing", False)
codes = [r["subject_code"] for r in clean] codes = [r["subject_code"] for r in clean]
existing = set() existing = set() # 本会话「可见」的同码 = 全局标准科目(NULL)∪ 本账套私有行
global_only = set() # 只以全局行(entity_id IS NULL)存在的同码:可见,但**不许被账套 UPSERT 改写**
if codes: if codes:
existing = {c for (c,) in db.query(Subject.subject_code) rows = (db.query(Subject.subject_code, Subject.entity_id)
.filter(Subject.subject_code.in_(codes)).all()} .filter(Subject.subject_code.in_(codes),
(Subject.entity_id.is_(None)) | (Subject.entity_id == entity_id))
.all())
by_code = {}
for c, eid in rows:
by_code.setdefault(c, set()).add(eid)
existing = set(by_code)
global_only = {c for c, eids in by_code.items() if eids == {None}}
to_insert = [r for r in clean if r["subject_code"] not in existing] to_insert = [r for r in clean if r["subject_code"] not in existing]
existed = [r for r in clean if r["subject_code"] in existing] existed = [r for r in clean if r["subject_code"] in existing]
to_update = existed if update_existing else [] # 可被本账套更新的只有「本账套私有行」;全局标准科目(NULL)一律不动(跨账套共享口径)
to_skip = [] if update_existing else existed updatable = [r for r in existed if r["subject_code"] not in global_only]
to_update = updatable if update_existing else []
updated_codes = {r["subject_code"] for r in to_update}
to_skip = [r for r in existed if r["subject_code"] not in updated_codes]
skip_global = [r for r in to_skip if r["subject_code"] in global_only]
before = db.query(func.count(Subject.id)).scalar() or 0 before = db.query(func.count(Subject.id)).scalar() or 0
if dry_run: if dry_run:
logger.info(f"subjects UPSERT dry_run: bot={bot.get('name')} rows={len(clean)} " logger.info(f"subjects UPSERT dry_run: bot={bot.get('name')} entity={entity_id} rows={len(clean)} "
f"would_insert={len(to_insert)} would_update={len(to_update)} " f"would_insert={len(to_insert)} would_update={len(to_update)} "
f"would_skip={len(to_skip)}") f"would_skip={len(to_skip)}")
return {"ok": True, "dry_run": True, "applied": False, "table": "subjects", return {"ok": True, "dry_run": True, "applied": False, "table": "subjects",
"entity_id": entity_id,
"before": before, "after": before, "before": before, "after": before,
"inserted": 0, "updated": 0, "skipped": 0, "upserted": 0, "inserted": 0, "updated": 0, "skipped": 0, "upserted": 0,
"would_insert": len(to_insert), "would_update": len(to_update), "would_insert": len(to_insert), "would_update": len(to_update),
"would_skip": len(to_skip), "validated": len(clean), "would_skip": len(to_skip), "validated": len(clean),
"would_skip_global": len(skip_global),
"errors": [], "update_existing": update_existing} "errors": [], "update_existing": update_existing}
try: try:
for r in to_insert: for r in to_insert:
db.add(Subject(subject_code=r["subject_code"], subject_name=r["subject_name"], db.add(Subject(subject_code=r["subject_code"], subject_name=r["subject_name"],
parent_code=r["parent_code"], level=r["level"], parent_code=r["parent_code"], level=r["level"],
category=r["category"], is_active=r["is_active"])) category=r["category"], is_active=r["is_active"],
entity_id=entity_id)) # 账套维度:落本会话账套(不再写全局 NULL 行)
for r in to_update: for r in to_update:
db.query(Subject).filter(Subject.subject_code == r["subject_code"]).update({ db.query(Subject).filter(
Subject.subject_code == r["subject_code"],
Subject.entity_id == entity_id, # 只更新本账套私有行(全局科目不被他账套改写)
).update({
Subject.subject_name: r["subject_name"], Subject.subject_name: r["subject_name"],
Subject.parent_code: r["parent_code"], Subject.parent_code: r["parent_code"],
Subject.level: r["level"], Subject.level: r["level"],
@@ -1250,11 +1281,13 @@ def bot_subjects_upsert(
if after != before + len(to_insert): if after != before + len(to_insert):
logger.warning(f"subjects UPSERT 行数口径差异: before={before} after={after} " logger.warning(f"subjects UPSERT 行数口径差异: before={before} after={after} "
f"inserted={len(to_insert)}") f"inserted={len(to_insert)}")
logger.info(f"subjects UPSERT: bot={bot.get('name')} inserted={len(to_insert)} " logger.info(f"subjects UPSERT: bot={bot.get('name')} entity={entity_id} inserted={len(to_insert)} "
f"updated={len(to_update)} skipped={len(to_skip)} before={before} after={after}") f"updated={len(to_update)} skipped={len(to_skip)} before={before} after={after}")
return {"ok": True, "dry_run": False, "applied": True, "table": "subjects", return {"ok": True, "dry_run": False, "applied": True, "table": "subjects",
"entity_id": entity_id,
"before": before, "after": after, "before": before, "after": after,
"inserted": after - before, "updated": len(to_update), "skipped": len(to_skip), "inserted": after - before, "updated": len(to_update), "skipped": len(to_skip),
"skipped_global": len(skip_global),
"upserted": (after - before) + len(to_update), "errors": [], "upserted": (after - before) + len(to_update), "errors": [],
"update_existing": update_existing} "update_existing": update_existing}
+11 -3
View File
@@ -107,7 +107,11 @@ def _query_items(
items: list[dict] = [] items: list[dict] = []
if not data_type or data_type == "subject": if not data_type or data_type == "subject":
q = db.query(Subject).filter(Subject.is_active == 1) # 账套分层(2026-09-23):全局标准准则科目 ∪ 本账套私有明细(与 KPI 侧同一账套口径)
q = db.query(Subject).filter(
Subject.is_active == 1,
(Subject.entity_id.is_(None)) | (Subject.entity_id == entity_id),
)
if data_level: if data_level:
q = q.filter(Subject.data_level == data_level) q = q.filter(Subject.data_level == data_level)
if important == 1: if important == 1:
@@ -181,9 +185,13 @@ def get_stats(
current_user = Depends(require_auth), current_user = Depends(require_auth),
): ):
"""数据分类分级统计概览(评估前准备看板)""" """数据分类分级统计概览(评估前准备看板)"""
subject_total = db.query(Subject).filter(Subject.is_active == 1).count() subject_total = db.query(Subject).filter(
Subject.is_active == 1,
(Subject.entity_id.is_(None)) | (Subject.entity_id == entity_id),
).count()
subject_marked = db.query(Subject).filter( subject_marked = db.query(Subject).filter(
Subject.is_active == 1, Subject.important_flag == 1 Subject.is_active == 1, Subject.important_flag == 1,
(Subject.entity_id.is_(None)) | (Subject.entity_id == entity_id),
).count() ).count()
kpi_total = db.query(KPIDefinition).filter( kpi_total = db.query(KPIDefinition).filter(
KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id
+11 -1
View File
@@ -45,12 +45,22 @@ def trace_ontology(
) )
# 第3层: 每个 KPI 依赖的科目 (kpi_subject_map) # 第3层: 每个 KPI 依赖的科目 (kpi_subject_map)
# ⚠️ 护栏(2026-09-23 账套维度收口批 ②,零结构改动、零读取逻辑改动):
# 该表是唯一一张「语义上必然账套相关、却无账套列」的表 —— **账套维度经 `kpi_id` 继承**
# (`kpi_definitions.entity_id`,本函数先按 `Objective.entity_id` 过滤目标、再按 kpi_id 取映射)。
# **禁止新增按 `subject_code` 反查的读口,禁止按 `subject_code` 批量写的入口**:
# 科目码在标准准则体系里两账套同义(同一 code 可属不同账套),按科目反查/批量写立刻跨账套。
subject_by_code = { subject_by_code = {
s.subject_code: s.subject_name s.subject_code: s.subject_name
for s in db.query(Subject).filter(Subject.is_active == 1).all() for s in db.query(Subject).filter(
Subject.is_active == 1,
# 账套分层(2026-09-23 subjects 加 entity_id 列):全局标准科目 ∪ 本账套私有明细
(Subject.entity_id.is_(None)) | (Subject.entity_id == entity_id),
).all()
} }
kpi_layer = [] kpi_layer = []
for link, kpi in links: for link, kpi in links:
# kpi_id 已由上方 Objective.entity_id 过滤链继承账套维度(见护栏注释),故此处不再按科目口径过滤
maps = ( maps = (
db.query(KPISubjectMap) db.query(KPISubjectMap)
.filter(KPISubjectMap.kpi_id == kpi.id) .filter(KPISubjectMap.kpi_id == kpi.id)
+13 -2
View File
@@ -5,6 +5,7 @@ from sqlalchemy import func
from typing import Optional, List from typing import Optional, List
from app.database import get_db from app.database import get_db
from app.auth_middleware import require_role from app.auth_middleware import require_role
from app.deps import get_entity_id
from app.models import Subject from app.models import Subject
import logging import logging
@@ -20,9 +21,19 @@ def list_subjects(
category: Optional[str] = Query(None, description="新30号准则分类过滤"), category: Optional[str] = Query(None, description="新30号准则分类过滤"),
keyword: Optional[str] = Query(None, description="科目名称/编码搜索"), keyword: Optional[str] = Query(None, description="科目名称/编码搜索"),
db: Session = Depends(get_db), db: Session = Depends(get_db),
entity_id: int = Depends(get_entity_id),
): ):
"""科目列表 — 支持新30号准则分类筛选""" """科目列表 — 支持新30号准则分类筛选
query = db.query(Subject).filter(Subject.is_active == 1)
账套分层(2026-09-23 `subjects` 加 `entity_id` 列):返回**全局标准准则科目**
(`entity_id IS NULL`)∪ **本账套私有明细**(`entity_id = 会话账套`)。
行为变更(预期):博海会话不再看到酣客智享通明细(`100106 酒窖备用金` / `100108 友赞商户` 等)。
返回体字段不删不加必填(前端契约不变)。
"""
query = db.query(Subject).filter(
Subject.is_active == 1,
(Subject.entity_id.is_(None)) | (Subject.entity_id == entity_id), # 分层读取口径
)
if category: if category:
query = query.filter(Subject.new_standard_category == category) query = query.filter(Subject.new_standard_category == category)
+18 -5
View File
@@ -8,6 +8,7 @@ from datetime import datetime
from app.database import get_db from app.database import get_db
from app.auth_middleware import require_role from app.auth_middleware import require_role
from app.deps import get_entity_id
from app.models import KPITemplate, KPIDefinition, OperationLog from app.models import KPITemplate, KPIDefinition, OperationLog
from app.api.kpi_governance import validate_kpi_payload, kpi_issues_message from app.api.kpi_governance import validate_kpi_payload, kpi_issues_message
@@ -107,8 +108,16 @@ def delete_template(template_id: int, db: Session = Depends(get_db), user=WRITE_
@router.post("/{template_id}/instantiate") @router.post("/{template_id}/instantiate")
def instantiate_template(template_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES): def instantiate_template(template_id: int, data: dict, db: Session = Depends(get_db),
"""从模板实例化创建KPI,复制模板快照到kpi_definitions""" user=WRITE_ROLES, entity_id: int = Depends(get_entity_id)):
"""从模板实例化创建KPI,复制模板快照到kpi_definitions
账套维度(2026-09-23 账套维度收口批 ①):新 KPI 显式挂**会话账套**,不做静默兜底。
历史缺陷:本构造点未传 `entity_id` ⇒ 模型 `default=1` 把任意账套实例化出的 KPI
静默挂到酣客(e1),与「建预算未写账套」同源。
判重口径:`kpi_code` 唯一性按账套判定(库内 `uk_entity_kpi(UNIQUE entity_id,kpi_code)`);
全表判重会让博海无法建与酣客同码的 KPI(反之则放过同账套重复)。
"""
t = db.query(KPITemplate).filter(KPITemplate.id == template_id).first() t = db.query(KPITemplate).filter(KPITemplate.id == template_id).first()
if not t: if not t:
raise HTTPException(404, "模板不存在") raise HTTPException(404, "模板不存在")
@@ -116,10 +125,13 @@ def instantiate_template(template_id: int, data: dict, db: Session = Depends(get
kpi_code = data.get("kpi_code", t.kpi_code) kpi_code = data.get("kpi_code", t.kpi_code)
kpi_name = data.get("kpi_name", t.kpi_name) kpi_name = data.get("kpi_name", t.kpi_name)
# 检查编码唯一性 # 检查编码唯一性(按账套收紧:仅本账套内判重)
existing = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == kpi_code).first() existing = db.query(KPIDefinition).filter(
KPIDefinition.kpi_code == kpi_code,
KPIDefinition.entity_id == entity_id,
).first()
if existing: if existing:
raise HTTPException(400, f"KPI编码 {kpi_code} 已存在,请修改") raise HTTPException(400, f"KPI编码 {kpi_code} 在本账套已存在,请修改")
# 数据治理:入库必检(规则1) + 编码规范(规则3) # 数据治理:入库必检(规则1) + 编码规范(规则3)
merged = { merged = {
@@ -136,6 +148,7 @@ def instantiate_template(template_id: int, data: dict, db: Session = Depends(get
raise HTTPException(422, detail={"message": "数据校验不通过", "errors": kpi_issues_message(gov_issues)}) raise HTTPException(422, detail={"message": "数据校验不通过", "errors": kpi_issues_message(gov_issues)})
kpi = KPIDefinition( kpi = KPIDefinition(
entity_id=entity_id, # 账套维度:显式挂会话账套(不得依赖模型 default=1)
template_id=t.id, template_id=t.id,
is_system=0, # 从模板实例化的KPI不是系统预置 is_system=0, # 从模板实例化的KPI不是系统预置
kpi_code=kpi_code, kpi_code=kpi_code,
+38 -5
View File
@@ -66,12 +66,22 @@ class StrategicMap(Base):
class KPIDefinition(Base): class KPIDefinition(Base):
"""KPI字典""" """KPI字典
账套维度:`kpi_code` 唯一性为 **`(entity_id, kpi_code)` 复合**(生产库 `uk_entity_kpi`,
2026-09-23 账套维度收口批与模型对齐——原模型写 `unique=True`(全表唯一)与现网 DDL 不一致,
会让 e1/e2 无法各建同码 KPI,也让 SQLite 测试库建不出真实形态)。
"""
__tablename__ = "kpi_definitions" __tablename__ = "kpi_definitions"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
entity_id = Column(Integer, default=1, comment="企业ID") entity_id = Column(Integer, default=1, comment="企业ID")
map_id = Column(Integer, ForeignKey("strategic_maps.id"), nullable=True, comment="关联战略地图") map_id = Column(Integer, ForeignKey("strategic_maps.id"), nullable=True, comment="关联战略地图")
kpi_code = Column(String(50), unique=True, nullable=False, comment="KPI编码") # 2026-09-23 账套维度收口批:生产库 kpi_definitions 有这两列、模型缺失 ⇒
# `instantiate_template` 的 `KPIDefinition(template_id=…, is_system=…)` 必抛
# TypeError(HTTP 500),「从模板实例化」实际不可用。按现网 DDL 补齐(模型↔库对齐)。
template_id = Column(Integer, nullable=True, comment="来源模板ID (kpi_templates.id)")
is_system = Column(Integer, default=0, comment="是否系统预置模板实例化(1=是)")
kpi_code = Column(String(50), nullable=False, comment="KPI编码(唯一性=(entity_id, kpi_code))")
kpi_name = Column(String(200), nullable=False, comment="KPI名称") kpi_name = Column(String(200), nullable=False, comment="KPI名称")
dimension = Column(String(50), comment="所属维度: finance/customer/process/learning") dimension = Column(String(50), comment="所属维度: finance/customer/process/learning")
objective = Column(String(200), comment="关联战略目标") objective = Column(String(200), comment="关联战略目标")
@@ -104,6 +114,9 @@ class KPIDefinition(Base):
created_by = Column(Integer, nullable=True) created_by = Column(Integer, nullable=True)
created_at = Column(DateTime, server_default=func.now()) created_at = Column(DateTime, server_default=func.now())
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
__table_args__ = (
UniqueConstraint("entity_id", "kpi_code", name="uk_entity_kpi"),
)
class KPIValue(Base): class KPIValue(Base):
@@ -558,9 +571,16 @@ class ReportHistory(Base):
class AnalysisResult(Base): class AnalysisResult(Base):
"""财务Bot分析结论 — 带置信度评分""" """财务Bot分析结论 — 带置信度评分
账套维度(2026-09-23 账套化,任务 0-P1-project-20260923-entity-dimension-convergence §三④):
`entity_id IS NULL` = 历史未归属行(存量 2 行)——**不参与任何账套过滤**,
读取端点单独登记(`unattributed`),删除端点一律 404(不猜账套、不引入 403 探测面)。
"""
__tablename__ = "analysis_results" __tablename__ = "analysis_results"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
entity_id = Column(Integer, nullable=True, index=True,
comment="企业ID(NULL=历史未归属行,不参与账套过滤)")
period = Column(String(20), nullable=False, comment="期间 YYYY-MM") period = Column(String(20), nullable=False, comment="期间 YYYY-MM")
conclusion = Column(String(1000), nullable=False, comment="分析结论") conclusion = Column(String(1000), nullable=False, comment="分析结论")
confidence = Column(Integer, nullable=False, comment="置信度 0-100") confidence = Column(Integer, nullable=False, comment="置信度 0-100")
@@ -608,10 +628,20 @@ class ScenarioSuggestion(Base):
# ============================================================ # ============================================================
class Subject(Base): class Subject(Base):
"""会计科目 — 新30号准则分类""" """会计科目 — 新30号准则分类
账套维度(2026-09-23 `subjects` 分层,任务 0-P1-project-20260923-entity-dimension-convergence):
`entity_id IS NULL` = 全局标准准则科目(两账套共享)
`entity_id = N` = 账套 N 私有明细(如酣客智享通 `remark='zxt-sync'` 明细)
唯一性 = `(entity_id, subject_code)`:不同账套允许同码;同一账套内同码仍拒绝(应用层 UPSERT 判重)。
⚠️ MySQL 唯一索引把 NULL 视为互不相同 ⇒ 全局行的同码重复只能靠写入口判重兜住,
新增「全局科目」写入路径时必须显式查重(不可依赖索引)。
"""
__tablename__ = "subjects" __tablename__ = "subjects"
id = Column(Integer, primary_key=True, index=True) id = Column(Integer, primary_key=True, index=True)
subject_code = Column(String(20), nullable=False, unique=True, comment="科目编码") entity_id = Column(Integer, nullable=True, index=True,
comment="企业ID(NULL=全局标准准则科目 / N=账套私有明细)")
subject_code = Column(String(20), nullable=False, comment="科目编码(唯一性=(entity_id, subject_code))")
subject_name = Column(String(200), nullable=False, comment="科目名称") subject_name = Column(String(200), nullable=False, comment="科目名称")
parent_code = Column(String(20), nullable=True, comment="上级科目编码") parent_code = Column(String(20), nullable=True, comment="上级科目编码")
level = Column(Integer, default=1, comment="科目级别 1-4") level = Column(Integer, default=1, comment="科目级别 1-4")
@@ -624,6 +654,9 @@ class Subject(Base):
remark = Column(String(500), nullable=True, comment="备注") remark = Column(String(500), nullable=True, comment="备注")
created_at = Column(DateTime, server_default=func.now()) created_at = Column(DateTime, server_default=func.now())
updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())
__table_args__ = (
UniqueConstraint("entity_id", "subject_code", name="uk_entity_subject"),
)
class VoucherDetail(Base): class VoucherDetail(Base):
+2 -1
View File
@@ -652,7 +652,8 @@ def unmapped_subjects(db, entity_id: int, period: str, basis: str, items: Dict[s
else: else:
q = q.filter(VoucherDetail.period == period) q = q.filter(VoucherDetail.period == period)
rows = apply_voucher_guard(db, q, VoucherDetail).group_by(VoucherDetail.subject_code).all() rows = apply_voucher_guard(db, q, VoucherDetail).group_by(VoucherDetail.subject_code).all()
cats = {c: cat for c, cat in db.query(Subject.subject_code, Subject.category).all()} cats = {c: cat for c, cat in db.query(Subject.subject_code, Subject.category)
.filter((Subject.entity_id.is_(None)) | (Subject.entity_id == entity_id)).all()}
except Exception: # noqa: BLE001 except Exception: # noqa: BLE001
return [] return []
+7
View File
@@ -85,6 +85,13 @@ def _subject_patterns(db: Session, kpi_id: int) -> list:
优先用已推导的取数映射 kpi_value_sources.filter_rule.subject_codes —— 那是把 优先用已推导的取数映射 kpi_value_sources.filter_rule.subject_codes —— 那是把
kpi_subject_map(新准则 6xxx) 机械桥接到 voucher_details 实际科目(5xxx/1xxx/2xxx) 的结果; kpi_subject_map(新准则 6xxx) 机械桥接到 voucher_details 实际科目(5xxx/1xxx/2xxx) 的结果;
无映射时退回 kpi_subject_map 原始科目(新老科目体系不一致时明细多为 0)。 无映射时退回 kpi_subject_map 原始科目(新老科目体系不一致时明细多为 0)。
⚠️ kpi_subject_map 护栏(2026-09-23 账套维度收口批 ②,零结构/逻辑改动):
该表是唯一一张「语义上必然账套相关、却无账套列」的表 —— **账套维度经 `kpi_id` 继承**
(调用方拿到的 kpi_id 来自已按 `entity_id` 过滤的 KPI,本函数两处查询**全部按 `kpi_id`**,
从不按 `subject_code` 反查)。**禁止新增按 `subject_code` 反查的读口,**
**禁止按 `subject_code` 批量写的入口**:同一 `subject_code` 可同时属多个账套(标准准则科目全局共享),
按科目反查或批量写会立刻跨账套串数。
""" """
from app.models import KPIValueSource, KPISubjectMap from app.models import KPIValueSource, KPISubjectMap
@@ -200,6 +200,11 @@ def main():
code_name, names, codes = load_voucher_index(db) code_name, names, codes = load_voucher_index(db)
subject_dict = load_subject_dict(db) subject_dict = load_subject_dict(db)
kpis = {k.id: k for k in db.query(KPIDefinition).filter(KPIDefinition.entity_id == ENTITY_ID).all()} kpis = {k.id: k for k in db.query(KPIDefinition).filter(KPIDefinition.entity_id == ENTITY_ID).all()}
# ⚠️ kpi_subject_map 护栏(2026-09-23 账套维度收口批 ②,零逻辑改动):
# 该表无账套列 —— **账套维度经 `kpi_id` 继承**。此处全表取映射后由 `kpis`(已按
# `entity_id == ENTITY_ID` 过滤)逐行滤掉非本账套 KPI ⇒ 现状安全。
# **禁止新增按 `subject_code` 反查的读口,禁止按 `subject_code` 批量写的入口**:
# 同一科目码可同时属多个账套(标准准则科目全局共享),按科目反查/批量写立刻跨账套。
rows = db.query(KPISubjectMap).order_by(KPISubjectMap.kpi_id, KPISubjectMap.id).all() rows = db.query(KPISubjectMap).order_by(KPISubjectMap.kpi_id, KPISubjectMap.id).all()
grouped = defaultdict(list) grouped = defaultdict(list)
for r in rows: for r in rows:
@@ -0,0 +1,203 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""`analysis_results` 账套化 DDL — 2026-09-23
任务:`0-P1-project-20260923-entity-dimension-convergence-fullstack` §三④
(裁定件 `project-entity-dimension-ruling-20260923` §3.2「读侧先止血、写侧随 DDL」)
背景
----
`analysis_results` 只有 2 行(`id=1 period=2026-06 kpi_code=TEST001` /
`id=2 period=2026-07 kpi_code=NULL`,均 2026-07-29 创建),**无账套列**;
`GET/POST/DELETE /api/cma/analysis/result*` 三端点原无 `get_entity_id`
⇒ 双 token 返回体**逐字节相同**(跨账套读,页面 `AnalysisConfidence.vue` 已上线)。
本脚本只做 DDL(加列 + 索引),不做任何数据订正:
· 存量 2 行**保持 `entity_id = NULL`** —— 归属存疑,按裁定件交 DAMA 取证 / 财务确认后另办;
· 代码侧口径(`app/api/analysis_results.py`):`NULL` 行 = 无归属,进 `unattributed` 登记池,
不参与任何账套过滤;列落地后读侧自动从 `kpi_code JOIN` 止血口径切到列口径。
用法
----
venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py --check
CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py \\
--apply --allow-prod
CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py \\
--down --allow-prod --yes
venv/bin/python3 scripts/migrate_analysis_results_entity_id_20260923.py --show-create
生产保护(生产资产派单纪律):目标库名 == 'cma' 时 `--apply/--down` 一律拒绝;
须 `--allow-prod` **且** `CMA_DDL_APPROVED=1`(= 已拿到 ops 审批单)才放行;
`--check/--show-create` 为只读,不受限但会打标。
"""
from __future__ import annotations
import argparse
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sqlalchemy import create_engine, inspect, text # noqa: E402
TABLE = "analysis_results"
COLUMN = "entity_id"
NEW_INDEX = "ix_analysis_results_entity_id"
PROD_DB_NAMES = ("cma",)
DDL_ADD_COLUMN = (
f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL "
"COMMENT '企业ID(NULL=历史未归属行,不参与账套过滤)' AFTER id"
)
DDL_DROP_COLUMN = f"ALTER TABLE {TABLE} DROP COLUMN {COLUMN}"
def is_sqlite(engine) -> bool:
"""沙箱演练(SQLite)与生产(MySQL)的索引/列 DDL 方言差异"""
return engine.dialect.name == "sqlite"
def stmt_add_column(engine) -> str:
"""生产形态 = MySQL(带 COMMENT / AFTER);沙箱 = SQLite 精简形态(语法不支持二者)"""
if is_sqlite(engine):
return f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL"
return DDL_ADD_COLUMN
def stmt_add_index(engine, name: str, cols: list) -> str:
cols_s = ", ".join(cols)
if is_sqlite(engine):
return f"CREATE INDEX {name} ON {TABLE} ({cols_s})"
return f"ALTER TABLE {TABLE} ADD INDEX {name} ({cols_s})"
def stmt_drop_index(engine, name: str) -> str:
return f"DROP INDEX {name}" if is_sqlite(engine) else f"ALTER TABLE {TABLE} DROP INDEX {name}"
def db_name_of(engine) -> str:
try:
return engine.url.database or ""
except Exception:
return ""
def state(engine) -> dict:
insp = inspect(engine)
tables = set(insp.get_table_names())
out = {"table_exists": TABLE in tables, "has_column": False, "indexes": [], "rows": []}
if not out["table_exists"]:
return out
out["has_column"] = COLUMN in {c["name"] for c in insp.get_columns(TABLE)}
out["indexes"] = sorted({i["name"] for i in insp.get_indexes(TABLE)})
with engine.connect() as conn:
out["total"] = conn.execute(text(f"SELECT COUNT(*) FROM {TABLE}")).scalar()
if out["has_column"]:
rows = conn.execute(text(
f"SELECT id, period, kpi_code, IFNULL({COLUMN},'NULL') FROM {TABLE} ORDER BY id")).fetchall()
else:
rows = conn.execute(text(
f"SELECT id, period, kpi_code, 'NO_COLUMN' FROM {TABLE} ORDER BY id")).fetchall()
out["rows"] = [[r[0], r[1], r[2], str(r[3])] for r in rows]
return out
def print_check(st: dict) -> None:
print("== 只读预检(inspector / SELECT,零写入)==")
print(f" table_exists={st['table_exists']} has_column={st['has_column']} total={st.get('total')}")
print(f" indexes={st['indexes']}")
for r in st.get("rows", []):
print(f" id={r[0]} period={r[1]} kpi_code={r[2]} entity_id={r[3]}")
def do_apply(engine) -> int:
st = state(engine)
if not st["table_exists"]:
print("!!! 目标库没有 analysis_results 表,拒绝 --apply")
return 3
done = []
with engine.begin() as conn:
if not st["has_column"]:
add_col = stmt_add_column(engine)
conn.execute(text(add_col))
done.append(add_col)
cur = state(engine)
if NEW_INDEX not in cur["indexes"]:
add_idx = stmt_add_index(engine, NEW_INDEX, [COLUMN])
conn.execute(text(add_idx))
done.append(add_idx)
print("== APPLY 完成,执行语句:")
for s in done:
print(" " + s)
if not done:
print(" (无语句执行:已是目标形态,幂等)")
print_check(state(engine))
return 0
def do_down(engine, yes: bool) -> int:
st = state(engine)
if not st["has_column"]:
print("== DOWN 跳过:entity_id 列不存在(已是回滚态)")
return 0
assigned = [r for r in st.get("rows", []) if r[3] not in ("NULL", "NO_COLUMN")]
if assigned and not yes:
print(f"!!! 已有 {len(assigned)} 行带账套归属 {assigned}:回滚会丢失归属,须显式 --yes")
return 3
with engine.begin() as conn:
if NEW_INDEX in st["indexes"]:
conn.execute(text(stmt_drop_index(engine, NEW_INDEX)))
conn.execute(text(DDL_DROP_COLUMN))
print("== DOWN 完成:删除 entity_id 索引 + 列")
print_check(state(engine))
return 0
def main():
ap = argparse.ArgumentParser(description="analysis_results 账套化 DDL")
g = ap.add_mutually_exclusive_group()
g.add_argument("--check", action="store_true", help="只读预检(默认)")
g.add_argument("--apply", action="store_true", help="执行 DDL(幂等)")
g.add_argument("--down", action="store_true", help="回滚:删索引 + 删列")
ap.add_argument("--show-create", action="store_true", help="打印 SHOW CREATE TABLE")
ap.add_argument("--url", default=None, help="目标库 SQLAlchemy URL(默认 backend/.env)")
ap.add_argument("--allow-prod", action="store_true", help="生产库放行开关(须 CMA_DDL_APPROVED=1)")
ap.add_argument("--yes", action="store_true", help="--down 有账套归属行时的确认开关")
args = ap.parse_args()
if args.url:
engine = create_engine(args.url, echo=False)
else:
from app.database import get_engine
engine = get_engine()
dbname = db_name_of(engine)
print(f"目标库: {dbname or '(未知)'} | dialect={engine.dialect.name} | host={engine.url.host or 'socket'}")
is_prod = dbname in PROD_DB_NAMES
if (args.apply or args.down) and is_prod:
if not args.allow_prod or os.getenv("CMA_DDL_APPROVED") != "1":
print(f"!!! 拒绝执行:目标库是生产库 {dbname}。")
print("!!! 生产 DDL 须 ops 审批后另派执行(本单边界:不得直连生产写入)。")
print("!!! 放行方式:--allow-prod 且 export CMA_DDL_APPROVED=1(持有审批单)。")
return 3
if args.apply:
return do_apply(engine)
if args.down:
return do_down(engine, args.yes)
st = state(engine)
print_check(st)
if is_prod:
print(" [注意] 目标是生产库:以上仅为只读预检,未做任何写入。")
if args.show_create:
if engine.dialect.name != "mysql":
print("== SHOW CREATE TABLE 跳过(非 MySQL 目标)==")
else:
with engine.connect() as conn:
print(conn.execute(text(f"SHOW CREATE TABLE {TABLE}")).fetchone()[1])
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,297 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""`subjects` 账套分层 DDL + 存量回填 — 2026-09-23
任务:`0-P1-project-20260923-entity-dimension-convergence-fullstack` §三③
(裁定件 `project-entity-dimension-ruling-20260923` §3.1「方案 A:分层」)
背景
----
`subjects` 207 行混装且 `subject_code` **全局唯一**(`uk_subject_code`):
· 43 行标准准则科目(1001 库存现金 …,`remark IS NULL`)= 天然全局
· 164 行 `remark='zxt-sync'` 酣客(智享通)企业明细 = 账套私有
双 token 实测两账套拿到**同一份 207 行** ⇒ 博海页面出现酣客的
`100106 酒窖备用金` / `100107 现金备用卡(唐)` / `100108 友赞商户`(结构性串配置)。
分层口径(`knowledge_articles` 同款已验模式)
------------------------------------------
`entity_id IS NULL` = 全局标准准则科目(两账套共享)
`entity_id = N` = 账套 N 私有明细
唯一键 `(subject_code)` → `(entity_id, subject_code)`
⚠️ MySQL 把 NULL 视为互不相同的索引值 ⇒ 「全局行同码重复」索引拦不住,
由写入口(`POST /api/cma/bot/subjects` UPSERT 判重)兜住,不可依赖索引。
动作(四段可独立跑,互不隐含)
----------------------------
--check 只读预检(默认):列/索引现状 + 行数分布 + 唯一键重复检查,零写入
--apply DDL:ADD COLUMN entity_id INT NULL + INDEX + UNIQUE(entity_id,subject_code)
− DROP INDEX uk_subject_code(幂等:已存在的步骤自动跳过)
--backfill 存量回填(**须 --evidence <json>**,与 DDL 分开放行):
`remark='zxt-sync'` 的 164 行 → `--target-entity`(默认 1);其余保持 NULL
--down 回滚:重建 `uk_subject_code`(先断言无同码重复,fail-closed)→ 删新唯一键 → 删列
--show-create 打印 SHOW CREATE TABLE subjects 原文(验收证据)
生产保护(生产资产派单纪律)
--------------------------
目标库名 == 'cma' 时 `--apply/--backfill/--down` 一律拒绝;须 `--allow-prod` **且**
环境变量 `CMA_DDL_APPROVED=1`(= 已拿到 ops 审批单)才放行。`--check/--show-create` 只读不受限。
用法
----
venv/bin/python3 scripts/migrate_subjects_entity_id_20260923.py --check
CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_subjects_entity_id_20260923.py \\
--apply --allow-prod
CMA_DDL_APPROVED=1 venv/bin/python3 scripts/migrate_subjects_entity_id_20260923.py \\
--backfill --allow-prod --evidence /path/to/evidence.json --target-entity 1
"""
from __future__ import annotations
import argparse
import json
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from sqlalchemy import create_engine, inspect, text # noqa: E402
TABLE = "subjects"
COLUMN = "entity_id"
NEW_UNIQUE = "uk_entity_subject"
OLD_UNIQUE = "uk_subject_code"
NEW_INDEX = "ix_subjects_entity_id"
PROD_DB_NAMES = ("cma",)
DDL_ADD_COLUMN = (
f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL "
"COMMENT '企业ID(NULL=全局标准准则科目 / N=账套私有明细)' AFTER id"
)
DDL_ADD_UNIQUE = f"ALTER TABLE {TABLE} ADD UNIQUE INDEX {NEW_UNIQUE} ({COLUMN}, subject_code)"
DDL_DROP_OLD_UNIQUE = f"ALTER TABLE {TABLE} DROP INDEX {OLD_UNIQUE}"
DDL_ADD_OLD_UNIQUE = f"ALTER TABLE {TABLE} ADD UNIQUE INDEX {OLD_UNIQUE} (subject_code)"
DDL_DROP_COLUMN = f"ALTER TABLE {TABLE} DROP COLUMN {COLUMN}"
def is_sqlite(engine) -> bool:
"""沙箱演练(SQLite)与生产(MySQL)的 DDL 方言差异:索引/列定义语句形态不同"""
return engine.dialect.name == "sqlite"
def stmt_add_column(engine) -> str:
"""生产形态 = MySQL(带 COMMENT / AFTER);沙箱 = SQLite 精简形态(语法不支持二者)"""
if is_sqlite(engine):
return f"ALTER TABLE {TABLE} ADD COLUMN {COLUMN} INT NULL"
return DDL_ADD_COLUMN
def stmt_add_index(engine, name: str, cols: list, unique: bool = False) -> str:
cols_s = ", ".join(cols)
if is_sqlite(engine):
return f"CREATE {'UNIQUE ' if unique else ''}INDEX {name} ON {TABLE} ({cols_s})"
return f"ALTER TABLE {TABLE} ADD {'UNIQUE ' if unique else ''}INDEX {name} ({cols_s})"
def stmt_drop_index(engine, name: str) -> str:
return f"DROP INDEX {name}" if is_sqlite(engine) else f"ALTER TABLE {TABLE} DROP INDEX {name}"
def db_name_of(engine) -> str:
try:
return engine.url.database or ""
except Exception:
return ""
def state(engine) -> dict:
insp = inspect(engine)
tables = set(insp.get_table_names())
out = {"table_exists": TABLE in tables, "has_column": False,
"indexes": [], "unique_keys": [], "counts": {}}
if not out["table_exists"]:
return out
out["has_column"] = COLUMN in {c["name"] for c in insp.get_columns(TABLE)}
out["indexes"] = sorted({i["name"] for i in insp.get_indexes(TABLE)})
out["unique_keys"] = sorted({uc["name"] for uc in insp.get_unique_constraints(TABLE)})
with engine.connect() as conn:
out["counts"]["total"] = conn.execute(text(f"SELECT COUNT(*) FROM {TABLE}")).scalar()
if out["has_column"]:
rows = conn.execute(text(
f"SELECT IFNULL({COLUMN},'NULL'), COUNT(*) FROM {TABLE} GROUP BY 1 ORDER BY 1")).fetchall()
out["counts"]["by_entity"] = [[str(r[0]), int(r[1])] for r in rows]
rows = conn.execute(text(
f"SELECT IFNULL(remark,'(null)'), COUNT(*) FROM {TABLE} GROUP BY 1 ORDER BY 2 DESC")).fetchall()
out["counts"]["by_remark"] = [[str(r[0]), int(r[1])] for r in rows]
dup_code = conn.execute(text(
f"SELECT COUNT(*) FROM (SELECT subject_code FROM {TABLE} GROUP BY subject_code "
f"HAVING COUNT(*) > 1) t")).scalar()
out["duplicate_subject_code_groups"] = int(dup_code)
if out["has_column"]:
dup_pair = conn.execute(text(
f"SELECT COUNT(*) FROM (SELECT {COLUMN}, subject_code FROM {TABLE} "
f"GROUP BY {COLUMN}, subject_code HAVING COUNT(*) > 1) t")).scalar()
out["duplicate_entity_code_groups"] = int(dup_pair)
return out
def print_check(st: dict) -> None:
print("== 只读预检(inspector / SELECT,零写入)==")
print(f" table_exists={st['table_exists']} has_column={st['has_column']}")
print(f" indexes={st['indexes']}")
print(f" unique_keys={st['unique_keys']}")
print(f" counts={st['counts']}")
print(f" duplicate_subject_code_groups={st.get('duplicate_subject_code_groups')}")
if "duplicate_entity_code_groups" in st:
print(f" duplicate_entity_code_groups={st['duplicate_entity_code_groups']}")
def do_apply(engine) -> int:
st = state(engine)
if not st["table_exists"]:
print("!!! 目标库没有 subjects 表,拒绝 --apply")
return 3
if st.get("duplicate_subject_code_groups"):
print("!!! subjects.subject_code 已有重复组,先处置数据再加/换唯一键")
return 3
done = []
with engine.begin() as conn:
if not st["has_column"]:
add_col = stmt_add_column(engine)
conn.execute(text(add_col))
done.append(add_col)
cur = state(engine)
present = set(cur["indexes"]) | set(cur["unique_keys"])
add_idx = stmt_add_index(engine, NEW_INDEX, [COLUMN])
if NEW_INDEX not in present:
conn.execute(text(add_idx))
done.append(add_idx)
add_uk = stmt_add_index(engine, NEW_UNIQUE, [COLUMN, "subject_code"], unique=True)
if NEW_UNIQUE not in present:
conn.execute(text(add_uk))
done.append(add_uk)
if OLD_UNIQUE in present:
drop_old = stmt_drop_index(engine, OLD_UNIQUE)
conn.execute(text(drop_old))
done.append(drop_old)
print("== APPLY 完成,执行语句:")
for s in done:
print(" " + s)
if not done:
print(" (无语句执行:已是目标形态,幂等)")
print_check(state(engine))
return 0
def do_backfill(engine, evidence_path: str, target: int) -> int:
if not evidence_path or not os.path.exists(evidence_path):
print("!!! --backfill 必须提供 --evidence <json>(DAMA 只读取证 + 财务确认件锚点),拒绝执行")
return 3
ev = json.load(open(evidence_path, encoding="utf-8"))
need = ("dama_evidence", "finance_confirmation", "counts")
missing = [k for k in need if not ev.get(k)]
if missing:
print(f"!!! 取证件缺字段 {missing},拒绝执行(fail-closed)")
return 3
st = state(engine)
if not st["has_column"]:
print("!!! entity_id 列未落地,先跑 --apply")
return 3
by_remark = dict(st["counts"]["by_remark"])
live_private = int(by_remark.get("zxt-sync", 0))
live_global = int(by_remark.get("(null)", 0))
exp = ev["counts"]
if int(exp.get("e1", -1)) != live_private or int(exp.get("global_null", -1)) != live_global:
print(f"!!! 取证件行数锚点与现网不符:evidence e1={exp.get('e1')}/global_null="
f"{exp.get('global_null')} vs live {live_private}/{live_global},拒绝执行")
return 3
with engine.begin() as conn:
n1 = conn.execute(text(
f"UPDATE {TABLE} SET {COLUMN} = :e WHERE remark = 'zxt-sync'"), {"e": target}).rowcount
n0 = conn.execute(text(
f"UPDATE {TABLE} SET {COLUMN} = NULL WHERE remark IS NULL AND {COLUMN} IS NOT NULL")).rowcount
print(f"== BACKFILL 完成:remark='zxt-sync' → entity_id={target} 命中 {n1} 行;"
f"标准准则科目置 NULL {n0} 行(evidence: {evidence_path})")
print_check(state(engine))
return 0
def do_down(engine, yes: bool) -> int:
st = state(engine)
if not st["has_column"]:
print("== DOWN 跳过:entity_id 列不存在(已是回滚态)")
return 0
if st.get("duplicate_subject_code_groups"):
print(f"!!! 已存在同码多行({st['duplicate_subject_code_groups']} 组)⇒ 无法重建全局唯一键 "
f"uk_subject_code,拒绝 --down(fail-closed:先人工处置同码行)")
return 3
if st["counts"]["by_entity"] and len(st["counts"]["by_entity"]) > 1 and not yes:
print(f"!!! 列内已有分账套数据 {st['counts']['by_entity']}:回滚会丢失账套归属,须显式 --yes")
return 3
with engine.begin() as conn:
present = set(st["indexes"]) | set(st["unique_keys"])
if NEW_UNIQUE in present:
conn.execute(text(stmt_drop_index(engine, NEW_UNIQUE)))
cur = state(engine)
present = set(cur["indexes"]) | set(cur["unique_keys"])
if NEW_INDEX in present:
conn.execute(text(stmt_drop_index(engine, NEW_INDEX)))
if OLD_UNIQUE not in present:
conn.execute(text(stmt_add_index(engine, OLD_UNIQUE, ["subject_code"], unique=True)))
conn.execute(text(DDL_DROP_COLUMN))
print("== DOWN 完成:重建 uk_subject_code + 删除 entity_id 列")
print_check(state(engine))
return 0
def main():
ap = argparse.ArgumentParser(description="subjects 账套分层 DDL/回填")
g = ap.add_mutually_exclusive_group()
g.add_argument("--check", action="store_true", help="只读预检(默认)")
g.add_argument("--apply", action="store_true", help="执行 DDL(幂等)")
g.add_argument("--backfill", action="store_true", help="存量回填(须 --evidence)")
g.add_argument("--down", action="store_true", help="回滚")
ap.add_argument("--show-create", action="store_true", help="打印 SHOW CREATE TABLE")
ap.add_argument("--url", default=None, help="目标库 SQLAlchemy URL(默认 backend/.env)")
ap.add_argument("--allow-prod", action="store_true", help="生产库放行开关(须 CMA_DDL_APPROVED=1)")
ap.add_argument("--evidence", default=None, help="--backfill 的取证件 JSON")
ap.add_argument("--target-entity", type=int, default=1, help="--backfill 目标账套(默认 1=酣客)")
ap.add_argument("--yes", action="store_true", help="--down 有分账套数据时的确认开关")
args = ap.parse_args()
if args.url:
engine = create_engine(args.url, echo=False)
else:
from app.database import get_engine
engine = get_engine()
dbname = db_name_of(engine)
print(f"目标库: {dbname or '(未知)'} | dialect={engine.dialect.name} | host={engine.url.host or 'socket'}")
is_prod = dbname in PROD_DB_NAMES
if (args.apply or args.backfill or args.down) and is_prod:
if not args.allow_prod or os.getenv("CMA_DDL_APPROVED") != "1":
print(f"!!! 拒绝执行:目标库是生产库 {dbname}。")
print("!!! 生产 DDL/回填须 ops 审批后另派执行(本单边界:不得直连生产写入)。")
print("!!! 放行方式:--allow-prod 且 export CMA_DDL_APPROVED=1(持有审批单)。")
return 3
if args.apply:
return do_apply(engine)
if args.backfill:
return do_backfill(engine, args.evidence, args.target_entity)
if args.down:
return do_down(engine, args.yes)
st = state(engine)
print_check(st)
if is_prod:
print(" [注意] 目标是生产库:以上仅为只读预检,未做任何写入。")
if args.show_create:
if engine.dialect.name != "mysql":
print("== SHOW CREATE TABLE 跳过(非 MySQL 目标)==")
else:
with engine.connect() as conn:
print(conn.execute(text(f"SHOW CREATE TABLE {TABLE}")).fetchone()[1])
return 0
if __name__ == "__main__":
sys.exit(main())
+4 -4
View File
@@ -554,7 +554,7 @@ class TestSubjectsWrite:
def test_update_existing_switch(self, client: TestClient, db: Session): def test_update_existing_switch(self, client: TestClient, db: Session):
db.add(Subject(subject_code="9301", subject_name="旧名称", level=1, db.add(Subject(subject_code="9301", subject_name="旧名称", level=1,
category="equity", is_active=1)) category="equity", is_active=1, entity_id=1)) # 账套维度:本账套私有行(分层后 NULL=全局标准科目)
db.commit() db.commit()
resp = client.post(self.URL, json={ resp = client.post(self.URL, json={
"subjects": [self._row("9301", "新名称", category="cost", level=2)], "subjects": [self._row("9301", "新名称", category="cost", level=2)],
@@ -603,9 +603,9 @@ class TestSubjectsWrite:
def test_revenue_existing_row_update_existing_true(self, client: TestClient, db: Session): def test_revenue_existing_row_update_existing_true(self, client: TestClient, db: Session):
"""原 400 复现用例:库内既有 revenue 行(6001 主营业务收入)+ update_existing=true""" """原 400 复现用例:库内既有 revenue 行(6001 主营业务收入)+ update_existing=true"""
db.add(Subject(subject_code="6001", subject_name="主营业务收入", level=1, db.add(Subject(subject_code="6001", subject_name="主营业务收入", level=1,
category="revenue", is_active=1)) category="revenue", is_active=1, entity_id=1)) # 本账套私有行(分层口径)
other = Subject(subject_code="6051", subject_name="其他业务收入", level=1, other = Subject(subject_code="6051", subject_name="其他业务收入", level=1,
category="revenue", is_active=1) category="revenue", is_active=1, entity_id=1)
db.add(other) db.add(other)
db.commit() db.commit()
before = db.query(Subject).count() before = db.query(Subject).count()
@@ -625,7 +625,7 @@ class TestSubjectsWrite:
def test_profit_loss_existing_row_update_existing_true(self, client: TestClient, db: Session): def test_profit_loss_existing_row_update_existing_true(self, client: TestClient, db: Session):
db.add(Subject(subject_code="6901", subject_name="终止经营损益", level=1, db.add(Subject(subject_code="6901", subject_name="终止经营损益", level=1,
category="profit_loss", is_active=1)) category="profit_loss", is_active=1, entity_id=1)) # 本账套私有行(分层口径)
db.commit() db.commit()
resp = client.post(self.URL, json={ resp = client.post(self.URL, json={
"subjects": [self._row("6901", "终止经营损益(V2)", category="profit_loss")], "subjects": [self._row("6901", "终止经营损益(V2)", category="profit_loss")],
@@ -0,0 +1,612 @@
"""账套维度收口批 · 2026-09-23 — 四段验收用例
任务:`0-P1-project-20260923-entity-dimension-convergence-fullstack`
(裁定件 `project-entity-dimension-ruling-20260923`)
A 段(零 DDL)
① `kpi_templates` instantiate 构造点显式挂会话账套 + `kpi_code` 判重按账套收紧
② `kpi_subject_map` 3 个读取点护栏注释(静态判据:`grep -n "账套维度经"` ≥3 命中)
③ `analysis_results` 读侧止血(`entity_id` 列未落地时按 `kpi_code` JOIN 判归属)
B 段(需 DDL,代码侧先落)
③ `subjects` 分层读取口径(全局 `entity_id IS NULL` ∪ 本账套私有)+ UPSERT 按账套判重
④ `analysis_results` 加列后的读/写/删三端点账套化
D 段:两个迁移脚本的旧形态/新形态/回填闸门/回滚/生产库守卫(SQLite 沙箱演练)
全部用例只跑内存 SQLite / 临时文件 SQLite(`tests/conftest.py` 机制),
**零生产库 / 零生产 Redis 连接**;迁移脚本用 `--url sqlite:///…` 沙箱,不碰 `cma` 库。
"""
import importlib
import json
import os
import sys
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session
from app import models as models_pkg
from app.api import analysis_results as ar_mod
from app.models import (
AnalysisResult, Entity, KPIDefinition, KPISubjectMap, KPITemplate, OperationLog, Subject,
)
from tests.conftest import TEST_USER_PASSWORD, auth_header, create_test_user
BACKEND = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TEMPLATES = "/api/cma/templates"
ANALYSIS = "/api/cma/analysis"
SUBJECTS = "/api/cma/subjects"
BOT_SUBJECTS = "/api/cma/bot/subjects"
DATA_CLS = "/api/cma/data-classification"
BOT_KEY = {"X-BOT-KEY": "cma-bot-finance-2026"}
MIG_SUBJECTS = "scripts.migrate_subjects_entity_id_20260923"
MIG_ANALYSIS = "scripts.migrate_analysis_results_entity_id_20260923"
# ──────────────────────────────────────────────
# 夹具
# ──────────────────────────────────────────────
def _ensure_entity(db: Session, eid: int, name: str, status: str = "active") -> Entity:
ent = db.query(Entity).filter(Entity.id == eid).first()
if not ent:
ent = Entity(id=eid, name=name, short_name=name, status=status)
db.add(ent)
db.commit()
db.refresh(ent)
return ent
def _seed_two_entities(db: Session):
_ensure_entity(db, 1, "陕西酣客文化传媒")
_ensure_entity(db, 2, "陕西博海网络科技")
def _login_as(client: TestClient, username: str, entity_id: int) -> str:
resp = client.post("/api/cma/auth/login", json={
"username": username, "password": TEST_USER_PASSWORD, "entity_id": entity_id,
})
assert resp.status_code == 200, f"登录 {username}@entity{entity_id} 失败: {resp.text[:300]}"
return resp.json()["token"]
def _two_tokens(client: TestClient, db: Session):
_seed_two_entities(db)
create_test_user(db)
return _login_as(client, "testadmin", 1), _login_as(client, "testadmin", 2)
def _seed_kpi(db: Session, code: str, entity_id: int, name: str = "测试KPI") -> KPIDefinition:
kpi = KPIDefinition(kpi_code=code, kpi_name=name, dimension="finance",
unit="元", target_value=100.0, status="active", entity_id=entity_id)
db.add(kpi)
db.commit()
db.refresh(kpi)
return kpi
def _seed_analysis(db: Session, kpi_code, entity_id, conclusion: str, period: str = "2026-06"):
row = AnalysisResult(period=period, conclusion=conclusion, confidence=70,
kpi_code=kpi_code, entity_id=entity_id)
db.add(row)
db.commit()
db.refresh(row)
return row
def _seed_subject(db: Session, code: str, name: str, entity_id=None, remark=None) -> Subject:
s = Subject(subject_code=code, subject_name=name, entity_id=entity_id, remark=remark,
is_active=1, level=1)
db.add(s)
db.commit()
db.refresh(s)
return s
# ══════════════════════════════════════════════
# ① kpi_templates instantiate 构造点
# ══════════════════════════════════════════════
class TestTemplateInstantiateEntityScope:
def _seed_template(self, db: Session, code: str = "F_TPL_REV") -> KPITemplate:
t = KPITemplate(kpi_code=code, kpi_name="模板_营业收入", dimension="finance",
unit="万元", formula="SUM(收入)", formula_desc="收入合计",
target_value=100.0, is_system=1, usage_count=0)
db.add(t)
db.commit()
db.refresh(t)
return t
def _instantiate(self, client: TestClient, token: str, tid: int, code: str):
return client.post(f"{TEMPLATES}/{tid}/instantiate", headers=auth_header(token),
json={"kpi_code": code, "kpi_name": "实例KPI", "dimension": "finance",
"unit": "万元", "target_value": 100.0})
def test_each_entity_gets_its_own_row(self, client: TestClient, db: Session):
"""e1 / e2 各实例化一次(同码)→ 回库各 1 行且归属正确"""
t1, t2 = _two_tokens(client, db)
tpl = self._seed_template(db, "F_TPL_REV")
r1 = self._instantiate(client, t1, tpl.id, "F_TPL_REV")
assert r1.status_code == 200, r1.text
assert r1.json()["entity_id"] == 1
# e2 允许建同码(uk_entity_kpi 复合唯一;全表判重会让博海建不了同码 KPI)
r2 = self._instantiate(client, t2, tpl.id, "F_TPL_REV")
assert r2.status_code == 200, r2.text
assert r2.json()["entity_id"] == 2
db.expire_all()
rows = db.query(KPIDefinition).filter(KPIDefinition.kpi_code == "F_TPL_REV").all()
assert len(rows) == 2, [r.id for r in rows]
assert sorted(r.entity_id for r in rows) == [1, 2]
def test_same_entity_duplicate_code_rejected(self, client: TestClient, db: Session):
"""同账套内同码 → 400(判重按账套收紧,不放过本账套重复)"""
t1, _ = _two_tokens(client, db)
tpl = self._seed_template(db, "F_TPL_DUP")
assert self._instantiate(client, t1, tpl.id, "F_TPL_DUP").status_code == 200
dup = self._instantiate(client, t1, tpl.id, "F_TPL_DUP")
assert dup.status_code == 400, dup.text
assert "在本账套已存在" in dup.text
db.expire_all()
assert db.query(KPIDefinition).filter(
KPIDefinition.kpi_code == "F_TPL_DUP", KPIDefinition.entity_id == 1).count() == 1
def test_constructor_carries_entity_id_in_source(self):
"""构造点显式传账套(防再次退回模型 default=1 的静默兜底)"""
src = open(os.path.join(BACKEND, "app/api/templates.py"), encoding="utf-8").read()
assert "entity_id=entity_id" in src
assert "KPIDefinition.kpi_code == kpi_code,\n KPIDefinition.entity_id == entity_id," in src
# ══════════════════════════════════════════════
# ② kpi_subject_map 护栏(静态判据)
# ══════════════════════════════════════════════
class TestKpiSubjectMapGuardrail:
GUARD_FILES = ("app/api/ontology.py", "app/utils/alert_attribution.py",
"scripts/kpi_value_source_derive.py")
def test_guard_comment_present_in_three_read_points(self):
hits = {}
for rel in self.GUARD_FILES:
src = open(os.path.join(BACKEND, rel), encoding="utf-8").read()
hits[rel] = src.count("账套维度经")
assert all(v >= 1 for v in hits.values()), hits
assert len(hits) == 3
def test_no_subject_code_reverse_lookup_introduced(self):
"""护栏语义:三个读取点都仍按 kpi_id 取数(不得出现按 subject_code 过滤的 kpi_subject_map 查询)"""
for rel in self.GUARD_FILES:
src = open(os.path.join(BACKEND, rel), encoding="utf-8").read()
assert "KPISubjectMap.subject_code ==" not in src, rel
# ══════════════════════════════════════════════
# ③④ analysis_results 账套化(列口径 + join 止血口径)
# ══════════════════════════════════════════════
class TestAnalysisResultEntityScope:
def _seed(self, db: Session):
_seed_kpi(db, "F_AN_1", 1, "酣客指标")
_seed_kpi(db, "F_AN_2", 2, "博海指标")
_seed_kpi(db, "F_AN_SHARED", 1, "同码A")
_seed_kpi(db, "F_AN_SHARED", 2, "同码B")
r1 = _seed_analysis(db, "F_AN_1", 1, "酣客结论")
r2 = _seed_analysis(db, "F_AN_2", 2, "博海结论")
r3 = _seed_analysis(db, None, None, "无归属(kpi_code 为空)", period="2026-07")
r4 = _seed_analysis(db, "F_AN_SHARED", None, "归属二义(同码跨账套)", period="2026-08")
r5 = _seed_analysis(db, "F_AN_NOPE", None, "库内无该码", period="2026-09")
return r1, r2, r3, r4, r5
def _ids(self, payload, key="results"):
return sorted(x["id"] for x in payload[key])
# ── 列口径(DDL 已落地) ──────────────────────────────
def test_get_two_tokens_intersection_empty(self, client: TestClient, db: Session):
r1, r2, r3, r4, r5 = self._seed(db)
t1, t2 = _two_tokens(client, db)
v1 = client.get(f"{ANALYSIS}/result", headers=auth_header(t1)).json()
v2 = client.get(f"{ANALYSIS}/result", headers=auth_header(t2)).json()
assert self._ids(v1) == [r1.id], v1
assert self._ids(v2) == [r2.id], v2
assert set(self._ids(v1)) & set(self._ids(v2)) == set()
assert v1["entity_id"] == 1 and v2["entity_id"] == 2
assert v1["entity_scope"] == "column"
# 无归属行(kpi_code 空 / 无定义 / 二义)对两账套都不进 results,只在登记池
assert sorted(x["id"] for x in v1["unattributed"]) == sorted([r3.id, r4.id, r5.id])
assert v1["unattributed"] == v2["unattributed"]
assert v1["unattributed_total"] == 3
assert all("结论" not in x for x in v1["unattributed"])
def test_get_join_stopgap_before_ddl(self, client: TestClient, db: Session, monkeypatch):
"""entity_id 列未落地时(DDL 前置)按 kpi_code JOIN 判归属,止血口径与列口径一致"""
r1, r2, r3, r4, r5 = self._seed(db)
t1, t2 = _two_tokens(client, db)
monkeypatch.setattr(ar_mod, "_has_entity_column", lambda db_: False)
v1 = client.get(f"{ANALYSIS}/result", headers=auth_header(t1)).json()
v2 = client.get(f"{ANALYSIS}/result", headers=auth_header(t2)).json()
assert v1["entity_scope"] == "kpi_join"
assert self._ids(v1) == [r1.id]
assert self._ids(v2) == [r2.id]
assert set(self._ids(v1)) & set(self._ids(v2)) == set()
# 二义码(F_AN_SHARED 属两账套)在 join 口径下不猜账套 → 登记池
assert sorted(x["id"] for x in v1["unattributed"]) == sorted([r3.id, r4.id, r5.id])
def test_delete_cross_entity_is_404_and_row_survives(self, client: TestClient, db: Session):
r1, r2, _, _, _ = self._seed(db)
t1, t2 = _two_tokens(client, db)
bad = client.delete(f"{ANALYSIS}/result/{r1.id}", headers=auth_header(t2))
assert bad.status_code == 404, bad.text
db.expire_all()
assert db.query(AnalysisResult).filter(AnalysisResult.id == r1.id).count() == 1, "对端行被删"
# 反向:e1 删 e2 的行同样 404
bad2 = client.delete(f"{ANALYSIS}/result/{r2.id}", headers=auth_header(t1))
assert bad2.status_code == 404
# 正对照:本账套删自己的行 → 200
ok = client.delete(f"{ANALYSIS}/result/{r1.id}", headers=auth_header(t1))
assert ok.status_code == 200, ok.text
db.expire_all()
assert db.query(AnalysisResult).filter(AnalysisResult.id == r1.id).count() == 0
def test_delete_unattributed_is_404(self, client: TestClient, db: Session):
_, _, r3, _, _ = self._seed(db)
t1, _ = _two_tokens(client, db)
resp = client.delete(f"{ANALYSIS}/result/{r3.id}", headers=auth_header(t1))
assert resp.status_code == 404
db.expire_all()
assert db.query(AnalysisResult).filter(AnalysisResult.id == r3.id).count() == 1
def test_post_writes_session_entity(self, client: TestClient, db: Session):
_two_tokens(client, db)
_seed_kpi(db, "F_AN_1", 1)
_seed_kpi(db, "F_AN_2", 2)
t1, t2 = _login_as(client, "testadmin", 1), _login_as(client, "testadmin", 2)
p2 = client.post(f"{ANALYSIS}/result", headers=auth_header(t2),
json={"period": "2026-09", "conclusion": "博海写入", "kpi_code": "F_AN_2"})
assert p2.status_code == 200, p2.text
assert p2.json()["entity_id"] == 2
assert p2.json()["entity_persisted"] == "column"
p1 = client.post(f"{ANALYSIS}/result", headers=auth_header(t1),
json={"period": "2026-09", "conclusion": "酣客写入", "kpi_code": "F_AN_1"})
assert p1.status_code == 200
db.expire_all()
new2 = db.query(AnalysisResult).filter(AnalysisResult.id == p2.json()["id"]).first()
assert new2.entity_id == 2, "create 未写会话账套"
# 交叉可见性:各自的 GET 只看到自己那一行
v1 = client.get(f"{ANALYSIS}/result", headers=auth_header(t1)).json()
v2 = client.get(f"{ANALYSIS}/result", headers=auth_header(t2)).json()
assert self._ids(v1) == [p1.json()["id"]]
assert self._ids(v2) == [p2.json()["id"]]
def test_post_pre_ddl_logs_entity_and_does_not_guess_column(self, client: TestClient, db: Session,
monkeypatch):
"""列未落地时:账套只落 OperationLog.detail,行内 entity_id 保持 NULL(不猜列)"""
_two_tokens(client, db)
_seed_kpi(db, "F_AN_1", 1)
t1 = _login_as(client, "testadmin", 1)
monkeypatch.setattr(ar_mod, "_has_entity_column", lambda db_: False)
resp = client.post(f"{ANALYSIS}/result", headers=auth_header(t1),
json={"period": "2026-09", "conclusion": "过渡期写入", "kpi_code": "F_AN_1"})
assert resp.status_code == 200, resp.text
assert resp.json()["entity_persisted"] == "operation_log"
db.expire_all()
row = db.query(AnalysisResult).filter(AnalysisResult.id == resp.json()["id"]).first()
assert row.entity_id is None
logs = db.query(OperationLog).filter(OperationLog.target_type == "analysis_result").all()
assert len(logs) == 1, [l.id for l in logs]
detail = logs[0].detail
detail = json.loads(detail) if isinstance(detail, str) else detail
assert detail["entity_id"] == 1
# ══════════════════════════════════════════════
# ③ subjects 分层(读取口径 + UPSERT 判重)
# ══════════════════════════════════════════════
class TestSubjectsLayering:
def _seed(self, db: Session):
_seed_subject(db, "1001", "库存现金", None) # 全局标准
_seed_subject(db, "1002", "银行存款", None) # 全局标准
_seed_subject(db, "100106", "酒窖备用金", 1, remark="zxt-sync") # 酣客私有
_seed_subject(db, "100107", "现金备用卡(唐)", 1, remark="zxt-sync")
_seed_subject(db, "200201", "博海专户", 2) # 博海私有
_seed_subject(db, "100199", "酣客同码科目", 1) # 同码不同账套
_seed_subject(db, "100199", "博海同码科目", 2)
def test_get_subjects_union_scope(self, client: TestClient, db: Session):
self._seed(db)
t1, t2 = _two_tokens(client, db)
v1 = client.get(SUBJECTS, headers=auth_header(t1)).json()
v2 = client.get(SUBJECTS, headers=auth_header(t2)).json()
c1 = {d["subject_code"] for d in v1["data"]}
c2 = {d["subject_code"] for d in v2["data"]}
assert {"1001", "1002", "100106", "100107", "100199"} <= c1
assert "200201" not in c1, "博海私有明细泄漏给酣客"
assert {"1001", "1002", "200201", "100199"} <= c2
assert "100106" not in c2 and "100107" not in c2, "酣客明细泄漏给博海"
# 同码不同账套 → 各看自己的名字
n1 = {d["subject_code"]: d["subject_name"] for d in v1["data"]}
n2 = {d["subject_code"]: d["subject_name"] for d in v2["data"]}
assert n1["100199"] == "酣客同码科目"
assert n2["100199"] == "博海同码科目"
# 全局标准科目两账套都在
assert {"1001", "1002"} <= c1 and {"1001", "1002"} <= c2
def test_data_classification_inventory_and_stats_scoped(self, client: TestClient, db: Session):
self._seed(db)
t1, t2 = _two_tokens(client, db)
inv1 = client.get(f"{DATA_CLS}/inventory", params={"data_type": "subject"},
headers=auth_header(t1)).json()
inv2 = client.get(f"{DATA_CLS}/inventory", params={"data_type": "subject"},
headers=auth_header(t2)).json()
codes = lambda v: {(i.get("subject_code") or i.get("code")) for i in (v.get("items") or v.get("data") or [])}
assert "100106" in codes(inv1) and "100106" not in codes(inv2)
assert "200201" in codes(inv2) and "200201" not in codes(inv1)
s1 = client.get(f"{DATA_CLS}/stats", headers=auth_header(t1)).json()
s2 = client.get(f"{DATA_CLS}/stats", headers=auth_header(t2)).json()
assert s1["subjects"]["total"] == 5, s1 # 2 全局 + 3 酣客私有
assert s2["subjects"]["total"] == 4, s2 # 2 全局 + 2 博海私有
assert s1["total"] == 5 and s2["total"] == 4 # 无 KPI 数据时总数为科目数
def test_bot_upsert_per_entity_judgement(self, client: TestClient, db: Session):
"""UPSERT 判重按 (entity_id, subject_code):不同账套可各有同码,同账套内才 skip"""
self._seed(db)
_seed_two_entities(db)
body = {"subjects": [{"subject_code": "300301", "subject_name": "博海新专户",
"parent_code": None, "level": 1, "category": "asset", "is_active": 1}],
"update_existing": False}
h2 = dict(BOT_KEY); h2["X-Entity-Id"] = "2"
h1 = dict(BOT_KEY); h1["X-Entity-Id"] = "1"
r2 = client.post(BOT_SUBJECTS, headers=h2, json=body)
assert r2.status_code == 200, r2.text
assert r2.json()["inserted"] == 1 and r2.json()["entity_id"] == 2
r1 = client.post(BOT_SUBJECTS, headers=h1, json=body)
assert r1.status_code == 200, r1.text
assert r1.json()["inserted"] == 1, "同码但不同账套应可插入(按账套判重)"
# 同账套重复 → skip
again = client.post(BOT_SUBJECTS, headers=h1, json=body)
assert again.json()["inserted"] == 0 and again.json()["skipped"] == 1
db.expire_all()
rows = db.query(Subject).filter(Subject.subject_code == "300301").all()
assert sorted(r.entity_id for r in rows) == [1, 2]
# 全局科目(entity_id IS NULL)不被他账套 update 改写
s = client.post(BOT_SUBJECTS, headers=h1,
json={"subjects": [{"subject_code": "1001", "subject_name": "库存现金改名",
"parent_code": None, "level": 1,
"category": "asset", "is_active": 1}],
"update_existing": True})
assert s.json()["updated"] == 0, "全局标准科目不该被计入 updated(分层后只更新本账套私有行)"
assert s.json()["skipped_global"] == 1
db.expire_all()
glob = db.query(Subject).filter(Subject.subject_code == "1001",
Subject.entity_id.is_(None)).first()
assert glob.subject_name == "库存现金", "全局标准科目被账套 UPSERT 改写"
def test_bot_upsert_global_row_is_never_rewritten_nor_duplicated(self, client: TestClient, db: Session):
"""全局标准科目(entity_id IS NULL):同码既不被账套 UPSERT 改写,也不插本账套副本
回归保护(本单 A③ 修过的一处缺陷):原先 `existing` 把全局行也算「已存在」,
但 UPDATE 只匹配 `entity_id == 会话账套` ⇒ 对全局行「报 updated=1 却零改动」(响应失真)。
分层口径:全局行 → 计入 `skipped_global`,永不改写、永不重复插入。
"""
self._seed(db)
_seed_two_entities(db)
h1 = dict(BOT_KEY); h1["X-Entity-Id"] = "1"
need = {"subject_code": "1001", "subject_name": "库存现金(改名)", "parent_code": None,
"level": 1, "category": "asset", "is_active": 1}
for update_existing in (False, True):
r = client.post(BOT_SUBJECTS, headers=h1,
json={"subjects": [need], "update_existing": update_existing})
assert r.status_code == 200, r.text
d = r.json()
assert (d["inserted"], d["updated"], d["skipped_global"]) == (0, 0, 1), d
db.expire_all()
rows = db.query(Subject).filter(Subject.subject_code == "1001").all()
assert len(rows) == 1, "全局标准科目被插入了账套副本(同码重复)"
assert rows[0].entity_id is None and rows[0].subject_name == "库存现金", "全局标准科目被改写"
# ── dry_run 口径与真实执行一致 ──
def test_bot_upsert_dry_run_reports_global_skip(self, client: TestClient, db: Session):
self._seed(db)
_seed_two_entities(db)
h1 = dict(BOT_KEY); h1["X-Entity-Id"] = "1"
r = client.post(BOT_SUBJECTS, headers=h1,
json={"subjects": [{"subject_code": "1001", "subject_name": "库存现金",
"parent_code": None, "level": 1,
"category": "asset", "is_active": 1}],
"update_existing": True, "dry_run": True})
assert r.status_code == 200, r.text
d = r.json()
assert d["dry_run"] is True and d["would_update"] == 0 and d["would_skip_global"] == 1, d
def test_read_sites_carry_layer_filter(self):
"""四个读取口都必须带分层口径(防回退成整表读)"""
for rel in ("app/api/subjects.py", "app/api/data_classification.py",
"app/services/report_reclass.py", "app/api/ontology.py"):
src = open(os.path.join(BACKEND, rel), encoding="utf-8").read()
assert "Subject.entity_id.is_(None)" in src, rel
# ══════════════════════════════════════════════
# 迁移脚本:沙箱演练(SQLite 旧形态 → apply → backfill → down)
# ══════════════════════════════════════════════
OLD_SUBJECTS_DDL = """
CREATE TABLE subjects (
id INTEGER PRIMARY KEY AUTOINCREMENT,
subject_code VARCHAR(20) NOT NULL UNIQUE,
subject_name VARCHAR(200) NOT NULL,
parent_code VARCHAR(20),
level INTEGER DEFAULT 1,
category VARCHAR(50),
new_standard_category VARCHAR(20),
important_flag INTEGER DEFAULT 0,
data_level VARCHAR(20),
data_category VARCHAR(50),
is_active INTEGER DEFAULT 1,
remark VARCHAR(500),
created_at DATETIME,
updated_at DATETIME
)
"""
OLD_ANALYSIS_DDL = """
CREATE TABLE analysis_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
period VARCHAR(20) NOT NULL,
conclusion VARCHAR(1000) NOT NULL,
confidence INTEGER NOT NULL,
data_source VARCHAR(500),
calculation_logic VARCHAR(1000),
comparable_benchmark VARCHAR(500),
limitations VARCHAR(1000),
has_actual INTEGER DEFAULT 0,
has_target INTEGER DEFAULT 0,
has_trend INTEGER DEFAULT 0,
has_review INTEGER DEFAULT 0,
kpi_code VARCHAR(50),
kpi_name VARCHAR(200),
created_at DATETIME
)
"""
class TestMigrationScriptsSandbox:
def _run(self, mod, argv):
old = sys.argv
sys.argv = ["migrate"] + argv
try:
return mod.main()
finally:
sys.argv = old
@pytest.fixture
def subj_env(self, tmp_path):
sys.path.insert(0, BACKEND)
mod = importlib.import_module(MIG_SUBJECTS)
eng = create_engine(f"sqlite:///{tmp_path/'subj.db'}")
with eng.begin() as conn:
conn.execute(text(OLD_SUBJECTS_DDL))
for i in range(164):
conn.execute(text("INSERT INTO subjects (subject_code, subject_name, remark) "
f"VALUES ('9{i:05d}', '酣客明细{i}', 'zxt-sync')"))
for i in range(43):
conn.execute(text("INSERT INTO subjects (subject_code, subject_name) "
f"VALUES ('1{i:03d}', '标准科目{i}')"))
return mod, eng, tmp_path
@pytest.fixture
def analysis_env(self, tmp_path):
sys.path.insert(0, BACKEND)
mod = importlib.import_module(MIG_ANALYSIS)
eng = create_engine(f"sqlite:///{tmp_path/'ar.db'}")
with eng.begin() as conn:
conn.execute(text(OLD_ANALYSIS_DDL))
conn.execute(text("INSERT INTO analysis_results (period, conclusion, confidence, kpi_code) "
"VALUES ('2026-06','测试',70,'TEST001')"))
conn.execute(text("INSERT INTO analysis_results (period, conclusion, confidence, kpi_code) "
"VALUES ('2026-07','测试2',70,NULL)"))
return mod, eng, tmp_path
def _cols(self, eng, table):
return {r[1] for r in eng.connect().execute(text(f"PRAGMA table_info({table})")).fetchall()}
# ── subjects ──────────────────────────────────
def test_subjects_check_is_readonly(self, subj_env):
mod, eng, _ = subj_env
assert self._run(mod, ["--check", "--url", str(eng.url)]) == 0
assert "entity_id" not in self._cols(eng, "subjects")
with eng.connect() as conn:
assert conn.execute(text("SELECT COUNT(*) FROM subjects")).scalar() == 207
def test_subjects_apply_backfill_down_roundtrip(self, subj_env):
mod, eng, tmp = subj_env
url = str(eng.url)
assert self._run(mod, ["--apply", "--url", url]) == 0
assert "entity_id" in self._cols(eng, "subjects")
# 二跑幂等
assert self._run(mod, ["--apply", "--url", url]) == 0
with eng.connect() as conn:
names = {r[0] for r in conn.execute(
text("SELECT name FROM sqlite_master WHERE type='index'")).fetchall()}
assert "uk_entity_subject" in names and "ix_subjects_entity_id" in names
assert conn.execute(text("SELECT COUNT(*) FROM subjects WHERE entity_id IS NOT NULL")).scalar() == 0
# 回填闸门:无取证件 / 取证件行数不符 → 拒绝(fail-closed)
assert self._run(mod, ["--backfill", "--url", url]) == 3
bad = tmp / "ev_bad.json"
bad.write_text(json.dumps({"dama_evidence": "x", "finance_confirmation": "y",
"counts": {"e1": 1, "global_null": 43}}), encoding="utf-8")
assert self._run(mod, ["--backfill", "--url", url, "--evidence", str(bad)]) == 3
with eng.connect() as conn:
assert conn.execute(text("SELECT COUNT(*) FROM subjects WHERE entity_id=1")).scalar() == 0
# 正确取证件 → 164 行归 e1、43 行留 NULL
good = tmp / "ev_ok.json"
good.write_text(json.dumps({"dama_evidence": "dama-...out", "finance_confirmation": "任总...",
"counts": {"e1": 164, "global_null": 43}}), encoding="utf-8")
assert self._run(mod, ["--backfill", "--url", url, "--evidence", str(good),
"--target-entity", "1"]) == 0
with eng.connect() as conn:
dist = dict(conn.execute(text(
"SELECT IFNULL(entity_id,'NULL'), COUNT(*) FROM subjects GROUP BY 1")).fetchall())
assert dist == {1: 164, "NULL": 43}, dist
# 回滚:有分账套数据须 --yes;回滚后列消失、code 唯一性恢复
assert self._run(mod, ["--down", "--url", url]) == 3
assert self._run(mod, ["--down", "--url", url, "--yes"]) == 0
assert "entity_id" not in self._cols(eng, "subjects")
with eng.connect() as conn:
assert conn.execute(text("SELECT COUNT(*) FROM subjects")).scalar() == 207
def test_subjects_prod_name_guard(self, subj_env, tmp_path, monkeypatch):
mod, _, _ = subj_env
monkeypatch.chdir(tmp_path)
assert self._run(mod, ["--apply", "--url", "sqlite:///cma"]) == 3 # 未放行
monkeypatch.setenv("CMA_DDL_APPROVED", "1")
assert self._run(mod, ["--apply", "--url", "sqlite:///cma", "--allow-prod"]) == 3 # 表不存在 → 3
# ── analysis_results ──────────────────────────
def test_analysis_apply_and_down(self, analysis_env):
mod, eng, _ = analysis_env
url = str(eng.url)
assert self._run(mod, ["--check", "--url", url]) == 0
assert self._run(mod, ["--apply", "--url", url]) == 0
assert "entity_id" in self._cols(eng, "analysis_results")
assert self._run(mod, ["--apply", "--url", url]) == 0 # 幂等
with eng.connect() as conn:
names = {r[0] for r in conn.execute(
text("SELECT name FROM sqlite_master WHERE type='index'")).fetchall()}
assert "ix_analysis_results_entity_id" in names
# 存量 2 行不被代猜账套
assert conn.execute(text(
"SELECT COUNT(*) FROM analysis_results WHERE entity_id IS NOT NULL")).scalar() == 0
assert conn.execute(text("SELECT COUNT(*) FROM analysis_results")).scalar() == 2
assert self._run(mod, ["--down", "--url", url, "--yes"]) == 0
assert "entity_id" not in self._cols(eng, "analysis_results")
def test_analysis_prod_name_guard(self, analysis_env, tmp_path, monkeypatch):
mod, _, _ = analysis_env
monkeypatch.chdir(tmp_path)
assert self._run(mod, ["--apply", "--url", "sqlite:///cma"]) == 3