一、背景与需求分析

企业工商信息（如统一社会信用代码、注册资本、经营范围、股东信息等）是市场调研、风险评估、供应链管理等领域的关键数据。传统方式通过国家企业信用信息公示系统等官网逐条查询效率低下，而Python的自动化能力可实现批量下载，显著提升效率。本文以某企业需获取1000家供应商的工商信息为例，探讨技术实现路径。

二、数据源选择与API接入

1. 官方数据源对比

• 国家企业信用信息公示系统：权威但无公开API，需模拟浏览器请求（风险：反爬机制严格）。
• 第三方数据平台（如天眼查、企查查）：提供付费API，支持批量查询（推荐：数据完整度高，稳定性强）。
• 开放数据接口：部分地方政府开放企业信息接口（如浙江政务服务网），需申请API密钥。

2. API调用示例（以某第三方平台为例）

import requests
import pandas as pd
def fetch_company_info(api_key, company_names):
    base_url = "https://api.example.com/v1/company/search"
    results = []
    for name in company_names:
        params = {
            "keyword": name,
            "api_key": api_key
        }
        response = requests.get(base_url, params=params)
        if response.status_code == 200:
            data = response.json()
            if data["code"] == 0:  # 成功响应
                results.append(data["result"])
        else:
            print(f"Error fetching {name}: {response.status_code}")
    return results
# 示例调用
api_key = "your_api_key_here"
companies = ["腾讯科技", "阿里巴巴"]
data = fetch_company_info(api_key, companies)

三、批量处理与异常处理

1. 批量请求优化

• 分批处理：避免单次请求数据量过大（如每次100条）。
• 并发请求：使用asyncio或multiprocessing提升速度（需注意API的QPS限制）。
  ```python
  import asyncio

async def fetch_async(api_key, company_names):
tasks = [asyncio.create_task(fetch_single(api_key, name)) for name in company_names]
return await asyncio.gather(*tasks)

async def fetch_single(api_key, name):

# 模拟异步请求
await asyncio.sleep(0.1)  # 模拟网络延迟
return {"name": name, "status": "success"}

#### 2. 异常处理机制
- **重试策略**：对失败请求自动重试（最多3次）。
- **日志记录**：记录失败原因（如API限额、网络错误）。
```python
def fetch_with_retry(api_key, name, max_retries=3):
    for attempt in range(max_retries):
        try:
            response = requests.get(base_url, params={"keyword": name, "api_key": api_key})
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            if attempt == max_retries - 1:
                print(f"Failed {name} after {max_retries} attempts: {e}")
                return None

四、数据存储与清洗

1. 存储格式选择

• CSV/Excel：适合结构化数据，便于Excel分析。
• 数据库（如MySQL、MongoDB）：适合长期存储与复杂查询。
  ```python

  存储为CSV

  df = pd.DataFrame(data)
  df.to_csv(“company_info.csv”, index=False, encoding=”utf-8-sig”)

存储到MySQL

import pymysql
conn = pymysql.connect(host=”localhost”, user=”root”, password=”123456”, database=”company_db”)
cursor = conn.cursor()
for item in data:
cursor.execute(“INSERT INTO companies VALUES (%s, %s, %s)”,
(item[“name”], item[“credit_code”], item[“reg_capital”]))
conn.commit()

#### 2. 数据清洗
- **去重**：根据统一社会信用代码去重。
- **字段标准化**：统一日期格式、金额单位。
```python
# 去重示例
df.drop_duplicates(subset=["credit_code"], inplace=True)
# 日期标准化
df["reg_date"] = pd.to_datetime(df["reg_date"]).dt.strftime("%Y-%m-%d")

五、合规性与注意事项

1. 数据使用合规：确保数据来源合法，避免侵犯商业秘密。
2. API调用限制：遵守第三方平台的QPS限制（如每秒最多10次）。
3. 反爬策略：若使用模拟浏览器方式，需设置随机User-Agent、代理IP池。

六、实战案例：供应商风险评估

场景：某制造企业需评估1000家供应商的注册资本与经营状态。
步骤：

1. 从ERP系统导出供应商名称列表。
2. 调用API批量获取工商信息。
3. 筛选注册资本低于100万或经营异常的供应商。
4. 生成风险报告并推送至业务部门。

代码片段：

# 筛选高风险供应商
high_risk = df[
    (df["reg_capital"] < 1000000) | 
    (df["status"] == "异常")
]
high_risk.to_excel("high_risk_suppliers.xlsx")

七、总结与建议

1. 优先使用官方API：若数据量小，可手动下载；若量大，选择第三方API更高效。
2. 自动化测试：在正式运行前，用少量数据验证流程正确性。
3. 定期更新：企业信息可能变更，建议每月更新一次数据。

通过Python实现企业工商信息的批量下载，不仅能节省人力成本，还能为业务决策提供数据支持。开发者可根据实际需求调整代码，灵活应对不同场景。