diff --git a/backend/app/api/analysis_results.py b/backend/app/api/analysis_results.py index fc40f051..2646fe33 100644 --- a/backend/app/api/analysis_results.py +++ b/backend/app/api/analysis_results.py @@ -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": "已删除"} diff --git a/backend/app/api/bot_bridge.py b/backend/app/api/bot_bridge.py index 2d561f80..1d2ae970 100644 --- a/backend/app/api/bot_bridge.py +++ b/backend/app/api/bot_bridge.py @@ -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} diff --git a/backend/app/api/data_classification.py b/backend/app/api/data_classification.py index 042e75ae..baca88e3 100644 --- a/backend/app/api/data_classification.py +++ b/backend/app/api/data_classification.py @@ -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 diff --git a/backend/app/api/ontology.py b/backend/app/api/ontology.py index ecd2dd99..85e372a4 100644 --- a/backend/app/api/ontology.py +++ b/backend/app/api/ontology.py @@ -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) diff --git a/backend/app/api/subjects.py b/backend/app/api/subjects.py index 9d0b53ae..3d10133a 100644 --- a/backend/app/api/subjects.py +++ b/backend/app/api/subjects.py @@ -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) diff --git a/backend/app/api/templates.py b/backend/app/api/templates.py index 16ef4a32..9079cc81 100644 --- a/backend/app/api/templates.py +++ b/backend/app/api/templates.py @@ -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, diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py index ca08d573..837b91b4 100644 --- a/backend/app/models/__init__.py +++ b/backend/app/models/__init__.py @@ -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): diff --git a/backend/app/services/report_reclass.py b/backend/app/services/report_reclass.py index 620254ab..26517d43 100644 --- a/backend/app/services/report_reclass.py +++ b/backend/app/services/report_reclass.py @@ -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 [] diff --git a/backend/app/utils/alert_attribution.py b/backend/app/utils/alert_attribution.py index 567465de..1ab9a16c 100644 --- a/backend/app/utils/alert_attribution.py +++ b/backend/app/utils/alert_attribution.py @@ -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 diff --git a/backend/scripts/kpi_value_source_derive.py b/backend/scripts/kpi_value_source_derive.py index 5851518d..10e0ca62 100644 --- a/backend/scripts/kpi_value_source_derive.py +++ b/backend/scripts/kpi_value_source_derive.py @@ -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: diff --git a/backend/scripts/migrate_analysis_results_entity_id_20260923.py b/backend/scripts/migrate_analysis_results_entity_id_20260923.py new file mode 100644 index 00000000..84247ca8 --- /dev/null +++ b/backend/scripts/migrate_analysis_results_entity_id_20260923.py @@ -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()) diff --git a/backend/scripts/migrate_subjects_entity_id_20260923.py b/backend/scripts/migrate_subjects_entity_id_20260923.py new file mode 100644 index 00000000..f7d7b9ec --- /dev/null +++ b/backend/scripts/migrate_subjects_entity_id_20260923.py @@ -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 **，与 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 （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()) diff --git a/backend/tests/test_bot_bridge.py b/backend/tests/test_bot_bridge.py index bf51890d..28282353 100644 --- a/backend/tests/test_bot_bridge.py +++ b/backend/tests/test_bot_bridge.py @@ -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")], diff --git a/backend/tests/test_entity_dimension_convergence_20260923.py b/backend/tests/test_entity_dimension_convergence_20260923.py new file mode 100644 index 00000000..063df1b4 --- /dev/null +++ b/backend/tests/test_entity_dimension_convergence_20260923.py @@ -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