"""数据对接 API""" import pandas as pd import io, json, hashlib from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File from sqlalchemy.orm import Session from sqlalchemy import func from typing import Optional from app.database import get_db from app.auth_middleware import require_auth, require_role from app.models import KPIValue, DataSourceConfig, OperationLog 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(...), db: Session = Depends(get_db)): content = await file.read() df = pd.read_excel(io.BytesIO(content)) required = ["kpi_code", "period", "actual_value"] if not all(c in df.columns for c in required): raise HTTPException(400, f"Excel必须包含列: {required}") 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).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_code", "")).strip() period = str(row.get("period", "")).strip() value = row.get("actual_value") 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, 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)} @router.get("/sources") def list_sources(db: Session = Depends(get_db)): sources = db.query(DataSourceConfig).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)): source = DataSourceConfig( 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)): source = db.query(DataSourceConfig).filter(DataSourceConfig.id == source_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)): source = db.query(DataSourceConfig).filter(DataSourceConfig.id == source_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": "删除成功"}