vincent
e43862be9f
功能一：生成记录预览 - 新增 API /api/records/<id>/preview 读取 Excel 返回号码数据 - 前端记录列表新增「预览」按钮，弹窗展示号码表格 - 支持最多显示100注，完整数据可通过下载按钮获取 功能二：中奖比对 - 新增 API /api/records/<id>/compare 根据生成时间匹配开奖数据 - 比对规则：生成时间 < 开奖日20:00 → 与当日开奖比对；≥ 20:00 → 与下次开奖日比对 - 按双色球官方规则判断一等奖至六等奖 - 前端记录列表新增「比对」按钮，弹窗展示比对结果 - 中奖号码高亮显示，未中奖号码半透明 - 若下次开奖尚未举行，返回等待状态提示 Co-authored-by: multica-agent <github@multica.ai>
925 lines
34 KiB
Python
925 lines
34 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
双色球 Web UI 服务
|
||
集成号码生成、历史数据查看、生成记录管理
|
||
监听 0.0.0.0，支持 PC 端和移动端响应式访问
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import uuid
|
||
import traceback
|
||
import threading
|
||
from datetime import datetime, timedelta
|
||
from flask import Flask, send_from_directory, jsonify, request, send_file, abort
|
||
from functools import wraps
|
||
|
||
import pandas as pd
|
||
import re
|
||
|
||
# 将项目目录加入路径，以便导入 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 load_latest_draw():
|
||
"""加载最近一期开奖数据，返回 dict 或 None。"""
|
||
try:
|
||
df = load_history_dataframe()
|
||
if df is None or df.empty:
|
||
return None
|
||
|
||
# 解析日期，按开奖时间降序找最近一期
|
||
df['_dt'] = pd.to_datetime(df['开奖时间'], errors='coerce')
|
||
df = df.sort_values('_dt', ascending=False).reset_index(drop=True)
|
||
|
||
if df.empty:
|
||
return None
|
||
|
||
row = df.iloc[0]
|
||
s = str(row.get('号码', '')).strip()
|
||
reds, blue = parse_number_string(s)
|
||
|
||
draw_date_str = str(row.get('开奖时间', '')).strip()
|
||
draw_date = None
|
||
try:
|
||
draw_date = datetime.strptime(draw_date_str[:10], '%Y-%m-%d')
|
||
except (ValueError, TypeError):
|
||
try:
|
||
dt = pd.to_datetime(draw_date_str, errors='coerce')
|
||
if pd.notna(dt):
|
||
draw_date = dt.to_pydatetime()
|
||
except Exception:
|
||
pass
|
||
|
||
return {
|
||
'date': draw_date_str,
|
||
'date_obj': draw_date,
|
||
'period': str(row.get('期数', '')).strip(),
|
||
'reds': reds,
|
||
'blue': blue,
|
||
'raw': {k: str(row.get(k, '')).strip() for k in ['开奖时间', '期数', '号码', '和值特征', '奇偶比', '大小比', '跨度']}
|
||
}
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return None
|
||
|
||
|
||
def find_next_draw_after(date_obj):
|
||
"""查找给定日期之后最近一期的开奖数据，返回 dict 或 None。"""
|
||
try:
|
||
df = load_history_dataframe()
|
||
if df is None or df.empty:
|
||
return None
|
||
|
||
df['_dt'] = pd.to_datetime(df['开奖时间'], errors='coerce')
|
||
# 找开奖日期 > date_obj 的最近一期
|
||
future = df[df['_dt'] > date_obj].sort_values('_dt', ascending=True)
|
||
if future.empty:
|
||
return None
|
||
|
||
row = future.iloc[0]
|
||
s = str(row.get('号码', '')).strip()
|
||
reds, blue = parse_number_string(s)
|
||
|
||
draw_date_str = str(row.get('开奖时间', '')).strip()
|
||
draw_date = None
|
||
try:
|
||
draw_date = datetime.strptime(draw_date_str[:10], '%Y-%m-%d')
|
||
except (ValueError, TypeError):
|
||
try:
|
||
dt = pd.to_datetime(draw_date_str, errors='coerce')
|
||
if pd.notna(dt):
|
||
draw_date = dt.to_pydatetime()
|
||
except Exception:
|
||
pass
|
||
|
||
return {
|
||
'date': draw_date_str,
|
||
'date_obj': draw_date,
|
||
'period': str(row.get('期数', '')).strip(),
|
||
'reds': reds,
|
||
'blue': blue,
|
||
'raw': {k: str(row.get(k, '')).strip() for k in ['开奖时间', '期数', '号码', '和值特征', '奇偶比', '大小比', '跨度']}
|
||
}
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return None
|
||
|
||
# ============================================================
|
||
# 认证装饰器（可选）
|
||
# ============================================================
|
||
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'])
|
||
|
||
|
||
# ============================================================
|
||
# 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/records/<record_id>/preview')
|
||
@require_auth
|
||
def api_preview_record(record_id):
|
||
"""预览生成记录中的号码数据（无需下载 Excel）"""
|
||
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 not os.path.exists(filepath):
|
||
return jsonify({'success': False, 'error': '文件不存在'}), 404
|
||
|
||
# 读取 Excel
|
||
df = pd.read_excel(filepath, sheet_name='生成号码', engine='openpyxl')
|
||
|
||
tickets = []
|
||
red_cols = [f'红球{i}' for i in range(1, 7)]
|
||
for _, row in df.iterrows():
|
||
reds = [int(row[c]) for c in red_cols if c in df.columns and pd.notna(row[c])]
|
||
blue = int(row['蓝球']) if '蓝球' in df.columns and pd.notna(row['蓝球']) else 0
|
||
ticket = {
|
||
'index': int(row['序号']) if '序号' in df.columns and pd.notna(row['序号']) else len(tickets) + 1,
|
||
'reds': reds,
|
||
'blue': blue,
|
||
'sum_value': int(row['和值']) if '和值' in df.columns and pd.notna(row['和值']) else sum(reds),
|
||
'odd_even': str(row['奇偶比']) if '奇偶比' in df.columns and pd.notna(row['奇偶比']) else '',
|
||
'size_ratio': str(row['大小比']) if '大小比' in df.columns and pd.notna(row['大小比']) else '',
|
||
'span': int(row['跨度']) if '跨度' in df.columns and pd.notna(row['跨度']) else (max(reds) - min(reds) if reds else 0),
|
||
}
|
||
tickets.append(ticket)
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'record': target,
|
||
'tickets': tickets,
|
||
'total': len(tickets)
|
||
}
|
||
})
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return jsonify({'success': False, 'error': '预览数据加载失败'}), 500
|
||
|
||
|
||
# ============================================================
|
||
# API：中奖比对
|
||
# ============================================================
|
||
@app.route('/api/records/<record_id>/compare')
|
||
@require_auth
|
||
def api_compare_record(record_id):
|
||
"""将生成记录与对应开奖数据进行中奖比对。
|
||
|
||
比对规则：
|
||
- 生成时间 < 开奖日 20:00 → 与当日开奖比对
|
||
- 生成时间 >= 开奖日 20:00 → 与下次开奖日比对
|
||
|
||
双色球中奖规则：
|
||
- 一等奖：6红+1蓝
|
||
- 二等奖：6红+0蓝
|
||
- 三等奖：5红+1蓝
|
||
- 四等奖：5红+0蓝 或 4红+1蓝
|
||
- 五等奖：4红+0蓝 或 3红+1蓝
|
||
- 六等奖：2红+1蓝 或 1红+1蓝 或 0红+1蓝
|
||
- 未中奖：其他
|
||
"""
|
||
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 not os.path.exists(filepath):
|
||
return jsonify({'success': False, 'error': '文件不存在'}), 404
|
||
|
||
# 解析生成时间
|
||
gen_time_str = target.get('created_at', '')
|
||
try:
|
||
gen_time = datetime.strptime(gen_time_str, '%Y-%m-%d %H:%M:%S')
|
||
except (ValueError, TypeError):
|
||
return jsonify({'success': False, 'error': '无法解析生成时间'}), 400
|
||
|
||
# 读取生成的号码
|
||
df = pd.read_excel(filepath, sheet_name='生成号码', engine='openpyxl')
|
||
red_cols = [f'红球{i}' for i in range(1, 7)]
|
||
tickets = []
|
||
for _, row in df.iterrows():
|
||
reds = [int(row[c]) for c in red_cols if c in df.columns and pd.notna(row[c])]
|
||
blue = int(row['蓝球']) if '蓝球' in df.columns and pd.notna(row['蓝球']) else 0
|
||
tickets.append({'reds': reds, 'blue': blue})
|
||
|
||
if not tickets:
|
||
return jsonify({'success': False, 'error': '无号码数据可比对'}), 400
|
||
|
||
# 加载历史开奖数据，找最近一期
|
||
latest_draw = load_latest_draw()
|
||
if not latest_draw or not latest_draw.get('reds'):
|
||
return jsonify({'success': False, 'error': '无法加载开奖数据'}), 500
|
||
|
||
# 确定比对的开奖期
|
||
# 规则：生成时间 < 开奖日 20:00 → 与当日开奖比对；>= 20:00 → 与下次开奖日比对
|
||
draw_to_compare = None
|
||
compare_type = ''
|
||
|
||
latest_date = latest_draw.get('date_obj')
|
||
if latest_date:
|
||
# 当日 20:00 的时间点
|
||
draw_day_8pm = latest_date.replace(hour=20, minute=0, second=0, microsecond=0)
|
||
|
||
if gen_time < draw_day_8pm:
|
||
# 生成时间在开奖日 20:00 之前 → 与当日（最近一期）比对
|
||
draw_to_compare = latest_draw
|
||
compare_type = 'current'
|
||
else:
|
||
# 生成时间在开奖日 20:00 之后 → 与下次开奖日比对
|
||
next_draw = find_next_draw_after(latest_date)
|
||
if next_draw:
|
||
draw_to_compare = next_draw
|
||
compare_type = 'next'
|
||
else:
|
||
# 暂无下次开奖数据，返回等待状态
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'status': 'waiting',
|
||
'message': f'生成时间 {gen_time_str} 在最近开奖日 {latest_draw["date"]} 20:00 之后，下次开奖尚未举行，请等待开奖后再次比对。',
|
||
'gen_time': gen_time_str,
|
||
'latest_draw_date': latest_draw['date'],
|
||
'latest_draw_period': latest_draw['period'],
|
||
'compare_rule': '生成时间 ≥ 开奖日 20:00 → 等待下次开奖日比对'
|
||
}
|
||
})
|
||
else:
|
||
# 无法解析开奖日期，直接用最近一期
|
||
draw_to_compare = latest_draw
|
||
compare_type = 'current_fallback'
|
||
|
||
# 执行比对
|
||
draw_reds = set(draw_to_compare['reds'])
|
||
draw_blue = draw_to_compare['blue']
|
||
|
||
results = []
|
||
for i, t in enumerate(tickets):
|
||
ticket_reds = set(t['reds'])
|
||
ticket_blue = t['blue']
|
||
|
||
red_matches = len(ticket_reds & draw_reds)
|
||
blue_match = (ticket_blue == draw_blue)
|
||
|
||
# 判断中奖等级
|
||
prize_level = '未中奖'
|
||
prize_desc = ''
|
||
|
||
if red_matches == 6 and blue_match:
|
||
prize_level = '一等奖'
|
||
prize_desc = '6红+1蓝 中大奖！'
|
||
elif red_matches == 6 and not blue_match:
|
||
prize_level = '二等奖'
|
||
prize_desc = '6红+0蓝'
|
||
elif red_matches == 5 and blue_match:
|
||
prize_level = '三等奖'
|
||
prize_desc = '5红+1蓝'
|
||
elif (red_matches == 5 and not blue_match) or (red_matches == 4 and blue_match):
|
||
prize_level = '四等奖'
|
||
prize_desc = f'{red_matches}红+{"1" if blue_match else "0"}蓝'
|
||
elif (red_matches == 4 and not blue_match) or (red_matches == 3 and blue_match):
|
||
prize_level = '五等奖'
|
||
prize_desc = f'{red_matches}红+{"1" if blue_match else "0"}蓝'
|
||
elif blue_match and red_matches <= 2:
|
||
prize_level = '六等奖'
|
||
prize_desc = f'{red_matches}红+1蓝'
|
||
|
||
results.append({
|
||
'index': i + 1,
|
||
'reds': t['reds'],
|
||
'blue': ticket_blue,
|
||
'red_matches': red_matches,
|
||
'blue_match': blue_match,
|
||
'prize_level': prize_level,
|
||
'prize_desc': prize_desc,
|
||
'is_win': prize_level != '未中奖'
|
||
})
|
||
|
||
win_count = sum(1 for r in results if r['is_win'])
|
||
|
||
return jsonify({
|
||
'success': True,
|
||
'data': {
|
||
'status': 'compared',
|
||
'gen_time': gen_time_str,
|
||
'compare_type': compare_type,
|
||
'compare_rule': '生成时间 < 开奖日 20:00 → 与当日开奖比对' if compare_type == 'current' else '生成时间 ≥ 开奖日 20:00 → 与下次开奖日比对',
|
||
'draw': {
|
||
'date': draw_to_compare['date'],
|
||
'period': draw_to_compare['period'],
|
||
'reds': draw_to_compare['reds'],
|
||
'blue': draw_to_compare['blue']
|
||
},
|
||
'results': results,
|
||
'total': len(results),
|
||
'win_count': win_count,
|
||
'win_tickets': [r for r in results if r['is_win']]
|
||
}
|
||
})
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return jsonify({'success': False, 'error': '中奖比对失败'}), 500
|
||
|
||
|
||
# ============================================================
|
||
# 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['大小比']),
|
||
'跨度': 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)
|