Python爬取双色球彩票开奖信息(二)

书接上回帖子：Python爬取双色球彩票开奖信息在上次帖子中，仅做了本次爬取（指定爬取条数），并未添加二次更新（爬取数据内容）时现有文件的追加，这次增加了本次爬取存储位置为当前目录下data的文件夹，本次修改内容足够后期使用中对该项目的终结，当然除非页面代码优化，无法获取实时数据，我会尝试更新代码

import requests
import json
import time
import random
import pandas as pd
import os
import re
from typing import Dict, List, Optional
from datetime import datetime, timedelta


class SSQSpider:
    def __init__(self):
        self.base_url = "https://www.cwl.gov.cn/cwl_admin/front/cwlkj/search/kjxx/findDrawNotice"
        self.headers = {
            "Accept": "application/json, text/javascript, */*; q=0.01",
            "Accept-Encoding": "gzip, deflate, br",
            "Accept-Language": "zh-CN,zh;q=0.9",
            "Connection": "keep-alive",
            "Host": "www.cwl.gov.cn",
            "Referer": "https://www.cwl.gov.cn/ygkj/wqkjgg/ssq/",
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
            "X-Requested-With": "XMLHttpRequest"
        }
        self.cookies = {
            "HMF_CI": "ccdeb225dfb17006dccc0241dfe329e21005024861f408cc58a70bd6fb877731381b1f33b9130951fda8c4cc26ba65dd51ce93586d3d79177bb66d4a320174ddcd",
            "21_vq": "46"
        }

    def get_ssq_data(self, page_no: int = 1, page_size: int = 30, **kwargs) -> Optional[Dict]:
        time.sleep(random.uniform(0.5, 2))
        params = {
            "name": "ssq",
            "pageNo": page_no,
            "pageSize": page_size,
            "systemType": "PC"
        }
        params.update({k: v for k, v in kwargs.items() if v is not None})
        try:
            response = requests.get(
                self.base_url,
                headers=self.headers,
                params=params,
                cookies=self.cookies,
                timeout=10
            )
            response.raise_for_status()
            return response.json()
        except Exception as e:
            print(f"第 {page_no} 页请求失败: {str(e)}")
            return None

    def parse_data(self, raw_data: Dict) -> List[Dict]:
        if not raw_data or not isinstance(raw_data.get("result"), list):
            return []
        structured_data = []
        for item in raw_data["result"]:
            try:
                date_str = item.get("date", "").split("(")[0].strip()
                try:
                    draw_date = datetime.strptime(date_str, "%Y-%m-%d").date()
                except ValueError:
                    continue
                red_balls = item.get("red", "").split(",")
                blue_ball = item.get("blue", "").strip()
                if len(red_balls) != 6 or not blue_ball:
                    continue
                first_prize = next(
                    (p for p in item.get("prizegrades", []) if p.get("type") == "1"),
                    {"bonus": "0", "num": "0"}
                )
                structured_data.append({
                    "期号": item.get("code", "").strip(),
                    "开奖日期": draw_date.strftime("%Y-%m-%d"),
                    "星期": item.get("week", "").strip(),
                    "红球1": red_balls[0],
                    "红球2": red_balls[1],
                    "红球3": red_balls[2],
                    "红球4": red_balls[3],
                    "红球5": red_balls[4],
                    "红球6": red_balls[5],
                    "蓝球": blue_ball,
                    "开奖号码": " ".join(red_balls) + " + " + blue_ball,
                    "奖池金额(元)": item.get("poolmoney", "0"),
                    "一等奖注数": first_prize.get("num", "0"),
                    "一等奖奖金(元)": first_prize.get("bonus", "0"),
                    "销售金额(元)": item.get("sales", "0")
                })
            except Exception as e:
                print(f"解析数据条目时出错: {str(e)}")
                continue
        return structured_data

    def save_to_excel(self, data: List[Dict], filename: str, overwrite: bool = False) -> bool:
        if not data:
            print("没有数据可保存")
            return False
        if not filename.endswith('.xlsx'):
            filename += '.xlsx'

        # 自动创建保存目录
        os.makedirs(os.path.dirname(filename) or '.', exist_ok=True)

        if os.path.exists(filename) and not overwrite:
            base_name, ext = os.path.splitext(filename)
            counter = 1
            while True:
                new_filename = f"{base_name}_{counter}{ext}"
                if not os.path.exists(new_filename):
                    filename = new_filename
                    print(f"将保存为新文件: {filename}")
                    break
                counter += 1
        try:
            df = pd.DataFrame(data)
            df["开奖日期"] = pd.to_datetime(df["开奖日期"], errors="coerce").dt.strftime("%Y-%m-%d")
            df = df.dropna(subset=["开奖日期"])
            # 确保期号为数值类型，排序安全
            if "期号" in df.columns:
                df["期号"] = df["期号"].astype(str).str.strip()
                df["期号"] = df["期号"].replace(['nan', 'None', ''], '0')
                df["期号"] = pd.to_numeric(df["期号"], errors='coerce').fillna(0).astype(int)
            df = df.sort_values("开奖日期", ascending=False)
            with pd.ExcelWriter(filename, engine="xlsxwriter") as writer:
                df.to_excel(writer, index=False, sheet_name="双色球开奖记录")
                # 简单格式化（可根据需要添加）
            print(f"成功保存 {len(df)} 条数据到 {filename}")
            return True
        except Exception as e:
            print(f"保存Excel失败: {str(e)}")
            return False

    # ========== 增量更新相关方法 ==========
    def fetch_by_date_range(self, start_date: str, end_date: str) -> List[Dict]:
        all_data = []
        page = 1
        page_size = 50
        max_pages = 100
        while page <= max_pages:
            print(f"  正在获取第 {page} 页...")
            raw = self.get_ssq_data(page_no=page, page_size=page_size,
                                    startDate=start_date, endDate=end_date)
            if raw is None:
                break
            parsed = self.parse_data(raw)
            if not parsed:
                break
            all_data.extend(parsed)
            if len(parsed) < page_size:
                break
            page += 1
            time.sleep(random.uniform(0.5, 1.5))
        filtered = [d for d in all_data if start_date <= d["开奖日期"] <= end_date]
        return filtered

    def incremental_update(self, filename: str = "ssq_results.xlsx") -> bool:
        if not os.path.exists(filename):
            print("本地数据文件不存在，请先执行全量爬取。")
            return False
        try:
            old_df = pd.read_excel(filename)
        except Exception as e:
            print(f"读取现有文件失败: {e}")
            return False
        if old_df.empty:
            print("现有文件为空，执行全量爬取...")
            return self.full_fetch_and_save(filename)
        old_df['开奖日期'] = pd.to_datetime(old_df['开奖日期'])
        latest_date = old_df['开奖日期'].max()
        start_date = (latest_date + timedelta(days=1)).strftime("%Y-%m-%d")
        end_date = datetime.now().strftime("%Y-%m-%d")
        if start_date > end_date:
            print("本地数据已是最新，无需更新。")
            return True
        print(f"🔄 正在获取 {start_date} 至 {end_date} 的新数据...")
        new_data = self.fetch_by_date_range(start_date, end_date)
        if not new_data:
            print("未获取到新数据。")
            return True
        new_df = pd.DataFrame(new_data)
        combined_df = pd.concat([old_df, new_df], ignore_index=True)
        combined_df.drop_duplicates(subset=["期号"], keep="last", inplace=True)
        # 统一期号类型
        if "期号" in combined_df.columns:
            combined_df["期号"] = combined_df["期号"].astype(str).str.strip()
            combined_df["期号"] = combined_df["期号"].replace(['nan', 'None', ''], '0')
            combined_df["期号"] = pd.to_numeric(combined_df["期号"], errors='coerce').fillna(0).astype(int)
        combined_df["开奖日期"] = pd.to_datetime(combined_df["开奖日期"]).dt.strftime("%Y-%m-%d")
        combined_df = combined_df.sort_values("开奖日期", ascending=False)
        with pd.ExcelWriter(filename, engine="xlsxwriter") as writer:
            combined_df.to_excel(writer, index=False, sheet_name="双色球开奖记录")
        print(f"✅ 增量更新完成，新增 {len(new_data)} 条，总计 {len(combined_df)} 条。")
        return True

    def full_fetch_and_save(self, filename: str, pages: int = 10) -> bool:
        print("执行全量爬取...")
        all_data = []
        for page in range(1, pages + 1):
            raw = self.get_ssq_data(page_no=page, page_size=50)
            if raw is None:
                continue
            parsed = self.parse_data(raw)
            if parsed:
                all_data.extend(parsed)
            time.sleep(random.uniform(1, 2))
        if all_data:
            return self.save_to_excel(all_data, filename, overwrite=True)
        return False


def get_user_input():
    print("\n双色球开奖数据工具")
    print("=" * 40)
    print("请选择操作模式:")
    print("1. 全量爬取（覆盖或新建）")
    print("2. 增量更新（仅追加新数据）")
    print("3. 退出")
    while True:
        choice = input("请输入选择(1/2/3): ").strip()
        if choice in ("1", "2", "3"):
            return choice
        print("无效输入，请重新选择！")


# def main():
#     spider = SSQSpider()
# 
#     # 确保 data 目录存在（基于脚本所在目录）
#     base_dir = os.path.dirname(os.path.abspath(__file__))
#     data_dir = os.path.join(base_dir, "data")
#     os.makedirs(data_dir, exist_ok=True)
# 
#     default_filename = os.path.join(data_dir, "ssq.xlsx")
#     while True:
#         choice = get_user_input()
#         if choice == "3":
#             print("退出程序。")
#             break
#         elif choice == "1":
#             confirm = input("确认全量爬取？将覆盖现有文件 (y/n): ").strip().lower()
#             if confirm != 'y':
#                 continue
#             pages = int(input("请输入要爬取的页数（每页50条，建议10页以上）: ") or "10")
#             spider.full_fetch_and_save(default_filename, pages)
#             break
#         elif choice == "2":
#             spider.incremental_update(default_filename)
#             break
def main():
    spider = SSQSpider()

    base_dir = os.path.dirname(os.path.abspath(__file__))
    data_dir = os.path.join(base_dir, "data")
    os.makedirs(data_dir, exist_ok=True)

    default_filename = os.path.join(data_dir, "ssq.xlsx")

    while True:
        choice = get_user_input()
        if choice == "3":
            print("退出程序。")
            break
        elif choice == "1":
            confirm = input("确认全量爬取？将覆盖现有文件 (y/n): ").strip().lower()
            if confirm != 'y':
                continue
            try:
                pages = int(input("请输入要爬取的页数（每页50条，建议10页以上）: ") or "10")
            except ValueError:
                pages = 10
            spider.full_fetch_and_save(default_filename, pages)
            break
        elif choice == "2":
            spider.incremental_update(default_filename)
            break

if __name__ == "__main__":
    main()