fetch_ssq.py

#!/usr/bin/env python3
"""
双色球开奖数据抓取脚本
数据来源：中彩网 JSONP 接口
"""

import re
import json
import time
import argparse
from datetime import datetime
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def _build_session() -> requests.Session:
    """创建忽略系统代理、带自动重试的 Session"""
    session = requests.Session()
    session.trust_env = False  # 忽略系统代理 / 环境变量代理
    retry = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session


_SESSION = _build_session()

BASE_URL = "https://jc.zhcw.com/port/client_json.php"

HEADERS = {
    "User-Agent": (
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0.0.0 Safari/537.36"
    ),
    "Referer": "https://www.zhcw.com/kjxx/ssq/",
    "Accept": "*/*",
    "Accept-Language": "zh-CN,zh;q=0.9",
}

PAGE_SIZE = 30


def fetch_page(page_num: int, issue_count: int, start_date: str = "", end_date: str = "") -> dict:
    """请求单页数据，返回解析后的 JSON 字典"""
    ts = int(time.time() * 1000)
    params = {
        "callback": f"cb{ts}",
        "transactionType": "10001001",
        "lotteryId": "1",
        "issueCount": issue_count,
        "startIssue": "",
        "endIssue": "",
        "startDate": start_date,
        "endDate": end_date,
        "type": "0",
        "pageNum": page_num,
        "pageSize": PAGE_SIZE,
        "tt": "0.1",
        "_": ts,
    }
    resp = _SESSION.get(BASE_URL, params=params, headers=HEADERS, timeout=15)
    resp.raise_for_status()

    # 剥离 JSONP 包装：cb123456({...})
    text = resp.text.strip()
    match = re.match(r"^[^(]+\((.+)\)\s*$", text, re.DOTALL)
    if not match:
        raise ValueError(f"无法解析 JSONP 响应：{text[:200]}")
    return json.loads(match.group(1))


def fetch_all(count: int = 100, start_date: str = "", end_date: str = "") -> list[dict]:
    """分页拉取，返回所有开奖记录列表"""
    all_records: list[dict] = []

    # 先拿第一页，获取总页数
    first = fetch_page(1, count, start_date, end_date)
    all_records.extend(first.get("data", []))
    total_pages = int(first.get("pages", 1))
    print(f"共 {first.get('total', '?')} 条记录，{total_pages} 页，正在抓取…")

    for page in range(2, total_pages + 1):
        data = fetch_page(page, count, start_date, end_date)
        all_records.extend(data.get("data", []))
        time.sleep(0.3)  # 适当间隔，避免请求过快

    return all_records[:count]  # 截取最多 count 条


def format_red_balls(num_str: str) -> str:
    """将红球号码格式化为带颜色标记的字符串"""
    balls = num_str.strip().split()
    return " ".join(f"`{b}`" for b in balls)


def format_blue_ball(num_str: str) -> str:
    """蓝球加粗标记"""
    return f"**{num_str.strip()}**"


def to_markdown(records: list[dict], start_date: str = "", end_date: str = "") -> str:
    """将记录列表转换为 Markdown 表格"""
    date_range = ""
    if start_date or end_date:
        date_range = f"（{start_date or '最早'} ~ {end_date or '最新'}）"

    lines = [
        f"# 双色球开奖数据{date_range}",
        "",
        f"> 共 {len(records)} 期，数据来源：[中彩网](https://www.zhcw.com/kjxx/ssq/)",
        f"> 导出时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        "",
        "| 期号 | 开奖日期 | 星期 | 红球号码 | 蓝球 |",
        "| :--: | :------: | :--: | :------: | :--: |",
    ]

    for r in records:
        issue = r.get("issue", "")
        open_time = r.get("openTime", "")
        week = r.get("week", "")
        red = format_red_balls(r.get("frontWinningNum", ""))
        blue = format_blue_ball(r.get("backWinningNum", ""))
        lines.append(f"| {issue} | {open_time} | {week} | {red} | {blue} |")

    return "\n".join(lines) + "\n"


def main():
    parser = argparse.ArgumentParser(description="抓取双色球开奖数据并导出为 Markdown")
    parser.add_argument("-n", "--count", type=int, default=100, help="抓取期数（默认 100）")
    parser.add_argument("-s", "--start-date", default="", help="开始日期，格式 YYYY-MM-DD")
    parser.add_argument("-e", "--end-date", default="", help="结束日期，格式 YYYY-MM-DD")
    parser.add_argument("-o", "--output", default="", help="输出文件名（默认自动生成）")
    args = parser.parse_args()

    print(f"开始抓取最近 {args.count} 期双色球数据…")
    records = fetch_all(args.count, args.start_date, args.end_date)
    print(f"成功获取 {len(records)} 条记录")

    md_content = to_markdown(records, args.start_date, args.end_date)

    if args.output:
        output_file = args.output
    else:
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = f"双色球开奖数据_{ts}.md"

    with open(output_file, "w", encoding="utf-8") as f:
        f.write(md_content)

    print(f"已导出到：{output_file}")


if __name__ == "__main__":
    main()