一、引言
在企业风控、背景调查、尽职调查等场景中，判决书查询是一个非常重要的环节。通过判决书查询，可以了解个人或企业的司法涉诉情况，为风险评估提供数据支持。本文将详细介绍如何开发和使用一个司法涉诉查询API接口，包括客户端实现、数据处理与风险评估等核心技术点。
二、API概述
本文介绍的"个人司法涉诉（详版）"API可查询个人的司法涉诉信息，包括失信被执行人、执行信息、刑事案件等多维度数据。该API通过加密传输，确保数据安全，返回标准化的风险评估结果。
- API代码：FLXG0V4B
- API名称：个人司法涉诉（详版）
- 请求端点：https://api.tianyuanapi.com/api/v1/FLXG0V4B
- 获取密钥：https://tianyuanapi.com
三、API客户端实现
3.1 ApiClient类
首先，我们需要实现一个API客户端类处理与
import json
import time
import base64
import requests
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad, unpad
from Crypto.Random import get_random_bytes
class ApiClient:
def __init__(self, use_mock=None):
"""初始化API客户端"""
# 从配置获取API设置
self.base_url = 'https://api.tianyuanapi.com'
self.access_id = '您的ACCESS_ID' # 请替换为实际的ACCESS_ID
self.encryption_key = '您的加密密钥' # 请替换为实际的加密密钥
self.use_mock = False if use_mock is None else use_mock
self.timeout = 30
def call_api(self, api_code, params, query_id=None):
"""
调用API接口
Args:
api_code: API代码
params: 请求参数
query_id: 查询ID，用于记录日志
Returns:
解密后的响应数据
"""
# 如果是模拟模式，直接返回模拟数据
if self.use_mock:
return self._get_mock_response(api_code, params)
try:
# 将参数转换为JSON字符串
params_json = json.dumps(params, ensure_ascii=False)
# 对请求参数进行加密
encrypted_data = self._encrypt_data(params_json)
# 构建请求头
headers = {
'Content-Type': 'application/json',
'Access-Id': self.access_id
}
# API请求地址
url = f"{self.base_url}/api/v1/{api_code}"
# 发送请求
response = requests.post(
url,
json={'data': encrypted_data},
headers=headers,
timeout=self.timeout
)
# 检查响应状态码
if response.status_code != 200:
return {
'success': False,
'code': response.status_code,
'message': f"API请求失败，状态码: {response.status_code}",
'data': None
}
# 解析响应数据
response_json = response.json()
# 检查响应格式
if 'code' not in response_json:
# 直接返回业务数据的API
return {
'success': True,
'code': 0,
'message': '成功',
'data': response_json
}
# 标准API响应处理
api_response_code = response_json.get('code')
api_message = response_json.get('message', '')
encrypted_response = response_json.get('data', '')
# 判断API响应码
if api_response_code != 0:
return {
'success': False,
'code': api_response_code,
'message': api_message,
'data': None
}
# 如果没有返回加密数据
if not encrypted_response:
return {
'success': True,
'code': 0,
'message': api_message,
'data': {}
}
# 解密响应数据
decrypted_data = self._decrypt_data(encrypted_response)
result = json.loads(decrypted_data)
return {
'success': True,
'code': 0,
'message': api_message,
'data': result
}
except Exception as e:
return {
'success': False,
'code': -1,
'message': f"调用API异常: {str(e)}",
'data': None
}
def _encrypt_data(self, data):
"""AES-128-CBC加密数据"""
# 将16进制密钥转换为字节
key = bytes.fromhex(self.encryption_key)
# 生成随机IV（初始化向量）
iv = get_random_bytes(16)
# 创建AES-CBC加密器
cipher = AES.new(key, AES.MODE_CBC, iv)
# 对数据进行填充并加密
padded_data = pad(data.encode('utf-8'), AES.block_size)
encrypted_data = cipher.encrypt(padded_data)
# 将IV和加密后的数据拼接，并进行Base64编码
result = base64.b64encode(iv + encrypted_data).decode('utf-8')
return result
def _decrypt_data(self, encrypted_data):
"""AES-128-CBC解密数据"""
# 将16进制密钥转换为字节
key = bytes.fromhex(self.encryption_key)
# Base64解码
encrypted_bytes = base64.b64decode(encrypted_data)
# 提取IV和加密数据
iv = encrypted_bytes[:16]
ciphertext = encrypted_bytes[16:]
# 创建AES-CBC解密器
cipher = AES.new(key, AES.MODE_CBC, iv)
# 解密并去除填充
padded_data = cipher.decrypt(ciphertext)
decrypted_data = unpad(padded_data, AES.block_size).decode('utf-8')
return decrypted_data服务器的通信，包括请求加密、发送和响应解密：
四、司法涉诉查询接口实现
4.1 查询函数实现
def query(name, id_card, query_id=None, use_mock=False):
"""
查询个人司法涉诉信息
Args:
name: 姓名
id_card: 身份证号码
query_id: 查询ID，用于日志记录
use_mock: 是否使用模拟数据
Returns:
查询结果字典
"""
start_time = time.time()
print(f"开始查询个人司法涉诉信息: name={name}, id_card={id_card[:6]}****{id_card[-4:]}")
# 参数验证
if not name or not id_card:
print("参数错误：姓名和身份证号不能为空")
return {
"api_code": "FLXG0V4B",
"状态": "失败",
"错误信息": "姓名和身份证号不能为空"
}
# 计算授权时间范围（当前日期前后三天）
today = datetime.now()
start_date = today - timedelta(days=3)
end_date = today + timedelta(days=3)
auth_date = f"{start_date.strftime('%Y%m%d')}-{end_date.strftime('%Y%m%d')}"
# 准备API请求参数
params = {
"name": name,
"id_card": id_card,
"auth_date": auth_date,
"description": "个人司法涉诉（详版）"
}
try:
# 调用API
api_client = ApiClient(use_mock=use_mock)
api_result = api_client.call_api("FLXG0V4B", params, query_id=query_id)
# 初始化基本返回结构
result = {
"api_code": "FLXG0V4B",
"状态": "失败",
"风险等级": "无风险",
"风险评分": 0,
"风险描述": "未查询到司法涉诉记录",
"风险详情": {
"案件统计": {},
"案件列表": [],
"各类案件统计": {
"刑事案件": 0,
"民事案件": 0,
"行政案件": 0,
"执行案件": 0,
"强制清算与破产案件": 0,
"非诉保全审查": 0,
"失信被执行人": 0,
"限制高消费": 0
}
}
}
# 处理API响应
if not api_result.get('success'):
error_message = api_result.get('message', '未知错误')
result["错误信息"] = error_message
return result
# 获取API返回的数据
response_data = api_result.get('data', {})
# 检查API状态码
api_code = None
api_message = None
api_id = None
if isinstance(response_data, dict):
api_code = response_data.get("code", "")
api_message = response_data.get("message", "")
api_id = response_data.get("id", "")
# 处理API状态码
if api_code and api_code != "00000" and api_code != "200":
result["状态"] = "失败"
result["错误信息"] = f"{api_message}"
result["API状态码"] = api_code
result["API流水号"] = api_id
return result
# API调用成功，解析结果
result["状态"] = "成功"
result["API状态码"] = api_code or "00000"
result["API流水号"] = api_id or ""
# 判断结果是否为空
if not response_data or (isinstance(response_data, dict) and "data" not in response_data):
print("API返回空结果")
result["风险描述"] = "未查询到司法涉诉记录"
return result
# 解析司法涉诉数据，评估风险
risk_assessment = assess_judicial_risk(response_data)
# 填充风险评估结果
result["风险等级"] = risk_assessment.get("风险等级", "无风险")
result["风险评分"] = risk_assessment.get("风险评分", 0)
result["风险描述"] = risk_assessment.get("风险描述", "未查询到司法涉诉记录")
result["风险详情"]["案件统计"] = risk_assessment.get("案件统计", {})
result["风险详情"]["案件列表"] = risk_assessment.get("案件列表", [])
result["风险详情"]["各类案件统计"] = risk_assessment.get("各类案件统计", {})
return result
except Exception as e:
print(f"司法涉诉查询异常: {str(e)}")
return {
"api_code": "FLXG0V4B",
"状态": "失败",
"错误信息": f"处理异常: {str(e)}",
"风险等级": "无风险",
"风险评分": 0,
"风险描述": "查询过程发生异常",
"风险详情": {
"各类案件统计": {
"刑事案件": 0,
"民事案件": 0,
"行政案件": 0,
"执行案件": 0,
"强制清算与破产案件": 0,
"非诉保全审查": 0,
"失信被执行人": 0,
"限制高消费": 0
}
}
}最佳实践
- 合规使用：根据《数据安全法》和《个人信息保护法》，查询个人司法涉诉信息需获得被查询人授权。本API设计了auth_date参数用于传递授权日期信息。
- 错误处理：在生产环境中，务必添加完善的错误处理机制，应对可能出现的网络超时、服务器错误等异常情况。
- 结果缓存：对于频繁查询的对象，考虑实现缓存机制，减少API调用次数，降低成本。
- 日志记录：记录每次API调用的请求参数和响应结果，便于后期问题排查。
- 响应解析：API返回的数据可能非常复杂，建议实现专门的解析函数，提取关键信息。
- 异步处理：对于大批量查询场景，考虑使用异步处理方式，如Celery任务队列，避免阻塞主流程。
八、总结
本文详细介绍了个人司法涉诉查询API的开发与使用，包括客户端实现、数据处理与风险评估算法。通过该API，可以全面了解个人的司法涉诉情况，为风险控制、尽职调查等业务场景提供数据支持。
开发过程中，我们需要重视数据安全和隐私保护，确保在合规的前提下使用API。同时，针对不同业务场景，可对风险评估算法进行调整，优化风险评分的精确度和实用性。
![[论文阅读] (38)基于大模型的威胁情报分析与知识图谱构建论文总结（读书笔记）](https://i-blog.csdnimg.cn/direct/f33212b2a7404aec8573d2bf5696ea93.png#pic_center)
