fix: 修复代码审查P2/P3问题
P2: - lottery.py 改为引用 history_loader.py 公共模块,删除独立 Excel 加载逻辑 - 删除 lottery.py 中不再使用的 _normalize_history_format 方法(已迁移到 history_loader) P3: - app.py: 删除未使用的 import shutil - app.py: 删除不可达代码 return stats - app.py: 修复 api_history 中奇偶形态字段映射错误(奇偶比→奇偶形态) - app.py: api_history columns 列表补充奇偶比和奇偶形态字段 - 开发文档目录结构补充 history_loader.py 和 web_executor.py 废弃标记 Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
@@ -10,7 +10,6 @@ import os
|
||||
import sys
|
||||
import json
|
||||
import uuid
|
||||
import shutil
|
||||
import traceback
|
||||
import threading
|
||||
from datetime import datetime
|
||||
@@ -207,8 +206,6 @@ def get_statistics_data(generator=None):
|
||||
"""获取统计数据 — 委托公共模块计算"""
|
||||
return compute_statistics(CONFIG['history_file'])
|
||||
|
||||
return stats
|
||||
|
||||
|
||||
# ============================================================
|
||||
# API:获取统计数据
|
||||
@@ -379,7 +376,8 @@ def api_history():
|
||||
'蓝球': row['蓝球'],
|
||||
'开机号': str(row['开机号']),
|
||||
'和值': str(row['和值特征']),
|
||||
'奇偶形态': str(row['奇偶比']),
|
||||
'奇偶比': str(row['奇偶比']),
|
||||
'奇偶形态': str(row['奇偶形态']),
|
||||
'大小比': str(row['大小比']),
|
||||
'跨度': str(row['跨度']),
|
||||
}
|
||||
@@ -392,7 +390,7 @@ def api_history():
|
||||
'total': total,
|
||||
'page': page,
|
||||
'page_size': page_size,
|
||||
'columns': ['开奖日期', '期号', '红球', '蓝球', '开机号', '和值', '大小比', '跨度']
|
||||
'columns': ['开奖日期', '期号', '红球', '蓝球', '开机号', '和值', '奇偶比', '奇偶形态', '大小比', '跨度']
|
||||
}
|
||||
})
|
||||
except Exception as e:
|
||||
|
||||
@@ -28,13 +28,15 @@ lottoData/
|
||||
├── app.py # Flask 主服务(统一入口)
|
||||
├── index.html # 前端 UI(响应式,4 Tab 页面)
|
||||
├── lottery.py # 号码生成核心逻辑
|
||||
├── history_loader.py # 公共 Excel 加载模块(统一格式兼容)
|
||||
├── fetch_data.py # 历史数据抓取脚本
|
||||
├── web_console.html # 数据抓取控制台前端
|
||||
├── web_executor.py # [已废弃] 旧版独立抓取服务,功能已整合到 app.py
|
||||
├── requirements.txt # Python 依赖
|
||||
├── 双色球历史数据.xlsx # 历史数据文件
|
||||
├── lottery/ # 号码生成结果输出目录
|
||||
├── .generation_records.json # 生成记录索引(JSON)
|
||||
├── .fetch_status.json # 抓取状态文件
|
||||
├── 双色球历史数据.xlsx # 历史数据文件(不纳入 git)
|
||||
├── lottery/ # 号码生成结果输出目录(不纳入 git)
|
||||
├── .generation_records.json # 生成记录索引(JSON,不纳入 git)
|
||||
├── .fetch_status.json # 抓取状态文件(不纳入 git)
|
||||
├── deploy/ # 部署相关文件
|
||||
│ ├── DEPLOY.md # 部署说明
|
||||
│ ├── lotto-app.service # systemd 服务文件
|
||||
|
||||
+7
-125
@@ -38,76 +38,24 @@ class DoubleColorBallGenerator:
|
||||
}
|
||||
|
||||
def load_history_data(self):
|
||||
"""加载历史数据"""
|
||||
"""加载历史数据 — 委托公共模块 history_loader 处理多格式兼容"""
|
||||
try:
|
||||
# 检查文件是否存在
|
||||
if not os.path.exists(self.history_file):
|
||||
print(f"错误: 文件 {self.history_file} 不存在")
|
||||
return False
|
||||
|
||||
# 读取Excel文件
|
||||
print(f"正在读取文件: {self.history_file}")
|
||||
try:
|
||||
raw_df = pd.read_excel(self.history_file, header=None)
|
||||
except Exception as excel_error:
|
||||
print(f"读取Excel文件失败: {excel_error}")
|
||||
return False
|
||||
|
||||
# 检查数据是否为空
|
||||
if raw_df.empty:
|
||||
print("错误: 历史数据文件为空")
|
||||
return False
|
||||
|
||||
# 兼容多种 Excel 格式:
|
||||
# 格式A(fetch_data.py 当前输出): Row0=新列名(期号|开奖日期|红球1...|蓝球|特别号), Row1=旧列名(开奖时间|期数|号码|...), Row2+=数据
|
||||
# 格式B(标准格式): Row0=列名(开奖时间|期数|号码|开机号|...), Row1+=数据
|
||||
# 格式C(分列含旧 header): Row0=旧列名, Row1+=数据 但无"号码"列
|
||||
|
||||
# 标准列名(lottery.py 期望的列)
|
||||
legacy_columns = ['开奖时间', '期数', '号码', '开机号', '和值特征', '奇偶比', '大小比', '奇偶形态', '跨度', '其他']
|
||||
|
||||
row0_vals = raw_df.iloc[0].astype(str).tolist() if len(raw_df) > 0 else []
|
||||
row1_vals = raw_df.iloc[1].astype(str).tolist() if len(raw_df) > 1 else []
|
||||
|
||||
# 检测各类格式
|
||||
has_legacy_header_in_row0 = any(col in row0_vals for col in ['开奖时间', '期数', '号码'])
|
||||
has_legacy_header_in_row1 = any(col in row1_vals for col in ['开奖时间', '期数', '号码'])
|
||||
has_new_header_in_row0 = any(col in row0_vals for col in ['期号', '开奖日期', '红球 1'])
|
||||
|
||||
if has_new_header_in_row0 and has_legacy_header_in_row1:
|
||||
# 格式A:Row0=新列名, Row1=旧列名, Row2+=数据
|
||||
# 用旧列名(Row1)作为列名,因为 lottery.py 期望"号码"列
|
||||
self.history_data = raw_df.iloc[2:].copy()
|
||||
num_cols = len(self.history_data.columns)
|
||||
self.history_data.columns = legacy_columns[:min(num_cols, len(legacy_columns))] + [f'col_{i}' for i in range(min(num_cols, len(legacy_columns)), num_cols)]
|
||||
self.history_data = self.history_data.reset_index(drop=True)
|
||||
print(f"加载成功(格式A: 新旧 header 双行),共{len(self.history_data)}条历史记录")
|
||||
print(f"数据列: {list(self.history_data.columns)}")
|
||||
|
||||
elif has_legacy_header_in_row0:
|
||||
# 格式B:Row0=标准列名, Row1+=数据
|
||||
self.history_data = raw_df.iloc[1:].copy()
|
||||
num_cols = len(self.history_data.columns)
|
||||
self.history_data.columns = legacy_columns[:min(num_cols, len(legacy_columns))] + [f'col_{i}' for i in range(min(num_cols, len(legacy_columns)), num_cols)]
|
||||
self.history_data = self.history_data.reset_index(drop=True)
|
||||
print(f"加载成功(格式B: 标准列名),共{len(self.history_data)}条历史记录")
|
||||
print(f"数据列: {list(self.history_data.columns)}")
|
||||
|
||||
else:
|
||||
# 格式C:检测不到旧列名,尝试直接用 pandas 读取
|
||||
self.history_data = pd.read_excel(self.history_file)
|
||||
print(f"加载成功(默认读取),共{len(self.history_data)}条历史记录")
|
||||
print(f"数据列: {list(self.history_data.columns)}")
|
||||
|
||||
# 如果没有"号码"列但有分列红球,尝试标准化
|
||||
if '号码' not in self.history_data.columns:
|
||||
if any(c in self.history_data.columns for c in ['红球 1', '红球1']):
|
||||
self._normalize_history_format()
|
||||
# 使用公共模块加载历史数据(统一处理格式A/B/C)
|
||||
from history_loader import load_history_dataframe
|
||||
self.history_data = load_history_dataframe(self.history_file)
|
||||
|
||||
if self.history_data.empty:
|
||||
print("错误: 历史数据文件为空")
|
||||
return False
|
||||
|
||||
print(f"加载成功,共 {len(self.history_data)} 条历史记录")
|
||||
print(f"数据列: {list(self.history_data.columns)}")
|
||||
|
||||
# 解析号码列
|
||||
def parse_numbers(row):
|
||||
"""解析单行号码数据
|
||||
@@ -211,72 +159,6 @@ class DoubleColorBallGenerator:
|
||||
print(traceback.format_exc())
|
||||
return False
|
||||
|
||||
def _normalize_history_format(self):
|
||||
"""将格式A(分列红球)转换为格式B(统一号码列 + 标准列名)。
|
||||
|
||||
格式A列名: 期号 | 开奖日期 | 红球 1 | 红球 2 | 红球 3 | 红球 4 | 红球 5 | 红球 6 | 蓝球 | 特别号
|
||||
格式B列名: 开奖时间 | 期数 | 号码 | 开机号 | 和值特征 | 奇偶比 | 大小比 | 奇偶形态 | 跨度 | 其他
|
||||
|
||||
在 self.history_data 上原地操作,构建 '号码' 列和标准列名。
|
||||
"""
|
||||
df = self.history_data
|
||||
standard_columns = ['开奖时间', '期数', '号码', '开机号', '和值特征', '奇偶比', '大小比', '奇偶形态', '跨度', '其他']
|
||||
|
||||
# 构建号码列:将 红球1~6 + 蓝球 拼接为 14 位字符串
|
||||
red_cols = [f'红球 {i}' for i in range(1, 7)]
|
||||
blue_col = '蓝球'
|
||||
|
||||
def build_number_string(row):
|
||||
parts = []
|
||||
for c in red_cols:
|
||||
val = row.get(c)
|
||||
if pd.isna(val):
|
||||
return None
|
||||
s = str(int(val)) if isinstance(val, (int, float)) else str(val).strip()
|
||||
parts.append(s.zfill(2))
|
||||
blue_val = row.get(blue_col)
|
||||
if pd.isna(blue_val):
|
||||
return None
|
||||
blue_s = str(int(blue_val)) if isinstance(blue_val, (int, float)) else str(blue_val).strip()
|
||||
return ''.join(parts) + blue_s.zfill(2)
|
||||
|
||||
df = df.copy()
|
||||
df['号码'] = df.apply(build_number_string, axis=1)
|
||||
|
||||
# 重命名列到标准列名 — 仅映射有语义对应关系的列
|
||||
# 格式A -> 格式B 语义映射:
|
||||
# 期号 -> 开奖时间(实际存的是日期)
|
||||
# 开奖日期 -> 期数(实际存的是期号数字)
|
||||
# 特别号 -> 跨度
|
||||
# 红球 1~6 和 蓝球 用于构建"号码"列后不再映射
|
||||
# 格式A不含开机号/和值特征/奇偶比/大小比/奇偶形态等字段,留空
|
||||
rename_map = {}
|
||||
if '期号' in df.columns:
|
||||
rename_map['期号'] = '开奖时间'
|
||||
if '开奖日期' in df.columns:
|
||||
rename_map['开奖日期'] = '期数'
|
||||
if '特别号' in df.columns:
|
||||
rename_map['特别号'] = '跨度'
|
||||
|
||||
df = df.rename(columns=rename_map)
|
||||
|
||||
# 删除已用于构建号码列的原始分列(避免数据重复)
|
||||
for col in [f'红球 {i}' for i in range(1, 7)] + ['蓝球']:
|
||||
if col in df.columns:
|
||||
df = df.drop(columns=[col])
|
||||
|
||||
# 确保所有标准列都存在(格式A缺失的字段留空)
|
||||
for col in standard_columns:
|
||||
if col not in df.columns:
|
||||
df[col] = ''
|
||||
|
||||
# 调整列顺序
|
||||
df = df[[c for c in standard_columns if c in df.columns] + [c for c in df.columns if c not in standard_columns]]
|
||||
|
||||
self.history_data = df.reset_index(drop=True)
|
||||
print(f"已标准化数据格式,共 {len(df)} 条记录")
|
||||
print(f"标准化后列名: {list(df.columns)}")
|
||||
|
||||
def _calculate_statistics(self):
|
||||
"""计算统计数据"""
|
||||
if self.history_data is None or len(self.history_data) == 0:
|
||||
|
||||
Reference in New Issue
Block a user