feat: 接入新华社工商信息同步

lsfx-mock-server/routers/enterprise_api.py

@@ -0,0 +1,115 @@
+import asyncio
+import hashlib
+from typing import Optional
+
+from fastapi import APIRouter, Form, Response
+from fastapi.responses import PlainTextResponse
+
+
+router = APIRouter()
+
+
+@router.post("/api/service/interface/invokeService/LSFXXHS")
+async def query_enterprise(
+    entName: Optional[str] = Form(None),
+    serialNum: Optional[str] = Form(None),
+    runType: Optional[str] = Form(None),
+    orgCode: Optional[str] = Form(None),
+):
+    missing = next((name for name, value in {
+        "entName": entName,
+        "serialNum": serialNum,
+        "runType": runType,
+        "orgCode": orgCode,
+    }.items() if not value), None)
+    if missing:
+        return business_error(f"缺少参数: {missing}")
+
+    name = entName.strip()
+    if name == "MOCK_HTTP_500":
+        return Response(content="mock server error", status_code=500)
+    if name == "MOCK_TIMEOUT":
+        await asyncio.sleep(5)
+        return success_wrapper(build_profile(name))
+    if name == "MOCK_EMPTY_RESPONSE":
+        return Response(content="", media_type="application/json")
+    if name == "MOCK_INVALID_JSON":
+        return PlainTextResponse("not-json")
+    if name == "MOCK_OUTER_FAILURE":
+        return {"success": False, "code": 50000, "data": None}
+    if name == "MOCK_BUSINESS_FAILURE":
+        return business_error("工商查询失败")
+    if name == "MOCK_MAPPING_MISSING":
+        return success_wrapper(None, include_mapping=False)
+    if name == "MOCK_MAPPING_NULL":
+        return success_wrapper(None)
+    if name == "MOCK_MAPPING_EMPTY":
+        return success_wrapper({})
+
+    profile = build_profile(name)
+    if name == "MOCK_NAME_MISMATCH":
+        profile["enterpriseName"] = "其他企业有限公司"
+    if name == "MOCK_CREDIT_CODE_MISSING":
+        profile["creditCode"] = " "
+    return success_wrapper(profile)
+
+
+def success_wrapper(profile, include_mapping=True):
+    data = {
+        "reasonMessage": "Running successfully",
+        "reasonCode": 200,
+        "status": 1,
+        "extensionMap": {},
+    }
+    if include_mapping:
+        data["mappingOutputFields"] = profile
+    return {"success": True, "code": 10000, "data": data}
+
+
+def business_error(message):
+    return {
+        "success": True,
+        "code": 10000,
+        "data": {
+            "mappingOutputFields": None,
+            "reasonMessage": message,
+            "reasonCode": 500,
+            "status": 0,
+            "extensionMap": {},
+        },
+    }
+
+
+def build_profile(name):
+    digest = hashlib.sha1(name.encode("utf-8")).hexdigest()
+    credit_code = "91" + "".join(str(int(ch, 16) % 10) for ch in digest[:16])
+    holders = [] if "空股东" in name else [
+        {"stock_name": "张三", "stock_percent": "60%", "should_capi": "600"},
+        {"stock_name": "李四", "stock_percent": "40%", "should_capi": "400"},
+    ]
+    if "多股东" in name:
+        holders = [
+            {"stock_name": f"股东{i}", "stock_percent": f"{i}%", "should_capi": str(i * 10)}
+            for i in range(1, 8)
+        ]
+
+    individual = "个体" in name
+    return {
+        "creditCode": credit_code,
+        "enterpriseName": name,
+        "baseCity": {
+            "province": "浙江省",
+            "province_code": "330000",
+            "city": "杭州市" if individual else None,
+            "county": "西湖区" if individual else None,
+        },
+        "regCapitalAmount": {"regist_capi_value": "1000", "regist_capi_unit": "万元"},
+        "estiblishTime": "2020-01-02",
+        "fromTime": "2020-01-01",
+        "companyOrgType": "个体" if individual else {"econ_kind": "有限责任公司", "econ_kind_code": "1100"},
+        "industry": {"industry_code": "I65", "industry": "软件和信息技术服务业"},
+        "regLocation": "浙江省杭州市西湖区测试路1号",
+        "employeeCount": None if individual else 25,
+        "legalPersonName": "王五",
+        "holders": holders,
+    }