data_loader.py

# © 中哥  All Rights Reserved
# 双色球历史统计分析器（Probaball）— 内部使用，未经授权禁止转载、商用。

""" data_loader.py — 数据加载、CSV/Excel 解析 """
import sys
import os
import re
import csv
import logging
from pathlib import Path
import pandas as pd

# 兼容 PyInstaller 打包后路径
def _get_base_path():
    if hasattr(sys, '_MEIPASS'):
        return sys._MEIPASS
    return os.path.dirname(os.path.abspath(__file__))

BASE_PATH = _get_base_path()

# 版权保护（打包运行时）
import importlib.util as _imp_util
_copyright_mod = None

def _find_copyright_module() -> str:
    candidates = []
    base = os.path.dirname(os.path.abspath(__file__))
    candidates.append(os.path.join(base, 'copyright_protection.py'))
    internal = os.path.join(base, '_internal')
    if os.path.isdir(internal):
        candidates.append(os.path.join(internal, 'copyright_protection.py'))
    candidates.append(os.path.join(base, '_internal', 'copyright_protection.py'))
    if os.path.basename(base) == '_internal':
        candidates.append(os.path.join(base, '..', 'copyright_protection.py'))
    for path in candidates:
        if os.path.isfile(path):
            return path
    return ""

_copyright_path = _find_copyright_module()
if _copyright_path:
    try:
        spec = _imp_util.spec_from_file_location("copyright_protection", _copyright_path)
        _copyright_mod = _imp_util.module_from_spec(spec)
        spec.loader.exec_module(_copyright_mod)
    except Exception:
        logging.debug("版权模块加载失败", exc_info=True)
        _copyright_mod = None

COPYRIGHT_TEXT = ""
if _copyright_mod and hasattr(_copyright_mod, 'get_copyright_text'):
    COPYRIGHT_TEXT = _copyright_mod.get_copyright_text()
if not COPYRIGHT_TEXT:
    COPYRIGHT_TEXT = "© 2026 双色球概率分析器 v3.0 — 仅供概率研究参考，不构成投注建议"

copyright_mod = _copyright_mod

# 资源与规模上限，防路径穿越 / 资源耗尽（DoS）
_MAX_FILE_BYTES = 50 * 1024 * 1024  # 单文件 50MB 上限
_MAX_RECORDS = 200000             # 解析记录上限


def _safe_resolve_path(filepath: str) -> str:
    """校验并规范化导入文件路径，防护路径穿越与非法输入。

    返回规范后的绝对路径；非法输入直接抛出明确异常。
    """
    if not filepath or "\x00" in filepath:
        raise ValueError("文件路径为空或包含非法字符")
    try:
        p = Path(filepath).resolve()
    except Exception as e:  # 路径无法解析
        raise ValueError(f"文件路径非法: {e}")
    if not p.exists():
        raise FileNotFoundError(f"文件不存在: {p.name}")
    if not p.is_file():
        raise ValueError(f"不是普通文件: {p.name}")
    if p.stat().st_size > _MAX_FILE_BYTES:
        raise ValueError(f"文件过大（超过 {_MAX_FILE_BYTES // (1024 * 1024)}MB）: {p.name}")
    return str(p)


HEADER_KEYWORDS = ['期号', '开奖', '日期', '红', '蓝', '号码']

def is_header_line(line: str) -> bool:
    count = sum(1 for kw in HEADER_KEYWORDS if kw in line)
    return count >= 2

def parse_data(raw: str) -> list:
    """解析用户粘贴的数据，返回 records 列表"""
    records = []
    for line in raw.strip().split("\n"):
        line = line.strip()
        if not line:
            continue
        if is_header_line(line):
            continue
        if not re.search(r"\d", line):
            continue
        nums = re.findall(r"\d+", line)
        all_nums = [int(n) for n in nums]

        # 策略1：数字很多，尝试找6连续红球+1蓝球
        if len(all_nums) > 10:
            found = False
            for i in range(len(all_nums) - 7):
                red = all_nums[i:i+6]
                blue = all_nums[i+6]
                if (all(1 <= n <= 33 for n in red) and 1 <= blue <= 16 and all(r > 0 for r in red)):
                    issue = str(all_nums[0]) if all_nums[0] > 2000 else "-"
                    records.append({"issue": issue, "red": red, "blue": blue})
                    found = True
                    break
            if found:
                continue

        # 策略2：取最后7个数字
        if len(all_nums) >= 7:
            red = all_nums[-7:-1]
            blue = all_nums[-1]
            if all(1 <= n <= 33 for n in red) and 1 <= blue <= 16:
                issue = str(all_nums[0]) if len(all_nums) > 7 else "-"
                records.append({"issue": issue, "red": red, "blue": blue})

    # 去重
    seen = set()
    unique = []
    for r in records:
        key = ",".join(str(n) for n in r["red"]) + str(r["blue"])
        if key not in seen:
            seen.add(key)
            unique.append(r)
    return unique[:_MAX_RECORDS]

def parse_excel_file(filepath: str) -> list:
    """从 Excel/CSV 文件解析数据"""
    filepath = _safe_resolve_path(filepath)
    ext = Path(filepath).suffix.lower()
    try:
        if ext == '.csv':
            for enc in ['utf-8-sig', 'utf-8', 'gbk', 'gb2312']:
                try:
                    df = pd.read_csv(filepath, encoding=enc)
                    break
                except Exception:
                    logging.debug(f"编码 {enc} 尝试失败", exc_info=True)
                    continue
            else:
                df = pd.read_csv(filepath, encoding='latin1')
        else:
            df = pd.read_excel(filepath, engine='openpyxl')
    except Exception as e:
        logging.warning(f"文件解析失败: {e}")
        raise Exception(f"文件解析失败: {e}")

    records = []
    # 尝试智能列匹配
    red_cols = []
    blue_col = None
    for c in df.columns:
        cl = str(c).strip().lower()
        if '红' in cl or 'red' in cl:
            red_cols.append(c)
        elif '蓝' in cl or 'blue' in cl:
            blue_col = c

    if not red_cols and blue_col is None:
        # 无明确列名：尝试用前6列作红球，第7列作蓝球
        all_cols = df.columns.tolist()
        if len(all_cols) >= 7:
            red_cols = all_cols[:6]
            blue_col = all_cols[6]
            for i in range(len(df)):
                row = df.iloc[i]
                try:
                    red = [int(float(row[c])) for c in red_cols]
                    blue = int(float(row[blue_col]))
                    if all(1 <= n <= 33 for n in red) and 1 <= blue <= 16:
                        records.append({"issue": "-", "red": red, "blue": blue})
                except (ValueError, TypeError):
                    pass
        if not records:
            raw_text = df.to_string(header=False, index=False)
            records = parse_data(raw_text)
        return _dedup_records(records)

    for i in range(len(df)):
        row = df.iloc[i]
        try:
            red = [int(float(row[c])) for c in red_cols]
        except (ValueError, TypeError):
            continue
        if not all(1 <= n <= 33 for n in red):
            continue

        blue = 0
        if blue_col:
            val = row[blue_col]
            if val is not None and not (isinstance(val, float) and pd.isna(val)):
                try:
                    blue = int(float(val))
                except (ValueError, TypeError):
                    pass
        if blue == 0:
            continue

        issue, date_str = '-', '-'
        for c in df.columns:
            cl = str(c).strip().lower()
            val = row[c]
            if val is None:
                continue
            vs = str(val).strip()
            if '期' in cl and re.match(r'^\d{5,7}$', vs):
                issue = vs
            elif '日期' in cl or '开奖' in cl:
                date_str = vs
            elif '期号' in cl:
                issue = vs
        records.append({"issue": issue, "date": date_str, "red": red, "blue": blue})

    if not records:
        raw_text = df.to_string(header=False, index=False)
        records = parse_data(raw_text)

    return _dedup_records(records)

def _dedup_records(records):
    seen = set()
    unique = []
    for r in records:
        if r["blue"] == 0:
            continue
        key = ",".join(str(n) for n in r["red"]) + str(r["blue"])
        if key not in seen:
            seen.add(key)
            unique.append(r)
    return unique