金蝶云星空fastapi对接

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

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

# ========================= 金蝶配置 =========================
K3_HOST = "http://localhost/K3Cloud"

# WebApi 登录接口（第三方系统登录授权：免密码 appId + appSecret）
K3_LOGIN_URL = "http://localhost/K3Cloud/Kingdee.BOS.WebApi.ServicesStub.AuthService.LoginByAppSecret.common.kdsvc"

# 通用业务接口基础地址
K3_WEBAPI_BASE = "http://localhost/K3Cloud"

# SimPas单点登录配置
SIMPAS_DBID = "6a7824cbcc2371"
SIMPAS_APPID = "356983_X+eq5/hsRMr43fSLwdRs4d1tUqx87Opp"
SIMPAS_APPSECRET = "69e79dbf6bca4dc99d8251c3060f284d"
SIMPAS_LCID = 2052

# WebApi 登录用户（免密，通过第三方系统登录授权）
K3_USERNAME = "用户"

# 金蝶会话
k3_session = requests.Session()
g_k3_session_id: Optional[str] = None
# ==========================================================


def generate_simpas_sign(app_secret: str, params: Dict[str, str]) -> str:
    """SimPas 签名：dbid + username + appid + timestamp + appSecret 的 HMAC-SHA256"""
    raw_str = f"{params['dbid']}{params['username']}{params['appid']}{params['timestamp']}{app_secret}"
    h = hmac.new(app_secret.encode("utf-8"), raw_str.encode("utf-8"), hashlib.sha256)
    return h.hexdigest()


def build_simpas_login_url(username: str = K3_USERNAME) -> Dict[str, str]:
    """构建SimPas单点登录长链接"""
    timestamp = str(int(time.time()))
    params = {
        "dbid": SIMPAS_DBID,
        "username": username,
        "appid": SIMPAS_APPID,
        "timestamp": timestamp,
    }
    signeddata = generate_simpas_sign(SIMPAS_APPSECRET, params)

    login_payload = {
        "dbid": SIMPAS_DBID,
        "username": username,
        "appid": SIMPAS_APPID,
        "signeddata": signeddata,
        "timestamp": timestamp,
        "lcid": SIMPAS_LCID,
        "origintype": "SimPas",
        "entryrole": "",
        "formid": "",
        "formtype": "",
        "pkid": "",
        "otherargs": "|{'permitcount':'0'}",
        "openmode": None,
        "loginthen": None,
        "loginAgain": ""
    }
    json_str = json.dumps(login_payload, ensure_ascii=False)
    b64_ud = base64.b64encode(json_str.encode("utf-8")).decode("utf-8")
    login_url = f"{K3_HOST}/html5/index.aspx?ud={b64_ud}"

    return {
        "timestamp": timestamp,
        "signeddata": signeddata,
        "base64_param": b64_ud,
        "sso_login_url": login_url
    }


def _parse_k3_response(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(result: Any) -> str:
    """从金蝶响应中提取可读错误信息"""
    if isinstance(result, str):
        if result.startswith("response_error:"):
            return result.replace("response_error:", "", 1).strip()
        return result
    if isinstance(result, dict):
        msg = result.get("Message") or result.get("msg")
        if msg:
            return msg
        status = result.get("Result", {}).get("ResponseStatus", {})
        if str(status.get("IsSuccess", "")).lower() == "false":
            errors = status.get("Errors", [])
            if errors:
                return json.dumps(errors, ensure_ascii=False)
            return "金蝶业务接口返回失败"
    return json.dumps(result, ensure_ascii=False)


def k3_login() -> str:
    """
    使用第三方系统登录授权（appId + appSecret）免密码登录金蝶 WebApi。
    登录成功后，k3_session 会带上 Cookie，后续调用可共用。
    """
    global g_k3_session_id

    login_data = {
        "acctid": SIMPAS_DBID,
        "username": K3_USERNAME,
        "appId": SIMPAS_APPID,
        "appSecret": SIMPAS_APPSECRET,
        "lcid": SIMPAS_LCID
    }

    resp = k3_session.post(K3_LOGIN_URL, data=login_data, timeout=30)
    result = _parse_k3_response(resp)

    if resp.status_code != 200:
        raise Exception(f"金蝶登录HTTP错误，status={resp.status_code}，返回={result}")

    if isinstance(result, dict):
        if result.get("IsSuccessByAPI") is True:
            g_k3_session_id = result.get("KDSVCSessionId") or ""
            return g_k3_session_id
        raise Exception(f"金蝶登录失败：{_extract_error(result)}")

    raise Exception(f"金蝶登录返回非预期内容：{result}")


def call_dynamic_form(method: str, formid: str, data: Dict[str, Any]) -> Any:
    """
    调用金蝶 DynamicFormService 业务接口。
    method: View / Save / Submit / Audit / ExecuteBillQuery / ...
    """
    global g_k3_session_id

    if not g_k3_session_id:
        k3_login()

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

    def do_call():
        r = k3_session.post(api_url, data=body, timeout=60)
        return _parse_k3_response(r)

    ret = do_call()

    # 票据失效重新登录
    need_relogin = False
    if isinstance(ret, dict):
        if ret.get("LoginResultType") == -1:
            need_relogin = True
        status = ret.get("Result", {}).get("ResponseStatus", {})
        if str(status.get("IsSuccess", "")).lower() == "false":
            err_code = status.get("ErrorCode") or status.get("MsgCode")
            if str(err_code) in ("1", "2"):
                need_relogin = True

    if need_relogin:
        k3_login()
        ret = do_call()

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

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

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


# ---------------------- 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:
        result = build_simpas_login_url(username)
        return {"username": username, **result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"生成单点链接失败:{str(e)}")


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


@app.post("/k3/bos/customer/list")
async def query_customer(page: int = 1, page_size: int = 10):
    """查询客户列表（使用 ExecuteBillQuery）"""
    data = {
        "FormId": "BD_Customer",
        "FieldKeys": "FCustId,FNumber,FName",
        "FilterString": "",
        "OrderString": "",
        "TopRowCount": 0,
        "StartRow": (page - 1) * page_size,
        "Limit": page_size
    }
    return call_dynamic_form("ExecuteBillQuery", "BD_Customer", data)


@app.post("/k3/bos/customer/view")
async def view_customer(number: str = "KHTEST001"):
    """查看客户"""
    data = {
        "CreateOrgId": 0,
        "Number": number,
        "Id": "",
        "IsSortBySeq": "false"
    }
    return call_dynamic_form("View", "BD_Customer", data)


@app.post("/k3/bos/customer/save")
async def save_customer():
    """保存客户"""
    data = {
        "Creator": K3_USERNAME,
        "NeedUpDateFields": [],
        "Model": [
            {
                "FNumber": "KHTEST001",
                "FName": "FastAPI测试客户",
                "FCreateOrgId": {"FNumber": "001"},
                "FUseOrgId": {"FNumber": "001"}
            }
        ]
    }
    return call_dynamic_form("Save", "BD_Customer", data)


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