金蝶对接fastapi优化版本

from fastapi import FastAPI, HTTPException, Body
import requests
import time
import base64
import json
import hmac
import hashlib
import threading
from typing import Dict, Any

app = FastAPI(title="金蝶云星空 FastAPI对接(SimPas单点+WebApi)")

# ========================= 金蝶配置 =========================
K3_HOST = "http://localhost/K3Cloud"
K3_LOGIN_URL = f"{K3_HOST}/Kingdee.BOS.WebApi.ServicesStub.AuthService.LoginByAppSecret.common.kdsvc"
K3_WEBAPI_BASE = K3_HOST

SIMPAS_DBID = "6a7824cbcc2371"
SIMPAS_APPID = "356983_X+eq5/hsRMr43fSLwdRs4d1tUqx87Opp"
SIMPAS_APPSECRET = "69e79dbf6bca4dc99d8251c3060f284d"
SIMPAS_LCID = 2052
K3_USERNAME = "XXXX"

# 复用同一会话；登录态由 kdservice-sessionid Cookie 维持
k3_session = requests.Session()
# 防止并发请求同时触发重登录、互相覆盖 Cookie
_login_lock = threading.Lock()


# ---------------------- SimPas 单点登录 ----------------------
def generate_simpas_sign(app_secret: str, params: Dict[str, str]) -> str:
    """SimPas 签名：dbid + username + appid + timestamp + appSecret 的 HMAC-SHA256（无分隔符拼接）"""
    raw = f"{params['dbid']}{params['username']}{params['appid']}{params['timestamp']}{app_secret}"
    return hmac.new(app_secret.encode(), raw.encode(), hashlib.sha256).hexdigest()


def build_simpas_login_url(username: str = K3_USERNAME) -> Dict[str, str]:
    ts = str(int(time.time()))
    params = {"dbid": SIMPAS_DBID, "username": username, "appid": SIMPAS_APPID, "timestamp": ts}
    signed = generate_simpas_sign(SIMPAS_APPSECRET, params)
    payload = {
        **params, "signeddata": signed, "lcid": SIMPAS_LCID, "origintype": "SimPas",
        "entryrole": "", "formid": "", "formtype": "", "pkid": "",
        "otherargs": "|{'permitcount':'0'}", "openmode": None, "loginthen": None, "loginAgain": "",
    }
    ud = base64.b64encode(json.dumps(payload, ensure_ascii=False).encode()).decode()
    return {"timestamp": ts, "signeddata": signed, "base64_param": ud,
            "sso_login_url": f"{K3_HOST}/html5/index.aspx?ud={ud}"}


# ---------------------- 金蝶 WebApi ----------------------
def _parse(resp: requests.Response) -> Any:
    """优先解析 JSON，失败返回原始文本"""
    text = resp.text.strip()
    if not text:
        raise Exception(f"金蝶返回空响应，HTTP={resp.status_code}")
    try:
        return resp.json()
    except Exception:
        return text


def _extract_error(res: Any) -> str:
    """从金蝶响应里提取可读错误信息"""
    if isinstance(res, str):
        return res.replace("response_error:", "", 1).strip() if res.startswith("response_error:") else res
    if isinstance(res, dict):
        if res.get("Message"):
            return res["Message"]
        status = res.get("Result", {}).get("ResponseStatus", {})
        if str(status.get("IsSuccess", "")).lower() == "false":
            errs = status.get("Errors") or []
            return json.dumps(errs, ensure_ascii=False) if errs else "金蝶业务接口返回失败"
    return json.dumps(res, ensure_ascii=False)


def _is_logged_in() -> bool:
    return bool(k3_session.cookies.get("kdservice-sessionid"))


def k3_login() -> None:
    """第三方系统登录授权（appId + appSecret）免密码登录，登录态写入 Cookie。并发安全。"""
    with _login_lock:
        # 已登录则直接返回，避免每次调用都重登录
        if _is_logged_in():
            return
        resp = k3_session.post(K3_LOGIN_URL, data={
            "acctid": SIMPAS_DBID, "username": K3_USERNAME,
            "appId": SIMPAS_APPID, "appSecret": SIMPAS_APPSECRET, "lcid": SIMPAS_LCID,
        }, timeout=30)
        res = _parse(resp)
        if not (isinstance(res, dict) and res.get("IsSuccessByAPI") is True):
            raise Exception(f"金蝶登录失败：{_extract_error(res)}")


def _session_lost(res: Any) -> bool:
    """判断是否会话失效（覆盖常见失效提示）"""
    text = _extract_error(res) if isinstance(res, (dict, str)) else json.dumps(res, ensure_ascii=False)
    return any(k in text for k in ("会话", "重新登录", "未登录", "LoginResultType"))


def call_dynamic_form(method: str, formid: str, data: Dict[str, Any]) -> Any:
    """
    调用金蝶 DynamicFormService 业务接口。
    method: View / Save / Submit / Audit / ExecuteBillQuery / ...
    业务失败时直接抛 HTTPException(400)，由 FastAPI 透传给前端。
    """
    if not _is_logged_in():
        k3_login()

    # Save/Submit/Audit 等写操作要求 data 内含 Model 对象。
    # 兼容两种调用写法：
    #   1) 显式传 {"Model": {...}}  —— 原样透传
    #   2) 直接传字段 {"FName": ...} —— 自动包一层 Model，方便调用
    # （View/ExecuteBillQuery 等读接口不需要 Model，method 不在列表则不处理）
    if method in ("Save", "Submit", "Audit", "Push", "Draft", "Enable", "Disable") and "Model" not in data:
        data = {"Model": data}

    url = f"{K3_WEBAPI_BASE}/Kingdee.BOS.WebApi.ServicesStub.DynamicFormService.{method}.common.kdsvc"
    body = {"formid": formid, "data": json.dumps(data, ensure_ascii=False)}

    res = _parse(k3_session.post(url, data=body, timeout=60))

    # 会话失效则重新登录一次再试
    if _session_lost(res):
        k3_login()
        res = _parse(k3_session.post(url, data=body, timeout=60))

    # ExecuteBillQuery 成功时直接返回二维数组 [[...], [...]]
    if isinstance(res, list):
        return {"rows": res, "count": len(res)}

    if isinstance(res, dict):
        status = res.get("Result", {}).get("ResponseStatus", {})
        if str(status.get("IsSuccess", "")).lower() == "false" or res.get("IsSuccessByAPI") is False:
            raise HTTPException(status_code=400, detail=_extract_error(res))
        return res.get("Result", res)

    raise HTTPException(status_code=500, detail=f"金蝶接口返回非JSON：{res}")


# ---------------------- FastAPI 接口 ----------------------
@app.get("/")
async def index():
    return {"msg": "金蝶云星空 FastAPI网关，支持SimPas单点登录 + BOS WebApi"}


@app.get("/k3/simpas/login_url")
async def get_sso_url(username: str = K3_USERNAME):
    """生成SimPas免登链接"""
    try:
        return {"username": username, **build_simpas_login_url(username)}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"生成单点链接失败:{e}")


@app.get("/k3/bos/test_token")
async def test_get_token():
    """测试登录并获取金蝶 WebApi 会话ID"""
    try:
        k3_login()
        return {"session_id": k3_session.cookies.get("kdservice-sessionid"),
                "cookies": {c.name: c.value for c in k3_session.cookies}}
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/k3/bos/{formid}/{method}")
async def dynamic_form(
    formid: str,
    method: str,
    payload: Dict[str, Any] = Body(
        ...,
        examples={
            "View查看客户": {
                "summary": "按编码查看客户（method=View）",
                "value": {"CreateOrgId": 0, "Number": "KHTEST001", "Id": "", "IsSortBySeq": "false"},
            },
            "View查看供应商": {
                "summary": "按编码查看供应商（method=View）",
                "value": {"CreateOrgId": 0, "Number": "VEN00001", "Id": "", "IsSortBySeq": "false"},
            },
            "ExecuteBillQuery查询客户": {
                "summary": "查询客户列表（method=ExecuteBillQuery）",
                "value": {
                    "FormId": "BD_Customer", "FieldKeys": "FNumber,FName",
                    "FilterString": "", "OrderString": "",
                    "TopRowCount": 0, "StartRow": 0, "Limit": 5,
                },
            },
            "Save保存客户": {
                "summary": "新增客户（method=Save，最简可用报文，其余字段金蝶按默认值处理；Creator 不填自动取当前登录对接账号）",
                "value": {
                    "Model": {
                        "FCreateOrgId": {"FNumber": "100"},
                        "FName": "chenggg",
                        "FINVOICETITLE": "chenggg",
                        "FTRADINGCURRID": {"FNumber": "PRE001"}
                    }
                },
            },
        },
    ),
):
    """
    通用动态表单接口（替代为每个表单写一遍接口）。
    - formid: 业务对象表单Id，如 BD_Customer / BD_Supplier / SAL_SaleOrder / BD_Material
              注意：formid 不要用数据库表名 t_XXX，直接用 BD_XXX。
    - method: View / Save / Submit / Audit / ExecuteBillQuery / ...
    - payload: 金蝶业务参数，原样作为 data 传过去
      View  {"Number": "DF1005A" }
      Save 所有基础资料引用字段（组织、币别、客户分类、税率等）都要包一层 {"FNumber": "xxx"}
        例如：{
                 "Model": {
                        "FName": "chenchen",
                        "FCreateOrgId": {"FNumber": "100"},
                        "FTRADINGCURRID": {"FNumber": "PRE001"}
                         }
            }
    """
    if not formid or not method:
        raise HTTPException(status_code=400, detail="formid 与 method 不能为空")
    return call_dynamic_form(method, formid, payload)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8001)