一、企业信息查询的常见场景与Python技术选型

企业信息查询是商业分析、风险控制和供应链管理中的高频需求，典型场景包括：

1. 工商信息核验：验证企业注册状态、股东结构、注册资本等基础信息
2. 信用风险评估：获取企业信用评级、司法诉讼、行政处罚等动态数据
3. 供应链尽调：批量查询上下游企业的经营状况与关联关系

Python凭借其丰富的第三方库和简洁的语法，成为实现企业信息查询的首选工具。核心依赖库包括：

• requests：处理HTTP请求，实现与第三方API的交互
• json：解析API返回的JSON格式数据
• pandas：对批量查询结果进行结构化存储与分析
• matplotlib/seaborn：可视化企业信息分布规律

二、主流企业信息API的接入与认证

1. 国家企业信用信息公示系统API

该系统提供官方权威数据，但需通过政府采购或授权服务商接入。典型API接口包括：

import requests
def query_company_by_name(api_key, company_name):
    url = "https://api.gsxt.gov.cn/company/search"
    params = {
        "key": api_key,
        "name": company_name,
        "page": 1,
        "size": 10
    }
    response = requests.get(url, params=params)
    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"API请求失败: {response.status_code}")

关键参数说明：

• api_key：需在政府平台申请的唯一标识
• name：支持模糊匹配的企业名称
• 返回数据包含统一社会信用代码、注册地址、法定代表人等核心字段

2. 第三方商业API（以天眼查为例）

商业API通常提供更丰富的字段和更高的调用频率。示例代码：

def query_company_by_credit_code(api_key, credit_code):
    url = "https://api.tianyancha.com/services/v3/open/company/search"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    data = {"creditCode": credit_code}
    response = requests.post(url, headers=headers, json=data)
    return response.json()

差异化优势：

• 支持通过统一社会信用代码精准查询
• 返回数据包含股东信息、对外投资、司法风险等深度字段
• 需注意商业API的调用配额（如天眼查免费版每日50次）

三、企业信息数据的解析与清洗

API返回的原始数据通常包含嵌套结构，需通过递归解析提取关键信息。示例代码：

def parse_company_info(raw_data):
    parsed_data = {
        "name": raw_data.get("name", ""),
        "credit_code": raw_data.get("creditCode", ""),
        "legal_person": raw_data.get("legalPersonName", ""),
        "reg_capital": raw_data.get("regCapital", ""),
        "shareholders": []
    }
    # 解析股东信息（嵌套结构）
    if "shareholderList" in raw_data:
        for shareholder in raw_data["shareholderList"]:
            parsed_data["shareholders"].append({
                "name": shareholder.get("name", ""),
                "type": shareholder.get("type", ""),
                "ratio": shareholder.get("ratio", "")
            })
    return parsed_data

数据清洗要点：

1. 字段缺失处理：使用.get()方法避免KeyError
2. 类型转换：将字符串类型的金额转换为数值（如float(reg_capital.replace("万", ""))）
3. 异常值过滤：剔除注册资本为负数或空值的记录

四、批量查询与性能优化

1. 多线程并发查询

当需查询大量企业时，单线程串行调用效率低下。可通过concurrent.futures实现并发：

from concurrent.futures import ThreadPoolExecutor
def batch_query(api_key, company_names, max_workers=5):
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(query_company_by_name, api_key, name) for name in company_names]
        for future in futures:
            try:
                results.append(future.result())
            except Exception as e:
                print(f"查询失败: {e}")
    return results

参数调优建议：

• max_workers设置为API允许的最大并发数（通常为5-10）
• 添加重试机制（如requests.Session()配合retry装饰器）

2. 本地缓存策略

为避免重复查询，可将结果存储至SQLite数据库：

import sqlite3
def init_db():
    conn = sqlite3.connect("company_data.db")
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS companies (
            id INTEGER PRIMARY KEY,
            name TEXT,
            credit_code TEXT UNIQUE,
            legal_person TEXT,
            reg_capital REAL,
            query_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    conn.close()
def save_to_db(company_data):
    conn = sqlite3.connect("company_data.db")
    cursor = conn.cursor()
    try:
        cursor.execute("""
            INSERT INTO companies (name, credit_code, legal_person, reg_capital)
            VALUES (?, ?, ?, ?)
        """, (
            company_data["name"],
            company_data["credit_code"],
            company_data["legal_person"],
            float(company_data["reg_capital"].replace("万", ""))
        ))
        conn.commit()
    except sqlite3.IntegrityError:
        print("记录已存在，跳过插入")
    finally:
        conn.close()

五、高级应用场景扩展

1. 企业关联关系分析

通过查询股东和对外投资数据，可构建企业关联图谱。示例代码：

import networkx as nx
import matplotlib.pyplot as plt
def build_relation_graph(companies):
    G = nx.Graph()
    for company in companies:
        G.add_node(company["name"])
        for shareholder in company["shareholders"]:
            if shareholder["type"] == "企业":
                G.add_edge(company["name"], shareholder["name"])
    nx.draw(G, with_labels=True, node_size=1000, node_color="skyblue")
    plt.show()

2. 实时风险预警系统

结合司法诉讼API，可监控目标企业的新增风险：

def monitor_risks(company_name, interval=3600):
    import time
    last_risk_count = 0
    while True:
        data = query_company_by_name(api_key, company_name)
        current_risk_count = len(data.get("lawsuits", []))
        if current_risk_count > last_risk_count:
            print(f"⚠️ 风险预警: {company_name}新增{current_risk_count - last_risk_count}条诉讼")
        last_risk_count = current_risk_count
        time.sleep(interval)

六、最佳实践与避坑指南

1. API选择策略：

   • 优先使用政府免费API（如国家企业信用信息公示系统）
   • 深度查询需购买商业API（推荐天眼查、企查查等）

2. 反爬虫应对：

   • 设置合理的User-Agent和请求间隔
   • 避免短时间内高频调用（建议QPS≤5）

3. 数据合规性：

   • 仅用于合法商业目的，不得用于非法爬取
   • 存储数据需符合《个人信息保护法》要求

4. 错误处理机制：

   def safe_query(api_func, *args, **kwargs):
       try:
           return api_func(*args, **kwargs)
       except requests.exceptions.RequestException as e:
           print(f"网络错误: {e}")
           return None
       except json.JSONDecodeError:
           print("返回数据解析失败")
           return None

通过上述方法，开发者可构建高效、稳定的企业信息查询系统，为商业决策提供数据支撑。实际开发中需根据具体需求调整API选择和数据处理逻辑，同时严格遵守相关法律法规。