当前位置：首页 > news >正文

记一个判决书查询API接口的开发文档

news 来源：原创 2025/6/7 16:02:19

一、引言

在企业风控、背景调查、尽职调查等场景中，判决书查询是一个非常重要的环节。通过判决书查询，可以了解个人或企业的司法涉诉情况，为风险评估提供数据支持。本文将详细介绍如何开发和使用一个司法涉诉查询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_bytesclass ApiClient:def __init__(self, use_mock=None):"""初始化API客户端"""# 从配置获取API设置self.base_url = 'https://api.tianyuanapi.com'self.access_id = '您的ACCESS_ID'  # 请替换为实际的ACCESS_IDself.encryption_key = '您的加密密钥'  # 请替换为实际的加密密钥self.use_mock = False if use_mock is None else use_mockself.timeout = 30def 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:# 直接返回业务数据的APIreturn {'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 resultdef _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:# 调用APIapi_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_messagereturn result# 获取API返回的数据response_data = api_result.get('data', {})# 检查API状态码api_code = Noneapi_message = Noneapi_id = Noneif 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_coderesult["API流水号"] = api_idreturn 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 resultexcept Exception as e:print(f"司法涉诉查询异常: {str(e)}")return {"api_code": "FLXG0V4B","状态": "失败","错误信息": f"处理异常: {str(e)}","风险等级": "无风险","风险评分": 0,"风险描述": "查询过程发生异常","风险详情": {"各类案件统计": {"刑事案件": 0,"民事案件": 0,"行政案件": 0,"执行案件": 0,"强制清算与破产案件": 0,"非诉保全审查": 0,"失信被执行人": 0,"限制高消费": 0}}}