From 193061cf4d86fd418c0d0ef4bbbbb814e6a9bac7 Mon Sep 17 00:00:00 2001 From: Hermes CI Fix Date: Tue, 8 Sep 2026 11:19:38 +0800 Subject: [PATCH] =?UTF-8?q?fix(data):=20=E6=95=B0=E6=8D=AE=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E6=A8=A1=E5=9D=97=E8=84=B1=E8=8A=82=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=20=E2=80=94=20Excel=E5=AF=BC=E5=85=A5/sources=E7=AB=AF?= =?UTF-8?q?=E7=82=B9=20entity=20=E9=9A=94=E7=A6=BB=20+=20ERP=E7=9B=B4?= =?UTF-8?q?=E8=BF=9E=E9=80=9A=E9=81=93=E8=AF=9A=E5=AE=9E=E5=8C=96(410)=20+?= =?UTF-8?q?=20=E7=9C=9F=E5=AE=9Ezxt=E9=93=BE=E8=B7=AF=E5=B1=95=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - import_excel/import_excel_smart KPI字典按 entity_id 过滤（防跨账套匹配串写，P0安全） - import_excel_smart 自动建KPI编码防冲突（_next_code 循环取号） - sources list/create/update/delete 全部按 entity_id 隔离（create 写入当前账套） - sync-kpis + admin/erp-sync 下线返回410（erp-gateway DB凭据2026-08-31已删，真实通道=zxt智享通注入） - 新增 GET /data/channel-status：展示真实zxt取数通道状态+遗留ERP数据源清单（诚实化） - DataManage.vue：数据源tab顶部展示真实通道(最近注入period/时间/行数/落盘目录mtime)、ERP遗留源标注'非当前通道'、实际取数映射(kpi_value_sources)表格 - 新增 test_data_entity_isolation_20260908.py（8条：双账套导入隔离/CRUD隔离/sync-kpis 410/channel-status） --- backend/app/api/data.py | 178 +++++++++----- backend/app/main.py | 23 +- .../test_data_entity_isolation_20260908.py | 230 ++++++++++++++++++ frontend/src/api/index.ts | 1 + frontend/src/views/DataManage.vue | 118 ++++++++- 5 files changed, 471 insertions(+), 79 deletions(-) create mode 100644 backend/tests/test_data_entity_isolation_20260908.py diff --git a/backend/app/api/data.py b/backend/app/api/data.py index ba107084..abe2dbb2 100644 --- a/backend/app/api/data.py +++ b/backend/app/api/data.py @@ -1,6 +1,6 @@ """数据对接 API""" import pandas as pd -import io, json, hashlib, re +import io, json, hashlib, re, os from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File, Form from sqlalchemy.orm import Session @@ -11,6 +11,17 @@ from app.deps import get_entity_id from app.auth_middleware import require_auth, require_role from app.models import KPIValue, DataSourceConfig, OperationLog, KPIDefinition +# ERP 直连同步通道停用标记（2026-08-31 OPS 安全加固：erp-gateway DB 凭据已删除） +# 陕西酣客财务原始数据的真实通道 = DAMA 读智享通网页 → 落盘 → 财务Bot 注入（source_type='zxt-sync'） +# 任何 ERP API 直连同步入口（/data/sync-kpis、/admin/erp-sync）一律拒绝执行，防止写 estimated 假数据 +ERP_DIRECT_CHANNEL_DISABLED = True +ERP_DIRECT_CHANNEL_REASON = ("ERP直连同步通道已于2026-08-31停用（erp-gateway 容器 DB 凭据已删除，zxbtest 账号停用）。" + "财务原始数据真实通道为智享通 zxt 通道：DAMA 读智享通网页 → 落盘 /root/WB_WS/_data/zxt-hanke → 财务Bot 注入 CMA" + "（subjects+kpi_values 双写, source_type='zxt-sync'）。如需恢复 ERP API 直连，请联系运维重新配置 erp-gateway DB 凭据。") + +# 智享通 zxt 落盘目录（真实通道文件来源；任总 2026-09-08 确认财务原始数据走 zxt_push） +ZXT_LANDING_DIR = "/root/WB_WS/_data/zxt-hanke" + router = APIRouter(prefix="/api/cma/data", tags=["数据对接"], dependencies=[Depends(require_role("ceo", "finance", "it"))], ) @@ -21,7 +32,8 @@ async def import_excel(file: UploadFile = File(...), period_col: str = Query("period", description="Excel中期间列名"), value_col: str = Query("actual_value", description="Excel中实际值列名"), default_period: str = Query(None, description="如文件无期间列，统一使用此值"), - db: Session = Depends(get_db)): + db: Session = Depends(get_db), + entity_id: int = Depends(get_entity_id)): content = await file.read() df = pd.read_excel(io.BytesIO(content)) @@ -38,8 +50,8 @@ async def import_excel(file: UploadFile = File(...), raise HTTPException(400, "Excel文件为空，没有数据行") from app.models import KPIDefinition - kpi_map = {k.kpi_code: k.id for k in db.query(KPIDefinition).all()} - kpi_entity_map = {k.kpi_code: k.entity_id for k in db.query(KPIDefinition).all()} # 账套隔离 P2 + kpi_map = {k.kpi_code: k.id for k in db.query(KPIDefinition).filter(KPIDefinition.entity_id == entity_id).all()} + kpi_entity_map = {k.kpi_code: k.entity_id for k in db.query(KPIDefinition).filter(KPIDefinition.entity_id == entity_id).all()} # 账套隔离 batch = hashlib.md5(str(datetime.now().timestamp()).encode()).hexdigest()[:12] count = 0 @@ -198,16 +210,16 @@ async def import_excel_smart( # 6. 检测报表类型（用于自动生成KPI编码前缀） stype = _smart_detect_statement_type(fname) - # 7. 预加载KPI字典 + # 7. 预加载KPI字典（★ 按当前账套过滤，防跨账套匹配 — 2026-09-08 P0 安全修复） from app.models import KPIDefinition - kpis = {k.kpi_code: k for k in db.query(KPIDefinition).all()} + kpis = {k.kpi_code: k for k in db.query(KPIDefinition).filter(KPIDefinition.entity_id == entity_id).all()} known_codes = set(kpis.keys()) # 构建别名映射（去掉空格/大小写/特殊字符） alias_map: dict[str, str] = {} for code in known_codes: clean = re.sub(r'[\s\-_（）()]', '', code).lower() alias_map[clean] = code - # 中文名映射（"营业收入"→F_REVENUE） + # 中文名映射（"营业收入"→F_REVENUE）——同账套内 kpi_name 冲突时后者覆盖，保持与原逻辑一致 name_map: dict[str, str] = {} for code, kpi_obj in kpis.items(): name_map[kpi_obj.kpi_name] = code @@ -218,6 +230,16 @@ async def import_excel_smart( imported = 0 created_kpis = 0 skipped_rows = [] + code_serial = len(kpis) + 1 + + def _next_code() -> str: + """生成当前账套内不冲突的新KPI编码""" + nonlocal code_serial + while True: + candidate = f"{stype_prefix}{code_serial:03d}" + code_serial += 1 + if candidate not in known_codes: + return candidate for idx, row in df.iterrows(): raw_kpi_raw = row.get(kpi_col, "") @@ -265,9 +287,9 @@ async def import_excel_smart( kpi_code = code break - # ⑤ 仍未匹配 → 自动创建KPI + # ⑤ 仍未匹配 → 自动创建KPI（★ 创建在当前账套 entity_id 下） if not kpi_code: - new_code = f"{stype_prefix}{len(kpis) + created_kpis + 1:03d}" + new_code = _next_code() new_kpi = KPIDefinition( entity_id=entity_id, kpi_code=new_code, @@ -300,7 +322,7 @@ async def import_excel_smart( db.add(KPIValue( kpi_id=kpis[kpi_code].id, - entity_id=kpis[kpi_code].entity_id, # 账套隔离 P2 + entity_id=entity_id, # 账套隔离：只写入当前账套 period=raw_period, actual_value=val, source_type="excel", @@ -323,13 +345,14 @@ async def import_excel_smart( return {"message": msg, "batch": batch, "total": imported, "skipped": len(skipped_rows)} @router.get("/sources") -def list_sources(db: Session = Depends(get_db)): - sources = db.query(DataSourceConfig).all() +def list_sources(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + sources = db.query(DataSourceConfig).filter(DataSourceConfig.entity_id == entity_id).all() return {"data": [{c.name: getattr(s, c.name) for c in DataSourceConfig.__table__.columns} for s in sources]} @router.post("/sources") -def create_source(data: dict, db: Session = Depends(get_db)): +def create_source(data: dict, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): source = DataSourceConfig( + entity_id=entity_id, # 账套隔离：创建在当前账套下 name=data.get("name", ""), source_type=data.get("source_type", "manual"), api_endpoint=data.get("api_endpoint"), @@ -347,8 +370,11 @@ def create_source(data: dict, db: Session = Depends(get_db)): return {"data": {c.name: getattr(source, c.name) for c in DataSourceConfig.__table__.columns}} @router.put("/sources/{source_id}") -def update_source(source_id: int, data: dict, db: Session = Depends(get_db)): - source = db.query(DataSourceConfig).filter(DataSourceConfig.id == source_id).first() +def update_source(source_id: int, data: dict, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + source = db.query(DataSourceConfig).filter( + DataSourceConfig.id == source_id, + DataSourceConfig.entity_id == entity_id, + ).first() if not source: raise HTTPException(404, "数据源不存在") for key in ["name", "source_type", "api_endpoint", "api_key", "query_sql", "sync_type", "status"]: @@ -361,8 +387,11 @@ def update_source(source_id: int, data: dict, db: Session = Depends(get_db)): return {"data": {c.name: getattr(source, c.name) for c in DataSourceConfig.__table__.columns}} @router.delete("/sources/{source_id}") -def delete_source(source_id: int, db: Session = Depends(get_db)): - source = db.query(DataSourceConfig).filter(DataSourceConfig.id == source_id).first() +def delete_source(source_id: int, db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + source = db.query(DataSourceConfig).filter( + DataSourceConfig.id == source_id, + DataSourceConfig.entity_id == entity_id, + ).first() if not source: raise HTTPException(404, "数据源不存在") db.add(OperationLog(action="delete_source", target_type="source", detail=source.name)) @@ -371,47 +400,78 @@ def delete_source(source_id: int, db: Session = Depends(get_db)): return {"message": "删除成功"} +# ── 数据链路状态（2026-09-08 P0 诚实化：展示真实取数通道，替代误导性 active ERP 数据源） ── + +def _zxt_channel_status(entity_id: int, db: Session) -> dict: + """真实取数通道状态：kpi_values source_type='zxt-sync' 最近注入信息 + 落盘目录""" + latest = db.query( + func.max(KPIValue.period).label("latest_period"), + func.max(KPIValue.calculated_at).label("latest_at"), + func.count(KPIValue.id).label("row_count"), + ).filter( + KPIValue.source_type == "zxt-sync", + KPIValue.entity_id == entity_id, + ).first() + dir_mtime = None + dir_exists = os.path.isdir(ZXT_LANDING_DIR) + if dir_exists: + try: + newest = max( + (os.path.getmtime(os.path.join(root, f)) + for root, _dirs, files in os.walk(ZXT_LANDING_DIR) + for f in files if f.endswith(".csv")), + default=0, + ) + if newest: + dir_mtime = datetime.fromtimestamp(newest).strftime("%Y-%m-%d %H:%M") + except Exception: + pass + return { + "channel": "zxt（智享通→落盘→注入）", + "enabled": True, + "latest_period": latest.latest_period if latest else None, + "latest_inject_at": latest.latest_at.strftime("%Y-%m-%d %H:%M:%S") if latest and latest.latest_at else None, + "injected_rows": latest.row_count if latest else 0, + "landing_dir": ZXT_LANDING_DIR, + "landing_dir_exists": dir_exists, + "landing_dir_mtime": dir_mtime, + } + + +@router.get("/channel-status") +def channel_status(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + """数据链路状态：真实取数通道(zxt) + 遗留 ERP 数据源说明（诚实化展示）""" + zxt = _zxt_channel_status(entity_id, db) + + # 当前账套的 data_source_config 遗留 ERP 源（有端点指向 erp-gateway 且从未同步成功） + legacy = db.query(DataSourceConfig).filter( + DataSourceConfig.entity_id == entity_id, + DataSourceConfig.source_type == "erp", + ).all() + legacy_rows = [{ + "id": s.id, + "name": s.name, + "api_endpoint": s.api_endpoint, + "status": s.status, + "last_sync_at": s.last_sync_at.strftime("%Y-%m-%d %H:%M:%S") if s.last_sync_at else None, + } for s in legacy] + + return { + "real_channel": zxt, + "erp_direct": { + "enabled": False, + "reason": ERP_DIRECT_CHANNEL_REASON, + "legacy_sources_count": len(legacy_rows), + "legacy_sources": legacy_rows, + }, + } + + @router.get("/sync-kpis") -def sync_kpis_from_erp(db: Session = Depends(get_db)): - """从ERP数据源同步KPI值（调用erp_sync模块）""" - from scripts.erp_sync import run_sync - import traceback - from datetime import datetime as dt +def sync_kpis_from_erp(db: Session = Depends(get_db), entity_id: int = Depends(get_entity_id)): + """[已停用] 从ERP数据源同步KPI值 - try: - # 获取所有标记为erp数据源的KPI - erp_kpis = db.query(KPIDefinition).filter( - KPIDefinition.status == "active", - KPIDefinition.data_source_type == "erp", - ).all() - kpi_count = len(erp_kpis) - kpi_codes = [k.kpi_code for k in erp_kpis] - - # 执行同步 (dry_run=False, use_api=False 使用本地fallback) - run_sync(dry_run=False, kpi_codes=kpi_codes, use_api=False) - - # 记录操作日志 - log = OperationLog( - action="sync_kpis", - target_type="kpi", - detail=f"ERP同步: {kpi_count}个KPI, 编码: {', '.join(kpi_codes[:10])}{'...' if kpi_count > 10 else ''}", - ) - db.add(log) - db.commit() - - return { - "message": f"ERP数据同步完成", - "total_kpis": kpi_count, - "kpi_codes": kpi_codes, - "synced_at": dt.now().isoformat(), - } - - except Exception as e: - log = OperationLog( - action="sync_kpis_error", - target_type="kpi", - detail=f"ERP同步失败: {str(e)[:500]}", - ) - db.add(log) - db.commit() - raise HTTPException(500, f"ERP同步失败: {str(e)}") + ERP 直连通道已于 2026-08-31 停用（erp-gateway DB 凭据删除）。 + 真实财务数据通道 = zxt 智享通注入（source_type='zxt-sync'），勿再调用本端点。 + """ + raise HTTPException(status_code=410, detail=ERP_DIRECT_CHANNEL_REASON) diff --git a/backend/app/main.py b/backend/app/main.py index c735b1de..c8f7c994 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -109,26 +109,21 @@ def startup(): logger.info("管理会计OS后端启动完成") +# ERP 直连同步通道停用（2026-08-31 起 erp-gateway DB 凭据删除；真实财务数据走 zxt 智享通注入） +_ERP_CHANNEL_DISABLED_REASON = ("ERP直连同步通道已于2026-08-31停用（erp-gateway DB 凭据已删除，zxbtest 账号停用）。" + "财务原始数据真实通道为 zxt 智享通注入（source_type='zxt-sync'），本端点不再可用。") + + @app.post("/api/cma/admin/erp-sync") def admin_erp_sync(user=Depends(require_auth), kpi_codes: str = None): - """手动触发ERP数据同步""" - kpi_list = kpi_codes.split(",") if kpi_codes else None - try: - run_erp_sync(dry_run=False, kpi_codes=kpi_list, use_api=True) - return {"message": "ERP同步完成", "kpis": kpi_list} - except Exception as e: - return JSONResponse(status_code=500, content={"detail": f"同步失败: {str(e)}"}) + """[已停用] 手动触发ERP数据同步 — ERP 直连通道 2026-08-31 停用，返回 410""" + return JSONResponse(status_code=410, content={"detail": _ERP_CHANNEL_DISABLED_REASON}) @app.get("/api/cma/admin/erp-sync/dry-run") def admin_erp_sync_dry_run(user=Depends(require_auth), kpi_codes: str = None): - "试运行，不写入数据库""" - kpi_list = kpi_codes.split(",") if kpi_codes else None - try: - run_erp_sync(dry_run=True, kpi_codes=kpi_list, use_api=True) - return {"message": "试运行完成"} - except Exception as e: - return JSONResponse(status_code=500, content={"detail": f"试运行失败: {str(e)}"}) + """[已停用] 试运行 — ERP 直连通道 2026-08-31 停用，返回 410""" + return JSONResponse(status_code=410, content={"detail": _ERP_CHANNEL_DISABLED_REASON}) @app.post("/api/cma/admin/alerts/check") diff --git a/backend/tests/test_data_entity_isolation_20260908.py b/backend/tests/test_data_entity_isolation_20260908.py new file mode 100644 index 00000000..630f931b --- /dev/null +++ b/backend/tests/test_data_entity_isolation_20260908.py @@ -0,0 +1,230 @@ +"""数据管理模块脱节修复测试（2026-09-08 P0，project-20260908-data-manage-fix） + +覆盖： +- Excel 导入不跨账套匹配（import-excel / import-excel-smart 双端点） +- 数据源管理端点 entity 隔离（list/create/update/delete + 跨账套 404） +- sync-kpis 已停用（410） +- channel-status 真实通道状态端点 +""" +import io +import pandas as pd +import pytest +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models import KPIDefinition, KPIValue, DataSourceConfig, Entity +from tests.conftest import create_test_user, get_token_for_user, auth_header + +DATA = "/api/cma/data" + + +def _ensure_entity(db: Session, eid: int, name: str = None) -> Entity: + ent = db.query(Entity).filter(Entity.id == eid).first() + if not ent: + ent = Entity(id=eid, name=name or f"企业{eid}", short_name=f"E{eid}", status="active") + db.add(ent) + db.commit() + db.refresh(ent) + return ent + + +def _login(client: TestClient, username: str, password: str, entity_id: int) -> str: + resp = client.post("/api/cma/auth/login", json={ + "username": username, "password": password, "entity_id": entity_id, + }) + assert resp.status_code == 200, f"登录 entity{entity_id} 失败: {resp.text[:300]}" + return resp.json()["token"] + + +def _excel_bytes(df: pd.DataFrame) -> io.BytesIO: + buf = io.BytesIO() + df.to_excel(buf, index=False) + buf.seek(0) + return buf + + +class TestExcelImportEntityIsolation: + """Excel 导入只匹配当前账套 KPI，防跨账套写入（P0 安全）""" + + def _mk_kpi(self, db: Session, code: str, eid: int) -> KPIDefinition: + kpi = KPIDefinition( + entity_id=eid, kpi_code=code, kpi_name=f"{code}-名称", + dimension="finance", category="financial_report", formula="-", + data_source_type="excel", data_source="Excel导入", data_owner="财务部", + frequency="monthly", unit="元", target_value=0, + kpi_level="operational", status="active", + ) + db.add(kpi) + db.commit() + db.refresh(kpi) + return kpi + + def test_import_excel_entity2_does_not_write_entity1_kpi(self, client: TestClient, db: Session): + """entity2 登录导入 F_REVENUE → entity1 的 F_REVENUE 零写入（旧实现会全表匹配串写）""" + _ensure_entity(db, 2) + k1 = self._mk_kpi(db, "F_REVENUE", 1) + _ensure_entity(db, 2) + create_test_user(db) + token2 = _login(client, "testadmin", "admin123", entity_id=2) + before = db.query(KPIValue).filter(KPIValue.kpi_id == k1.id).count() + + df = pd.DataFrame({ + "kpi_code": ["F_REVENUE"], + "period": ["2026-08"], + "actual_value": [999.0], + }) + resp = client.post( + f"{DATA}/import-excel", + headers=auth_header(token2), + files={"file": ("t.xlsx", _excel_bytes(df), + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}, + ) + assert resp.status_code == 200 + # entity2 无 F_REVENUE → 匹配不到 → 跳过 0 条（不是跨账套写入 entity1） + assert "0 条" in resp.json()["message"] + assert "不存在" in resp.json()["message"] + after = db.query(KPIValue).filter(KPIValue.kpi_id == k1.id).count() + assert after == before, "entity2 导入不得写入 entity1 的 KPI" + + def test_import_excel_smart_creates_kpi_in_current_entity_only(self, client: TestClient, db: Session): + """entity2 智能导入不存在的 F_REVENUE → 自动创建在 entity2（不匹配 entity1 的同名KPI）""" + _ensure_entity(db, 2) + k1 = self._mk_kpi(db, "F_REVENUE", 1) + create_test_user(db) + token2 = _login(client, "testadmin", "admin123", entity_id=2) + + df = pd.DataFrame({ + "kpi_code": ["F_REVENUE"], + "period": ["2026-08"], + "actual_value": [999.0], + }) + resp = client.post( + f"{DATA}/import-excel-smart", + headers=auth_header(token2), + files={"file": ("测试利润表.xlsx", _excel_bytes(df), + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}, + ) + assert resp.status_code == 200, resp.text + body = resp.json() + # entity2 自动创建 KPI（PL_ 前缀因为文件名含"利润表"），不写 entity1 的 F_REVENUE + assert "自动创建1个新KPI" in body["message"] + created = db.query(KPIDefinition).filter( + KPIDefinition.entity_id == 2, + KPIDefinition.kpi_code.like("PL_%"), + ).first() + assert created is not None, "entity2 导入应自动创建 PL_xxx KPI 在 entity2 账套" + # entity1 的 F_REVENUE 及其 KPIValue 未被触碰 + assert db.query(KPIValue).filter(KPIValue.kpi_id == k1.id).count() == 0 + + def test_import_excel_entity1_own_kpi_still_works(self, client: TestClient, db: Session): + """正常路径：entity1 导入自己的 F_REVENUE 成功""" + k1 = self._mk_kpi(db, "F_REVENUE", 1) + create_test_user(db) + token1 = get_token_for_user(client) # entity1 + df = pd.DataFrame({ + "kpi_code": ["F_REVENUE"], + "period": ["2026-08"], + "actual_value": [100.0], + }) + resp = client.post( + f"{DATA}/import-excel", + headers=auth_header(token1), + files={"file": ("t.xlsx", _excel_bytes(df), + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")}, + ) + assert resp.status_code == 200 + assert "导入成功 1 条" in resp.json()["message"] + row = db.query(KPIValue).filter(KPIValue.kpi_id == k1.id).first() + assert row is not None and row.actual_value == 100.0 + + +class TestSourceCRUDEntityIsolation: + """数据源管理端点 entity 隔离（P1）""" + + def _mk_entity2_source(self, client, db, token2) -> int: + resp = client.post( + f"{DATA}/sources", + headers=auth_header(token2), + json={"name": "entity2数据源", "source_type": "manual"}, + ) + assert resp.status_code == 200 + assert resp.json()["data"]["entity_id"] == 2 + return resp.json()["data"]["id"] + + def test_list_sources_isolated_by_entity(self, client: TestClient, db: Session): + """entity2 登录 list_sources 看不到 entity1 的数据源""" + _ensure_entity(db, 2) + create_test_user(db) + token1 = get_token_for_user(client) + token2 = _login(client, "testadmin", "admin123", entity_id=2) + + r1 = client.post(f"{DATA}/sources", headers=auth_header(token1), + json={"name": "entity1数据源", "source_type": "manual"}) + assert r1.status_code == 200 + + # entity2 登录看不到 entity1 的源 + resp2 = client.get(f"{DATA}/sources", headers=auth_header(token2)) + assert resp2.status_code == 200 + names2 = [s["name"] for s in resp2.json()["data"]] + assert "entity1数据源" not in names2 + + # entity1 登录能看到自己的 + resp1 = client.get(f"{DATA}/sources", headers=auth_header(token1)) + names1 = [s["name"] for s in resp1.json()["data"]] + assert "entity1数据源" in names1 + + def test_cross_entity_update_delete_404(self, client: TestClient, db: Session): + """entity2 不能改/删 entity1 的数据源（404 越权防护）""" + _ensure_entity(db, 2) + create_test_user(db) + token1 = get_token_for_user(client) + token2 = _login(client, "testadmin", "admin123", entity_id=2) + + r1 = client.post(f"{DATA}/sources", headers=auth_header(token1), + json={"name": "entity1的源", "source_type": "manual"}) + e1_source_id = r1.json()["data"]["id"] + + # entity2 尝试更新 entity1 的源 → 404（隔离生效，不再允许越权操作） + resp = client.put(f"{DATA}/sources/{e1_source_id}", headers=auth_header(token2), + json={"name": "篡改"}) + assert resp.status_code == 404 + resp = client.delete(f"{DATA}/sources/{e1_source_id}", headers=auth_header(token2)) + assert resp.status_code == 404 + # entity1 自己仍可删除 + resp = client.delete(f"{DATA}/sources/{e1_source_id}", headers=auth_header(token1)) + assert resp.status_code == 200 + + +class TestSyncKpisDecommissioned: + """sync-kpis 收口：ERP 直连通道 2026-08-31 停用，端点返回 410（下线说明）""" + + def test_sync_kpis_returns_410(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + resp = client.get(f"{DATA}/sync-kpis", headers=auth_header(token)) + assert resp.status_code == 410 + assert "2026-08-31" in resp.json()["detail"] + assert "zxt" in resp.json()["detail"] + + def test_admin_erp_sync_returns_410(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + resp = client.post("/api/cma/admin/erp-sync", headers=auth_header(token)) + assert resp.status_code == 410 + resp2 = client.get("/api/cma/admin/erp-sync/dry-run", headers=auth_header(token)) + assert resp2.status_code == 410 + + +class TestChannelStatus: + """channel-status 真实取数链路状态端点""" + + def test_channel_status_shape(self, client: TestClient, db: Session): + create_test_user(db) + token = get_token_for_user(client) + resp = client.get(f"{DATA}/channel-status", headers=auth_header(token)) + assert resp.status_code == 200 + d = resp.json() + assert "real_channel" in d and "erp_direct" in d + assert d["real_channel"]["channel"].startswith("zxt") + assert d["erp_direct"]["enabled"] is False + assert "2026-08-31" in d["erp_direct"]["reason"] diff --git a/frontend/src/api/index.ts b/frontend/src/api/index.ts index dbb586d0..5fe189d3 100644 --- a/frontend/src/api/index.ts +++ b/frontend/src/api/index.ts @@ -144,6 +144,7 @@ export const dataApi = { createSource: (data: any) => api.post('/data/sources', data), updateSource: (id: number, data: any) => api.put(`/data/sources/${id}`, data), deleteSource: (id: number) => api.delete(`/data/sources/${id}`), + channelStatus: () => api.get('/data/channel-status'), } export const alertApi = { diff --git a/frontend/src/views/DataManage.vue b/frontend/src/views/DataManage.vue index b8182a8b..08e62b0c 100644 --- a/frontend/src/views/DataManage.vue +++ b/frontend/src/views/DataManage.vue @@ -20,6 +20,30 @@ + + + + - + @@ -40,9 +64,11 @@ {{ row.sync_type }} - + @@ -54,21 +80,52 @@ +
+ 💡 上表中「ERP」类型数据源为遗留演示配置：其端点指向 erp-gateway 容器（2026-08-31 起 DB 凭据删除、已停用）。 + 新增数据源请选择非 ERP 类型；企业财务数据由上方 zxt 智享通通道真实供给，无需在此配置。 +
+ + + + + +
06:30 定时采集（scripts/kpi_value_collector.py）按此映射写入 kpi_values（source_type='auto_collect'），与上方 ERP 遗留数据源相互独立。
+ + + + + + + + + + + - + + + + @@ -92,7 +149,7 @@ import { ref, onMounted } from 'vue' import { ElMessage, ElMessageBox } from 'element-plus' import { UploadFilled } from '@element-plus/icons-vue' -import { dataApi } from '../api/index' +import { dataApi, budgetApi } from '../api/index' import MyDialog from '../components/MyDialog.vue' const activeTab = ref('import') @@ -132,6 +189,38 @@ async function doImport() { uploading.value = false } +// === 数据链路状态（2026-09-08 诚实化） === +const channelLoading = ref(false) +const realChannel = ref(null) +const erpDirect = ref(null) +const legacyErpIds = ref>(new Set()) +const valueSources = ref([]) + +function isLegacyErpSource(row: any): boolean { + return row.source_type === 'erp' && legacyErpIds.value.has(Number(row.id)) +} + +async function loadChannelStatus() { + channelLoading.value = true + try { + const r: any = await dataApi.channelStatus() + const d = r.data || r + realChannel.value = d.real_channel || null + erpDirect.value = d.erp_direct || null + const ids = (d.erp_direct?.legacy_sources || []).map((s: any) => Number(s.id)) + legacyErpIds.value = new Set(ids) + } catch (e) { /* 非阻塞：状态卡片加载失败不影响列表 */ } + channelLoading.value = false +} + +async function loadValueSources() { + try { + const r: any = await budgetApi.valueSources({ status: 'active' }) + const d = Array.isArray(r) ? r : (r.data || []) + valueSources.value = d + } catch (e) { valueSources.value = [] } +} + // === 数据源管理 === const sources = ref([]) const sourcesLoading = ref(false) @@ -143,6 +232,11 @@ function sourceTypeTag(t: string) { return ({ erp: 'primary', business: 'warning', excel: 'success', manual: 'info' } as any)[t] || 'info' } +function onSourceTypeChange() { + // 触发表单内 ERP 遗留说明显示 + sourceForm.value = { ...sourceForm.value } +} + async function loadSources() { sourcesLoading.value = true try { @@ -173,6 +267,7 @@ async function saveSource() { } showSourceForm.value = false loadSources() + loadChannelStatus() } catch (e) { ElMessage.error('保存失败') } sourceSaving.value = false } @@ -183,12 +278,17 @@ async function doDeleteSource(row: any) { await dataApi.deleteSource(row.id) ElMessage.success('已删除') loadSources() + loadChannelStatus() } catch (e: any) { if (e !== 'cancel') ElMessage.error('删除失败') } } -onMounted(loadSources) +onMounted(() => { + loadSources() + loadChannelStatus() + loadValueSources() +})