#!/usr/bin/env python3 """ 批量调用远程 FLXG7E8F，分析执行/失信/限高三类案件中的金额字段。 金额字段含义（见 个人司法涉诉查询_返回字段说明）: 执行案件 implement.cases[]: n_sqzxbdje 申请执行标的金额 n_sjdwje 实际到位金额（已还款/已执行到位，repaidAmount 应对应此字段） n_wzxje 未执行金额（未还款/未执行到位，正确的「未还款金额」应对应此字段） n_jabdje 结案标的金额（民事结案金额，不能当作还款金额） 失信 breachCaseList[]: estimatedJudgementAmount 判决金额估计（非未还款字段） 无 n_wzxje；可尝试按案号关联 implement 案件取 n_wzxje 限高 consumptionRestrictionList[]: 无金额字段；可尝试按案号关联 implement 案件取 n_wzxje 依赖: pip install pycryptodome requests 示例: python batch_flxg7e8f_analyze.py \\ --input ..\\query_1-2026-06-27_32042_decrypted.json \\ --access-id <你的AccessId> \\ --secret-key f507c58537aac8227b5f4a99cbd317ff \\ --output ..\\flxg7e8f_analysis.json \\ --limit 3 获取 access_id: SELECT access_id, secret_key FROM api_users WHERE user_id = '2acfef92-0700-4638-b386-bd03a8ad09c3'; """ from __future__ import annotations import argparse import base64 import json import os import re import sys import time from pathlib import Path from typing import Any try: import requests from Crypto.Cipher import AES from Crypto.Random import get_random_bytes except ImportError: print("请先安装: pip install pycryptodome requests", file=sys.stderr) sys.exit(1) DEFAULT_BASE_URL = "https://api.tianyuanapi.com" API_NAME = "FLXG7E8F" def aes_encrypt(plain_text: bytes, key_hex: str) -> str: key = bytes.fromhex(key_hex.strip()) block = 16 pad = block - len(plain_text) % block plain_text = plain_text + bytes([pad]) * pad iv = get_random_bytes(16) ct = AES.new(key, AES.MODE_CBC, iv).encrypt(plain_text) return base64.b64encode(iv + ct).decode() def aes_decrypt(cipher_b64: str, key_hex: str) -> Any: key = bytes.fromhex(key_hex.strip()) raw = base64.b64decode(cipher_b64.strip()) iv, ciphertext = raw[:16], raw[16:] plain = AES.new(key, AES.MODE_CBC, iv).decrypt(ciphertext) pad = plain[-1] plain = plain[:-pad] return json.loads(plain.decode("utf-8")) def to_float(val: Any) -> float | None: if val is None or val == "": return None if isinstance(val, (int, float)): return float(val) if isinstance(val, str): s = val.strip().replace(",", "") if not s: return None try: return float(s) except ValueError: return None return None def normalize_case_no(case_no: str) -> str: if not case_no: return "" s = case_no.strip().upper() s = re.sub(r"\s+", "", s) return s def pick_amount_fields(case_map: dict) -> dict[str, Any]: sqzx = to_float(case_map.get("n_sqzxbdje")) sjdw = to_float(case_map.get("n_sjdwje")) wzx = to_float(case_map.get("n_wzxje")) jabd = to_float(case_map.get("n_jabdje")) computed_unpaid = None if sqzx is not None and sjdw is not None: computed_unpaid = max(sqzx - sjdw, 0.0) has_correct_unpaid_field = wzx is not None has_correct_repaid_field = sjdw is not None wrong_if_use_jabdje_as_repaid = ( not has_correct_repaid_field and jabd is not None and jabd > 0 ) return { "n_sqzxbdje": sqzx, "n_sjdwje": sjdw, "n_wzxje": wzx, "n_jabdje": jabd, "computed_unpaid_sqzx_minus_sjdw": computed_unpaid, "has_correct_unpaid_field": has_correct_unpaid_field, "has_correct_repaid_field": has_correct_repaid_field, "wrong_if_use_jabdje_as_repaid": wrong_if_use_jabdje_as_repaid, "unpaid_amount": wzx if wzx is not None else computed_unpaid, } def build_implement_index(implement_cases: list[dict]) -> dict[str, dict]: index: dict[str, dict] = {} for case in implement_cases: case_no = normalize_case_no(str(case.get("c_ah") or "")) if not case_no: continue amounts = pick_amount_fields(case) index[case_no] = { "case_number": case.get("c_ah"), "case_status": case.get("n_ajjzjd"), "filing_date": case.get("d_larq"), "close_date": case.get("d_jarq"), **amounts, } return index def is_case_fully_paid(amounts: dict) -> bool: """执行案件是否已全部到位。""" wzx = amounts.get("n_wzxje") if wzx is not None: return wzx <= 0 sqzx = amounts.get("n_sqzxbdje") sjdw = amounts.get("n_sjdwje") if sqzx is not None and sjdw is not None: return sjdw >= sqzx unpaid = amounts.get("unpaid_amount") if unpaid is not None: return unpaid <= 0 return True def has_outstanding_unpaid(amounts: dict) -> bool: wzx = amounts.get("n_wzxje") if wzx is not None: return wzx > 0 unpaid = amounts.get("unpaid_amount") if unpaid is not None: return unpaid > 0 sqzx = amounts.get("n_sqzxbdje") sjdw = amounts.get("n_sjdwje") if sqzx is not None and sjdw is not None and sqzx > sjdw: return True return False def classify_analysis(analysis: dict) -> tuple[str, str]: """ 判定是否有「金额映射问题」需要关注: NO_ISSUE - 无执行/失信/限高，或金额已全部到位 HAS_OUTSTANDING_UNPAID - 存在未结/未执行金额，需用 n_wzxje / n_sjdwje 正确映射 """ if not analysis.get("hit"): return "NO_ISSUE", "未命中司法数据" implement_cases = analysis.get("implement_cases") or [] breach_cases = analysis.get("breach_cases") or [] limit_cases = analysis.get("limit_cases") or [] if not implement_cases and not breach_cases and not limit_cases: return "NO_ISSUE", "无执行/失信/限高案件" outstanding_details: list[str] = [] for case in implement_cases: if has_outstanding_unpaid(case): outstanding_details.append(f"执行 {case.get('case_number')} 未执行金额={case.get('n_wzxje') or case.get('unpaid_amount')}") for case in breach_cases: fulfill = (case.get("fulfill_status") or "").strip() linked = case.get("linked_implement") or {} if fulfill == "全部未履行": if linked and has_outstanding_unpaid(linked): outstanding_details.append(f"失信 {case.get('case_number')} 关联未执行={linked.get('n_wzxje')}") elif not linked: outstanding_details.append(f"失信 {case.get('case_number')} 全部未履行(无关联执行明细)") elif linked and has_outstanding_unpaid(linked): outstanding_details.append(f"失信 {case.get('case_number')} 关联未执行={linked.get('n_wzxje')}") for case in limit_cases: linked = case.get("linked_implement") or {} if linked and has_outstanding_unpaid(linked): outstanding_details.append(f"限高 {case.get('case_number')} 关联未执行={linked.get('n_wzxje')}") if outstanding_details: return "HAS_OUTSTANDING_UNPAID", "; ".join(outstanding_details) return "NO_ISSUE", "有执行/失信/限高案件，但金额已全部到位或无未结金额" def analyze_judicial_data(judicial_data: dict) -> dict[str, Any]: if not judicial_data or judicial_data == -1: return {"hit": False, "message": "未命中司法数据"} lawsuit_stat = judicial_data.get("lawsuitStat") or {} implement_section = lawsuit_stat.get("implement") or {} implement_raw = implement_section.get("cases") or [] implement_cases = [] for item in implement_raw: if isinstance(item, dict): case_no = item.get("c_ah") amounts = pick_amount_fields(item) implement_cases.append( { "case_number": case_no, "case_status": item.get("n_ajjzjd"), "filing_date": item.get("d_larq"), "close_date": item.get("d_jarq"), **amounts, } ) implement_index = build_implement_index( [c for c in implement_raw if isinstance(c, dict)] ) breach_cases = [] for item in judicial_data.get("breachCaseList") or []: if not isinstance(item, dict): continue case_no = normalize_case_no(str(item.get("caseNumber") or "")) linked = implement_index.get(case_no) est = to_float(item.get("estimatedJudgementAmount")) breach_cases.append( { "case_number": item.get("caseNumber"), "fulfill_status": item.get("fulfillStatus"), "estimated_judgement_amount": est, "issue_date": item.get("issueDate"), "linked_implement": linked, "has_correct_unpaid_via_link": bool( linked and linked.get("has_correct_unpaid_field") ), "unpaid_amount": linked.get("unpaid_amount") if linked else None, } ) limit_cases = [] for item in judicial_data.get("consumptionRestrictionList") or []: if not isinstance(item, dict): continue case_no = normalize_case_no(str(item.get("caseNumber") or "")) linked = implement_index.get(case_no) limit_cases.append( { "case_number": item.get("caseNumber"), "issue_date": item.get("issueDate"), "file_date": item.get("fileDate"), "court": item.get("executiveCourt"), "linked_implement": linked, "has_correct_unpaid_via_link": bool( linked and linked.get("has_correct_unpaid_field") ), "unpaid_amount": linked.get("unpaid_amount") if linked else None, } ) implement_with_outstanding = [c for c in implement_cases if has_outstanding_unpaid(c)] breach_with_outstanding = [ c for c in breach_cases if (c.get("fulfill_status") == "全部未履行") or (c.get("linked_implement") and has_outstanding_unpaid(c["linked_implement"])) ] limit_with_outstanding = [ c for c in limit_cases if c.get("linked_implement") and has_outstanding_unpaid(c["linked_implement"]) ] result = { "hit": True, "summary": { "implement_count": len(implement_cases), "breach_count": len(breach_cases), "limit_count": len(limit_cases), "implement_with_outstanding_unpaid": len(implement_with_outstanding), "breach_with_outstanding": len(breach_with_outstanding), "limit_with_outstanding": len(limit_with_outstanding), }, "implement_cases": implement_cases, "breach_cases": breach_cases, "limit_cases": limit_cases, } flag, reason = classify_analysis(result) result["flag"] = flag result["flag_reason"] = reason return result def call_flxg7e8f( base_url: str, access_id: str, secret_key: str, name: str, id_card: str, timeout: int = 60, ) -> tuple[bool, dict[str, Any]]: url = f"{base_url.rstrip('/')}/api/v1/{API_NAME}" params = {"name": name, "id_card": id_card} encrypted = aes_encrypt(json.dumps(params, ensure_ascii=False).encode("utf-8"), secret_key) payload = {"data": encrypted, "options": {"json": True}} resp = requests.post( url, headers={"Access-Id": access_id, "Content-Type": "application/json"}, json=payload, timeout=timeout, ) resp.raise_for_status() body = resp.json() if body.get("code") != 0: return False, { "error_code": body.get("code"), "error_message": body.get("message"), "transaction_id": body.get("transaction_id"), } decrypted = aes_decrypt(body["data"], secret_key) return True, { "transaction_id": body.get("transaction_id"), "response": decrypted, } def load_input_records(path: Path) -> list[dict]: data = json.loads(path.read_text(encoding="utf-8-sig")) if not isinstance(data, list): raise ValueError("输入 JSON 应为数组") return data def dedupe_persons(records: list[dict]) -> list[dict]: seen: set[str] = set() persons = [] for rec in records: rp = rec.get("request_params") or {} if isinstance(rp, str): continue id_card = (rp.get("id_card") or "").strip().upper() name = (rp.get("name") or "").strip() if not id_card or not name: continue if id_card in seen: continue seen.add(id_card) persons.append( { "name": name, "id_card": id_card, "source_ids": [rec.get("id")], "source_transaction_ids": [rec.get("transaction_id")], } ) # attach all source ids for duplicates id_to_person = {p["id_card"]: p for p in persons} for rec in records: rp = rec.get("request_params") or {} if isinstance(rp, str): continue id_card = (rp.get("id_card") or "").strip().upper() if id_card in id_to_person: p = id_to_person[id_card] rid, rtx = rec.get("id"), rec.get("transaction_id") if rid and rid not in p["source_ids"]: p["source_ids"].append(rid) if rtx and rtx not in p["source_transaction_ids"]: p["source_transaction_ids"].append(rtx) return persons def main() -> None: parser = argparse.ArgumentParser(description="批量 FLXG7E8F + 未还款金额分析") parser.add_argument("--input", required=True, help="解密后的 COMBMY01 JSON") parser.add_argument("--output", default="flxg7e8f_analysis.json") parser.add_argument("--raw-output", default="flxg7e8f_raw.json", help="FLXG7E8F 解密后源数据汇总") parser.add_argument("--access-id", default=os.getenv("TYAPI_ACCESS_ID")) parser.add_argument("--secret-key", default=os.getenv("TYAPI_SECRET_KEY")) parser.add_argument("--base-url", default=os.getenv("TYAPI_BASE_URL", DEFAULT_BASE_URL)) parser.add_argument("--delay", type=float, default=0.5, help="每次请求间隔秒数") parser.add_argument("--limit", type=int, default=0, help="仅处理前 N 个唯一身份证（0=全部）") parser.add_argument("--resume", help="已有结果 JSON，跳过已成功的人员") parser.add_argument("--dry-run", action="store_true", help="只列出待调用人员，不请求 API") args = parser.parse_args() if not args.secret_key: print("请提供 --secret-key 或环境变量 TYAPI_SECRET_KEY", file=sys.stderr) sys.exit(1) input_path = Path(args.input) records = load_input_records(input_path) persons = dedupe_persons(records) if args.limit > 0: persons = persons[: args.limit] print(f"输入记录 {len(records)} 条, 去重后 {len(persons)} 人") existing: dict[str, dict] = {} if args.resume and Path(args.resume).is_file(): prev = json.loads(Path(args.resume).read_text(encoding="utf-8")) for item in prev.get("persons", []): if item.get("api_success"): existing[item["id_card"]] = item if args.dry_run: for p in persons: print(f"{p['name']}\t{p['id_card']}\tsources={len(p['source_ids'])}") return if not args.access_id: print("请提供 --access-id 或环境变量 TYAPI_ACCESS_ID", file=sys.stderr) sys.exit(1) results = [] raw_records = [] ok_count = fail_count = no_issue_count = outstanding_count = 0 for idx, person in enumerate(persons, start=1): id_card = person["id_card"] if id_card in existing: cached = existing[id_card] results.append(cached) if cached.get("raw_response"): raw_records.append( { "name": cached.get("name"), "id_card": id_card, "transaction_id": cached.get("transaction_id"), "raw_response": cached.get("raw_response"), } ) print(f"[{idx}/{len(persons)}] 跳过(已存在) {person['name']} {id_card}") continue print(f"[{idx}/{len(persons)}] 调用 {person['name']} {id_card} ...", flush=True) item = { "name": person["name"], "id_card": id_card, "source_ids": person["source_ids"], "source_transaction_ids": person["source_transaction_ids"], "api_success": False, } try: success, api_result = call_flxg7e8f( args.base_url, args.access_id, args.secret_key, person["name"], id_card ) item["api_success"] = success if not success: item.update(api_result) fail_count += 1 else: item["transaction_id"] = api_result.get("transaction_id") item["raw_response"] = api_result.get("response") raw_records.append( { "name": person["name"], "id_card": id_card, "transaction_id": item["transaction_id"], "source_ids": person["source_ids"], "raw_response": item["raw_response"], } ) judicial = (api_result.get("response") or {}).get("judicial_data") analysis = analyze_judicial_data(judicial) item["analysis"] = analysis item["flag"] = analysis.get("flag", "NO_ISSUE") item["flag_reason"] = analysis.get("flag_reason", "") ok_count += 1 if item["flag"] == "HAS_OUTSTANDING_UNPAID": outstanding_count += 1 else: no_issue_count += 1 except Exception as exc: item["api_success"] = False item["error_message"] = str(exc) fail_count += 1 results.append(item) # 增量保存，防止中断丢数据 out_path = Path(args.output) raw_path = Path(args.raw_output) payload = { "meta": { "input": str(input_path), "total_records": len(records), "unique_persons": len(persons), "processed": len(results), }, "persons": results, } out_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") raw_path.write_text( json.dumps( { "meta": payload["meta"], "records": raw_records, }, ensure_ascii=False, indent=2, ), encoding="utf-8", ) if idx < len(persons) and args.delay > 0: time.sleep(args.delay) summary = { "api_ok": ok_count, "api_fail": fail_count, "no_issue": no_issue_count, "has_outstanding_unpaid": outstanding_count, } final = { "meta": { "input": str(input_path), "total_records": len(records), "unique_persons": len(persons), "summary": summary, "raw_output": str(Path(args.raw_output).resolve()), }, "persons": results, } Path(args.output).write_text(json.dumps(final, ensure_ascii=False, indent=2), encoding="utf-8") Path(args.raw_output).write_text( json.dumps({"meta": final["meta"], "records": raw_records}, ensure_ascii=False, indent=2), encoding="utf-8", ) print("\n=== 完成 ===") print(json.dumps(summary, ensure_ascii=False, indent=2)) print(f"分析结果: {Path(args.output).resolve()}") print(f"源数据: {Path(args.raw_output).resolve()}") if __name__ == "__main__": main()