# -*- coding: utf-8 -*-
"""
项目名称：帕依提提 PayItiti 数据集API调用工具
版权所有：©2026 帕依提提 保留所有权利
安全重要提示：
1、AK/SK为个人账户付费凭证，泄露会导致Token被盗刷扣费，请勿转发、上传公网、代码托管平台
2、禁止将AK/SK明文写入代码上传网络，推荐使用.env本地文件或系统环境变量注入密钥
3、单密钥每分钟最多请求60次，超限将临时封禁5分钟
4、单次读取条数限制 1~5000，超出会返回参数错误
5、线上公网环境强制使用HTTPS，HTTP仅允许本地127.0.0.1调试，明文传输会导致密钥劫持
6、SK仅本地参与签名计算，不会通过网络传输，抓包无法单独盗用密钥
7、所有接口请求永久留存审计日志，用于对账、风控溯源
调用模式：
模式1：统一limit（支持分页，WANT_MAX_ROWS控制最大获取条数，不会拉全量）
模式2：limit_list多数据集自定义条数（不分页，少量查询使用）
使用指引：
方式1（推荐本地）：项目根目录新建.env填入PAYITITI_AK、PAYITITI_SK、API_BASE_URL
方式2（服务器）：设置系统环境变量 PAYITITI_AK / PAYITITI_SK / API_BASE_URL
"""
import requests
import json
import time
import hashlib
import re
import os
from typing import Dict, List, Any
from dotenv import load_dotenv
# 加载本地.env文件，优先级：.env文件 > 系统全局环境变量
load_dotenv()
# ==================== 【全局配置区 仅修改此处】 ====================
# 密钥：优先.env > 系统环境变量 > 空字符串兜底
API_AK = os.getenv("PAYITITI_AK", "")
API_SK = os.getenv("PAYITITI_SK", "")
# 接口地址：支持.env自定义覆盖，兜底本地调试地址
BASE_URL = os.getenv("API_BASE_URL", "")
# 请求超时时间
TIMEOUT = 30
# 单次最大行数限制
SERVER_MAX_LIMIT = 5000
# 分页单页大小
PAGE_SIZE = 1
# 自定义最多拉取多少条，达到立即停止分页
WANT_MAX_ROWS = 5
# 接口失败重试次数
RETRY_COUNT = 2
# 基础重试间隔，采用指数退避
BASE_RETRY_DELAY = 2
# 【新增开关】True=保存本地JSON文件；False=仅控制台打印完整JSON不存文件
SAVE_TO_LOCAL = False
EXPORT_FILE_NAME = "../dataset_full_data.json"
# 模式开关 True=多数据集自定义条数；False=统一limit分页模式
USE_CUSTOM_LIMIT = False
# 模式1 统一limit配置（分页可用）
TARGET_DATASET_IDS = "10160"
GLOBAL_LIMIT = PAGE_SIZE
# 模式2 多数据集自定义条数（不分页，ID与条数一一对应）
CUSTOM_DATASET_IDS = "10160,10105"
CUSTOM_LIMIT_LIST = "10,20"
# 签名时间戳有效期：服务端允许5分钟误差(300秒)
SIGN_EXPIRE_SECONDS = 300
# =================================================================
class DatasetAPIClient:
def __init__(self):
self.all_full_data: List[Dict[str, Any]] = []
self.dataset_total_map: Dict[int, int] = {}
self.dataset_deduct_map: Dict[int, int] = {}
self.want_max = WANT_MAX_ROWS
self.save_file_switch = SAVE_TO_LOCAL
self.export_name = EXPORT_FILE_NAME
# 清理密钥、地址首尾空格
self.ak = API_AK.strip()
self.sk = API_SK.strip()
self.base_url = BASE_URL.strip()
# 前置安全校验 + 参数合法性校验
self._security_check()
self._check_config()
def _security_check(self):
"""运行时安全自检，拦截高危调用场景"""
# 公网环境禁止HTTP明文传输
if not self.base_url.startswith("http://127.0.0.1") and self.base_url.startswith("http://"):
raise Exception(
"安全拦截：公网环境禁止使用HTTP明文接口！\n"
"请替换为平台HTTPS正式域名，明文传输会导致AK/SK被劫持盗刷Token"
)
# 密钥为空拦截提示
if not self.ak or not self.sk:
raise Exception(
"密钥未配置！三种配置方式：\n"
"1. 项目根目录新建.env文件，填写PAYITITI_AK、PAYITITI_SK（本地推荐）\n"
"2. 设置系统环境变量 PAYITITI_AK、PAYITITI_SK（服务器使用）\n"
"3. 本地临时填写API_AK/API_SK，使用完成立刻清空，禁止分享代码文件"
)
def _check_config(self):
"""配置合法性校验，提前抛出错误阻断运行"""
# AK 32位十六进制强校验
if not re.fullmatch(r"[0-9a-fA-F]{32}", self.ak):
raise Exception("AK格式错误，标准AK为32位十六进制字符串，请重新复制无空格密钥")
# SK 64位十六进制强校验
if not re.fullmatch(r"[0-9a-fA-F]{64}", self.sk):
raise Exception("SK格式错误，标准SK为64位十六进制字符串，请重新复制无空格密钥")
if PAGE_SIZE <= 0 or PAGE_SIZE > SERVER_MAX_LIMIT:
raise Exception(f"PAGE_SIZE 范围必须 1~{SERVER_MAX_LIMIT}")
if WANT_MAX_ROWS <= 0:
raise Exception("WANT_MAX_ROWS 必须大于0")
if not self.base_url:
raise Exception("API接口地址BASE_URL不能为空，请检查.env或代码默认配置")
def _generate_signature(self, dataset_ids: str, limit: int, offset: int, limit_list: str = None) -> tuple[str, str]:
"""
生成时间戳 + 请求签名
return: timestamp, signature
"""
timestamp = str(int(time.time()))
# 拼接原始签名字符串，无参数填空字符串
lmt_str = str(limit) if limit else ""
off_str = str(offset)
ll_str = limit_list if limit_list else ""
raw_sign = f"{timestamp}{dataset_ids}{lmt_str}{off_str}{ll_str}"
# 拼接SK做sha256
content = raw_sign + self.sk
sha256_obj = hashlib.sha256(content.encode("utf-8"))
signature = sha256_obj.hexdigest().lower()
return timestamp, signature
def _build_headers(self, timestamp: str, signature: str) -> Dict[str, str]:
"""构造请求头，仅传输AK、时间戳、签名，SK不上传网络"""
return {
"X-AK": self.ak,
"X-Timestamp": timestamp,
"X-Signature": signature,
"User-Agent": "PayItiti-Dataset-Client/2.0 CustomScript"
}
def single_page_request(self, dataset_ids: str, limit: int, offset: int = 0, limit_list: str = None) -> Dict[str, Any]:
"""单页请求，指数退避重试，敏感信息不打印防止密钥泄露"""
params = {
"dataset_id": dataset_ids,
"offset": offset
}
if limit_list:
params["limit_list"] = limit_list
else:
safe_limit = min(limit, SERVER_MAX_LIMIT)
params["limit"] = safe_limit
for retry in range(RETRY_COUNT + 1):
current_delay = BASE_RETRY_DELAY * (2 ** retry)
try:
# 每次请求实时生成全新签名与时间戳，防重放
ts, sign = self._generate_signature(dataset_ids, safe_limit if not limit_list else 0, offset, limit_list)
headers = self._build_headers(ts, sign)
resp = requests.get(
url=self.base_url,
headers=headers,
params=params,
timeout=TIMEOUT
)
print(f"\n===== 请求 offset={offset} | 重试轮次 {retry}/{RETRY_COUNT} 延迟{current_delay}s =====")
print(f"HTTP 状态码: {resp.status_code}")
# 非200状态码仅打印返回文本，屏蔽请求头避免密钥输出控制台
if resp.status_code != 200:
print(f"【服务端拒绝请求】原始返回内容：{resp.text.strip()[:500]}")
if retry < RETRY_COUNT:
print(f"{current_delay}s后进行下一次重试...")
time.sleep(current_delay)
continue
raise Exception(f"接口返回非200状态码 {resp.status_code}，已耗尽重试次数")
# 正常200解析JSON
resp_data = resp.json()
print("【当前分页完整接口返回JSON】")
print(json.dumps(resp_data, ensure_ascii=False, indent=2))
return resp_data
except requests.exceptions.Timeout:
print(f"【请求超时】{current_delay}s后重试")
time.sleep(current_delay)
except json.JSONDecodeError:
resp_text = resp.text if "resp" in locals() else "无返回内容"
print(f"【JSON解析失败】接口原始返回：{resp_text[:500]}")
time.sleep(current_delay)
except Exception as e:
# 异常截断打印，不输出底层原始请求对象，杜绝密钥泄露
print(f"【网络请求异常】{str(e)[:200]}，{current_delay}s后重试")
time.sleep(current_delay)
raise Exception("全部重试失败，终止任务")
def fetch_custom_limit_batch(self):
"""模式2：多数据集自定义条数，单次请求不分页"""
print("===== 进入【多数据集自定义条数】模式（无分页） =====")
resp_data = self.single_page_request(
dataset_ids=CUSTOM_DATASET_IDS,
limit=0,
offset=0,
limit_list=CUSTOM_LIMIT_LIST
)
self.handle_response(resp_data)
self.print_statistic()
self.output_result()
def fetch_single_dataset_page(self, ds_id: int):
"""模式1：单数据集分页，严格控制最大获取条数"""
print(f"\n===== 开始读取数据集 ID={ds_id}，最大获取 {self.want_max} 条 =====")
offset = 0
total_rows = 0
total_deduct = 0
while True:
# 达到设定条数上限，直接停止分页
if total_rows >= self.want_max:
print(f"✅ 已到达设定上限 {self.want_max} 条，停止分页请求")
break
resp_data = self.single_page_request(str(ds_id), GLOBAL_LIMIT, offset)
batch_list = resp_data.get("data", {}).get("batch", [])
batch_info = batch_list[0] if batch_list else {}
# 单数据集读取失败直接跳出循环，不中断整体任务
if batch_info.get("status") != "ok":
print(f"❌ 数据集{ds_id}读取失败：{batch_info.get('msg')}")
break
page_data = batch_info.get("list", [])
page_cost = batch_info.get("deduct_token", 0)
has_next = batch_info.get("has_more", False)
# 截断超出最大获取条数的数据
remain = self.want_max - total_rows
if len(page_data) > remain:
page_data = page_data[:remain]
self.all_full_data.extend(page_data)
curr_count = len(page_data)
total_rows += curr_count
total_deduct += page_cost
print(f"本页获取 {curr_count} 条 | 累计 {total_rows}/{self.want_max} | 本页扣费 {page_cost} Token")
# 无下一页直接结束当前数据集分页
if not has_next:
print(f"✅ 数据集{ds_id}无更多数据，本次合计读取 {total_rows} 条")
self.dataset_total_map[ds_id] = total_rows
self.dataset_deduct_map[ds_id] = total_deduct
break
offset += GLOBAL_LIMIT
# 轻微节流降低限流触发概率
time.sleep(0.3)
def fetch_unified_batch(self):
"""模式1：统一limit分页入口，支持多数据集循环读取"""
ds_str_arr = TARGET_DATASET_IDS.split(",")
ds_id_arr = [int(ds.strip()) for ds in ds_str_arr if ds.strip().isdigit()]
if not ds_id_arr:
raise Exception("TARGET_DATASET_IDS 数据集ID配置非法")
for ds in ds_id_arr:
self.fetch_single_dataset_page(ds)
self.print_statistic()
self.output_result()
def handle_response(self, resp_data: Dict[str, Any]):
"""处理不分页批量模式返回结果"""
code = resp_data.get("code", -1)
msg = resp_data.get("message", "")
batch_arr = resp_data.get("data", {}).get("batch", [])
print(f"接口返回 code={code} 提示：{msg}")
for item in batch_arr:
ds_id = item.get("dataset_id")
status = item.get("status")
data_list = item.get("list", [])
cost = item.get("deduct_token", 0)
if status == "ok":
self.all_full_data.extend(data_list)
cnt = len(data_list)
self.dataset_total_map[ds_id] = self.dataset_total_map.get(ds_id, 0) + cnt
self.dataset_deduct_map[ds_id] = self.dataset_deduct_map.get(ds_id, 0) + cost
print(f"✅ 数据集{ds_id}读取成功，获取 {cnt} 条，扣费 {cost} Token")
else:
print(f"❌ 数据集{ds_id}读取失败：{item.get('msg')}")
def print_statistic(self):
"""打印全量读取汇总统计"""
print("\n==================== 读取汇总统计 ====================")
print(f"预设最大获取条数：{self.want_max}")
print(f"本次实际获取总条数：{len(self.all_full_data)}")
print("分数据集明细：")
for ds, cnt in self.dataset_total_map.items():
print(f" 数据集{ds}：{cnt} 条，累计扣费 {self.dataset_deduct_map[ds]} Token")
print("======================================================")
def output_result(self):
"""统一输出逻辑：控制台打印 + 可选本地文件导出（覆盖二次确认）"""
print("\n==================== 全部分页拼接完成，完整数据JSON输出 ====================")
full_json_str = json.dumps(self.all_full_data, ensure_ascii=False, indent=2)
# 控制台固定打印拼接后完整数据
print(full_json_str)
# 开启导出时增加文件覆盖确认弹窗
if self.save_file_switch:
if os.path.exists(self.export_name):
choice = input(f"警告：文件 {self.export_name} 已存在，是否覆盖？(y/n)").strip().lower()
if choice != "y":
print("取消文件导出")
return
try:
with open(self.export_name, "w", encoding="utf-8") as f:
f.write(full_json_str)
print(f"\n✅ 已同步保存至本地文件：{self.export_name}")
except Exception as e:
print(f"\n❌ 文件保存失败：{str(e)}")
else:
print("\nℹ️ 未开启本地保存，仅控制台输出JSON")
def run(self):
"""程序统一入口"""
try:
if USE_CUSTOM_LIMIT:
self.fetch_custom_limit_batch()
else:
self.fetch_unified_batch()
except Exception as err:
print(f"\n❌ 任务执行终止：{str(err)}")
if __name__ == "__main__":
client = DatasetAPIClient()
client.run()
