vincent/Lottery
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Lottery/app.py
T
vincent 5cebbfa433 fix: 修复代码审查反馈全部问题 
审查意见修复清单:

P1 列映射语义修复 (lottery.py):
- _normalize_history_format() 不再将红球2-6映射到开机号/和值特征/奇偶比等
  格式A不含这些特征字段，缺失列留空，前端做降级显示
- 删除已用于构建号码列的原始分列，避免数据重复

P2 架构优化:
- 提取 Excel 兼容逻辑到公共模块 history_loader.py
  lottery.py 和 app.py 共同引用，消除三处重复代码
- web_executor.py 标记为已废弃，功能已整合到 app.py

部署修复:
- 删除 deploy/lotto-web.service (旧服务)，仅保留 lotto-app.service
- 更新 deploy/DEPLOY.md: 端口5000→8085, 接口清单更新, 添加迁移说明

安全加固:
- API Token 改为环境变量读取: os.environ.get('LOTTO_API_TOKEN')
- 错误信息不再暴露内部异常，改为通用错误消息+日志记录
- 目录遍历防护改用 os.path.realpath 检查最终路径

其他:
- .gitignore 补充排除 双色球历史数据.xlsx
- app.py 引用公共模块，简化 get_statistics_data 和 load_history_dataframe

测试验证: 全部 API 测试通过，120条历史数据正确解析

Issue: BIZ-75
2026-07-04 01:28:57 +08:00

615 lines
22 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
双色球 Web UI 服务
集成号码生成、历史数据查看、生成记录管理
监听 0.0.0.0，支持 PC 端和移动端响应式访问
"""
import os
import sys
import json
import uuid
import shutil
import traceback
import threading
from datetime import datetime
from flask import Flask, send_from_directory, jsonify, request, send_file, abort
from functools import wraps
# 将项目目录加入路径，以便导入 lottery.py 和 history_loader.py
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# 导入公共历史数据加载模块
from history_loader import load_history_dataframe as _load_history, parse_number_string, compute_statistics
sys.path.insert(0, BASE_DIR)
# 导入号码生成器
from lottery import DoubleColorBallGenerator
app = Flask(__name__)
# ============================================================
# 配置
# ============================================================
CONFIG = {
'host': '0.0.0.0',
'port': 8085,
'history_file': os.path.join(BASE_DIR, '双色球历史数据.xlsx'),
'lottery_output_dir': os.path.join(BASE_DIR, 'lottery'),
'records_file': os.path.join(BASE_DIR, '.generation_records.json'),
'api_token': os.environ.get('LOTTO_API_TOKEN', 'lotto2026'),
'auth_enabled': False,
'max_tickets': 1000,
'default_tickets': 10,
# 数据抓取配置（原 web_executor.py 功能）
'fetch_script': os.path.join(BASE_DIR, 'fetch_data.py'),
'fetch_status_file': os.path.join(BASE_DIR, '.fetch_status.json'),
'fetch_timeout': 300, # 抓取超时秒数
}
# ============================================================
# 生成记录管理（线程安全）
# ============================================================
# 全局锁：保护 .generation_records.json 的并发读写
records_lock = threading.Lock()
def load_records():
"""加载生成记录（线程安全读取）"""
with records_lock:
if os.path.exists(CONFIG['records_file']):
try:
with open(CONFIG['records_file'], 'r', encoding='utf-8') as f:
return json.load(f)
except (json.JSONDecodeError, IOError):
return []
return []
def save_records(records):
"""保存生成记录（线程安全写入）"""
with records_lock:
os.makedirs(os.path.dirname(CONFIG['records_file']), exist_ok=True)
# 先写临时文件再原子替换，防止写入中途崩溃导致数据损坏
tmp_path = CONFIG['records_file'] + '.tmp'
with open(tmp_path, 'w', encoding='utf-8') as f:
json.dump(records, f, ensure_ascii=False, indent=2)
os.replace(tmp_path, CONFIG['records_file'])
def add_record(strategy, num_tickets, filename):
"""添加一条生成记录（原子操作：读-改-写全程持锁）"""
with records_lock:
# 读取现有记录
if os.path.exists(CONFIG['records_file']):
try:
with open(CONFIG['records_file'], 'r', encoding='utf-8') as f:
records = json.load(f)
except (json.JSONDecodeError, IOError):
records = []
else:
records = []
# 插入新记录
new_record = {
'id': str(uuid.uuid4())[:8],
'created_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'strategy': '高级策略' if strategy == 'advanced' else '基础策略',
'num_tickets': num_tickets,
'filename': filename,
'filesize': os.path.getsize(os.path.join(BASE_DIR, filename)) if os.path.exists(os.path.join(BASE_DIR, filename)) else 0
}
records.insert(0, new_record)
# 原子写入
os.makedirs(os.path.dirname(CONFIG['records_file']), exist_ok=True)
tmp_path = CONFIG['records_file'] + '.tmp'
with open(tmp_path, 'w', encoding='utf-8') as f:
json.dump(records, f, ensure_ascii=False, indent=2)
os.replace(tmp_path, CONFIG['records_file'])
return new_record
# ============================================================
# Excel 历史数据读取辅助
# ============================================================
# 历史数据加载 — 使用公共模块 history_loader.py
# ============================================================
def load_history_dataframe():
"""加载历史数据 Excel，委托公共模块处理多格式兼容。"""
return _load_history(CONFIG['history_file'])
# ============================================================
# 认证装饰器（可选）
# ============================================================
def require_auth(f):
@wraps(f)
def decorated(*args, **kwargs):
if CONFIG['auth_enabled']:
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if token != CONFIG['api_token']:
return jsonify({'success': False, 'error': '未授权访问'}), 401
return f(*args, **kwargs)
return decorated
# ============================================================
# API：号码生成
# ============================================================
@app.route('/api/generate', methods=['POST'])
@require_auth
def api_generate():
"""生成双色球号码"""
try:
data = request.get_json(force=True, silent=True) or {}
num_tickets = int(data.get('num_tickets', CONFIG['default_tickets']))
strategy = data.get('strategy', 'advanced')
# 参数校验
if num_tickets < 1 or num_tickets > CONFIG['max_tickets']:
return jsonify({
'success': False,
'error': f'注数必须在 1-{CONFIG["max_tickets"]} 之间'
}), 400
if strategy not in ('advanced', 'basic'):
return jsonify({'success': False, 'error': '策略参数无效，请使用 advanced 或 basic'}), 400
# 初始化生成器
generator = DoubleColorBallGenerator(CONFIG['history_file'])
if not generator.load_history_data():
return jsonify({'success': False, 'error': '无法加载历史数据'}), 500
# 生成号码
tickets_df = generator.generate_multiple_tickets(num_tickets, strategy)
if tickets_df.empty:
return jsonify({'success': False, 'error': '号码生成失败'}), 500
# 保存到 Excel
filename = generator.save_to_excel(tickets_df, num_tickets, strategy)
if not filename:
return jsonify({'success': False, 'error': '文件保存失败'}), 500
# 获取相对路径
rel_path = os.path.relpath(filename, BASE_DIR)
# 添加生成记录
record = add_record(strategy, num_tickets, rel_path)
# 构建返回数据
tickets_data = []
for _, row in tickets_df.iterrows():
tickets_data.append({
'index': int(row['序号']),
'reds': [int(row[f'红球{i}']) for i in range(1, 7)],
'blue': int(row['蓝球']),
'sum_value': int(row['和值']),
'odd_even': row['奇偶比'],
'size_ratio': row['大小比'],
'span': int(row['跨度'])
})
# 获取统计信息
stats = get_statistics_data(generator)
return jsonify({
'success': True,
'data': {
'tickets': tickets_data[:50], # 前端最多展示 50 注
'total': len(tickets_data),
'filename': rel_path,
'download_url': f'/api/download/{rel_path}',
'record': record,
'statistics': stats
}
})
except Exception as e:
traceback.print_exc()
return jsonify({'success': False, 'error': '号码生成失败，请检查历史数据文件是否完整'}), 500
def get_statistics_data(generator=None):
"""获取统计数据 — 委托公共模块计算"""
return compute_statistics(CONFIG['history_file'])
return stats
# ============================================================
# API：获取统计数据
# ============================================================
@app.route('/api/statistics')
@require_auth
def api_statistics():
"""获取统计数据"""
try:
stats = get_statistics_data()
return jsonify({'success': True, 'data': stats})
except Exception as e:
traceback.print_exc()
return jsonify({'success': False, 'error': '获取统计数据失败'}), 500
# ============================================================
@app.route('/api/records')
@require_auth
def api_records():
"""获取生成记录列表"""
try:
page = int(request.args.get('page', 1))
page_size = int(request.args.get('page_size', 20))
records = load_records()
total = len(records)
start = (page - 1) * page_size
end = start + page_size
page_records = records[start:end]
return jsonify({
'success': True,
'data': {
'records': page_records,
'total': total,
'page': page,
'page_size': page_size
}
})
except Exception as e:
traceback.print_exc()
return jsonify({'success': False, 'error': '获取生成记录失败'}), 500
# ============================================================
# API：删除生成记录
# ============================================================
@app.route('/api/records/<record_id>', methods=['DELETE'])
@require_auth
def api_delete_record(record_id):
"""删除生成记录"""
try:
records = load_records()
target = None
for r in records:
if r['id'] == record_id:
target = r
break
if not target:
return jsonify({'success': False, 'error': '记录不存在'}), 404
# 删除文件
filepath = os.path.join(BASE_DIR, target['filename'])
if os.path.exists(filepath):
os.remove(filepath)
# 删除记录（加锁保护读-改-写）
with records_lock:
if os.path.exists(CONFIG['records_file']):
try:
with open(CONFIG['records_file'], 'r', encoding='utf-8') as f:
records = json.load(f)
except (json.JSONDecodeError, IOError):
records = []
else:
records = []
records = [r for r in records if r['id'] != record_id]
tmp_path = CONFIG['records_file'] + '.tmp'
with open(tmp_path, 'w', encoding='utf-8') as f:
json.dump(records, f, ensure_ascii=False, indent=2)
os.replace(tmp_path, CONFIG['records_file'])
return jsonify({'success': True, 'message': '记录已删除'})
except Exception as e:
traceback.print_exc()
return jsonify({'success': False, 'error': '删除记录失败'}), 500
# ============================================================
@app.route('/api/download/<path:filepath>')
@require_auth
def api_download(filepath):
"""下载文件"""
try:
# 安全检查：防止目录遍历
safe_path = os.path.normpath(filepath)
full_path = os.path.realpath(os.path.join(BASE_DIR, safe_path))
# 使用 realpath 检查最终路径是否仍在 BASE_DIR 内
if not full_path.startswith(os.path.realpath(BASE_DIR)):
abort(403)
if not os.path.exists(full_path):
abort(404)
return send_file(full_path, as_attachment=True)
except Exception:
abort(404)
# ============================================================
# API：历史数据查看
# ============================================================
@app.route('/api/history')
@require_auth
def api_history():
"""获取双色球历史开奖数据"""
try:
page = int(request.args.get('page', 1))
page_size = int(request.args.get('page_size', 20))
search = request.args.get('search', '').strip()
# 读取历史数据
if not os.path.exists(CONFIG['history_file']):
return jsonify({'success': False, 'error': '历史数据文件不存在'}), 404
import pandas as pd
import re
# 使用智能加载函数
data_df = load_history_dataframe()
# 解析红球 (号码列是 6 红球+1 蓝球的拼接字符串，如 '09101316192108')
def parse_red_balls(val):
s = str(val).strip()
if len(s) >= 12:
return [int(s[i:i+2]) for i in range(0, 12, 2)]
return []
def parse_blue_ball(val):
s = str(val).strip()
if len(s) >= 14:
return int(s[12:14])
return None
data_df['红球列表'] = data_df['号码'].apply(parse_red_balls)
data_df['蓝球'] = data_df['号码'].apply(parse_blue_ball)
# 搜索过滤
if search:
mask = data_df.astype(str).apply(
lambda row: row.astype(str).str.contains(search, na=False).any(), axis=1
)
data_df = data_df[mask]
total = len(data_df)
# 分页
start = (page - 1) * page_size
end = start + page_size
page_df = data_df.iloc[start:end]
# 转换为 JSON
records = []
for _, row in page_df.iterrows():
reds = row['红球列表']
record = {
'开奖日期': str(row['开奖时间']),
'期号': str(row['期数']),
'红球': reds if len(reds) == 6 else [],
'蓝球': row['蓝球'],
'开机号': str(row['开机号']),
'和值': str(row['和值特征']),
'奇偶形态': str(row['奇偶比']),
'大小比': str(row['大小比']),
'跨度': str(row['跨度']),
}
records.append(record)
return jsonify({
'success': True,
'data': {
'records': records,
'total': total,
'page': page,
'page_size': page_size,
'columns': ['开奖日期', '期号', '红球', '蓝球', '开机号', '和值', '大小比', '跨度']
}
})
except Exception as e:
traceback.print_exc()
return jsonify({'success': False, 'error': '获取历史数据失败'}), 500
# ============================================================
# API：系统状态
# ============================================================
@app.route('/api/status')
def api_status():
"""获取系统状态"""
try:
history_exists = os.path.exists(CONFIG['history_file'])
history_size = os.path.getsize(CONFIG['history_file']) if history_exists else 0
lottery_files = []
if os.path.exists(CONFIG['lottery_output_dir']):
lottery_files = [f for f in os.listdir(CONFIG['lottery_output_dir']) if f.endswith('.xlsx')]
records = load_records()
return jsonify({
'success': True,
'data': {
'server_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'history_exists': history_exists,
'history_size': history_size,
'total_generations': len(records),
'total_lottery_files': len(lottery_files),
'config': {
'port': CONFIG['port'],
'auth_enabled': CONFIG['auth_enabled'],
'max_tickets': CONFIG['max_tickets']
}
}
})
except Exception as e:
traceback.print_exc()
return jsonify({'success': False, 'error': '获取系统状态失败'}), 500
# ============================================================
@app.route('/')
def index():
"""首页 - 双色球 Web UI"""
return send_from_directory(BASE_DIR, 'index.html')
@app.route('/api/config')
def api_config():
"""获取前端配置"""
return jsonify({
'success': True,
'data': {
'max_tickets': CONFIG['max_tickets'],
'default_tickets': CONFIG['default_tickets'],
'auth_enabled': CONFIG['auth_enabled']
}
})
# ============================================================
# 数据抓取控制台（原 web_executor.py 功能整合）
# ============================================================
# 全局抓取状态
fetch_status = {
"is_running": False,
"last_update": None,
"last_record_count": 0,
"last_error": None
}
fetch_lock = threading.Lock()
def load_fetch_status():
"""从文件加载抓取状态"""
global fetch_status
if os.path.exists(CONFIG['fetch_status_file']):
try:
with open(CONFIG['fetch_status_file'], 'r', encoding='utf-8') as f:
saved = json.load(f)
with fetch_lock:
# 保留当前 is_running 状态（运行中不覆盖）
running = fetch_status.get('is_running', False)
fetch_status = saved
fetch_status['is_running'] = running
except (json.JSONDecodeError, IOError):
pass
def save_fetch_status():
"""保存抓取状态到文件"""
with fetch_lock:
with open(CONFIG['fetch_status_file'], 'w', encoding='utf-8') as f:
json.dump(fetch_status, f, ensure_ascii=False, indent=2)
@app.route('/fetch')
def fetch_console():
"""数据抓取控制台页面"""
return send_from_directory(BASE_DIR, 'web_console.html')
@app.route('/api/fetch/status')
def api_fetch_status():
"""获取抓取执行状态"""
with fetch_lock:
return jsonify({
"success": True,
"isRunning": fetch_status.get("is_running", False),
"lastUpdate": fetch_status.get("last_update"),
"recordCount": fetch_status.get("last_record_count", 0),
"lastError": fetch_status.get("last_error")
})
@app.route('/api/fetch/execute', methods=['POST'])
def api_fetch_execute():
"""触发数据抓取"""
global fetch_status
with fetch_lock:
if fetch_status.get("is_running", False):
return jsonify({
"success": False,
"error": "任务正在执行中，请稍后再试"
}), 409
# 启动后台执行线程
def run_fetch_script():
global fetch_status
with fetch_lock:
fetch_status["is_running"] = True
fetch_status["last_error"] = None
save_fetch_status()
try:
import subprocess
print(f"[{datetime.now()}] 开始执行抓取脚本...")
result = subprocess.run(
[sys.executable, CONFIG['fetch_script']],
capture_output=True,
text=True,
timeout=CONFIG['fetch_timeout']
)
if result.returncode == 0:
# 解析输出获取记录数
record_count = 0
for line in result.stdout.split('\n'):
if '共保存' in line and '条记录' in line:
try:
record_count = int(line.split('共保存')[1].split('条记录')[0].strip())
except ValueError:
pass
elif '成功解析' in line and '条数据' in line:
try:
record_count = int(line.split('成功解析')[1].split('条数据')[0].strip())
except ValueError:
pass
with fetch_lock:
fetch_status["last_update"] = datetime.now().isoformat()
fetch_status["last_record_count"] = record_count
fetch_status["is_running"] = False
save_fetch_status()
print(f"✅ 抓取成功，共 {record_count} 条数据")
else:
error_msg = result.stderr or f"脚本执行失败，返回码：{result.returncode}"
with fetch_lock:
fetch_status["last_error"] = error_msg
fetch_status["is_running"] = False
save_fetch_status()
print(f"❌ {error_msg}")
except subprocess.TimeoutExpired:
error_msg = f"脚本执行超时（超过 {CONFIG['fetch_timeout']} 秒）"
with fetch_lock:
fetch_status["last_error"] = error_msg
fetch_status["is_running"] = False
save_fetch_status()
print(f"❌ {error_msg}")
except Exception as e:
error_msg = f"执行异常：{str(e)}"
with fetch_lock:
fetch_status["last_error"] = error_msg
fetch_status["is_running"] = False
save_fetch_status()
print(f"❌ {error_msg}")
thread = threading.Thread(target=run_fetch_script, daemon=True)
thread.start()
return jsonify({
"success": True,
"message": "任务已启动，正在执行中..."
})
# ============================================================
# 启动服务
# ============================================================
if __name__ == '__main__':
# 加载抓取状态
load_fetch_status()
print('=' * 60)
print('🎯 双色球 Web UI 服务（统一）')
print('=' * 60)
print(f'\n📂 项目路径: {BASE_DIR}')
print(f'📁 历史数据: {CONFIG["history_file"]}')
print(f'📁 生成目录: {CONFIG["lottery_output_dir"]}')
print(f'📁 抓取脚本: {CONFIG["fetch_script"]}')
print(f'\n🌐 服务地址: http://{CONFIG["host"]}:{CONFIG["port"]}')
print(f' 局域网访问: http://<本机IP>:{CONFIG["port"]}')
print(f' 抓取控制台: http://<本机IP>:{CONFIG["port"]}/fetch')
print(f'\n✅ 服务就绪！')
print('=' * 60)
app.run(host=CONFIG['host'], port=CONFIG['port'], debug=False, threaded=True)
Reference in New Issue View Git Blame Copy Permalink