fix(data): 数据管理模块脱节修复 — Excel导入/sources端点 entity 隔离 + ERP直连通道诚实化(410) + 真实zxt链路展示
- 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）
This commit is contained in:
Hermes CI Fix
+119
-59
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user