Hermes CI Fix
193061cf4d
- 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）
478 lines
19 KiB
Python
478 lines
19 KiB
Python
"""数据对接 API"""
|
||
import pandas as pd
|
||
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
|
||
from sqlalchemy import func
|
||
from typing import Optional
|
||
from app.database import get_db
|
||
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"))],
|
||
)
|
||
|
||
@router.post("/import-excel")
|
||
async def import_excel(file: UploadFile = File(...),
|
||
kpi_col: str = Query("kpi_code", description="Excel中KPI编码列名"),
|
||
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),
|
||
entity_id: int = Depends(get_entity_id)):
|
||
content = await file.read()
|
||
df = pd.read_excel(io.BytesIO(content))
|
||
|
||
required = [kpi_col, value_col]
|
||
if not default_period:
|
||
required.append(period_col)
|
||
|
||
missing = [c for c in required if c not in df.columns]
|
||
if missing:
|
||
raise HTTPException(400,
|
||
f"Excel缺少列: {missing}。当前文件列: {list(df.columns)}")
|
||
|
||
if len(df) == 0:
|
||
raise HTTPException(400, "Excel文件为空，没有数据行")
|
||
|
||
from app.models import KPIDefinition
|
||
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
|
||
skipped = []
|
||
for idx, row in df.iterrows():
|
||
kpi_code = str(row.get(kpi_col, "")).strip()
|
||
period = str(row.get(period_col, default_period or "")).strip() if period_col in df.columns else (default_period or "").strip()
|
||
value = row.get(value_col)
|
||
|
||
if not kpi_code or not period or pd.isna(value):
|
||
skipped.append(f"第{idx+2}行: 缺少必填字段")
|
||
continue
|
||
|
||
kid = kpi_map.get(kpi_code)
|
||
if not kid:
|
||
skipped.append(f"第{idx+2}行: KPI编码「{kpi_code}」不存在")
|
||
continue
|
||
|
||
db.add(KPIValue(
|
||
kpi_id=kid,
|
||
entity_id=kpi_entity_map.get(kpi_code),
|
||
period=period,
|
||
actual_value=float(value),
|
||
source_type="excel",
|
||
source_batch=batch,
|
||
data_status="verified",
|
||
))
|
||
count += 1
|
||
|
||
db.commit()
|
||
|
||
msg = f"✅ 导入成功 {count} 条数据"
|
||
if skipped:
|
||
msg += f"，{len(skipped)}条跳过:\n" + "\n".join(skipped[:10])
|
||
if len(skipped) > 10:
|
||
msg += f"\n...还有{len(skipped)-10}条"
|
||
return {"message": msg, "batch": batch, "total": count, "skipped": len(skipped)}
|
||
|
||
|
||
# ── 智能导入（BOT自动识别，无需手动映射） ──
|
||
|
||
_SMART_MAP = {
|
||
# KPI名称/编码列匹配模式（顺序重要：名称类列优先，避免科目编码被当名称）
|
||
"kpi_code_patterns": [
|
||
re.compile(r'^(科目名称|项目名称|指标名称?|kpi名称?|name|名称)$', re.I),
|
||
re.compile(r'^(科目编码|科目代码|kpi_?code|指标编码|编码)$', re.I),
|
||
re.compile(r'^(科目|项目|账户|报表项目)$'),
|
||
],
|
||
# 期间列匹配
|
||
"period_patterns": [
|
||
re.compile(r'^(period|期间|月份?|年月|日期|会计期间)$', re.I),
|
||
re.compile(r'^(报表期[间]?|所属期)$'),
|
||
re.compile(r'^年|^月'), # 以"年"或"月"开头的列
|
||
],
|
||
# 数值列匹配
|
||
"value_patterns": [
|
||
re.compile(r'^(actual_?value|数值|实际值|实际金额)$', re.I),
|
||
re.compile(r'^(本期金额|本月数|本期|期末余额|期末数|余额)$'),
|
||
re.compile(r'^(金额|数据|value|本年累计|本期发生[额]?)$', re.I),
|
||
],
|
||
# 文件名→期间提取
|
||
"period_in_filename_patterns": [
|
||
re.compile(r'[-_]?(\d{4})[-_]?(\d{1,2})'), # 2026-06 / 2026_06 / 2606
|
||
re.compile(r'(\d{4})年(\d{1,2})月(?:至(\d{4})年(\d{1,2})月)?'), # 2026年01月 / 2026年01月至2026年05月
|
||
re.compile(r'(\d{4})(\d{2})'), # 202606 (纯数字6-8位)
|
||
],
|
||
# 文件名→报表类型
|
||
"statement_types": {
|
||
"利润表": "PL",
|
||
"利润": "PL",
|
||
"income": "PL",
|
||
"现金流量表": "CF",
|
||
"现金流": "CF",
|
||
"cashflow": "CF",
|
||
"cash_flow": "CF",
|
||
"资产负债表": "BS",
|
||
"资产负": "BS",
|
||
"balance": "BS",
|
||
},
|
||
}
|
||
|
||
|
||
def _smart_detect_kpi_col(cols: list[str]) -> str | None:
|
||
for pat in _SMART_MAP["kpi_code_patterns"]:
|
||
for c in cols:
|
||
if pat.match(c.strip()):
|
||
return c
|
||
return None
|
||
|
||
|
||
def _smart_detect_period_col(cols: list[str]) -> str | None:
|
||
for pat in _SMART_MAP["period_patterns"]:
|
||
for c in cols:
|
||
if pat.match(c.strip()):
|
||
return c
|
||
return None
|
||
|
||
|
||
def _smart_detect_value_col(cols: list[str]) -> str | None:
|
||
for pat in _SMART_MAP["value_patterns"]:
|
||
for c in cols:
|
||
if pat.match(c.strip()):
|
||
return c
|
||
return None
|
||
|
||
|
||
def _smart_extract_period_from_filename(filename: str) -> str | None:
|
||
for pat in _SMART_MAP["period_in_filename_patterns"]:
|
||
m = pat.search(filename)
|
||
if m:
|
||
groups = m.groups()
|
||
if len(groups) == 4 and groups[2]: # 2026年01月至2026年05月 → 取结束月
|
||
return f"{groups[2]}-{int(groups[3]):02d}"
|
||
if len(groups) >= 2: # 2026-06 或 2026年01月
|
||
return f"{int(groups[0])}-{int(groups[1]):02d}"
|
||
return None
|
||
|
||
|
||
def _smart_detect_statement_type(filename: str) -> str | None:
|
||
for kw, tp in _SMART_MAP["statement_types"].items():
|
||
if kw in filename:
|
||
return tp
|
||
return None
|
||
|
||
|
||
@router.post("/import-excel-smart")
|
||
async def import_excel_smart(
|
||
file: UploadFile = File(...),
|
||
entity_id: int = Depends(get_entity_id),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
"""智能导入 — BOT自动识别列名/期间/报表类型，无需手动映射"""
|
||
content = await file.read()
|
||
fname = file.filename or "未知文件"
|
||
|
||
try:
|
||
df = pd.read_excel(io.BytesIO(content))
|
||
except Exception as e:
|
||
raise HTTPException(400, f"无法读取Excel文件: {e}")
|
||
|
||
if len(df) == 0:
|
||
raise HTTPException(400, "Excel文件为空")
|
||
|
||
cols = list(df.columns)
|
||
if len(cols) < 2:
|
||
raise HTTPException(400, f"Excel列数过少: {cols}")
|
||
|
||
# 4. 智能检测列
|
||
kpi_col = _smart_detect_kpi_col(cols) or cols[0]
|
||
value_col = _smart_detect_value_col(cols) or cols[-1]
|
||
period_col = _smart_detect_period_col(cols)
|
||
|
||
# 5. 从文件名提取期间
|
||
period = _smart_extract_period_from_filename(fname) if not period_col else None
|
||
|
||
# 6. 检测报表类型（用于自动生成KPI编码前缀）
|
||
stype = _smart_detect_statement_type(fname)
|
||
|
||
# 7. 预加载KPI字典（★ 按当前账套过滤，防跨账套匹配 — 2026-09-08 P0 安全修复）
|
||
from app.models import KPIDefinition
|
||
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）——同账套内 kpi_name 冲突时后者覆盖，保持与原逻辑一致
|
||
name_map: dict[str, str] = {}
|
||
for code, kpi_obj in kpis.items():
|
||
name_map[kpi_obj.kpi_name] = code
|
||
|
||
# 8. 遍历导入（匹配不上的自动创建KPI）
|
||
stype_prefix = {"PL": "PL_", "CF": "CF_", "BS": "BS_"}.get(stype or "", "EXT_")
|
||
batch = hashlib.md5(str(datetime.now().timestamp()).encode()).hexdigest()[:12]
|
||
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, "")
|
||
raw_kpi = str(raw_kpi_raw).strip()
|
||
raw_val = row.get(value_col)
|
||
raw_period = str(row.get(period_col, period or "")).strip() if period_col else (period or "")
|
||
|
||
# 名称防护：NaN/空/None 或 纯数字（疑似科目编码被误当名称）→ 跳过，避免创建垃圾KPI
|
||
if raw_kpi.lower() in ("nan", "none") or not raw_kpi:
|
||
skipped_rows.append(f"第{idx+2}行: KPI名称为空")
|
||
continue
|
||
if re.fullmatch(r"\d+(\.\d+)?", raw_kpi):
|
||
skipped_rows.append(f"第{idx+2}行: KPI名称疑似科目编码「{raw_kpi}」，跳过")
|
||
continue
|
||
if pd.isna(raw_val):
|
||
skipped_rows.append(f"第{idx+2}行: 缺数据")
|
||
continue
|
||
if not raw_period:
|
||
skipped_rows.append(f"第{idx+2}行: 无法确定期间")
|
||
continue
|
||
|
||
# 清理科目名（去掉"一、""减：""加："等前缀）
|
||
clean_name = re.sub(r'^[一二三四五六七八九十、\s\+]+', '', raw_kpi)
|
||
clean_name = re.sub(r'^[减加]?[：:]\s*', '', clean_name).strip()
|
||
if not clean_name:
|
||
clean_name = raw_kpi
|
||
|
||
# 匹配KPI
|
||
kpi_code = None
|
||
|
||
# ① 精确编码匹配（极少情况）
|
||
if raw_kpi in known_codes:
|
||
kpi_code = raw_kpi
|
||
# ② 别名匹配（去符号小写）
|
||
if not kpi_code:
|
||
clean_key = re.sub(r'[\s\-_（）()]', '', raw_kpi).lower()
|
||
kpi_code = alias_map.get(clean_key)
|
||
# ③ 中文名精确匹配
|
||
if not kpi_code:
|
||
kpi_code = name_map.get(clean_name)
|
||
# ④ 中文名模糊匹配
|
||
if not kpi_code:
|
||
for code, kpi_obj in kpis.items():
|
||
if clean_name in kpi_obj.kpi_name or kpi_obj.kpi_name in clean_name:
|
||
kpi_code = code
|
||
break
|
||
|
||
# ⑤ 仍未匹配 → 自动创建KPI（★ 创建在当前账套 entity_id 下）
|
||
if not kpi_code:
|
||
new_code = _next_code()
|
||
new_kpi = KPIDefinition(
|
||
entity_id=entity_id,
|
||
kpi_code=new_code,
|
||
kpi_name=clean_name,
|
||
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(new_kpi)
|
||
db.flush()
|
||
kpis[new_code] = new_kpi
|
||
known_codes.add(new_code)
|
||
name_map[clean_name] = new_code
|
||
kpi_code = new_code
|
||
created_kpis += 1
|
||
|
||
try:
|
||
val = float(raw_val)
|
||
except:
|
||
skipped_rows.append(f"第{idx+2}行: 数值格式错误「{raw_val}」")
|
||
continue
|
||
|
||
db.add(KPIValue(
|
||
kpi_id=kpis[kpi_code].id,
|
||
entity_id=entity_id, # 账套隔离：只写入当前账套
|
||
period=raw_period,
|
||
actual_value=val,
|
||
source_type="excel",
|
||
source_batch=batch,
|
||
data_status="verified",
|
||
))
|
||
imported += 1
|
||
|
||
db.commit()
|
||
|
||
# 9. 返回汇总
|
||
stype_label = {"PL": "利润表", "CF": "现金流量表", "BS": "资产负债表"}.get(stype or "", "数据表")
|
||
msg = f"✅ {stype_label}识别成功，导入{imported}条"
|
||
if created_kpis:
|
||
msg += f"，自动创建{created_kpis}个新KPI"
|
||
if skipped_rows:
|
||
msg += f"，{len(skipped_rows)}条跳过:\n" + "\n".join(skipped_rows[:8])
|
||
if len(skipped_rows) > 8:
|
||
msg += f"\n...还有{len(skipped_rows) - 8}条"
|
||
return {"message": msg, "batch": batch, "total": imported, "skipped": len(skipped_rows)}
|
||
|
||
@router.get("/sources")
|
||
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), 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"),
|
||
api_key=data.get("api_key"),
|
||
query_sql=data.get("query_sql"),
|
||
sync_type=data.get("sync_type", "manual"),
|
||
status="active",
|
||
)
|
||
db.add(source)
|
||
db.commit()
|
||
db.refresh(source)
|
||
# 操作日志
|
||
db.add(OperationLog(action="create_source", target_type="source", detail=source.name))
|
||
db.commit()
|
||
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), 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"]:
|
||
if key in data:
|
||
setattr(source, key, data[key])
|
||
db.commit()
|
||
db.refresh(source)
|
||
db.add(OperationLog(action="update_source", target_type="source", detail=source.name))
|
||
db.commit()
|
||
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), 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))
|
||
db.delete(source)
|
||
db.commit()
|
||
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), entity_id: int = Depends(get_entity_id)):
|
||
"""[已停用] 从ERP数据源同步KPI值
|
||
|
||
ERP 直连通道已于 2026-08-31 停用（erp-gateway DB 凭据删除）。
|
||
真实财务数据通道 = zxt 智享通注入（source_type='zxt-sync'），勿再调用本端点。
|
||
"""
|
||
raise HTTPException(status_code=410, detail=ERP_DIRECT_CHANNEL_REASON)
|