# -*- coding: utf-8 -*- """ 大乐透历史开奖结果抓取程序 数据来源：中国体育彩票官方网站 https://www.lottery.gov.cn/kj/kjlb.html?dlt 该页面通过 iframe 加载 static.sporttery.cn/res_1_0/jcw/default/html/kj/dlt.html 真实数据由 webapi.sporttery.cn 的 getHistoryPageListV1.qry 接口返回 输出： 1. SQLite 数据库（默认 dlt.db） 表：dlt_draw 开奖主表 / dlt_ball 号码长表 / dlt_prize 奖项明细 视图：v_ball_freq、v_front_freq、v_back_freq、v_pair_freq、 v_draw_stat、v_draw_seq、v_ball_gap、v_ball_miss 2. Excel 文件（默认 dlt_results.xlsx） 工作表：开奖结果 / 奖项明细 / 号码频率 用法： python dlt_spider.py # 增量抓取（遇到已存在的期号即停止） python dlt_spider.py --full # 全量抓取（2925 期） python dlt_spider.py --max-pages 5 # 最多抓取 5 页 python dlt_spider.py --stats # 抓取后打印前区/后区冷热号统计 python dlt_spider.py --print-schema # 打印库结构 + 常用 SQL，可直接贴给 AI python dlt_spider.py --db out.db --excel out.xlsx """ import argparse import json import os import sqlite3 import sys import time from datetime import datetime import requests from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter # ---------------------------------------------------------------- 配置 API_URL = "https://webapi.sporttery.cn/gateway/lottery/getHistoryPageListV1.qry" GAME_NO = "85" # 85 = 超级大乐透 PROVINCE_ID = "0" # 0 = 全国 HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36" ), "Referer": "https://static.sporttery.cn/res_1_0/jcw/default/html/kj/dlt.html", "Accept": "application/json, text/plain, */*", } BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DEFAULT_DB = os.path.join(BASE_DIR, "dlt.db") DEFAULT_EXCEL = os.path.join(BASE_DIR, "dlt_results.xlsx") # ---------------------------------------------------------------- 工具函数 def to_number(value, default=None): """'831,053,727.13' -> 831053727.13 ；'---' / '-1' / '' -> default""" if value is None: return default text = str(value).strip().replace(",", "") if text in ("", "---", "-", "-1"): return default try: return float(text) except ValueError: return default def as_text(value): """历史老数据里部分字段是 {} 而非字符串，统一转成文本；空对象/空串转 None""" if value is None: return None if isinstance(value, (dict, list)): return json.dumps(value, ensure_ascii=False) if value else None text = str(value).strip() return text or None def to_int(value, default=None): num = to_number(value, None) return int(num) if num is not None else default def now_str(): return datetime.now().strftime("%Y-%m-%d %H:%M:%S") # ---------------------------------------------------------------- 抓取 def fetch_page(page_no, page_size=30, retries=3, timeout=20): """抓取一页开奖数据，返回 (records, pages, total)""" params = { "gameNo": GAME_NO, "provinceId": PROVINCE_ID, "pageSize": page_size, "isVerify": 1, "pageNo": page_no, } last_err = None for attempt in range(retries): try: resp = requests.get(API_URL, params=params, headers=HEADERS, timeout=timeout) resp.encoding = "utf-8" payload = resp.json() if not payload.get("success"): raise RuntimeError(payload.get("errorMessage") or "接口返回失败") value = payload.get("value") or {} return value.get("list") or [], int(value.get("pages") or 1), int(value.get("total") or 0) except Exception as exc: # noqa: BLE001 last_err = exc time.sleep(1.5 * (attempt + 1)) raise RuntimeError(f"第 {page_no} 页抓取失败：{last_err}") def parse_record(item): """把接口返回的单条记录解析成 (开奖主记录 dict, 奖项明细 list[dict], 号码明细 list[tuple])""" numbers = str(item.get("lotteryDrawResult") or "").split() # 大乐透为 5 前区（1~35）+ 2 后区（1~12），存整数便于统计；不足时补 None numbers = [to_int(n) for n in numbers] numbers += [None] * (7 - len(numbers)) if len(numbers) < 7 else [] front = numbers[:5] back = numbers[5:7] draw = { "issue": str(item.get("lotteryDrawNum") or ""), "draw_date": as_text(item.get("lotteryDrawTime")), "front_1": front[0], "front_2": front[1], "front_3": front[2], "front_4": front[3], "front_5": front[4], "back_1": back[0], "back_2": back[1], "draw_result": as_text(item.get("lotteryDrawResult")), "unsorted_result": as_text(item.get("lotteryUnsortDrawresult")), "sale_amount": to_number(item.get("drawFlowFund")), "pool_balance": to_number(item.get("poolBalance")), "pool_balance_after": to_number(item.get("poolBalanceAfterdraw")), "sale_begin_time": as_text(item.get("lotterySaleBeginTime")), "sale_end_time": as_text(item.get("lotterySaleEndtime")), "paid_begin_time": as_text(item.get("lotteryPaidBeginTime")), "paid_end_time": as_text(item.get("lotteryPaidEndTime")), "draw_pdf_url": as_text(item.get("drawPdfUrl")), "raw_json": json.dumps(item, ensure_ascii=False), "updated_at": now_str(), } prizes = [] for p in item.get("prizeLevelList") or []: prizes.append({ "issue": draw["issue"], "sort": to_int(p.get("sort")), "prize_level": as_text(p.get("prizeLevel")), "stake_count": to_int(p.get("stakeCount")), "stake_amount": to_number(p.get("stakeAmount")), "total_prize_amount": to_number(p.get("totalPrizeamount")), }) # 号码明细（长表）：一行一个球，频率统计的主力表 balls = [(draw["issue"], draw["draw_date"], "front", i + 1, n) for i, n in enumerate(front) if n is not None] balls += [(draw["issue"], draw["draw_date"], "back", i + 1, n) for i, n in enumerate(back) if n is not None] return draw, prizes, balls # ---------------------------------------------------------------- 数据库 DRAW_COLUMNS_DDL = """ issue TEXT PRIMARY KEY, -- 期号，如 26107 draw_date TEXT, -- 开奖日期 YYYY-MM-DD front_1 INTEGER, -- 前区第 1 个号（1~35） front_2 INTEGER, front_3 INTEGER, front_4 INTEGER, front_5 INTEGER, back_1 INTEGER, -- 后区第 1 个号（1~12） back_2 INTEGER, draw_result TEXT, -- 完整开奖号码（已排序） unsorted_result TEXT, -- 开奖号码（出球顺序） sale_amount REAL, -- 本期销售额（元） pool_balance REAL, -- 开奖前奖池（元） pool_balance_after REAL, -- 开奖后奖池（元） sale_begin_time TEXT, -- 销售开始时间 sale_end_time TEXT, -- 销售截止时间 paid_begin_time TEXT, -- 兑奖开始时间 paid_end_time TEXT, -- 兑奖截止时间 draw_pdf_url TEXT, -- 官方开奖公告 PDF raw_json TEXT, -- 接口原始 JSON，便于追溯 updated_at TEXT -- 入库/更新时间 """ # dlt_ball：号码长表，一行 = 一期的一个球。频率/冷热号/连号统计都基于它 BALL_TABLE_DDL = """ CREATE TABLE IF NOT EXISTS dlt_ball ( id INTEGER PRIMARY KEY AUTOINCREMENT, issue TEXT, -- 期号，关联 dlt_draw.issue draw_date TEXT, -- 开奖日期，冗余便于按日期区间统计 zone TEXT, -- 'front' 前区 / 'back' 后区 position INTEGER, -- 位置：前区 1~5、后区 1~2 ball INTEGER, -- 号码：前区 1~35、后区 1~12 UNIQUE (issue, zone, position) ) """ # 面向 AI / 分析查询的常用视图 VIEW_DDL = """ CREATE VIEW IF NOT EXISTS v_ball_freq AS SELECT zone, ball, COUNT(*) AS hit_count, MIN(draw_date) AS first_date, MAX(draw_date) AS last_date FROM dlt_ball b GROUP BY zone, ball; CREATE VIEW IF NOT EXISTS v_front_freq AS SELECT * FROM v_ball_freq WHERE zone = 'front'; CREATE VIEW IF NOT EXISTS v_back_freq AS SELECT * FROM v_ball_freq WHERE zone = 'back'; CREATE VIEW IF NOT EXISTS v_pair_freq AS SELECT a.zone AS zone, a.ball AS ball_a, b.ball AS ball_b, COUNT(*) AS co_count FROM dlt_ball a JOIN dlt_ball b ON a.issue = b.issue AND a.zone = b.zone AND a.ball < b.ball GROUP BY a.zone, a.ball, b.ball; CREATE VIEW IF NOT EXISTS v_draw_stat AS SELECT issue, draw_date, (front_1 + front_2 + front_3 + front_4 + front_5) AS front_sum, (back_1 + back_2) AS back_sum, (MAX(front_1, front_2, front_3, front_4, front_5) - MIN(front_1, front_2, front_3, front_4, front_5)) AS front_span, (back_2 - back_1) AS back_span, ((front_1 % 2) + (front_2 % 2) + (front_3 % 2) + (front_4 % 2) + (front_5 % 2)) AS front_odd_count, ((back_1 % 2) + (back_2 % 2)) AS back_odd_count FROM dlt_draw; -- 期号不是连续数字（跨年跳号），用开奖序号 seq 计算遗漏才准确 CREATE VIEW IF NOT EXISTS v_draw_seq AS SELECT issue, ROW_NUMBER() OVER (ORDER BY issue) AS seq FROM dlt_draw; -- 每个号码每次出现时，距上次出现隔了多少期 CREATE VIEW IF NOT EXISTS v_ball_gap AS SELECT b.issue, b.zone, b.ball, s.seq - (SELECT MAX(s2.seq) FROM dlt_ball p JOIN v_draw_seq s2 ON s2.issue = p.issue WHERE p.zone = b.zone AND p.ball = b.ball AND s2.seq < s.seq) AS gap_issues FROM dlt_ball b JOIN v_draw_seq s ON s.issue = b.issue; -- 各号码"当前遗漏"：距最近一期已多少期未出（越大越冷） CREATE VIEW IF NOT EXISTS v_ball_miss AS SELECT b.zone AS zone, b.ball AS ball, (SELECT MAX(seq) FROM v_draw_seq) - MAX(s.seq) AS current_miss, f.hit_count AS hit_count, f.last_date AS last_date FROM dlt_ball b JOIN v_draw_seq s ON s.issue = b.issue JOIN v_ball_freq f ON f.zone = b.zone AND f.ball = b.ball GROUP BY b.zone, b.ball; """ SCHEMA = f""" CREATE TABLE IF NOT EXISTS dlt_draw ({DRAW_COLUMNS_DDL}); {BALL_TABLE_DDL}; CREATE TABLE IF NOT EXISTS dlt_prize ( id INTEGER PRIMARY KEY AUTOINCREMENT, issue TEXT, -- 期号，关联 dlt_draw.issue sort INTEGER, -- 奖项排序（101 一等、201 一等追加…） prize_level TEXT, -- 奖项名称 stake_count INTEGER, -- 中奖注数 stake_amount REAL, -- 单注奖金（元） total_prize_amount REAL, -- 本奖级总奖金（元） UNIQUE (issue, sort) ); CREATE INDEX IF NOT EXISTS idx_dlt_draw_date ON dlt_draw (draw_date); CREATE INDEX IF NOT EXISTS idx_dlt_prize_issue ON dlt_prize (issue); CREATE INDEX IF NOT EXISTS idx_dlt_ball_zone_ball ON dlt_ball (zone, ball); CREATE INDEX IF NOT EXISTS idx_dlt_ball_issue ON dlt_ball (issue); """ def init_db(db_path): conn = sqlite3.connect(db_path) conn.executescript(SCHEMA) migrate_draw_columns(conn) conn.executescript(VIEW_DDL) # 视图依赖表，必须在迁移之后再建 conn.commit() return conn def migrate_draw_columns(conn): """旧库号码列是 TEXT，重建为 INTEGER，并把号码明细回填到 dlt_ball""" cols = {row[1]: (row[2] or "").upper() for row in conn.execute("PRAGMA table_info(dlt_draw)")} if cols and cols.get("front_1") == "INTEGER": return rows = conn.execute("SELECT issue, raw_json FROM dlt_draw WHERE raw_json IS NOT NULL").fetchall() conn.executescript(f""" DROP VIEW IF EXISTS v_ball_freq; DROP VIEW IF EXISTS v_front_freq; DROP VIEW IF EXISTS v_back_freq; DROP VIEW IF EXISTS v_pair_freq; DROP VIEW IF EXISTS v_draw_stat; DROP VIEW IF EXISTS v_draw_seq; DROP VIEW IF EXISTS v_ball_gap; DROP VIEW IF EXISTS v_ball_miss; DROP TABLE IF EXISTS dlt_draw_new; CREATE TABLE dlt_draw_new ({DRAW_COLUMNS_DDL}); INSERT INTO dlt_draw_new SELECT issue, draw_date, CAST(front_1 AS INTEGER), CAST(front_2 AS INTEGER), CAST(front_3 AS INTEGER), CAST(front_4 AS INTEGER), CAST(front_5 AS INTEGER), CAST(back_1 AS INTEGER), CAST(back_2 AS INTEGER), draw_result, unsorted_result, sale_amount, pool_balance, pool_balance_after, sale_begin_time, sale_end_time, paid_begin_time, paid_end_time, draw_pdf_url, raw_json, updated_at FROM dlt_draw; DROP TABLE dlt_draw; ALTER TABLE dlt_draw_new RENAME TO dlt_draw; CREATE INDEX IF NOT EXISTS idx_dlt_draw_date ON dlt_draw (draw_date); """) # 用原始 JSON 重建号码长表（不重新联网） for issue, raw in rows: try: item = json.loads(raw) except (TypeError, ValueError): continue _, _, balls = parse_record(item) conn.execute("DELETE FROM dlt_ball WHERE issue = ?", (issue,)) conn.executemany( "INSERT OR REPLACE INTO dlt_ball (issue, draw_date, zone, position, ball) VALUES (?,?,?,?,?)", balls, ) conn.commit() print(f" 已迁移旧库结构：号码列改为整数，并从原始 JSON 重建 {len(rows)} 期的号码明细") def save_record(conn, draw, prizes, balls): conn.execute( """ INSERT INTO dlt_draw ( issue, draw_date, front_1, front_2, front_3, front_4, front_5, back_1, back_2, draw_result, unsorted_result, sale_amount, pool_balance, pool_balance_after, sale_begin_time, sale_end_time, paid_begin_time, paid_end_time, draw_pdf_url, raw_json, updated_at ) VALUES ( :issue, :draw_date, :front_1, :front_2, :front_3, :front_4, :front_5, :back_1, :back_2, :draw_result, :unsorted_result, :sale_amount, :pool_balance, :pool_balance_after, :sale_begin_time, :sale_end_time, :paid_begin_time, :paid_end_time, :draw_pdf_url, :raw_json, :updated_at ) ON CONFLICT(issue) DO UPDATE SET draw_date=excluded.draw_date, front_1=excluded.front_1, front_2=excluded.front_2, front_3=excluded.front_3, front_4=excluded.front_4, front_5=excluded.front_5, back_1=excluded.back_1, back_2=excluded.back_2, draw_result=excluded.draw_result, unsorted_result=excluded.unsorted_result, sale_amount=excluded.sale_amount, pool_balance=excluded.pool_balance, pool_balance_after=excluded.pool_balance_after, sale_begin_time=excluded.sale_begin_time, sale_end_time=excluded.sale_end_time, paid_begin_time=excluded.paid_begin_time, paid_end_time=excluded.paid_end_time, draw_pdf_url=excluded.draw_pdf_url, raw_json=excluded.raw_json, updated_at=excluded.updated_at """, draw, ) conn.executemany( """ INSERT INTO dlt_prize (issue, sort, prize_level, stake_count, stake_amount, total_prize_amount) VALUES (:issue, :sort, :prize_level, :stake_count, :stake_amount, :total_prize_amount) ON CONFLICT(issue, sort) DO UPDATE SET prize_level=excluded.prize_level, stake_count=excluded.stake_count, stake_amount=excluded.stake_amount, total_prize_amount=excluded.total_prize_amount """, prizes, ) conn.execute("DELETE FROM dlt_ball WHERE issue = ?", (draw["issue"],)) conn.executemany( "INSERT OR REPLACE INTO dlt_ball (issue, draw_date, zone, position, ball) VALUES (?,?,?,?,?)", balls, ) def existing_issues(conn): return {row[0] for row in conn.execute("SELECT issue FROM dlt_draw")} # ---------------------------------------------------------------- Excel DRAW_COLUMNS = [ ("期号", "issue", 12), ("开奖日期", "draw_date", 14), ("前区1", "front_1", 8), ("前区2", "front_2", 8), ("前区3", "front_3", 8), ("前区4", "front_4", 8), ("前区5", "front_5", 8), ("后区1", "back_1", 8), ("后区2", "back_2", 8), ("开奖号码", "draw_result", 22), ("出球顺序", "unsorted_result", 22), ("本期销售额(元)", "sale_amount", 18), ("开奖前奖池(元)", "pool_balance", 18), ("开奖后奖池(元)", "pool_balance_after", 18), ("销售截止时间", "sale_end_time", 20), ("兑奖截止时间", "paid_end_time", 20), ("开奖公告", "draw_pdf_url", 40), ] PRIZE_COLUMNS = [ ("期号", "issue", 12), ("开奖日期", "draw_date", 14), ("奖项", "prize_level", 16), ("中奖注数", "stake_count", 12), ("单注奖金(元)", "stake_amount", 16), ("总奖金(元)", "total_prize_amount", 18), ] FREQ_COLUMNS = [ ("区域", "zone_cn", 10), ("号码", "ball", 8), ("出现次数", "hit_count", 12), ("出现频率", "hit_rate", 12), ("首次出现", "first_date", 14), ("最近出现", "last_date", 14), ] HEADER_FILL = PatternFill("solid", fgColor="DDEBF7") HEADER_FONT = Font(bold=True, color="1F4E78") def write_sheet(ws, columns, rows): ws.append([title for title, _, _ in columns]) for cell in ws[1]: cell.fill = HEADER_FILL cell.font = HEADER_FONT cell.alignment = Alignment(horizontal="center", vertical="center") for row in rows: ws.append([row.get(key) for _, key, _ in columns]) for idx, (_, _, width) in enumerate(columns, start=1): ws.column_dimensions[get_column_letter(idx)].width = width ws.freeze_panes = "A2" ws.auto_filter.ref = ws.dimensions def export_excel(conn, excel_path): wb = Workbook() draw_rows = [ dict(zip( ["issue", "draw_date", "front_1", "front_2", "front_3", "front_4", "front_5", "back_1", "back_2", "draw_result", "unsorted_result", "sale_amount", "pool_balance", "pool_balance_after", "sale_end_time", "paid_end_time", "draw_pdf_url"], row, )) for row in conn.execute( """ SELECT issue, draw_date, printf('%02d', front_1), printf('%02d', front_2), printf('%02d', front_3), printf('%02d', front_4), printf('%02d', front_5), printf('%02d', back_1), printf('%02d', back_2), draw_result, unsorted_result, sale_amount, pool_balance, pool_balance_after, sale_end_time, paid_end_time, draw_pdf_url FROM dlt_draw ORDER BY issue DESC """ ) ] write_sheet(wb.active, DRAW_COLUMNS, draw_rows) wb.active.title = "开奖结果" prize_rows = [ dict(zip(["issue", "draw_date", "prize_level", "stake_count", "stake_amount", "total_prize_amount"], row)) for row in conn.execute( """ SELECT p.issue, d.draw_date, p.prize_level, p.stake_count, p.stake_amount, p.total_prize_amount FROM dlt_prize p LEFT JOIN dlt_draw d ON d.issue = p.issue ORDER BY p.issue DESC, p.sort ASC """ ) ] write_sheet(wb.create_sheet("奖项明细"), PRIZE_COLUMNS, prize_rows) total_draws = conn.execute("SELECT COUNT(*) FROM dlt_draw").fetchone()[0] or 1 freq_rows = [ dict(zip(["zone_cn", "ball", "hit_count", "hit_rate", "first_date", "last_date"], row)) for row in conn.execute( """ SELECT CASE zone WHEN 'front' THEN '前区' ELSE '后区' END, printf('%02d', ball), hit_count, printf('%.2f%%', hit_count * 100.0 / ?), first_date, last_date FROM v_ball_freq ORDER BY zone DESC, ball ASC """, (total_draws,), ) ] write_sheet(wb.create_sheet("号码频率"), FREQ_COLUMNS, freq_rows) wb.save(excel_path) # ---------------------------------------------------------------- 主流程 def crawl(conn, full=False, max_pages=None, page_size=30, delay=0.4): """返回 (新增期数, 更新期数, 抓取页数)""" known = existing_issues(conn) added = updated = page_count = 0 page_no, total_pages = 1, None while True: records, pages, total = fetch_page(page_no, page_size) total_pages = pages page_count += 1 if not records: break stop = False for item in records: issue = str(item.get("lotteryDrawNum") or "") if not issue: continue # 增量模式：遇到已入库期号说明后面都是旧数据，直接结束 if not full and issue in known: stop = True continue draw, prizes, balls = parse_record(item) save_record(conn, draw, prizes, balls) if issue in known: updated += 1 else: known.add(issue) added += 1 conn.commit() new_issues = [str(i.get("lotteryDrawNum")) for i in records] print(f" 第 {page_no}/{total_pages} 页完成（累计 {total} 期）：{new_issues[0]} ~ {new_issues[-1]} 新增 {added} 更新 {updated}") if stop: print(" 增量模式：已遇到数据库中已有的期号，停止抓取。") break if max_pages and page_count >= max_pages: break if page_no >= total_pages: break page_no += 1 time.sleep(delay) return added, updated, page_count # ---------------------------------------------------------------- 分析辅助 def print_schema(conn): """打印库结构，可直接贴给 AI 让它写 SQL""" print("-- 表 --") for (ddl,) in conn.execute( "SELECT sql FROM sqlite_master WHERE type = 'table' AND name LIKE 'dlt%' ORDER BY name" ): print(ddl.strip()) print() print("-- 视图 --") for (ddl,) in conn.execute( "SELECT sql FROM sqlite_master WHERE type = 'view' ORDER BY name" ): print(ddl.strip()) print() print("-- 常用查询示例 --") for line in QUERY_EXAMPLES: print(line) QUERY_EXAMPLES = """ SELECT ball, hit_count FROM v_front_freq ORDER BY hit_count DESC LIMIT 10; -- 前区热号 SELECT ball, hit_count FROM v_back_freq ORDER BY hit_count DESC LIMIT 10; -- 后区热号 SELECT zone, ball, hit_count FROM v_ball_freq ORDER BY zone DESC, ball; -- 全号频率 SELECT ball_a, ball_b, co_count FROM v_pair_freq WHERE zone='front' ORDER BY co_count DESC LIMIT 10; -- 前区最常见同现组合 SELECT zone, ball, MAX(gap_issues) AS max_gap FROM v_ball_gap WHERE gap_issues IS NOT NULL GROUP BY zone, ball ORDER BY max_gap DESC LIMIT 10; -- 最大遗漏 SELECT AVG(front_sum), AVG(front_span) FROM v_draw_stat; -- 和值/跨度均值 -- 指定期号区间：SELECT ball, COUNT(*) FROM dlt_ball -- WHERE zone='front' AND issue BETWEEN '25001' AND '26107' GROUP BY ball ORDER BY 2 DESC; """.strip().splitlines() def print_stats(conn, top=10): total = conn.execute("SELECT COUNT(*) FROM dlt_draw").fetchone()[0] if not total: print("数据库为空，无统计数据") return print(f"统计范围：{total} 期") print(f" {conn.execute('SELECT MIN(draw_date) FROM dlt_draw').fetchone()[0]} ~ " f"{conn.execute('SELECT MAX(draw_date) FROM dlt_draw').fetchone()[0]}") for zone, label in (("front", "前区(1~35)"), ("back", "后区(1~12)")): rows = conn.execute( "SELECT ball, hit_count FROM v_ball_freq WHERE zone = ? ORDER BY hit_count DESC, ball", (zone,), ).fetchall() hot = " ".join(f"{b:02d}:{c}" for b, c in rows[:top]) cold = " ".join(f"{b:02d}:{c}" for b, c in rows[-top:]) print(f"\n{label} 出现最多的 {top} 个号：\n {hot}") print(f"{label} 出现最少的 {top} 个号：\n {cold}") # ---------------------------------------------------------------- 查询辅助（供 REST / LLM 调用） def query_ball_frequency(conn, zone=None, year=None, top=None): """号码出现频率。zone=None 时返回全部；year=None 时不做年份过滤。""" sql = "SELECT ball, COUNT(*) AS cnt FROM dlt_ball WHERE 1=1" params = [] if zone: sql += " AND zone = ?" params.append(zone) if year: sql += " AND substr(draw_date, 1, 4) = ?" params.append(str(year)) sql += " GROUP BY ball ORDER BY cnt DESC, ball ASC" if top: sql += " LIMIT ?" params.append(int(top)) return [(b, c) for b, c in conn.execute(sql, params).fetchall()] def query_latest_draws(conn, limit=5): return [ dict(zip( ["issue", "draw_date", "front_1", "front_2", "front_3", "front_4", "front_5", "back_1", "back_2", "draw_result"], row, )) for row in conn.execute( "SELECT issue, draw_date, front_1, front_2, front_3, front_4, front_5, " "back_1, back_2, draw_result FROM dlt_draw ORDER BY issue DESC LIMIT ?", (int(limit),), ) ] def query_year_summary(conn, year): row = conn.execute( "SELECT COUNT(*), MIN(draw_date), MAX(draw_date) FROM dlt_draw WHERE substr(draw_date,1,4)=?", (str(year),), ).fetchone() avg = conn.execute( "SELECT AVG(front_sum), AVG(front_span) FROM v_draw_stat WHERE substr(draw_date,1,4)=?", (str(year),), ).fetchone() return { "year": int(year), "draw_count": row[0], "first_date": row[1], "last_date": row[2], "avg_front_sum": round(avg[0], 1) if avg[0] else None, "avg_front_span": round(avg[1], 1) if avg[1] else None, } def run_sync(db_path=DEFAULT_DB, excel_path=DEFAULT_EXCEL, full=False, **kwargs): """供定时任务/REST 调用的同步入口：增量抓取并刷新 Excel，返回统计信息。""" conn = init_db(db_path) try: added, updated, pages = crawl(conn, full=full, **kwargs) if not kwargs.get("no_excel", False): export_excel(conn, excel_path) return {"pages": pages, "added": added, "updated": updated, "total_draws": conn.execute("SELECT COUNT(*) FROM dlt_draw").fetchone()[0]} finally: conn.close() def main(): if hasattr(sys.stdout, "reconfigure"): # Windows 控制台中文不乱码 sys.stdout.reconfigure(encoding="utf-8") parser = argparse.ArgumentParser(description="抓取大乐透历史开奖结果到 SQLite 与 Excel") parser.add_argument("--db", default=DEFAULT_DB, help=f"SQLite 数据库路径（默认 {DEFAULT_DB}）") parser.add_argument("--excel", default=DEFAULT_EXCEL, help=f"Excel 文件路径（默认 {DEFAULT_EXCEL}）") parser.add_argument("--full", action="store_true", help="全量抓取（默认增量，遇到已存在期号即停止）") parser.add_argument("--max-pages", type=int, default=None, help="最多抓取的页数") parser.add_argument("--page-size", type=int, default=30, help="每页期数（默认 30）") parser.add_argument("--delay", type=float, default=0.4, help="翻页间隔秒数（默认 0.4）") parser.add_argument("--no-excel", action="store_true", help="只写数据库，不导出 Excel") parser.add_argument("--stats", action="store_true", help="抓取后打印前区/后区号码频率统计") parser.add_argument("--print-schema", action="store_true", help="只打印库结构（可贴给 AI 写 SQL），不抓取") args = parser.parse_args() db_path = os.path.abspath(args.db) excel_path = os.path.abspath(args.excel) if args.print_schema: conn = init_db(db_path) try: print_schema(conn) finally: conn.close() return 0 print(f"数据库：{db_path}") print(f"Excel ：{'（跳过）' if args.no_excel else excel_path}") print(f"模式 ：{'全量' if args.full else '增量'}") conn = init_db(db_path) try: added, updated, page_count = crawl( conn, full=args.full, max_pages=args.max_pages, page_size=args.page_size, delay=args.delay, ) total_draws = conn.execute("SELECT COUNT(*) FROM dlt_draw").fetchone()[0] total_prizes = conn.execute("SELECT COUNT(*) FROM dlt_prize").fetchone()[0] print(f"\n抓取完成：{page_count} 页，新增 {added} 期，更新 {updated} 期") print(f"数据库现有：{total_draws} 期开奖记录 / {total_prizes} 条奖项明细") if not args.no_excel: export_excel(conn, excel_path) print(f"Excel 已导出：{excel_path}") latest = conn.execute( "SELECT issue, draw_date, draw_result FROM dlt_draw ORDER BY issue DESC LIMIT 3" ).fetchall() if latest: print("最新三期：") for issue, draw_date, result in latest: print(f" {issue} {draw_date} {result}") if args.stats: print("\n=== 号码频率统计 ===") print_stats(conn) finally: conn.close() return 0 if __name__ == "__main__": sys.exit(main())