f

scripts/export_ivyz4y27_calls.py

@@ -0,0 +1,238 @@
+#!/usr/bin/env python3
+"""
+导出「河南钟馗科技有限公司」学历信息高级版(IVYZ4Y27)成功调用记录为 CSV。
+解密 request_params，仅输出脱敏后的姓名、身份证。
+
+依赖: pip install pycryptodome psycopg2-binary
+
+用法 A - 直连数据库（在可访问生产/测试库的环境执行）:
+  python export_ivyz4y27_calls.py \\
+    --from-db \\
+    --db-host tyapi-postgres-prod --db-name tyapi --db-user postgres --db-password <pwd> \\
+    -o ivyz4y27_zhongkui_calls.csv
+
+  或使用环境变量:
+  set PGHOST=... PGDATABASE=... PGUSER=... PGPASSWORD=...
+  python export_ivyz4y27_calls.py --from-db -o ivyz4y27_zhongkui_calls.csv
+
+用法 B - 先 SQL 导出 JSON，再本地转 CSV:
+  python export_ivyz4y27_calls.py -i ivyz4y27_zhongkui_raw.json -o ivyz4y27_zhongkui_calls.csv
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import csv
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Any
+
+try:
+    from Crypto.Cipher import AES
+except ImportError:
+    print("请先安装: pip install pycryptodome", file=sys.stderr)
+    sys.exit(1)
+
+COMPANY_NAME = "河南钟馗科技有限公司"
+PRODUCT_CODE = "IVYZ4Y27"
+
+SQL = """
+SELECT
+    ac.id,
+    ac.transaction_id,
+    ac.user_id,
+    ei.company_name,
+    p.code          AS product_code,
+    p.name          AS product_name,
+    ac.request_params,
+    au.secret_key,
+    ac.status,
+    ac.cost,
+    ac.start_at,
+    ac.created_at
+FROM api_calls ac
+INNER JOIN products p ON ac.product_id = p.id
+INNER JOIN api_users au ON ac.user_id = au.user_id
+INNER JOIN enterprise_infos ei ON ac.user_id = ei.user_id
+WHERE ei.company_name = %s
+  AND ei.deleted_at IS NULL
+  AND p.code = %s
+  AND ac.status = 'success'
+ORDER BY ac.created_at DESC
+"""
+
+
+def aes_decrypt(cipher_b64: str, key_hex: str) -> dict[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 mask_name(name: str) -> str:
+    """与 dwbg8b4d maskName 一致：张*三 / 李*"""
+    name = (name or "").strip()
+    if not name:
+        return ""
+    chars = list(name)
+    if len(chars) <= 1:
+        return name
+    if len(chars) == 2:
+        return chars[0] + "*"
+    return chars[0] + "*" + chars[-1]
+
+
+def mask_id_card(id_card: str) -> str:
+    """与 query_whitelist_helpers MaskIDCard 一致：350681********0611"""
+    id_card = (id_card or "").strip().upper()
+    if len(id_card) <= 10:
+        return id_card
+    return id_card[:6] + "********" + id_card[-4:]
+
+
+def extract_name_id_card(params: dict[str, Any]) -> tuple[str, str]:
+    name = params.get("name") or params.get("Name") or ""
+    id_card = params.get("id_card") or params.get("idCard") or params.get("IDCard") or ""
+    if isinstance(name, str):
+        name = name.strip()
+    else:
+        name = str(name)
+    if isinstance(id_card, str):
+        id_card = id_card.strip().upper()
+    else:
+        id_card = str(id_card).strip().upper()
+    return name, id_card
+
+
+def load_rows_from_json(path: Path) -> list[dict]:
+    text = path.read_text(encoding="utf-8-sig").strip()
+    data = json.loads(text)
+    if isinstance(data, list):
+        return data
+    if isinstance(data, dict):
+        for key in ("records", "rows", "data", "items"):
+            if key in data and isinstance(data[key], list):
+                return data[key]
+    raise ValueError("JSON 应为数组，或 export SQL 的 json_agg 结果")
+
+
+def fetch_rows_from_db(args: argparse.Namespace) -> list[dict]:
+    try:
+        import psycopg2
+        import psycopg2.extras
+    except ImportError:
+        print("直连数据库需要: pip install psycopg2-binary", file=sys.stderr)
+        sys.exit(1)
+
+    host = args.db_host or os.getenv("PGHOST", "localhost")
+    port = args.db_port or os.getenv("PGPORT", "5432")
+    dbname = args.db_name or os.getenv("PGDATABASE", "tyapi_dev")
+    user = args.db_user or os.getenv("PGUSER", "postgres")
+    password = args.db_password or os.getenv("PGPASSWORD", "")
+
+    conn = psycopg2.connect(
+        host=host,
+        port=port,
+        dbname=dbname,
+        user=user,
+        password=password,
+    )
+    try:
+        with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
+            cur.execute(SQL, (COMPANY_NAME, PRODUCT_CODE))
+            rows = cur.fetchall()
+            return [dict(r) for r in rows]
+    finally:
+        conn.close()
+
+
+def row_to_csv_record(row: dict) -> dict[str, str]:
+    encrypted = row.get("request_params") or ""
+    secret_key = row.get("secret_key") or ""
+    name_masked = ""
+    id_card_masked = ""
+    decrypt_error = ""
+
+    if encrypted and secret_key:
+        try:
+            params = aes_decrypt(encrypted, secret_key)
+            name, id_card = extract_name_id_card(params)
+            name_masked = mask_name(name)
+            id_card_masked = mask_id_card(id_card)
+        except Exception as exc:
+            decrypt_error = str(exc)
+
+    return {
+        "company_name": str(row.get("company_name") or COMPANY_NAME),
+        "product_code": str(row.get("product_code") or PRODUCT_CODE),
+        "product_name": str(row.get("product_name") or ""),
+        "api_call_id": str(row.get("api_call_id") or row.get("id") or ""),
+        "transaction_id": str(row.get("transaction_id") or ""),
+        "call_time": str(row.get("call_time") or row.get("created_at") or ""),
+        "start_at": str(row.get("start_at") or ""),
+        "cost": str(row.get("cost") or ""),
+        "name_masked": name_masked,
+        "id_card_masked": id_card_masked,
+        "decrypt_error": decrypt_error,
+    }
+
+
+def write_csv(records: list[dict[str, str]], output: Path) -> None:
+    fields = [
+        "company_name",
+        "product_code",
+        "product_name",
+        "api_call_id",
+        "transaction_id",
+        "call_time",
+        "start_at",
+        "cost",
+        "name_masked",
+        "id_card_masked",
+        "decrypt_error",
+    ]
+    with output.open("w", encoding="utf-8-sig", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=fields)
+        writer.writeheader()
+        writer.writerows(records)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="导出 IVYZ4Y27 学历高级版调用记录 CSV")
+    parser.add_argument("-i", "--input", help="SQL 导出的 JSON 文件")
+    parser.add_argument("-o", "--output", default="ivyz4y27_zhongkui_calls.csv")
+    parser.add_argument("--from-db", action="store_true", help="直连 PostgreSQL 查询")
+    parser.add_argument("--db-host")
+    parser.add_argument("--db-port", default="5432")
+    parser.add_argument("--db-name")
+    parser.add_argument("--db-user")
+    parser.add_argument("--db-password")
+    args = parser.parse_args()
+
+    if args.from_db:
+        rows = fetch_rows_from_db(args)
+    elif args.input:
+        rows = load_rows_from_json(Path(args.input))
+    else:
+        print("请指定 --from-db 或 -i <json文件>", file=sys.stderr)
+        sys.exit(1)
+
+    records = [row_to_csv_record(r) for r in rows]
+    out = Path(args.output)
+    write_csv(records, out)
+
+    failed = sum(1 for r in records if r.get("decrypt_error"))
+    print(f"企业: {COMPANY_NAME}")
+    print(f"产品: {PRODUCT_CODE} (学历信息高级版)")
+    print(f"成功记录: {len(records)} 条, 解密失败: {failed} 条")
+    print(f"输出: {out.resolve()}")
+
+
+if __name__ == "__main__":
+    main()

scripts/export_ivyz4y27_zhongkui.sql

@@ -0,0 +1,33 @@
+-- 河南钟馗科技有限公司 · 学历信息高级版(IVYZ4Y27) · 调用成功记录
+-- 在 DBeaver / Navicat 中执行 → 结果集右键「导出 CSV」即可。
+--
+-- 说明:
+--   request_params 为 AES 加密密文，客户端无法直接看到姓名/身份证。
+--   导出后可用 scripts/export_ivyz4y27_calls.py -i <导出的json> 解密并脱敏。
+--
+-- 若 JOIN products 报错，把 products 改成 product 再试。
+
+SELECT
+    ac.id                   AS api_call_id,
+    ac.transaction_id,
+    ac.created_at           AS call_time,
+    ac.start_at,
+    ac.end_at,
+    ac.status,
+    ac.cost,
+    ei.company_name,
+    p.code                  AS product_code,
+    p.name                  AS product_name,
+    ac.user_id,
+    ac.access_id,
+    ac.request_params,
+    au.secret_key
+FROM api_calls ac
+INNER JOIN products p ON ac.product_id = p.id
+INNER JOIN api_users au ON ac.user_id = au.user_id
+INNER JOIN enterprise_infos ei ON ac.user_id = ei.user_id
+WHERE ei.company_name = '河南钟馗科技有限公司'
+  AND ei.deleted_at IS NULL
+  AND p.code = 'IVYZ4Y27'
+  AND ac.status = 'success'
+ORDER BY ac.created_at DESC;