#!/usr/bin/env python3 """ 批量获取抖音榜单数据（近30天） - 每日作品热榜 (likesRank) — 按点赞总量排名 - 每日点赞榜 (hotContentRank source=每日点赞飙升榜) — 按单日新增点赞排名 - 7日点赞榜 (hotContentRank source=七日点赞飙升榜) — 按七日新增点赞排名 输出：RedFox数据/榜单数据/YYYY-MM/ 目录，JSON格式保留完整原始响应 用法： python batch_fetch_rankings.py """ import os, json, time, sys from datetime import datetime, timedelta from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError # ── 配置 ── API_KEY = os.environ.get("REDFOX_API_KEY", "ak_662511b4e9a74de59dc0e77d9edbca43") LIKES_RANK_URL = "https://redfox.hk/story/api/dy/search/likesRank" HOT_CONTENT_URL = "https://redfox.hk/story/api/dy/search/hotContentRank" OUTPUT_BASE = "D:/github/技能百宝箱/ThirdPartySkills/RedFox数据/榜单数据" # 近30天 END_DATE = datetime(2026, 8, 9) DAYS = 30 REQUEST_DELAY = 0.5 # 每次请求间隔秒数，避免限流 def post_json(url, body): """发送 POST 请求，返回解析后的 JSON""" data = json.dumps(body).encode("utf-8") req = Request(url, data=data, method="POST") req.add_header("X-API-KEY", API_KEY) req.add_header("Content-Type", "application/json") req.add_header("Accept", "application/json") with urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) def save_json(data, directory, filename): """保存 JSON 文件，返回 (路径, KB数, 记录数)""" os.makedirs(directory, exist_ok=True) path = os.path.join(directory, filename) with open(path, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) size_kb = os.path.getsize(path) / 1024 return path, size_kb def fetch_daily_likes_rank(date_str): """每日作品热榜：按点赞总量排名""" body = { "source": "抖音每日热门作品榜-GitHub", "type": "全部", "startTime": date_str, "endTime": date_str } return post_json(LIKES_RANK_URL, body) def fetch_daily_surge(date_str): """每日点赞榜：按单日新增点赞排名""" body = { "source": "抖音每日点赞飙升榜", "type": "全部", "startTime": date_str } return post_json(HOT_CONTENT_URL, body) def fetch_weekly_surge(date_str): """7日点赞榜：按七日新增点赞排名""" body = { "source": "抖音七日点赞飙升榜-GitHub", "type": "全部", "startTime": date_str } return post_json(HOT_CONTENT_URL, body) def main(): print(f"🚀 开始批量获取近{DAYS}天榜单数据") print(f" 时间范围：{(END_DATE - timedelta(days=DAYS-1)).strftime('%Y-%m-%d')} ~ {END_DATE.strftime('%Y-%m-%d')}") print(f" 输出目录：{OUTPUT_BASE}") print() stats = {"success": 0, "empty": 0, "error": 0, "total_records": 0, "total_kb": 0.0} # ── 第一轮：每日作品热榜（30天）── print("=" * 60) print("📊 一、每日作品热榜（likesRank）") print("=" * 60) for i in range(DAYS): d = END_DATE - timedelta(days=i) date_str = d.strftime('%Y-%m-%d') month = d.strftime('%Y-%m') out_dir = os.path.join(OUTPUT_BASE, month) fname = f"每日作品热榜_{date_str}.json" try: resp = fetch_daily_likes_rank(date_str) code = resp.get("code", -1) items = resp.get("data", []) if isinstance(items, list): count = len(items) elif isinstance(items, dict): count = len(items.get("dailyRank", items.get("weeklyRank", []))) else: count = 0 if code != 2000: print(f" ⚠️ {date_str} API错误码={code} msg={resp.get('msg', '')}") stats["error"] += 1 time.sleep(REQUEST_DELAY) continue if count == 0: print(f" 📭 {date_str} 暂无数据") stats["empty"] += 1 time.sleep(REQUEST_DELAY) continue path, kb = save_json(resp, out_dir, fname) stats["success"] += 1 stats["total_records"] += count stats["total_kb"] += kb print(f" ✅ {date_str} {count}条 {kb:.1f}KB → {fname}") except HTTPError as e: print(f" ❌ {date_str} HTTP {e.code}") stats["error"] += 1 except URLError as e: print(f" ❌ {date_str} 网络错误: {e.reason}") stats["error"] += 1 except Exception as e: print(f" ❌ {date_str} {e}") stats["error"] += 1 time.sleep(REQUEST_DELAY) print() # ── 第二轮：每日点赞榜（30天）── print("=" * 60) print("📊 二、每日点赞榜（hotContentRank - 日飙升）") print("=" * 60) for i in range(DAYS): d = END_DATE - timedelta(days=i) date_str = d.strftime('%Y-%m-%d') month = d.strftime('%Y-%m') out_dir = os.path.join(OUTPUT_BASE, month) fname = f"每日点赞榜_{date_str}.json" try: resp = fetch_daily_surge(date_str) code = resp.get("code", -1) data_obj = resp.get("data", {}) items = data_obj.get("dailyRank", []) if isinstance(data_obj, dict) else [] count = len(items) if code != 2000: print(f" ⚠️ {date_str} API错误码={code} msg={resp.get('msg', '')}") stats["error"] += 1 time.sleep(REQUEST_DELAY) continue if count == 0: print(f" 📭 {date_str} 暂无数据") stats["empty"] += 1 time.sleep(REQUEST_DELAY) continue path, kb = save_json(resp, out_dir, fname) stats["success"] += 1 stats["total_records"] += count stats["total_kb"] += kb print(f" ✅ {date_str} {count}条 {kb:.1f}KB → {fname}") except HTTPError as e: print(f" ❌ {date_str} HTTP {e.code}") stats["error"] += 1 except URLError as e: print(f" ❌ {date_str} 网络错误: {e.reason}") stats["error"] += 1 except Exception as e: print(f" ❌ {date_str} {e}") stats["error"] += 1 time.sleep(REQUEST_DELAY) print() # ── 第三轮：7日点赞榜（30天）── print("=" * 60) print("📊 三、7日点赞榜（hotContentRank - 周飙升）") print("=" * 60) for i in range(DAYS): d = END_DATE - timedelta(days=i) date_str = d.strftime('%Y-%m-%d') month = d.strftime('%Y-%m') out_dir = os.path.join(OUTPUT_BASE, month) fname = f"7日点赞榜_{date_str}.json" try: resp = fetch_weekly_surge(date_str) code = resp.get("code", -1) data_obj = resp.get("data", {}) items = data_obj.get("weeklyRank", []) if isinstance(data_obj, dict) else [] count = len(items) if code != 2000: print(f" ⚠️ {date_str} API错误码={code} msg={resp.get('msg', '')}") stats["error"] += 1 time.sleep(REQUEST_DELAY) continue if count == 0: print(f" 📭 {date_str} 暂无数据") stats["empty"] += 1 time.sleep(REQUEST_DELAY) continue path, kb = save_json(resp, out_dir, fname) stats["success"] += 1 stats["total_records"] += count stats["total_kb"] += kb print(f" ✅ {date_str} {count}条 {kb:.1f}KB → {fname}") except HTTPError as e: print(f" ❌ {date_str} HTTP {e.code}") stats["error"] += 1 except URLError as e: print(f" ❌ {date_str} 网络错误: {e.reason}") stats["error"] += 1 except Exception as e: print(f" ❌ {date_str} {e}") stats["error"] += 1 time.sleep(REQUEST_DELAY) # ── 汇总 ── print() print("=" * 60) print("📋 汇总统计") print("=" * 60) print(f" ✅ 成功: {stats['success']} 次") print(f" 📭 空数据: {stats['empty']} 次") print(f" ❌ 错误: {stats['error']} 次") print(f" 📊 总记录数: {stats['total_records']} 条") print(f" 💾 总数据量: {stats['total_kb']:.1f} KB ({stats['total_kb']/1024:.2f} MB)") if __name__ == "__main__": main()