feat(entity-dim): 账套维度收口批 — subjects/analysis_results 加列 + 分析结论账套化 + 模板实例化构造点补账套
ci/woodpecker/push/main Pipeline failed
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
@@ -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 sqlalchemy.orm import Session
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy import func, inspect
|
||||
from app.database import get_db
|
||||
from app.deps import get_entity_id
|
||||
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 datetime import datetime
|
||||
|
||||
@@ -12,6 +27,95 @@ router = APIRouter(prefix="/api/cma/analysis", tags=["置信度评分"],
|
||||
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:
|
||||
"""基于数据完整度自动计算置信度"""
|
||||
@@ -31,8 +135,14 @@ async def get_analysis_results(
|
||||
period: Optional[str] = Query(None, description="期间 YYYY-MM"),
|
||||
kpi_code: Optional[str] = Query(None, description="KPI编码"),
|
||||
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())
|
||||
|
||||
if period:
|
||||
@@ -41,36 +151,34 @@ async def get_analysis_results(
|
||||
query = query.filter(AnalysisResult.kpi_code == kpi_code)
|
||||
|
||||
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 {
|
||||
"total": len(results),
|
||||
"total": len(mine),
|
||||
"period": period,
|
||||
"results": [
|
||||
{
|
||||
"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,
|
||||
}
|
||||
for r in results
|
||||
],
|
||||
"entity_id": entity_id,
|
||||
"entity_scope": "column" if _has_entity_column(db) else "kpi_join",
|
||||
"unattributed_total": len(unattributed),
|
||||
"unattributed": [_unattributed_dict(r) for r in unattributed],
|
||||
"results": [_to_dict(r) for r in mine],
|
||||
}
|
||||
|
||||
|
||||
@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:
|
||||
data = {}
|
||||
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)
|
||||
|
||||
has_column = _has_entity_column(db)
|
||||
result = AnalysisResult(
|
||||
period=period,
|
||||
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_review=1 if has_review else 0,
|
||||
)
|
||||
if has_column:
|
||||
result.entity_id = entity_id
|
||||
db.add(result)
|
||||
db.commit()
|
||||
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 {
|
||||
"id": result.id,
|
||||
"period": result.period,
|
||||
@@ -117,6 +241,8 @@ def create_analysis_result(data: dict = None, db: Session = Depends(get_db)):
|
||||
"可比基准": result.comparable_benchmark,
|
||||
"局限": result.limitations,
|
||||
"confidence_score": result.confidence,
|
||||
"entity_id": entity_id,
|
||||
"entity_persisted": "column" if has_column else "operation_log",
|
||||
"message": "分析结论已保存",
|
||||
}
|
||||
|
||||
@@ -125,11 +251,16 @@ def create_analysis_result(data: dict = None, db: Session = Depends(get_db)):
|
||||
async def delete_analysis_result(
|
||||
result_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
entity_id: int = Depends(get_entity_id),
|
||||
):
|
||||
"""删除分析结论"""
|
||||
"""删除分析结论 — 归属校验：非本账套 或 无归属一律 **404**（不引入 403，避免探测面）"""
|
||||
result = db.query(AnalysisResult).filter(AnalysisResult.id == result_id).first()
|
||||
if not result:
|
||||
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.commit()
|
||||
return {"message": "已删除"}
|
||||
|
||||
@@ -1036,13 +1036,16 @@ def _norm_subj_active(v):
|
||||
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 且零落库），返回标准化行 [{字段}]
|
||||
|
||||
规则（任务 dama-cma-subjects-write-endpoint-20260912）：
|
||||
subject_code 非空且 ≤_SUBJ_CODE_MAXLEN（=20，DB varchar(20) 列宽）+ 批次内去重；subject_name 非空（DB NOT NULL）且 ≤200；
|
||||
level ∈ {1,2,3}（缺省 1）；parent_code 为空或存在于本表（或本批次内，支持层级同批导入）；
|
||||
category 限枚举；is_active ∈ {0,1}（缺省 1）。
|
||||
|
||||
账套维度（2026-09-23 `subjects` 分层）：parent_code 存在性只认
|
||||
**全局科目（`entity_id IS NULL`）∪ 本会话账套私有科目**——他账套私有父科目不构成合法父级。
|
||||
"""
|
||||
if not isinstance(rows, list) or not rows:
|
||||
raise HTTPException(400, '请求体必须是非空行数组: {"subjects": [{...}]}')
|
||||
@@ -1134,11 +1137,13 @@ def _validate_subject_rows(rows, db):
|
||||
if not bad:
|
||||
clean.append(item)
|
||||
|
||||
# parent_code 存在性：库内（本表）∪ 本批次
|
||||
# parent_code 存在性：库内（本表，按账套分层）∪ 本批次
|
||||
unknown_parents = sorted(parents - set(seen.keys()))
|
||||
if unknown_parents:
|
||||
found = {c for (c,) in db.query(Subject.subject_code)
|
||||
.filter(Subject.subject_code.in_(unknown_parents)).all()}
|
||||
pq = db.query(Subject.subject_code).filter(Subject.subject_code.in_(unknown_parents))
|
||||
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]
|
||||
if missing:
|
||||
errs.append(f"parent_code 不存在于 subjects 表且不在本批次内: {missing}")
|
||||
@@ -1181,50 +1186,76 @@ def bot_subjects_upsert(
|
||||
bot: dict = Depends(verify_bot_key),
|
||||
payload: dict = Depends(_read_subjects_body),
|
||||
db: Session = Depends(get_db),
|
||||
entity_id: int = Depends(get_entity_id),
|
||||
):
|
||||
"""会计科目写入（subjects 幂等 UPSERT，单事务；绝不 DELETE）
|
||||
|
||||
body: {"subjects": [{"subject_code","subject_name","parent_code","level","category","is_active"}],
|
||||
"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 且库内零变化
|
||||
- 已存在行：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 → 只校验与回显，零写入
|
||||
返回: {"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)
|
||||
update_existing = payload.get("update_existing", False)
|
||||
codes = [r["subject_code"] for r in clean]
|
||||
|
||||
existing = set()
|
||||
existing = set() # 本会话「可见」的同码 = 全局标准科目（NULL）∪ 本账套私有行
|
||||
global_only = set() # 只以全局行（entity_id IS NULL）存在的同码：可见，但**不许被账套 UPSERT 改写**
|
||||
if codes:
|
||||
existing = {c for (c,) in db.query(Subject.subject_code)
|
||||
.filter(Subject.subject_code.in_(codes)).all()}
|
||||
rows = (db.query(Subject.subject_code, Subject.entity_id)
|
||||
.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]
|
||||
existed = [r for r in clean if r["subject_code"] in existing]
|
||||
to_update = existed if update_existing else []
|
||||
to_skip = [] if update_existing else existed
|
||||
# 可被本账套更新的只有「本账套私有行」；全局标准科目（NULL）一律不动（跨账套共享口径）
|
||||
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
|
||||
|
||||
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_skip={len(to_skip)}")
|
||||
return {"ok": True, "dry_run": True, "applied": False, "table": "subjects",
|
||||
"entity_id": entity_id,
|
||||
"before": before, "after": before,
|
||||
"inserted": 0, "updated": 0, "skipped": 0, "upserted": 0,
|
||||
"would_insert": len(to_insert), "would_update": len(to_update),
|
||||
"would_skip": len(to_skip), "validated": len(clean),
|
||||
"would_skip_global": len(skip_global),
|
||||
"errors": [], "update_existing": update_existing}
|
||||
|
||||
try:
|
||||
for r in to_insert:
|
||||
db.add(Subject(subject_code=r["subject_code"], subject_name=r["subject_name"],
|
||||
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:
|
||||
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.parent_code: r["parent_code"],
|
||||
Subject.level: r["level"],
|
||||
@@ -1250,11 +1281,13 @@ def bot_subjects_upsert(
|
||||
if after != before + len(to_insert):
|
||||
logger.warning(f"subjects UPSERT 行数口径差异: before={before} after={after} "
|
||||
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}")
|
||||
return {"ok": True, "dry_run": False, "applied": True, "table": "subjects",
|
||||
"entity_id": entity_id,
|
||||
"before": before, "after": after,
|
||||
"inserted": after - before, "updated": len(to_update), "skipped": len(to_skip),
|
||||
"skipped_global": len(skip_global),
|
||||
"upserted": (after - before) + len(to_update), "errors": [],
|
||||
"update_existing": update_existing}
|
||||
|
||||
|
||||
@@ -107,7 +107,11 @@ def _query_items(
|
||||
items: list[dict] = []
|
||||
|
||||
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:
|
||||
q = q.filter(Subject.data_level == data_level)
|
||||
if important == 1:
|
||||
@@ -181,9 +185,13 @@ def get_stats(
|
||||
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.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()
|
||||
kpi_total = db.query(KPIDefinition).filter(
|
||||
KPIDefinition.status == "active", KPIDefinition.entity_id == entity_id
|
||||
|
||||
@@ -45,12 +45,22 @@ def trace_ontology(
|
||||
)
|
||||
|
||||
# 第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 = {
|
||||
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 = []
|
||||
for link, kpi in links:
|
||||
# kpi_id 已由上方 Objective.entity_id 过滤链继承账套维度（见护栏注释），故此处不再按科目口径过滤
|
||||
maps = (
|
||||
db.query(KPISubjectMap)
|
||||
.filter(KPISubjectMap.kpi_id == kpi.id)
|
||||
|
||||
@@ -5,6 +5,7 @@ from sqlalchemy import func
|
||||
from typing import Optional, List
|
||||
from app.database import get_db
|
||||
from app.auth_middleware import require_role
|
||||
from app.deps import get_entity_id
|
||||
from app.models import Subject
|
||||
import logging
|
||||
|
||||
@@ -20,9 +21,19 @@ def list_subjects(
|
||||
category: Optional[str] = Query(None, description="新30号准则分类过滤"),
|
||||
keyword: Optional[str] = Query(None, description="科目名称/编码搜索"),
|
||||
db: Session = Depends(get_db),
|
||||
entity_id: int = Depends(get_entity_id),
|
||||
):
|
||||
"""科目列表 — 支持新30号准则分类筛选"""
|
||||
query = db.query(Subject).filter(Subject.is_active == 1)
|
||||
"""科目列表 — 支持新30号准则分类筛选
|
||||
|
||||
账套分层（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:
|
||||
query = query.filter(Subject.new_standard_category == category)
|
||||
|
||||
@@ -8,6 +8,7 @@ from datetime import datetime
|
||||
|
||||
from app.database import get_db
|
||||
from app.auth_middleware import require_role
|
||||
from app.deps import get_entity_id
|
||||
from app.models import KPITemplate, KPIDefinition, OperationLog
|
||||
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")
|
||||
def instantiate_template(template_id: int, data: dict, db: Session = Depends(get_db), user=WRITE_ROLES):
|
||||
"""从模板实例化创建KPI，复制模板快照到kpi_definitions"""
|
||||
def instantiate_template(template_id: int, data: dict, db: Session = Depends(get_db),
|
||||
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()
|
||||
if not t:
|
||||
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_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:
|
||||
raise HTTPException(400, f"KPI编码 {kpi_code} 已存在，请修改")
|
||||
raise HTTPException(400, f"KPI编码 {kpi_code} 在本账套已存在，请修改")
|
||||
|
||||
# 数据治理：入库必检(规则1) + 编码规范(规则3)
|
||||
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)})
|
||||
|
||||
kpi = KPIDefinition(
|
||||
entity_id=entity_id, # 账套维度：显式挂会话账套（不得依赖模型 default=1）
|
||||
template_id=t.id,
|
||||
is_system=0, # 从模板实例化的KPI不是系统预置
|
||||
kpi_code=kpi_code,
|
||||
|
||||
@@ -66,12 +66,22 @@ class StrategicMap(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"
|
||||
id = Column(Integer, primary_key=True, index=True)
|
||||
entity_id = Column(Integer, default=1, comment="企业ID")
|
||||
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名称")
|
||||
dimension = Column(String(50), comment="所属维度: finance/customer/process/learning")
|
||||
objective = Column(String(200), comment="关联战略目标")
|
||||
@@ -104,6 +114,9 @@ class KPIDefinition(Base):
|
||||
created_by = Column(Integer, nullable=True)
|
||||
created_at = Column(DateTime, server_default=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):
|
||||
@@ -558,9 +571,16 @@ class ReportHistory(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"
|
||||
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")
|
||||
conclusion = Column(String(1000), nullable=False, comment="分析结论")
|
||||
confidence = Column(Integer, nullable=False, comment="置信度 0-100")
|
||||
@@ -608,10 +628,20 @@ class ScenarioSuggestion(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"
|
||||
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="科目名称")
|
||||
parent_code = Column(String(20), nullable=True, comment="上级科目编码")
|
||||
level = Column(Integer, default=1, comment="科目级别 1-4")
|
||||
@@ -624,6 +654,9 @@ class Subject(Base):
|
||||
remark = Column(String(500), nullable=True, comment="备注")
|
||||
created_at = Column(DateTime, server_default=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):
|
||||
|
||||
@@ -652,7 +652,8 @@ def unmapped_subjects(db, entity_id: int, period: str, basis: str, items: Dict[s
|
||||
else:
|
||||
q = q.filter(VoucherDetail.period == period)
|
||||
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
|
||||
return []
|
||||
|
||||
|
||||
@@ -85,6 +85,13 @@ def _subject_patterns(db: Session, kpi_id: int) -> list:
|
||||
优先用已推导的取数映射 kpi_value_sources.filter_rule.subject_codes —— 那是把
|
||||
kpi_subject_map(新准则 6xxx) 机械桥接到 voucher_details 实际科目(5xxx/1xxx/2xxx) 的结果；
|
||||
无映射时退回 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
|
||||
|
||||
|
||||
@@ -200,6 +200,11 @@ def main():
|
||||
code_name, names, codes = load_voucher_index(db)
|
||||
subject_dict = load_subject_dict(db)
|
||||
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()
|
||||
grouped = defaultdict(list)
|
||||
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())
|
||||
@@ -554,7 +554,7 @@ class TestSubjectsWrite:
|
||||
|
||||
def test_update_existing_switch(self, client: TestClient, db: Session):
|
||||
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()
|
||||
resp = client.post(self.URL, json={
|
||||
"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):
|
||||
"""原 400 复现用例：库内既有 revenue 行（6001 主营业务收入）+ update_existing=true"""
|
||||
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,
|
||||
category="revenue", is_active=1)
|
||||
category="revenue", is_active=1, entity_id=1)
|
||||
db.add(other)
|
||||
db.commit()
|
||||
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):
|
||||
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()
|
||||
resp = client.post(self.URL, json={
|
||||
"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
|
||||
Reference in New Issue
Block a user