使用Python免费批量查询企业司法信息
在某些情况下需要查询一些企业的司法信息，比如：
了解企业的信用情况：通过查询企业司法信息，可以了解企业的诉讼情况、失信被执行人维度等信息，进而判断企业的信用情况。判断企业是否存在风险：企业的诉讼情况是判断一家企业是否存在风险的重要角度之一，透过企业所涉及到的诉讼可以了解到该企业的财力、信用度等情况，还能判断其是否存在爆雷的风险。为投资决策提供参考：无论是机构投资者还是个人投资者，在做出投资决策之前，都需要对目标企业的诚信度、生产经营情况、财务情况有一定的了解，以帮助其判断可能需要承担的风险和收益。对于个人的话就是相当于对企业的背调，这个企业是否有价值。
总之，对于不同的人有不同的需要。那么如何快速获取这些信息呢？答案就是找一个免费的查询网站，使用来实现。
逻辑分析
本文以钉钉企典为例，实现对企业的查询以及信息的存储。
输入基本信息
在首页输入企业名称后即可，以宁德时代为例。
获取企业列表
如图所示，这里获取到了若干个企业的名称，因为咱们是模糊查询的，如果企业名称更精确，展示的会更精准。
经过对比分析，只有 这个接口是我们想要的结果。令人开心的事这里的查询参数是明文的，没有做任何加密，只有企业名称是变量，其他参数都可以固定不变。接下来分析一些响应数据：
其实数据还有很多，但是标注的这些可能是能用得到的，但具体是哪个还不清楚，待会再继续看。
获取企业详情
这里可以看到URL里跟着的一个参数很眼熟，就是前面获取的数据中的ocid字段。
其实也就是这个接口里的参数拼接。但是这里有一个坑，ch接口里的数据并没有案号。只有一些别的信息，比如工商信息、高管信息、股东信息和变更信息等。明明页面里展示了司法信息的数量，但是并没有司法数据，这是为什么呢？
可以看到，当前选择的是基本信息，高亮的也是这个板块，所以会不会是因为没有选中司法风险导致的呢？
当我们点击司法风险版块的时候，发出了第二次请求，但是请求参数变了，所以会不会是增加了司法信息的参数查询，接下来看一下响应数据就可以确定了。
的确，在这里找到了多条案号信息。
至此，整套逻辑梳理完了，接下来就可以写代码了。
代码编写 搜索企业获取ocid
这里的就是企业名称。
async def get_company(c_name):async with semaphore:async with aiohttp.ClientSession() as session:data = {"keyword": c_name,"orderByType": 5,"pageNo": 1,"pageSize": 10}async with session.post(host_search, data=json.dumps(data), ssl=False, headers=headers,) as resp:if resp.status == 200:try:res = await resp.json()ocid = res.get('data').get('data')[0]['ocid']print(c_name, ocid)com_info = {'name': c_name, 'ocid': ocid}await get_card_batch(com_info)except Exception as e:print(e)else:# 处理状态码不是 200 的情况print(f"Request to {host_search} failed with status code {resp.status} {resp.json()}")获取企业案号
async def get_card_batch(info):async with aiohttp.ClientSession() as session:data = {"dataModuleIds": [460, 456, 455, 462, 457, 490, 470, 458, 495], "type": "web", "pageNo": 1,"pageSize": 10, "params": [{"key": "onecomp_id", "value": info.get('ocid')}]}async with session.post(card_batch, data=json.dumps(data), ssl=False, headers=headers,) as resp:if resp.status == 200:try:res = await resp.json()rows = res.get('data')[1]['rows']if rows:tmp_list = [row[4] for row in rows]info['rows'] = ",".join(tmp_list)else:info['rows'] = ""save_to_csv([[info.get('name'), info.get('rows')]])print(info['rows'])except Exception as e:print(e)else:# 处理状态码不是 200 的情况print(f"Request to {host_search} failed with status code {resp.status} {resp.json()}")await asyncio.sleep(random.uniform(1.0, 10.0))
数据存储
def save_to_csv(data):with open(output_file, 'a', newline='') as file:writer = csv.writer(file)for sublist in data:writer.writerow(sublist)
