""" 短道速滑竞赛成绩 Dashboard 后端API服务 """ APP_VERSION = "1.1.96" import os import sys import json import time import hashlib import logging import functools from datetime import datetime, timedelta from functools import wraps from flask import Flask, request, jsonify, send_from_directory, send_file, g, session from flask_cors import CORS from flask_wtf.csrf import CSRFProtect from flask_limiter import Limiter from flask_limiter.util import get_remote_address from flask_caching import Cache import bcrypt import database import scraper sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from database import ( init_db, get_chaoyang_athletes, get_athlete_results, get_athlete_events, get_event_timeseries, get_opponent_results, get_current_matches, get_all_matches, get_seasons, get_last_update, get_match_race_groups, get_race_group_results, get_athlete_categories, get_team_aliases, add_team_alias, delete_team_alias, get_athlete_overrides, set_athlete_override, delete_athlete_override, add_athlete, get_all_athletes, soft_delete_athlete, restore_athlete, delete_athlete, get_athlete_names, get_active_athlete_names, is_chaoyang_team, recalculate_chaoyang_flags, get_db, get_active_team, get_all_teams, set_active_team, get_team_config, get_athlete_photo, set_athlete_photo, delete_athlete_photo, get_training_types, add_training_type, delete_training_type, get_training_records, add_training_record, delete_training_record, add_training_records_batch, update_training_record, submit_training_change, get_pending_changes, approve_pending_change, reject_pending_change, get_athlete_attendance_detail, get_attendance_stats, get_attendance_detail, get_fee_config, set_fee_config, get_athlete_exemptions, set_athlete_exemption, delete_athlete_exemption, calculate_fees, create_user, get_user_by_username, get_all_users, update_user_role, update_user_password, delete_user, update_user_bio, get_user_public_profile, approve_user, reject_user, get_user_avatar, set_user_avatar, get_messages, add_message, delete_message, like_message, get_message_likes_by_user, log_visit, get_visit_stats, set_athlete_family, delete_athlete_family, get_athlete_family, get_user_family, get_all_families, is_family_of, get_athlete_honors, add_athlete_honor, update_athlete_honor, delete_athlete_honor, get_athlete_profile, update_athlete_profile, get_athlete_custom_results, add_athlete_custom_result, update_athlete_custom_result, delete_athlete_custom_result, get_relay_members, set_relay_members, delete_relay_member, auto_populate_relay_members, get_athlete_relay_results, get_weekly_schedule, set_weekly_schedule, delete_weekly_schedule, get_schedule_for_date, # New fee management functions get_athlete_fee_config, set_athlete_fee_config, delete_athlete_fee_config, get_prepaid_balances, set_prepaid_balance, delete_prepaid_balance, get_prepaid_transactions, get_fee_settlements, add_fee_settlement, delete_fee_settlement, get_accessible_athletes, # Nickname functions get_all_nicknames, add_nickname, delete_nickname, get_nickname_map, resolve_nickname, # 小程序微信登录 get_wx_binding_by_openid, create_wx_binding, approve_wx_binding, get_pending_wx_bindings, create_user_token, get_user_by_token, get_or_create_wx_user, ) from scraper import crawl_update as crawl_update_zhitikeji from scraper_chinacsa import crawl_update as crawl_update_chinacsa from database import close_db from config import get_config from structured_logger import StructuredLogger # 加载配置 config = get_config() app = Flask(__name__, static_folder="static", static_url_path="") app.secret_key = config.SECRET_KEY app.config['SESSION_COOKIE_SAMESITE'] = config.SESSION_COOKIE_SAMESITE app.config['SESSION_COOKIE_HTTPONLY'] = config.SESSION_COOKIE_HTTPONLY app.config['SESSION_COOKIE_SECURE'] = config.SESSION_COOKIE_SECURE app.config['PERMANENT_SESSION_LIFETIME'] = config.PERMANENT_SESSION_LIFETIME # 初始化结构化日志 StructuredLogger.configure(config) # 多服务器同步配置 SERVER_ROLE = os.environ.get("SERVER_ROLE", "master") # "master" or "slave" PEER_URL = os.environ.get("PEER_URL", "") SYNC_SECRET = os.environ.get("SYNC_SECRET", "st_sync_secret_2026") DB_PUSH_DEBOUNCE_SECONDS = 3.0 import threading import requests as http_requests _push_timer = None _push_lock = threading.Lock() # DB 重建信号文件：当 db_health_check 替换了 DB 文件后，通知所有 worker 重建连接 DB_REBUILD_SIGNAL = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", ".db_rebuild_signal") def push_db_to_peer(): """推送完整 short_track.db 到对端服务器（带安全校验）""" from database import get_db db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "short_track.db") if not os.path.exists(db_path): print(f"[sync] DB file not found at {db_path}, skipping push") return # 1. WAL checkpoint: 确保所有写入已合并到主文件 try: db = get_db() db.execute("PRAGMA wal_checkpoint(TRUNCATE)") except Exception as e: print(f"[sync] WAL checkpoint failed: {e}") # 2. 完整性校验：损坏的库坚决不推送 try: db = get_db() result = db.execute("PRAGMA integrity_check").fetchone() if result[0] != "ok": print(f"[sync] DB integrity check FAILED: {result[0]}, refusing to push corrupted DB") return except Exception as e: print(f"[sync] DB integrity check error: {e}, refusing to push") return # 3. 计算 hash（checkpoint 后文件一致） sha = hashlib.sha256() with open(db_path, 'rb') as f: while chunk := f.read(8192): sha.update(chunk) db_hash = sha.hexdigest() try: hash_resp = http_requests.post( f"{PEER_URL}/api/sync/check_hash", json={"hash": db_hash}, headers={"Authorization": f"Bearer {SYNC_SECRET}", "Content-Type": "application/json"}, timeout=10, ) if hash_resp.status_code == 200 and hash_resp.json().get("match"): print(f"[sync] Peer already has DB hash {db_hash[:12]}, skipping transfer") return except Exception as e: print(f"[sync] Hash check failed ({e}), proceeding with full push") try: with open(db_path, 'rb') as f: push_resp = http_requests.post( f"{PEER_URL}/api/sync/push_db", files={"db_file": ("short_track.db", f, "application/octet-stream")}, headers={"Authorization": f"Bearer {SYNC_SECRET}"}, timeout=120, ) if push_resp.status_code == 200: print(f"[sync] DB pushed to peer successfully ({os.path.getsize(db_path)} bytes)") else: print(f"[sync] Peer returned error: {push_resp.status_code} {push_resp.text[:200]}") except Exception as e: print(f"[sync] Failed to push DB to peer: {e}") def schedule_db_push(): """防抖：DB_PUSH_DEBOUNCE_SECONDS 秒内的多次写入只触发一次推送（仅 Master）""" global _push_timer if SERVER_ROLE != 'master' or not PEER_URL: return with _push_lock: if _push_timer is not None: _push_timer.cancel() _push_timer = threading.Timer(DB_PUSH_DEBOUNCE_SECONDS, push_db_to_peer) _push_timer.daemon = True _push_timer.start() def _get_db_integrity(): """返回 DB 完整性状态：ok / 错误信息（使用独立连接，避免受 thread-local 连接池污染）""" import sqlite3 as _sqlite3 db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "short_track.db") conn = None try: conn = _sqlite3.connect(db_path, timeout=5) result = conn.execute("PRAGMA integrity_check").fetchone() return result[0] if result else "unknown" except Exception as e: return str(e) finally: if conn: try: conn.close() except Exception: pass def _restore_from_local_backup(data_dir, db_path): """尝试从本地备份恢复数据库，返回 True 若成功""" backups = sorted( [f for f in os.listdir(data_dir) if f.startswith("short_track.db.bak_")], reverse=True ) import sqlite3 for bak_file in backups: bak_path = os.path.join(data_dir, bak_file) try: test_conn = sqlite3.connect(bak_path) result = test_conn.execute("PRAGMA integrity_check").fetchone() test_conn.close() if result[0] == "ok": if os.path.exists(db_path): os.rename(db_path, db_path + ".corrupted_" + datetime.now().strftime("%Y%m%d_%H%M%S")) with open(bak_path, 'rb') as src, open(db_path, 'wb') as dst: dst.write(src.read()) print(f"[health] Restored DB from local backup: {bak_file}") # 通知所有 worker/gunicorn 进程重建数据库连接 try: with open(DB_REBUILD_SIGNAL, 'w') as sf: sf.write(str(time.time())) except Exception: pass return True except Exception as e: print(f"[health] Local backup {bak_file} failed: {e}") return False def _pull_db_from_peer(): """从对端服务器拉取数据库，返回 True 若成功""" if not PEER_URL: return False try: resp = http_requests.get( f"{PEER_URL}/api/sync/pull_db", headers={"Authorization": f"Bearer {SYNC_SECRET}"}, timeout=120, ) if resp.status_code != 200: print(f"[health] Pull from peer failed: {resp.status_code} {resp.text[:100]}") return False db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "short_track.db") temp_path = db_path + ".incoming" with open(temp_path, 'wb') as f: f.write(resp.content) import sqlite3 test_conn = sqlite3.connect(temp_path) result = test_conn.execute("PRAGMA integrity_check").fetchone() test_conn.close() if result[0] != "ok": os.remove(temp_path) print(f"[health] Peer DB failed integrity check: {result[0]}") return False if os.path.exists(db_path): os.rename(db_path, db_path + ".corrupted_" + datetime.now().strftime("%Y%m%d_%H%M%S")) os.rename(temp_path, db_path) # 清理 WAL/SHM for suffix in ("-wal", "-shm"): wf = db_path + suffix if os.path.exists(wf): try: os.remove(wf) except Exception: pass print(f"[health] Successfully recovered DB from peer ({os.path.getsize(db_path)} bytes)") # 通知所有 worker/gunicorn 进程重建数据库连接 try: with open(DB_REBUILD_SIGNAL, 'w') as sf: sf.write(str(time.time())) except Exception: pass return True except Exception as e: print(f"[health] Pull from peer error: {e}") return False def db_health_check(): """DB 完整性检查 + 自动恢复（Master 和 Slave 均运行） 使用独立 sqlite3 连接检测，不与业务请求共享连接池。 恢复成功后写入 DB_REBUILD_SIGNAL 文件，get_db() 检测到后自动重建连接。 整体超时约 90 秒（本地备份扫描 + 对端拉取 60s timeout）。 """ import sqlite3 as _sqlite3 db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "short_track.db") data_dir = os.path.dirname(db_path) # 清理过期重建信号（如果 DB 完好，信号无意义） integrity = _get_db_integrity() if integrity == "ok": if os.path.exists(DB_REBUILD_SIGNAL): try: os.remove(DB_REBUILD_SIGNAL) except Exception: pass return print(f"[health] ALERT: DB integrity check FAILED: {integrity}") print(f"[health] Starting auto-recovery on {SERVER_ROLE}...") recovered = False # Step 1: 尝试本地备份恢复 if _restore_from_local_backup(data_dir, db_path): new_integrity = _get_db_integrity() if new_integrity == "ok": print(f"[health] Local backup restored successfully") recovered = True else: print(f"[health] Restored DB still fails integrity: {new_integrity}") # Step 2: 从对端拉取 if not recovered and _pull_db_from_peer(): new_integrity = _get_db_integrity() if new_integrity == "ok": print(f"[health] Peer recovery successful") recovered = True else: print(f"[health] Peer-recovered DB still fails integrity: {new_integrity}") if recovered: # 通知当前进程内 thread-local 连接重建 from database import close_db try: close_db() except Exception: pass # Master 恢复后推送到 Slave if SERVER_ROLE == 'master': push_db_to_peer() return print(f"[health] CRITICAL: All recovery attempts FAILED on {SERVER_ROLE}") # Initialize cache（生产环境使用 Redis，开发环境使用 simple） cache_config = { 'CACHE_TYPE': config.CACHE_TYPE, 'CACHE_DEFAULT_TIMEOUT': config.CACHE_DEFAULT_TIMEOUT, } if config.CACHE_TYPE == 'RedisCache': cache_config['CACHE_REDIS_URL'] = config.CACHE_REDIS_URL cache = Cache(app, config=cache_config) CORS(app, supports_credentials=True, origins=[ 'http://localhost:5000', 'http://127.0.0.1:5000', 'http://122.51.80.140:5000', 'capacitor://localhost', # Capacitor iOS 'http://localhost', # Capacitor Android 'https://localhost', # Capacitor Android (some versions) 'https://chaofenghui-short-track.hf.space', # HF Space slave ]) # API 限流 limiter = Limiter( app=app, key_func=get_remote_address, default_limits=["200 per minute"], storage_uri="memory://", ) # CSRF 保护：mutating 请求必须有合法的 Content-Type 或空 body @app.before_request def csrf_protect(): """轻量级 CSRF 保护：验证 mutating 请求的 Content-Type""" if request.method in ('GET', 'HEAD', 'OPTIONS'): return None # 空 body 的请求（如 DELETE、简单 POST 触发操作）允许通过 # SameSite=Lax cookie 提供实际 CSRF 保护 if not request.content_length: return None content_type = request.content_type or '' if not any(ct in content_type for ct in ('application/json', 'multipart/form-data', 'application/x-www-form-urlencoded')): return jsonify({"error": "不支持的请求类型"}), 415 def login_required(f): """登录验证装饰器 — 支持 Session Cookie 和 Bearer Token""" @wraps(f) def decorated(*args, **kwargs): if 'user_id' in session: return f(*args, **kwargs) # 小程序 Bearer Token 鉴权 token = request.headers.get("Authorization", "").replace("Bearer ", "") if token: user = get_user_by_token(token) if user: session['user_id'] = user['id'] session['username'] = user['username'] session['role'] = user['role'] return f(*args, **kwargs) return jsonify({"error": "请先登录"}), 401 return decorated def admin_required(f): """管理员权限装饰器""" @wraps(f) def decorated(*args, **kwargs): if 'user_id' not in session: return jsonify({"error": "请先登录"}), 401 if session.get('role') not in ('admin', 'superadmin'): return jsonify({"error": "需要管理员权限"}), 403 return f(*args, **kwargs) return decorated def superadmin_required(f): """超级管理员权限装饰器""" @wraps(f) def decorated(*args, **kwargs): if 'user_id' not in session: return jsonify({"error": "请先登录"}), 401 if session.get('role') != 'superadmin': return jsonify({"error": "需要超级管理员权限"}), 403 return f(*args, **kwargs) return decorated def _build_user_payload(user, family=None): """构建统一用户返回结构""" family = family or get_user_family(user['id']) return { "id": user['id'], "username": user['username'], "role": user['role'], "is_superadmin": user['role'] == 'superadmin', "athlete_name": user.get('athlete_name'), "relationship": user.get('relationship'), "family_approved": family.get('approved') if family else None, "family_athlete": family.get('athlete_name') if family else None, "bio": user.get('bio') or '', "team_id": user.get('team_id') or 'chaoyang', } def _get_issued_team(): """获取当前发行队伍配置""" return get_team_config(database.get_default_issued_team_id()) def _resolve_effective_team(user=None): """解析当前请求实际生效的队伍上下文""" issued_team = _get_issued_team() if not user: return issued_team user_team = get_team_config(user.get('team_id') or issued_team['team_id']) if user.get('role') == 'superadmin': view_team_id = session.get('view_team_id') or user_team['team_id'] return get_team_config(view_team_id) return user_team def _build_team_context_payload(user=None): """构建队伍上下文返回结构""" issued_team = _get_issued_team() effective_team = _resolve_effective_team(user) user_team = get_team_config(user.get('team_id') or issued_team['team_id']) if user else issued_team return { "issued_team": issued_team, "effective_team": effective_team, "user_team": user_team, "view_team_id": effective_team['team_id'], "can_switch": bool(user and user.get('role') == 'superadmin'), } def _get_request_user(): """获取当前请求用户""" if 'username' not in session: return None return get_user_by_username(session['username']) def _get_effective_team_id(user=None): """获取当前请求生效队伍 ID""" return _resolve_effective_team(user or _get_request_user())['team_id'] def _get_team_alias_values(conn, team_id): """获取队伍别名列表""" rows = conn.execute( "SELECT alias FROM team_aliases WHERE is_active = 1 AND team_id = ? ORDER BY alias", (team_id,) ).fetchall() aliases = [r['alias'] for r in rows if r['alias']] team = get_team_config(team_id) if team and team.get('team_name'): aliases.append(team['team_name']) aliases.append(team_id) seen = set() result = [] for alias in aliases: if alias and alias not in seen: seen.add(alias) result.append(alias) return result def _build_team_filter_clause(conn, team_id, result_alias='r', include_override_join=False): """构建队伍过滤 SQL 条件""" aliases = _get_team_alias_values(conn, team_id) params = [team_id, team_id] team_checks = [] for alias in aliases: team_checks.append(f"{result_alias}.team LIKE ?") params.append(f"%{alias}%") clauses = [ f"EXISTS (SELECT 1 FROM athlete_overrides ao_team_in WHERE ao_team_in.name = {result_alias}.name AND ao_team_in.team_id = ? AND ao_team_in.is_chaoyang = 1)", f"NOT EXISTS (SELECT 1 FROM athlete_overrides ao_team_out WHERE ao_team_out.name = {result_alias}.name AND ao_team_out.team_id = ? AND ao_team_out.is_chaoyang = 0)" ] if team_checks: clauses.append("(" + " OR ".join(team_checks) + ")") if team_id == 'chaoyang': clauses.append(f"{result_alias}.is_chaoyang = 1") clause = "(" + " OR ".join([ clauses[0], "(" + clauses[1] + (" AND (" + " OR ".join(team_checks + ([f"{result_alias}.is_chaoyang = 1"] if team_id == 'chaoyang' else [])) + ")" if (team_checks or team_id == 'chaoyang') else "") + ")" ]) + ")" return clause, params def _row_belongs_to_team(row, team_id, aliases=None): """判断单条结果是否属于当前队伍""" aliases = aliases if aliases is not None else [] name = (row.get('athlete_name') or row.get('name') or '') if isinstance(row, dict) else '' team_text = (row.get('team') or '') if isinstance(row, dict) else '' if not isinstance(row, dict): return False conn = get_db() override_in = conn.execute( "SELECT 1 FROM athlete_overrides WHERE name = ? AND team_id = ? AND is_chaoyang = 1 LIMIT 1", (name, team_id) ).fetchone() if override_in: return True override_out = conn.execute( "SELECT 1 FROM athlete_overrides WHERE name = ? AND team_id = ? AND is_chaoyang = 0 LIMIT 1", (name, team_id) ).fetchone() if override_out: return False if team_id == 'chaoyang' and row.get('is_chaoyang'): return True if any(alias and alias in team_text for alias in aliases): return True return False def _get_team_athlete_names(conn, team_id, include_override_only=True): """获取当前队伍的运动员名单""" filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') rows = conn.execute(f""" SELECT DISTINCT r.name FROM results r WHERE r.name IS NOT NULL AND r.name != '' AND {filter_clause} ORDER BY r.name """, filter_params).fetchall() seen = set() names = [] for row in rows: name = row['name'] if name and name not in seen: seen.add(name) names.append(name) if team_id == 'chaoyang': manual_rows = conn.execute( "SELECT name FROM athletes WHERE deleted_at IS NULL ORDER BY name" ).fetchall() for row in manual_rows: name = row['name'] if name and name not in seen: seen.add(name) names.append(name) for name in get_active_athlete_names(): if name and name not in seen: seen.add(name) names.append(name) if include_override_only: override_rows = conn.execute( "SELECT name FROM athlete_overrides WHERE team_id = ? AND is_chaoyang = 1 ORDER BY name", (team_id,) ).fetchall() for row in override_rows: name = row['name'] if name and name not in seen: seen.add(name) names.append(name) return names def _get_team_sync_payload(conn, team_id): """获取当前队伍同步所需的运动员与成绩数据""" filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') aliases = _get_team_alias_values(conn, team_id) athlete_rows = conn.execute(f""" SELECT DISTINCT r.name, a.gender, a.category FROM results r LEFT JOIN athletes a ON a.name = r.name AND a.deleted_at IS NULL WHERE {filter_clause} ORDER BY r.name """, filter_params).fetchall() athletes = [] seen_names = set() for row in athlete_rows: if row['name'] in seen_names: continue seen_names.add(row['name']) athletes.append({ 'name': row['name'], 'gender': row['gender'], 'category': row['category'], 'is_chaoyang': 1, }) if team_id == 'chaoyang': manual_rows = conn.execute("SELECT name, gender, category FROM athletes WHERE deleted_at IS NULL ORDER BY name").fetchall() for row in manual_rows: if row['name'] not in seen_names: seen_names.add(row['name']) athletes.append({ 'name': row['name'], 'gender': row['gender'], 'category': row['category'], 'is_chaoyang': 1, }) override_rows = conn.execute( "SELECT name FROM athlete_overrides WHERE team_id = ? AND is_chaoyang = 1 ORDER BY name", (team_id,) ).fetchall() for row in override_rows: if row['name'] not in seen_names: manual = conn.execute( "SELECT gender, category FROM athletes WHERE name = ? AND deleted_at IS NULL", (row['name'],) ).fetchone() seen_names.add(row['name']) athletes.append({ 'name': row['name'], 'gender': manual['gender'] if manual else None, 'category': manual['category'] if manual else None, 'is_chaoyang': 1, }) results = conn.execute(f""" SELECT r.name as athlete_name, r.team, r.place, r.time_result, r.qual, r.is_chaoyang, rg.race_date, rg.event, rg.category, rg.gender as race_gender, rg.round_type, m.title as match_title, m.season, m.venue FROM results r LEFT JOIN race_groups rg ON r.rcbh = rg.rcbh LEFT JOIN matches m ON rg.match_id = m.id WHERE {filter_clause} ORDER BY rg.race_date DESC """, filter_params).fetchall() result_dicts = [dict(r) for r in results] opponents = [] seen_opponents = set() for row in result_dicts: if row['athlete_name'] in seen_names: continue if _row_belongs_to_team(row, team_id, aliases): continue key = (row['athlete_name'], row.get('team') or '') if key in seen_opponents: continue seen_opponents.add(key) opponents.append({ 'name': row['athlete_name'], 'team': row.get('team'), 'is_chaoyang': 0, }) return athletes, result_dicts, opponents # 全局错误处理中间件 @app.errorhandler(500) def internal_error(error): """统一500错误处理，返回JSON而非HTML""" StructuredLogger.error("internal_error", error=str(error), path=request.path) return jsonify({"error": "服务器内部错误，请稍后重试"}), 500 @app.errorhandler(404) def not_found(error): """统一404错误处理""" if request.path.startswith('/api/'): return jsonify({"error": "资源不存在"}), 404 return send_from_directory('static', 'index.html') @app.errorhandler(405) def method_not_allowed(error): """统一405错误处理""" return jsonify({"error": "请求方法不允许"}), 405 @app.errorhandler(Exception) def handle_exception(error): """捕获所有未处理异常，返回JSON""" if request.path.startswith('/api/'): StructuredLogger.error("unhandled_exception", error=str(error), path=request.path, method=request.method) return jsonify({"error": "服务器内部错误，请稍后重试"}), 500 raise error @app.teardown_appcontext def shutdown_session(exception=None): """请求结束时关闭数据库连接""" close_db() @app.after_request def trigger_db_sync(response): """写操作成功后，仅 Master 异步推送DB到 Slave""" if SERVER_ROLE == 'master' and PEER_URL and request.method in ("POST", "PUT", "DELETE", "PATCH"): if 200 <= response.status_code < 300: schedule_db_push() return response @app.after_request def add_no_cache_headers(response): """禁止浏览器缓存HTML页面，确保总是获取最新版本""" if response.content_type and 'text/html' in response.content_type: response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate' response.headers['Pragma'] = 'no-cache' response.headers['Expires'] = '0' return response @app.route("/") def index(): return send_from_directory("static", "index.html") @app.route("/api/ping", methods=["GET"]) def api_ping(): """健康检查 — 供前端测速和角色判断""" from database import get_db try: db = get_db() db.execute("SELECT 1") db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "short_track.db") db_size = os.path.getsize(db_path) if os.path.exists(db_path) else 0 return jsonify({ "status": "ok", "server_role": SERVER_ROLE, "version": APP_VERSION, "db_size_bytes": db_size }), 200 except Exception as e: return jsonify({"status": "error", "message": str(e)}), 500 @app.route("/api/visit", methods=["POST"]) def api_log_visit(): """前端记录访问（避免Service Worker缓存导致不记录）""" try: ip = request.headers.get('X-Forwarded-For', request.remote_addr or '').split(',')[0].strip() if not ip: ip = request.remote_addr or 'unknown' user_agent = request.headers.get('User-Agent', '')[:500] page = request.json.get('page', '/') if request.is_json else '/' log_visit(ip, user_agent, page) except Exception: pass return jsonify({"status": "ok"}) def is_athlete_active(conn, name, team_id='chaoyang'): """判断队员是否仍在当前队伍的活跃名单中""" if team_id == 'chaoyang': manual = conn.execute("SELECT id FROM athletes WHERE name = ? AND deleted_at IS NULL", (name,)).fetchone() if manual: return True override = conn.execute( "SELECT is_chaoyang FROM athlete_overrides WHERE name = ? AND team_id = ?", (name, team_id) ).fetchone() if override and override["is_chaoyang"]: return True if team_id == 'chaoyang': cutoff = (datetime.now() - timedelta(days=180)).strftime('%Y-%m-%d') has_training = conn.execute( """SELECT 1 FROM training_records tr LEFT JOIN athlete_nicknames an ON tr.athlete_name = an.nickname WHERE (tr.athlete_name = ? OR an.real_name = ?) AND tr.training_date >= ? LIMIT 1""", (name, name, cutoff) ).fetchone() if has_training: return True filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') last_team_row = conn.execute(f""" SELECT m.season FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.name = ? AND {filter_clause} ORDER BY m.date_range DESC, rg.race_date DESC LIMIT 1 """, [name] + filter_params).fetchone() if not last_team_row: return False seasons = [s["season"] for s in conn.execute( "SELECT DISTINCT season FROM matches ORDER BY date_range DESC" ).fetchall()] if not seasons: return True return last_team_row["season"] in seasons[:2] @app.route("/api/athletes") @login_required def api_athletes(): """获取当前队伍当前活跃运动员列表，按荣誉排名""" try: from database import get_db conn = get_db() team_id = _get_effective_team_id() team_names = _get_team_athlete_names(conn, team_id, include_override_only=False) all_athletes_data = database.get_athletes_with_details(team_names) athlete_data_map = {a['name']: a for a in all_athletes_data} athlete_map = {} for name in sorted(team_names): if name in athlete_map: continue if _is_team_name(name): continue if not is_athlete_active(conn, name, team_id): continue row = conn.execute(""" SELECT team FROM results r WHERE r.name = ? ORDER BY CASE WHEN r.team IS NULL OR r.team = '' THEN 1 ELSE 0 END, r.id DESC LIMIT 1 """, (name,)).fetchone() team_name = row['team'] if row and row['team'] else get_team_config(team_id)['team_name'] athlete_map[name] = {"name": name, "team": team_name, "categories": [], "genders": []} filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') for name in athlete_map: cat_rows = conn.execute(f""" SELECT rg.category, rg.gender, m.date_range FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.name = ? AND {filter_clause} AND rg.category IS NOT NULL """, [name] + filter_params).fetchall() def parse_end_date(dr): try: end_part = dr.split('—')[-1].strip() parts = end_part.split('/') return f"{parts[0]}-{int(parts[1]):02d}-{int(parts[2]):02d}" except Exception: return dr cat_rows_sorted = sorted(cat_rows, key=lambda r: parse_end_date(r['date_range']), reverse=True) seen_cats = set() seen_genders = set() for lc in cat_rows_sorted: if lc["category"] and lc["category"] not in seen_cats: athlete_map[name]["categories"].append(lc["category"]) seen_cats.add(lc["category"]) if lc["gender"] and lc["gender"] not in seen_genders: athlete_map[name]["genders"].append(lc["gender"]) seen_genders.add(lc["gender"]) if team_id == 'chaoyang': manual_athletes = conn.execute("SELECT name, gender, category FROM athletes WHERE deleted_at IS NULL").fetchall() for ma in manual_athletes: name = ma["name"] if name not in athlete_map and not _is_team_name(name): athlete_map[name] = {"name": name, "team": get_team_config(team_id)['team_name'], "categories": [], "genders": []} if ma["category"]: athlete_map[name]["categories"].insert(0, ma["category"]) if ma["gender"]: athlete_map[name]["genders"].insert(0, ma["gender"]) active_training_names = get_active_athlete_names() for name in active_training_names: if name not in athlete_map and not _is_team_name(name): athlete_map[name] = {"name": name, "team": get_team_config(team_id)['team_name'], "categories": [], "genders": []} all_honors = get_athlete_honors() honors_map = {} for h in all_honors: if h['athlete_name'] in athlete_map: honors_map.setdefault(h['athlete_name'], []).append(h) relay_filter_clause, relay_filter_params = _build_team_filter_clause(conn, team_id, 'r') relay_medals_raw = conn.execute(f""" SELECT rm.athlete_name, r.place FROM relay_members rm JOIN results r ON r.name = rm.relay_team_name JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE {relay_filter_clause} AND r.place IN ('1', '2', '3') AND (rg.round_type = '决赛/Finals' OR rg.round_type = '决赛A' OR rg.round_type = '决赛' OR rg.round_type = '决赛B') AND (rm.season IS NULL OR rm.season = m.season) AND (rm.event IS NULL OR rm.event = rg.event) AND rm.athlete_name IS NOT NULL """, relay_filter_params).fetchall() relay_gold_map = {} relay_medal_map = {} for rm in relay_medals_raw: n = rm['athlete_name'] p = str(rm['place']) if p == '1': relay_gold_map[n] = relay_gold_map.get(n, 0) + 1 relay_medal_map[n] = relay_medal_map.get(n, 0) + 1 for name, info in athlete_map.items(): gold = 0 total = 0 athlete_results = conn.execute(f""" SELECT r.place, rg.round_type FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE r.name = ? AND {filter_clause} """, [name] + filter_params).fetchall() for result_row in athlete_results: p = str(result_row['place']) rt = result_row['round_type'] or '' is_final = rt in ('决赛', '决赛/Finals', '决赛A', '决赛B') if p == '1' and is_final: gold += 1 if p in ('1', '2', '3') and is_final: total += 1 gold += relay_gold_map.get(name, 0) total += relay_medal_map.get(name, 0) info['gold_medals'] = gold info['total_medals'] = total info['total_results'] = len(athlete_results) info['honors'] = honors_map.get(name, []) result_list = list(athlete_map.values()) result_list.sort(key=lambda x: (-x['gold_medals'], -x['total_medals'], x['name'])) return jsonify(result_list) except Exception as e: logging.error(f"Error fetching athletes: {e}") return jsonify({'error': 'Failed to fetch athletes'}), 500 @app.route("/api/athlete/") @login_required @cache.cached(timeout=300, query_string=True) def api_athlete_detail(name): # 获取运动员参加的项目列表 events = get_athlete_events(name) # 获取运动员参加的项目类别 categories = get_athlete_categories(name) # 获取运动员比赛成绩 results = get_athlete_results(name) # 获取运动员接力比赛成绩 relay_results = get_athlete_relay_results(name) # 获取运动员代表过的所有队伍 from database import get_db as _get_db conn = _get_db() teams = conn.execute(""" SELECT DISTINCT r.team, r.is_chaoyang, COUNT(*) as race_count FROM results r WHERE r.name = ? GROUP BY r.team, r.is_chaoyang ORDER BY race_count DESC """, (name,)).fetchall() # 获取手动添加的队员信息 manual_info = conn.execute("SELECT gender, category, notes FROM athletes WHERE name = ? AND deleted_at IS NULL", (name,)).fetchone() conn.close() # 获取运动员荣誉 honors = get_athlete_honors(name) # 获取运动员自定义成绩（同步显示到运动成绩页） custom_results = get_athlete_custom_results(name) return jsonify({ "name": name, "events": events, "categories": categories, "results": results, "relay_results": relay_results, "teams": [dict(t) for t in teams], "manual_info": dict(manual_info) if manual_info else None, "honors": honors, "custom_results": custom_results, }) @app.route("/api/athlete//comparison") @login_required def api_athlete_comparison(name): """获取运动员与对手的对比数据""" event = request.args.get("event", "") category = request.args.get("category", "") gender = request.args.get("gender", "") if not all([event, category, gender]): return jsonify({"error": "需要提供 event, category, gender 参数"}), 400 # 获取运动员自己的数据 athlete_data = get_event_timeseries(name, event) for d in athlete_data: d["seconds"] = parse_time_to_seconds(d.get("time_result", "")) # 获取对手数据 opponents = get_opponent_results(event, category, gender, exclude_name=name) # 按对手名字分组 opponent_map = {} for o in opponents: oname = o["name"] if oname not in opponent_map: opponent_map[oname] = {"name": oname, "team": o["team"], "data": []} seconds = parse_time_to_seconds(o.get("time_result", "")) if seconds is not None: opponent_map[oname]["data"].append({ **o, "seconds": seconds, }) return jsonify({ "athlete": {"name": name, "data": athlete_data}, "opponents": list(opponent_map.values()), }) @app.route("/api/opponents") @login_required def api_opponents(): """获取指定项目/组别的对手列表""" event = request.args.get("event", "") category = request.args.get("category", "") gender = request.args.get("gender", "") if not all([event, category, gender]): return jsonify({"error": "需要提供 event, category, gender 参数"}), 400 opponents = get_opponent_results(event, category, gender) # 按名字分组 opponent_map = {} for o in opponents: oname = o["name"] if oname not in opponent_map: opponent_map[oname] = {"name": oname, "team": o["team"]} return jsonify(list(opponent_map.values())) @app.route("/api/opponent//timeseries") @login_required def api_opponent_timeseries(name): """获取对手的成绩时间序列""" event = request.args.get("event", "") if not event: return jsonify({"error": "需要提供 event 参数"}), 400 from database import get_db conn = get_db() rows = conn.execute(""" SELECT r.time_result, r.place, r.qual, rg.race_date, rg.round_type, rg.category, rg.gender, m.title as match_title, m.season FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id INNER JOIN ( SELECT rg2.match_id, MIN(r2.time_result) as best_time FROM results r2 JOIN race_groups rg2 ON r2.rcbh = rg2.rcbh WHERE r2.name = ? AND rg2.event = ? AND r2.time_result NOT LIKE '%00:00.000%' AND r2.qual != '犯规' GROUP BY rg2.match_id ) best ON rg.match_id = best.match_id AND r.time_result = best.best_time WHERE r.name = ? AND rg.event = ? AND r.time_result NOT LIKE '%00:00.000%' AND r.qual != '犯规' GROUP BY rg.match_id ORDER BY m.season, rg.race_date """, (name, event, name, event)).fetchall() conn.close() data = [] for r in rows: d = dict(r) d["seconds"] = parse_time_to_seconds(d.get("time_result", "")) if d["seconds"] is not None: d["_sort_date"] = parse_chinese_date(d.get("race_date", "")) data.append(d) # 按赛季+真实日期排序 data.sort(key=lambda x: (x.get("season", ""), x.get("_sort_date", ""))) for d in data: d.pop("_sort_date", None) return jsonify(data) @app.route("/api/opponent//summary") @login_required def api_opponent_summary(name): """获取对手所有项目的最好成绩概要""" from database import get_db conn = get_db() rows = conn.execute(""" SELECT rg.event, rg.category, MIN(r.time_result) as best_time, MIN(CAST(r.place AS INTEGER)) as best_place, COUNT(DISTINCT m.id) as match_count, COUNT(*) as race_count FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.name = ? AND r.time_result NOT LIKE '%00:00.000%' AND r.qual != '犯规' GROUP BY rg.event, rg.category ORDER BY rg.event, rg.category """, (name,)).fetchall() conn.close() # 按项目分组 event_map = {} for r in rows: ev = r["event"] if ev not in event_map: event_map[ev] = [] event_map[ev].append({ "category": r["category"], "best_time": r["best_time"], "best_place": r["best_place"], "match_count": r["match_count"], "race_count": r["race_count"], }) return jsonify(event_map) @app.route("/api/athlete//personal_bests") @login_required def api_athlete_personal_bests(name): """获取运动员各项目的个人最好成绩(PB)""" from database import get_db conn = get_db() rows = conn.execute(""" SELECT rg.event, MIN(r.time_result) as best_time, MIN(CAST(r.place AS INTEGER)) as best_place, COUNT(DISTINCT m.id) as match_count, COUNT(*) as race_count FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.name = ? AND r.time_result NOT LIKE '%00:00.000%' AND r.time_result NOT LIKE '%NO TIME%' AND r.time_result != '' AND r.qual != '犯规' GROUP BY rg.event ORDER BY rg.event """, (name,)).fetchall() # Get the match info for each PB (the specific race where PB was achieved) pb_matches = conn.execute(""" SELECT rg.event, m.title as pb_match_title, rg.race_date as pb_race_date, rg.round_type as pb_round_type, rg.category as pb_category FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.name = ? AND r.qual != '犯规' AND r.time_result NOT LIKE '%00:00.000%' AND r.time_result NOT LIKE '%NO TIME%' AND r.time_result != '' AND r.time_result IN ( SELECT MIN(r2.time_result) FROM results r2 JOIN race_groups rg2 ON r2.rcbh = rg2.rcbh WHERE r2.name = ? AND r2.qual != '犯规' AND r2.time_result NOT LIKE '%00:00.000%' AND r2.time_result NOT LIKE '%NO TIME%' AND r2.time_result != '' AND rg2.event = rg.event ) """, (name, name)).fetchall() pb_match_map = {r["event"]: dict(r) for r in pb_matches} conn.close() result = [] for r in rows: pb_info = pb_match_map.get(r["event"], {}) result.append({ "event": r["event"], "best_time": r["best_time"], "best_place": r["best_place"], "match_count": r["match_count"], "race_count": r["race_count"], "pb_match_title": pb_info.get("pb_match_title", ""), "pb_race_date": pb_info.get("pb_race_date", ""), "pb_round_type": pb_info.get("pb_round_type", ""), "pb_category": pb_info.get("pb_category", ""), }) return jsonify(result) # ========== 运动员荣誉 API ========== @app.route("/api/athlete//honors") @login_required def api_athlete_honors(name): """获取运动员荣誉列表""" honors = get_athlete_honors(name) return jsonify(honors) @app.route("/api/admin/athlete_honors", methods=["GET"]) @admin_required def api_all_honors(): """获取所有运动员荣誉（管理员）""" honors = get_athlete_honors() return jsonify(honors) @app.route("/api/admin/athlete_honors", methods=["POST"]) @admin_required def api_add_honor(): """添加运动员荣誉""" data = request.json athlete_name = data.get("athlete_name", "").strip() title = data.get("title", "").strip() awarded_date = data.get("awarded_date", "").strip() or None source = data.get("source", "").strip() or None if not athlete_name or not title: return jsonify({"error": "队员姓名和荣誉名称不能为空"}), 400 if add_athlete_honor(athlete_name, title, awarded_date, source): return jsonify({"status": "success"}) return jsonify({"error": "添加失败"}), 400 @app.route("/api/admin/athlete_honors/", methods=["PUT"]) @admin_required def api_update_honor(honor_id): """更新运动员荣誉""" data = request.json title = data.get("title") awarded_date = data.get("awarded_date") source = data.get("source") update_athlete_honor(honor_id, title, awarded_date, source) return jsonify({"status": "success"}) @app.route("/api/admin/athlete_honors/", methods=["DELETE"]) @admin_required def api_delete_honor(honor_id): """删除运动员荣誉""" delete_athlete_honor(honor_id) return jsonify({"status": "success"}) # ========== 运动员个人中心 API ========== @app.route("/api/athlete//profile") @login_required def api_athlete_profile(name): """获取运动员个人中心数据""" profile = get_athlete_profile(name) custom_results = get_athlete_custom_results(name) honors = get_athlete_honors(name) events = get_athlete_events(name) # 获取运动员代表过的所有队伍 from database import get_db as _get_db conn = _get_db() teams = conn.execute(""" SELECT DISTINCT r.team, r.is_chaoyang, COUNT(*) as race_count FROM results r WHERE r.name = ? GROUP BY r.team, r.is_chaoyang ORDER BY race_count DESC """, (name,)).fetchall() # 获取手动添加的队员信息 manual_info = conn.execute("SELECT gender, category, notes FROM athletes WHERE name = ? AND deleted_at IS NULL", (name,)).fetchone() # 获取运动员荣誉 honors = get_athlete_honors(name) # 获取运动员自定义成绩（同步显示到运动成绩页） custom_results = get_athlete_custom_results(name) # 获取运动员爬虫比赛成绩（用于个人中心成绩TAB） results = conn.execute(""" SELECT r.time_result, r.place, r.qual, r.is_chaoyang, r.team, rg.race_date, rg.round_type, rg.event, rg.category, rg.gender, m.title as match_title, m.season, m.id as match_id FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.name = ? ORDER BY COALESCE(m.date_range, '') DESC, rg.race_date DESC """, (name,)).fetchall() conn.close() # Sort results by date (newest first) results_list = [dict(r) for r in results] results_list.sort(key=lambda r: ( r.get('season', ''), parse_chinese_date(r.get('race_date', '')) ), reverse=True) # 获取运动员接力成绩 relay_results = get_athlete_relay_results(name) return jsonify({ "name": name, "profile": profile, "events": events, "teams": [dict(t) for t in teams], "manual_info": dict(manual_info) if manual_info else None, "honors": honors, "custom_results": custom_results, "results": results_list, "relay_results": relay_results, "custom_medals": { "gold": sum(1 for r in custom_results if str(r.get("place", "")) == "1"), "silver": sum(1 for r in custom_results if str(r.get("place", "")) == "2"), "bronze": sum(1 for r in custom_results if str(r.get("place", "")) == "3"), }, }) @app.route("/api/athlete//profile/bio", methods=["PUT"]) @login_required def api_update_athlete_bio(name): """更新队员简介""" if not can_edit_athlete(name): return jsonify({"error": "没有权限编辑该队员资料"}), 403 data = request.json bio = (data.get("bio") or "").strip() if len(bio) > 500: return jsonify({"error": "简介不能超过500字"}), 400 update_athlete_profile(name, bio, updated_by=session.get('username')) return jsonify({"status": "success"}) @app.route("/api/athlete//custom_results", methods=["POST"]) @login_required def api_add_custom_result(name): """添加自定义成绩""" if not can_edit_athlete(name): return jsonify({"error": "没有权限编辑该队员资料"}), 403 data = request.json competition_date = (data.get("competition_date") or "").strip() competition_name = (data.get("competition_name") or "").strip() result_text = (data.get("result_text") or "").strip() event = (data.get("event") or "").strip() if not competition_name or not result_text: return jsonify({"error": "比赛名称和成绩不能为空"}), 400 add_athlete_custom_result( name, competition_date or None, competition_name, result_text, created_by=session.get('username'), event=event or None, round_type=(data.get("round_type") or "").strip() or None, team=(data.get("team") or "").strip() or None, time_result=(data.get("time_result") or "").strip() or None, place=data.get("place"), season=(data.get("season") or "").strip() or None ) return jsonify({"status": "success"}) @app.route("/api/athlete//events") @login_required def api_athlete_events_list(name): """获取运动员参赛项目列表""" events = get_athlete_events(name) return jsonify(events) @app.route("/api/athlete//timeseries/") @login_required def api_athlete_timeseries(name, event): """获取运动员某项目的时间序列成绩（用于曲线图）""" from database import get_db conn = get_db() rows = conn.execute(""" SELECT r.time_result, r.place, r.qual, rg.race_date, rg.round_type, rg.category, rg.gender, m.title as match_title, m.season FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id INNER JOIN ( SELECT rg2.match_id, MIN(r2.time_result) as best_time FROM results r2 JOIN race_groups rg2 ON r2.rcbh = rg2.rcbh WHERE r2.name = ? AND rg2.event = ? AND r2.time_result NOT LIKE '%00:00.000%' AND r2.qual != '犯规' GROUP BY rg2.match_id ) best ON rg.match_id = best.match_id AND r.time_result = best.best_time WHERE r.name = ? AND rg.event = ? AND r.time_result NOT LIKE '%00:00.000%' AND r.qual != '犯规' GROUP BY rg.match_id ORDER BY m.season, rg.race_date """, (name, event, name, event)).fetchall() data = [] for r in rows: d = dict(r) d["seconds"] = parse_time_to_seconds(d.get("time_result", "")) if d["seconds"] is not None: d["_sort_date"] = parse_chinese_date(d.get("race_date", "")) data.append(d) # Include custom results for the same event custom_rows = conn.execute(""" SELECT cr.time_result, CAST(cr.place AS TEXT) as place, cr.round_type as qual, cr.competition_date as race_date, cr.round_type, NULL as category, NULL as gender, cr.competition_name as match_title, cr.season FROM athlete_custom_results cr WHERE cr.athlete_name = ? AND cr.event = ? AND cr.time_result IS NOT NULL AND cr.time_result != '' """, (name, event)).fetchall() for r in custom_rows: d = dict(r) d["seconds"] = parse_time_to_seconds(d.get("time_result", "")) if d["seconds"] is not None: d["_sort_date"] = d.get("race_date", "") or "" d["source"] = "custom" data.append(d) # 按赛季+真实日期排序 data.sort(key=lambda x: (x.get("season", ""), x.get("_sort_date", ""))) for d in data: d.pop("_sort_date", None) conn.close() return jsonify(data) @app.route("/api/admin/recalculate", methods=["POST"]) @admin_required def api_recalculate_chaoyang(): """重新计算朝阳队标记""" recalculate_chaoyang_flags() return jsonify({"status": "success"}) @app.route("/api/athlete//custom_results/", methods=["DELETE"]) @login_required def api_delete_custom_result(name, result_id): """删除自定义成绩""" if not can_edit_athlete(name): return jsonify({"error": "没有权限编辑该队员资料"}), 403 delete_athlete_custom_result(result_id, name) return jsonify({"status": "success"}) @app.route("/api/athlete//custom_results/", methods=["PUT"]) @login_required def api_update_custom_result(name, result_id): """修改自定义成绩""" if not can_edit_athlete(name): return jsonify({"error": "没有权限编辑该队员资料"}), 403 data = request.json competition_date = (data.get("competition_date") or "").strip() competition_name = (data.get("competition_name") or "").strip() result_text = (data.get("result_text") or "").strip() event = (data.get("event") or "").strip() round_type = (data.get("round_type") or "").strip() team = (data.get("team") or "").strip() time_result = (data.get("time_result") or "").strip() place = data.get("place") season = (data.get("season") or "").strip() if not competition_name: return jsonify({"error": "比赛名称不能为空"}), 400 if place is not None: try: place = int(place) except (ValueError, TypeError): place = None update_athlete_custom_result(result_id, name, competition_date=competition_date or None, competition_name=competition_name or None, result_text=result_text or None, event=event or None, round_type=round_type or None, team=team or None, time_result=time_result or None, place=place, season=season or None) return jsonify({"status": "success"}) def can_edit_athlete(name): """判断当前用户是否可以编辑指定运动员资料""" if session.get('role') in ('admin', 'superadmin'): return True return is_family_of(session['user_id'], name) @app.route("/api/athlete//honors/family", methods=["POST"]) @login_required def api_family_add_honor(name): """家属添加荣誉（待审核）""" if not can_edit_athlete(name): return jsonify({"error": "没有权限编辑该队员资料"}), 403 data = request.json title = (data.get("title") or "").strip() awarded_date = (data.get("awarded_date") or "").strip() or None if not title: return jsonify({"error": "荣誉名称不能为空"}), 400 add_athlete_honor(name, title, awarded_date, source="家属添加") return jsonify({"status": "success"}) @app.route("/api/athlete//honors/", methods=["DELETE"]) @login_required def api_family_delete_honor(name, honor_id): """家属删除荣誉""" if not can_edit_athlete(name): return jsonify({"error": "没有权限编辑该队员资料"}), 403 delete_athlete_honor(honor_id) return jsonify({"status": "success"}) @app.route("/api/matches") @login_required def api_matches(): """获取比赛列表（默认只返回朝阳队有人参加的比赛）""" season = request.args.get("season", "") matches = get_all_matches(season if season else None, chaoyang_only=True) return jsonify(matches) @app.route("/api/matches/current") @login_required def api_current_matches(): """获取正在进行的比赛""" matches = get_current_matches() return jsonify(matches) @app.route("/api/match//groups") @login_required def api_match_groups(match_id): """获取比赛的分组信息""" groups = get_match_race_groups(match_id) return jsonify(groups) @app.route("/api/match//chaoyang_opponents") @login_required def api_match_chaoyang_opponents(match_id): """获取当前比赛中所有分组信息，包括有朝阳队员和没有朝阳队员的分组""" from database import get_db conn = get_db() groups = get_match_race_groups(match_id) result = [] for g in groups: rcbh = g["rcbh"] race_results = conn.execute(""" SELECT r.name, r.team, r.place, r.time_result, r.qual, r.is_chaoyang, r.heat_number FROM results r WHERE r.rcbh = ? ORDER BY r.heat_number, r.place """, (rcbh,)).fetchall() # 按 heat_number 分组 heat_map = {} for r in race_results: hn = r["heat_number"] or 1 if hn not in heat_map: heat_map[hn] = [] heat_map[hn].append(dict(r)) # 检查是否有朝阳队员 has_chaoyang = any(r["is_chaoyang"] for r in race_results) if has_chaoyang: # 为每个小组中的朝阳队员收集同组对手 for heat_num, heat_results in heat_map.items(): chaoyang_results = [r for r in heat_results if r["is_chaoyang"] and is_athlete_active(conn, r["name"])] if not chaoyang_results: continue for cr in chaoyang_results: opponents = [r for r in heat_results if not r["is_chaoyang"] and "00:00.000" not in (r["time_result"] or "")] entry = { "event": g["event"], "category": g["category"], "gender": g["gender"], "round_type": g["round_type"], "race_date": g.get("race_date", ""), "heat_number": heat_num, "has_chaoyang": True, "chaoyang_athlete": { "name": cr["name"], "place": cr["place"], "time_result": cr["time_result"], }, "opponents": [{"name": o["name"], "team": o["team"], "place": o["place"], "time_result": o["time_result"]} for o in opponents] } result.append(entry) else: # 没有朝阳队员的分组：仍然返回，让前端显示完整赛程 all_athletes = [dict(r) for r in race_results] def _safe_place(p): try: return int(p) if p else 999 except (ValueError, TypeError): return 999 all_athletes.sort(key=lambda r: (_safe_place(r["place"]))) # Use the actual heat_number from results (rcbh is now heat-specific) actual_heat = race_results[0]["heat_number"] if race_results else 1 entry = { "event": g["event"], "category": g["category"], "gender": g["gender"], "round_type": g["round_type"], "race_date": g.get("race_date", ""), "heat_number": actual_heat, "has_chaoyang": False, "chaoyang_athlete": None, "opponents": [{"name": a["name"], "team": a["team"], "place": a["place"], "time_result": a["time_result"]} for a in all_athletes if "00:00.000" not in (a["time_result"] or "")] } result.append(entry) conn.close() return jsonify(result) @app.route("/api/match//chaoyang_results") @login_required def api_match_chaoyang_results(match_id): """获取某个比赛中当前队伍队员的成绩详情""" from database import get_db conn = get_db() team_id = _get_effective_team_id() filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') # 获取比赛信息 match = conn.execute("SELECT * FROM matches WHERE id = ?", (match_id,)).fetchone() if not match: conn.close() return jsonify({"error": "比赛不存在"}), 404 match_info = dict(match) # 获取该比赛中当前队伍队员的成绩 rows = conn.execute(f""" SELECT r.name, r.team, r.place, r.lane, r.time_result, r.qual, r.heat_number, rg.event, rg.category, rg.gender, rg.round_type, rg.race_date, rg.race_time, rg.rcbh FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE rg.match_id = ? AND {filter_clause} ORDER BY r.name, rg.event, rg.race_date, rg.race_time, CASE rg.round_type WHEN '决赛/Finals' THEN 0 WHEN '决赛A' THEN 0 WHEN '决赛' THEN 0 WHEN '决赛B' THEN 1 WHEN '半决赛/Semifinals' THEN 2 WHEN '半决赛' THEN 2 WHEN '1/4赛/Quarterfinals' THEN 3 WHEN '1/4赛' THEN 3 WHEN '预赛/Heats' THEN 4 WHEN '预赛' THEN 4 WHEN '初赛' THEN 4 ELSE 5 END, r.heat_number, r.place """, [match_id] + filter_params).fetchall() conn.close() results = [dict(r) for r in rows] # 按队员名字分组 athlete_map = {} for r in results: name = r["name"] if name not in athlete_map: athlete_map[name] = { "name": name, "team": r["team"], "events": {} } event = r["event"] if event not in athlete_map[name]["events"]: athlete_map[name]["events"] = athlete_map[name].get("events", {}) athlete_map[name]["events"][event] = [] athlete_map[name]["events"][event].append(r) return jsonify({ "match": match_info, "athletes": list(athlete_map.values()) }) @app.route("/api/match//summary") @login_required def api_match_summary(match_id): """获取或生成比赛总结（专业体育记者风格）""" import random as _random from database import get_db conn = get_db() match = conn.execute("SELECT * FROM matches WHERE id = ?", (match_id,)).fetchone() if not match: conn.close() return jsonify({"error": "比赛不存在"}), 404 match_info = dict(match) force = request.args.get("force", "0") if match_info.get("summary") and force != "1": conn.close() return jsonify({"summary": match_info["summary"], "generated": False}) team_id = _get_effective_team_id() filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') # --- 基础数据 --- athletes = conn.execute(f""" SELECT DISTINCT r.name FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE rg.match_id = ? AND {filter_clause} """, [match_id] + filter_params).fetchall() athlete_names = [a["name"] for a in athletes] athlete_count = len(athlete_names) if athlete_count == 0: conn.close() return jsonify({"summary": "", "generated": False, "message": "该比赛无本队队员参赛"}) # --- 奖牌统计 --- medals = conn.execute(f""" SELECT r.name, r.place, rg.event, rg.round_type, r.time_result FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE rg.match_id = ? AND {filter_clause} AND r.place IN ('1','2','3') AND (rg.round_type = '决赛' OR rg.round_type = '决赛/Finals' OR rg.round_type = '决赛A' OR rg.round_type = '决赛B') AND r.qual != '犯规' AND r.time_result NOT LIKE '%00:00.000%' """, [match_id] + filter_params).fetchall() gold_athletes = {} silver_athletes = {} bronze_athletes = {} event_medals = {} # event -> {gold,silver,bronze} for m in medals: name = m["name"] evt = m["event"] if evt not in event_medals: event_medals[evt] = {"gold": 0, "silver": 0, "bronze": 0} if m["place"] == "1": gold_athletes[name] = gold_athletes.get(name, 0) + 1 event_medals[evt]["gold"] += 1 elif m["place"] == "2": silver_athletes[name] = silver_athletes.get(name, 0) + 1 event_medals[evt]["silver"] += 1 elif m["place"] == "3": bronze_athletes[name] = bronze_athletes.get(name, 0) + 1 event_medals[evt]["bronze"] += 1 gold_total = sum(gold_athletes.values()) silver_total = sum(silver_athletes.values()) bronze_total = sum(bronze_athletes.values()) medal_total = gold_total + silver_total + bronze_total # --- 奖牌运动员排名 --- medal_counts = {} for name in athlete_names: g = gold_athletes.get(name, 0) s = silver_athletes.get(name, 0) b = bronze_athletes.get(name, 0) if g + s + b > 0: medal_counts[name] = (g, s, b) top_athletes = sorted(medal_counts.items(), key=lambda x: (x[1][0], x[1][1], x[1][2]), reverse=True)[:5] # --- 首次获奖牌者 --- first_medalists = [] for name in medal_counts: prev_medals = conn.execute(""" SELECT COUNT(*) FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m2 ON rg.match_id = m2.id WHERE r.name = ? AND m2.date_range < ? AND r.place IN ('1','2','3') AND (rg.round_type = '决赛' OR rg.round_type = '决赛/Finals' OR rg.round_type = '决赛A' OR rg.round_type = '决赛B') AND r.qual != '犯规' """, (name, match_info.get("date_range", "9999"))).fetchone()[0] if prev_medals == 0: first_medalists.append(name) # --- 各项最佳成绩 --- best_times = conn.execute(f""" SELECT r.name, rg.event, r.time_result, rg.round_type FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE rg.match_id = ? AND {filter_clause} AND r.time_result != '' AND r.time_result NOT LIKE '%00:00.000%' AND r.qual != '犯规' ORDER BY r.time_result ASC """, [match_id] + filter_params).fetchall() # group best per event event_best = {} for bt in best_times: evt = bt["event"] if evt not in event_best: event_best[evt] = bt # --- 参赛项目统计 --- events = conn.execute(f""" SELECT DISTINCT rg.event FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE rg.match_id = ? AND {filter_clause} """, [match_id] + filter_params).fetchall() event_count = len(events) event_list = [e["event"] for e in events] # --- 进入决赛人数 --- finalists = conn.execute(f""" SELECT COUNT(DISTINCT r.name) FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE rg.match_id = ? AND {filter_clause} AND (rg.round_type = '决赛' OR rg.round_type = '决赛/Finals' OR rg.round_type = '决赛A' OR rg.round_type = '决赛B') """, [match_id] + filter_params).fetchone()[0] # ======== 生成记者风格总结 ======== match_title = match_info["title"] or "本次比赛" venue = match_info.get("venue", "") date_range = match_info.get("date_range", "") match_type = match_info.get("match_type", "") # 提取比赛地点简称 venue_short = venue.replace("北京市", "").replace("北京", "").strip() if venue else "" # 提取赛季 season = match_info.get("season", "") parts = [] # ==== 导语：多样化的开场方式 ==== if medal_total >= 10: openings = [ f"{match_title}于{date_range}在{venue}落下帷幕，我队在本站比赛中掀起夺牌狂潮，交出了一份令人振奋的成绩单。", f"随着{date_range}{match_title}的终场哨响，我队以压倒性的表现完成了本站征程，奖牌榜上捷报频传。", f"{venue}的冰面见证了又一场酣畅淋漓的战役——{match_title}收官，我队将士用一枚枚奖牌书写了属于自己的荣光。", ] elif medal_total >= 4: openings = [ f"{match_title}于{date_range}在{venue}圆满收官，我队运动员在多个项目中展现出扎实的竞技实力，收获颇丰。", f"{date_range}，{match_title}在{venue}画上句号。我队选手稳扎稳打，在激烈竞争中拼出了属于自己的位置。", f"为期数日的{match_title}在{venue}落下帷幕，我队健儿不畏强手、敢打敢拼，交出了一份亮眼的答卷。", ] elif medal_total > 0: openings = [ f"{match_title}于{date_range}在{venue}结束全部赛程，我队运动员在有限的参赛项目中抓住机会，拼下宝贵奖牌。", f"在{venue}举办的{match_title}（{date_range}）已告一段落，我队小将们在实战中经受住了考验。", ] else: openings = [ f"{match_title}于{date_range}在{venue}落幕。虽然本站未能站上领奖台，但队员们在冰场上展现出的拼搏精神同样值得铭记。", f"{date_range}，{match_title}在{venue}收官。每一场比赛都是成长的阶梯，我队运动员在本次赛事中积累了宝贵的实战经验。", ] parts.append(_random.choice(openings)) # ==== 队伍规模与参赛面 ==== scale_phrases = [ f"本次比赛我队派出**{athlete_count}名**运动员，覆盖**{event_count}个**项目（{'、'.join(event_list)}），展现了队伍在各距离、各项目上的全面布局。", f"本站我队**{athlete_count}人**出战，在{'、'.join(event_list)}等**{event_count}个**项目中全面出击，从短距离爆发到长距离耐力均有涉猎。", f"面对{match_type or '本次赛事'}的考验，我队**{athlete_count}名**健儿在{'、'.join(event_list) if event_count <= 3 else f'{event_count}个项目'}中轮番上阵，阵容齐整。", ] parts.append(_random.choice(scale_phrases)) # ==== 奖牌战报 ==== if medal_total > 0: if gold_total >= 5: medal_lead = f"最终，我队以**{gold_total}金{silver_total}银{bronze_total}铜**的优异战绩傲视群雄，" if gold_total > 0: medal_lead += f"在{'、'.join(e for e in event_medals if event_medals[e]['gold'] > 0)}等项目上均有金牌入账，" medal_lead += f"全队累计斩获**{medal_total}枚**奖牌。" elif gold_total >= 1: medal_lead = f"奖牌榜方面，我队共收获**{gold_total}金{silver_total}银{bronze_total}铜**，总计**{medal_total}枚**奖牌。" if gold_total == 1 and len(gold_athletes) == 1: gold_name = list(gold_athletes.keys())[0] medal_lead += f"**{gold_name}**在决赛中摘得桂冠，为全队赢得了宝贵金牌。" else: medal_lead = f"虽然没有金牌入账，但我队仍拼下**{silver_total}银{bronze_total}铜**，每一枚奖牌都凝结着队员们的汗水与坚持。" parts.append(medal_lead) # 奖牌分布 medal_events = [e for e in event_medals if sum(event_medals[e].values()) > 0] if len(medal_events) >= 2: top_evt = max(medal_events, key=lambda e: event_medals[e]["gold"] * 3 + event_medals[e]["silver"] * 2 + event_medals[e]["bronze"]) top_evt_data = event_medals[top_evt] parts.append(f"在**{top_evt}**项目上，我队表现最为抢眼，贡献了{top_evt_data['gold']}金{top_evt_data['silver']}银{top_evt_data['bronze']}铜的亮眼成绩单。") # ==== 奖牌运动员亮点（团队视角，不突出个人）==== if top_athletes: top_names = [name for name, _ in top_athletes[:3]] if len(top_athletes) == 1: name, (g, s, b) = top_athletes[0] star_phrases = [ f"**{name}**本站斩获{g}金{s}银{b}铜，用扎实的成绩为团队做出了重要贡献。", f"**{name}**收获{g}金{s}银{b}铜，每一枚奖牌背后都是日复一日的刻苦训练。", ] elif len(top_athletes) <= 3: star_phrases = [ f"{'、'.join(f'**{n}**' for n in top_names)}等队员在各自项目中发挥出色，为团队贡献了宝贵奖牌。", f"{'、'.join(f'**{n}**' for n in top_names)}在本站比赛中均有亮眼发挥，在各自项目上展现了扎实的竞技水平。", ] else: star_phrases = [ f"**{'、'.join(top_names[:3])}**等多名队员均有奖牌入账，全队多点开花。{'、'.join(f'**{n}**' for n in top_names)}在各自的主攻项目上均展现出了明显的进步。", f"多名队员在不同项目上交替闪耀，{'、'.join(f'**{n}**' for n in top_names[:3])}等人均有斩获，展现了队伍深厚的人才储备与均衡的竞争力。", ] parts.append(_random.choice(star_phrases)) # ==== 首次获奖牌者 ==== if first_medalists and len(first_medalists) <= 5: if len(first_medalists) == 1: parts.append(f"本站还有一个令人欣喜的突破——**{first_medalists[0]}**首次站上了领奖台，实现了个人奖牌零的突破。这一刻，所有训练中的汗水都化作了最灿烂的笑容。") else: names_str = "、".join(f"**{n}**" for n in first_medalists) parts.append(f"更令人振奋的是，{names_str}在本站比赛中均收获了个人首枚奖牌，新人的崛起让我队的未来更加可期。") # ==== 最佳成绩亮点 ==== if event_best: # pick 2-3 events with meaningful times best_items = [] for evt, data in list(event_best.items())[:3]: time_str = data["time_result"] if time_str and ":" in time_str: best_items.append(f"{evt}**{data['name']}**跑出**{time_str}**") if best_items: parts.append(f"成绩方面，{'，'.join(best_items)}，这些成绩在同期选手中具有相当的竞争力。") # ==== 决赛参与度 ==== if finalists > 0: finalist_ratio = finalists / athlete_count if athlete_count > 0 else 0 if finalist_ratio >= 0.6: parts.append(f"全队**{finalists}人**闯入各项目决赛，决赛参与率高达{int(finalist_ratio * 100)}%，充分体现了我队在各项目上的整体竞争力。") elif finalist_ratio >= 0.3: parts.append(f"共有**{finalists}名**队员打进决赛轮次，在与各路高手的正面交锋中积累了宝贵的实战经验。") # ==== 收尾：多样化结语（绝不重复）==== if medal_total >= 10: closings = [ "这是一次堪称完美的出征。但短道速滑的冰面从不停歇，下一站的起点线已在眼前。愿队员们带着这份自信与锐气，向着更高的目标全速滑行。⛸️", "奖牌是最好的证明，但绝不是终点。每一场比赛都是一块磨刀石，磨出更锋利的刃、更沉稳的心。期待将士们在接下来的赛事中续写辉煌。", "从预赛到决赛，这支队伍用一场又一场的硬仗证明了自己的实力。冰场不会忘记每一个拼尽全力的弯道，我们下一站再见。", ] elif medal_total >= 4: closings = [ "这份成绩单有亮点也有遗憾，而这恰恰是竞技体育的魅力所在——永远有下一个弯道可以超越，永远有下一次出发值得期待。", "站上领奖台的瞬间固然耀眼，但真正令人动容的是每一轮预赛、每一次弯道中队员们眼神里的坚定。这支队伍正在一步一个脚印地向上攀登。", "比赛落幕，征程继续。每一次出征都是一次淬炼，每一次对决都是一面镜子。愿队员们看清优势也正视差距，在接下来的训练中有的放矢、更进一步。", ] elif medal_total > 0: closings = [ "奖牌虽少，分量不轻。在短道速滑的世界里，每一个弯道都可能改写历史，每一次蹬冰都蕴藏着翻盘的希望。这支年轻的队伍，正在实战中悄然生长。", "对于一支正在爬坡的队伍来说，每一枚奖牌都是对汗水的最好回馈，每一次参赛都是走向更强的必经之路。抬起头，下一站会更好。", "竞技体育的字典里没有「容易」二字。今天的每一份付出都会在未来的某一天开花结果。保持信念，保持专注，属于你们的高光时刻正在路上。", ] else: closings = [ "没有奖牌不代表没有收获。每一位站上赛场的队员，都在与更强的对手交锋中看到了自己的不足与可能。这是一种比奖牌更珍贵的成长。", "短道速滑的冰面上，摔倒过的人才知道如何更好地站立。本站虽未夺牌，但队员们在实战中积累的经验、暴露的问题，都是下一阶段训练的方向。", "真正的强者从不畏惧暂时的低谷。每一次大赛都是成长的阶梯，每一次失败都为成功埋下伏笔。愿队员们从中汲取力量，在下一站绽放。", ] parts.append(_random.choice(closings)) summary = "\n\n".join(parts) try: conn.execute("UPDATE matches SET summary = ? WHERE id = ?", (summary, match_id)) conn.commit() except Exception: pass conn.close() return jsonify({"summary": summary, "generated": True}) @app.route("/api/race//results") @login_required def api_race_results(rcbh): """获取分组比赛结果""" results = get_race_group_results(rcbh) return jsonify(results) @app.route("/api/match//final_results") @login_required def api_match_final_results(match_id): """获取指定比赛的决赛分组全部成绩（含所有队伍）""" event = request.args.get("event", "") category = request.args.get("category", "") athlete_name = request.args.get("athlete_name", "").strip() conn = get_db() # Find the final race group(s) for this event groups = conn.execute(""" SELECT rg.rcbh, rg.event, rg.round_type, rg.category, rg.gender, rg.race_date FROM race_groups rg WHERE rg.match_id = ? AND rg.event = ? AND (rg.round_type = '决赛' OR rg.round_type = '决赛/Finals' OR rg.round_type = '决赛A' OR rg.round_type = '决赛B') """, (match_id, event)).fetchall() if category: groups = [g for g in groups if g["category"] == category] if athlete_name: filtered = [] for g in groups: # Check individual results if conn.execute( "SELECT 1 FROM results WHERE rcbh = ? AND name = ? LIMIT 1", (g["rcbh"], athlete_name) ).fetchone(): filtered.append(g) continue # For relay events, check relay_members if '接力' in g["event"]: if conn.execute( "SELECT 1 FROM relay_members WHERE athlete_name = ? AND event = ? LIMIT 1", (athlete_name, g["event"]) ).fetchone(): filtered.append(g) groups = filtered if not groups: conn.close() return jsonify({"error": "未找到决赛分组"}), 404 # Get results for each group result_groups = [] for g in groups: rows = conn.execute(""" SELECT r.name, r.place, r.time_result, r.qual, r.team, r.is_chaoyang, rg.round_type, rg.category, rg.gender, rg.event FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE r.rcbh = ? ORDER BY CAST(r.place AS INTEGER) """, (g["rcbh"],)).fetchall() result_groups.append({ "round_type": g["round_type"], "category": g["category"], "gender": g["gender"], "results": [dict(r) for r in rows] }) # Get match info match = conn.execute("SELECT title, date_range, venue FROM matches WHERE id = ?", (match_id,)).fetchone() conn.close() return jsonify({ "match": dict(match) if match else {}, "groups": result_groups }) @app.route("/api/seasons") @login_required def api_seasons(): """获取赛季列表""" seasons = get_seasons() return jsonify(seasons) @app.route("/api/leaderboard") @login_required def api_leaderboard(): """队内排行榜 — 按项目/组别/性别筛选，返回当前队伍队员的最佳成绩排名""" event = request.args.get("event", "") category = request.args.get("category", "") gender = request.args.get("gender", "") season = request.args.get("season", "") conn = get_db() team_id = _get_effective_team_id() filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') conditions = [filter_clause, "r.place != ''", "(r.qual IS NULL OR r.qual NOT IN ('犯规','红牌','黄牌','弃权','伤病','DNF','DNS','DQ','退赛'))"] params = list(filter_params) if event: conditions.append("rg.event = ?") params.append(event) if category: conditions.append("rg.category = ?") params.append(category) if gender: conditions.append("rg.gender = ?") params.append(gender) if season: conditions.append("m.season = ?") params.append(season) where = " AND ".join(conditions) rows = conn.execute(f""" SELECT r.name, rg.event, rg.category, rg.gender, MIN(r.time_result) as best_time, MIN(CAST(r.place AS INTEGER)) as best_place, COUNT(DISTINCT m.id) as race_count FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE {where} AND r.time_result != '' AND r.time_result NOT LIKE '%00:00.000%' AND r.time_result NOT LIKE '%NO TIME%' GROUP BY r.name, rg.event """, params).fetchall() results = [] for r in rows: seconds = parse_time_to_seconds(r["best_time"]) if seconds is not None: results.append({ "name": r["name"], "event": r["event"], "category": r["category"], "best_time": r["best_time"], "best_place": r["best_place"], "race_count": r["race_count"], "seconds": seconds, "trend": 0, }) results.sort(key=lambda x: x["seconds"]) trend_filter_clause, trend_filter_params = _build_team_filter_clause(conn, team_id, 'r') for res in results: name = res["name"] evt = res["event"] try: recent = conn.execute(f""" SELECT r.time_result, m.date_range FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.name = ? AND rg.event = ? AND {trend_filter_clause} ORDER BY m.date_range DESC LIMIT 3 """, [name, evt] + trend_filter_params).fetchall() if len(recent) >= 2: s1 = parse_time_to_seconds(recent[0]["time_result"]) s2 = parse_time_to_seconds(recent[-1]["time_result"]) if s1 is not None and s2 is not None: diff = round(s2 - s1, 2) res["trend"] = diff if abs(diff) < 100 else 0 except Exception: pass total = len(results) for i, r in enumerate(results): r["rank"] = i + 1 r["total"] = total if total > 1: r["percentile"] = round((r["rank"] / total) * 100, 1) else: r["percentile"] = 100.0 # Season comparison: if a specific season is selected, compare with previous season if season: try: parts = season.split("-") if len(parts) == 2: prev_season = f"{int(parts[0]) - 1}-{int(parts[1]) - 1}" prev_rows = conn.execute(f""" SELECT r.name, rg.event, MIN(r.time_result) as best_time, MIN(CAST(r.place AS INTEGER)) as best_place FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE {filter_clause} AND r.place != '' AND r.time_result != '' AND r.time_result NOT LIKE '%00:00.000%' AND r.time_result NOT LIKE '%NO TIME%' AND m.season = ? {"AND rg.event = ?" if event else ""} {"AND rg.category = ?" if category else ""} {"AND rg.gender = ?" if gender else ""} AND (r.qual IS NULL OR r.qual NOT IN ('犯规','红牌','黄牌','弃权','伤病','DNF','DNS','DQ','退赛')) GROUP BY r.name, rg.event """, [prev_season] + params[2:]).fetchall() prev_map = {} prev_results = [] for r2 in prev_rows: s2 = parse_time_to_seconds(r2["best_time"]) if s2 is not None: prev_results.append({"name": r2["name"], "event": r2["event"], "seconds": s2, "best_place": r2["best_place"]}) prev_results.sort(key=lambda x: x["seconds"]) for j, pr in enumerate(prev_results): prev_map[(pr["name"], pr["event"])] = {"prev_rank": j + 1, "prev_total": len(prev_results)} for r in results: key = (r["name"], r["event"]) if key in prev_map: p = prev_map[key] r["prev_rank"] = p["prev_rank"] r["prev_total"] = p["prev_total"] r["rank_change"] = p["prev_rank"] - r["rank"] except Exception: pass return jsonify(results) @app.route("/api/export/csv") @login_required def api_export_csv(): """导出数据为 CSV 格式 参数: type=leaderboard|athletes|results """ import csv import io export_type = request.args.get("type", "leaderboard") conn = get_db() output = io.StringIO() writer = csv.writer(output) if export_type == "leaderboard": event = request.args.get("event", "") category = request.args.get("category", "") gender = request.args.get("gender", "") season = request.args.get("season", "") team_id = _get_effective_team_id() filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') conditions = [filter_clause, "r.place != ''", "(r.qual IS NULL OR r.qual NOT IN ('犯规','红牌','黄牌','弃权','伤病','DNF','DNS','DQ','退赛'))"] params = list(filter_params) if event: conditions.append("rg.event = ?"); params.append(event) if category: conditions.append("rg.category = ?"); params.append(category) if gender: conditions.append("rg.gender = ?"); params.append(gender) if season: conditions.append("m.season = ?"); params.append(season) where = " AND ".join(conditions) writer.writerow(["排名", "姓名", "项目", "组别", "性别", "最佳成绩", "参赛次数"]) rows = conn.execute(f""" SELECT r.name, rg.event, rg.category, rg.gender, MIN(r.time_result) as best_time, COUNT(DISTINCT m.id) as race_count FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE {where} AND r.time_result != '' AND r.time_result NOT LIKE '%00:00.000%' GROUP BY r.name, rg.event ORDER BY best_time """, params).fetchall() for i, r in enumerate(rows, 1): writer.writerow([i, r["name"], r["event"], r["category"] or "", r["gender"] or "", r["best_time"] or "", r["race_count"]]) elif export_type == "athletes": writer.writerow(["姓名", "组别", "性别"]) athletes_info = conn.execute( "SELECT name, category, gender FROM athletes WHERE deleted_at IS NULL ORDER BY name" ).fetchall() for a in athletes_info: writer.writerow([a["name"], a["category"] or "", a["gender"] or ""]) elif export_type == "results": athlete_name = request.args.get("athlete_name", "") writer.writerow(["姓名", "项目", "组别", "性别", "轮次", "成绩", "名次", "队伍", "比赛名称", "赛季"]) query = """ SELECT r.name, rg.event, rg.category, rg.gender, rg.round_type, r.time_result, r.place, r.team, m.title, rg.race_date, m.season FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.is_chaoyang = 1 """ params = [] if athlete_name: query += " AND r.name = ?" params.append(athlete_name) query += " ORDER BY m.date_range DESC, rg.race_date DESC" for row in conn.execute(query, params).fetchall(): writer.writerow([row["name"], row["event"], row["category"] or "", row["gender"] or "", row["round_type"] or "", row["time_result"] or "", row["place"] or "", row["team"] or "", row["title"] or "", row["season"] or ""]) else: conn.close() return jsonify({"error": "不支持的导出类型"}), 400 conn.close() csv_content = output.getvalue() output.close() from flask import Response return Response( csv_content, mimetype="text/csv; charset=utf-8-sig", headers={"Content-Disposition": f"attachment; filename=short_track_{export_type}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"} ) @app.route("/api/achievements") @login_required def api_achievements(): """成就系统 — 计算当前队伍解锁的成就""" conn = get_db() team_id = _get_effective_team_id() filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') team_athlete_names = set(_get_team_athlete_names(conn, team_id)) athlete_stats = conn.execute(f""" SELECT r.name, COUNT(DISTINCT m.id) as match_count, SUM(CASE WHEN CAST(r.place AS INTEGER) = 1 THEN 1 ELSE 0 END) as gold_count, SUM(CASE WHEN CAST(r.place AS INTEGER) IN (1,2,3) THEN 1 ELSE 0 END) as medal_count FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE {filter_clause} AND (rg.round_type = '决赛/Finals' OR rg.round_type = '决赛A' OR rg.round_type = '决赛' OR rg.round_type = '决赛B') GROUP BY r.name """, filter_params).fetchall() stats_map = {r["name"]: dict(r) for r in athlete_stats} training_data = conn.execute(""" SELECT athlete_name, COUNT(DISTINCT training_date) as training_days FROM training_records GROUP BY athlete_name """).fetchall() training_map = {r["athlete_name"]: r["training_days"] for r in training_data if r["athlete_name"] in team_athlete_names} attendance_data = conn.execute(""" SELECT athlete_name, training_date, COUNT(*) as sessions FROM training_records GROUP BY athlete_name, training_date """).fetchall() achievements = [] first_race_names = list(stats_map.keys()) if first_race_names: achievements.append({ "id": "first_race", "name": "初出茅庐", "desc": "首次参加正式比赛", "unlocked": True, "athletes": first_race_names[:5], }) triple = [n for n, s in stats_map.items() if s["gold_count"] >= 3] if triple: achievements.append({ "id": "triple_crown", "name": "三冠王", "desc": "累计获得3枚金牌", "unlocked": True, "athletes": triple, }) veterans = [n for n, s in stats_map.items() if s["match_count"] >= 10] if veterans: achievements.append({ "id": "veteran", "name": "百战老兵", "desc": "参加10场以上比赛", "unlocked": True, "athletes": veterans, }) five_races = [n for n, s in stats_map.items() if s["match_count"] >= 5] if five_races: achievements.append({ "id": "five_races", "name": "五战健将", "desc": "参加5场以上比赛", "unlocked": True, "athletes": five_races, }) collectors = [n for n, s in stats_map.items() if s["medal_count"] >= 10] if collectors: achievements.append({ "id": "medal_collector", "name": "奖牌收藏家", "desc": "累计获得10枚奖牌", "unlocked": True, "athletes": collectors, }) ten_golds = [n for n, s in stats_map.items() if s["gold_count"] >= 10] if ten_golds: achievements.append({ "id": "ten_golds", "name": "十金传奇", "desc": "累计获得10枚金牌", "unlocked": True, "athletes": ten_golds, }) iron = [n for n, d in training_map.items() if d >= 30] if iron: achievements.append({ "id": "iron_training", "name": "铁人训练", "desc": "累计训练30天以上", "unlocked": True, "athletes": iron, }) full_attendance_athletes = [] monthly_attendance = {} for row in attendance_data: athlete_name = row["athlete_name"] if athlete_name not in team_athlete_names: continue training_date = row["training_date"] or '' month_key = training_date[:7] if not month_key: continue monthly_attendance.setdefault((athlete_name, month_key), 0) monthly_attendance[(athlete_name, month_key)] += 1 full_attendance_athletes = sorted({name for (name, month_key), sessions in monthly_attendance.items() if month_key and sessions >= 20}) achievements.append({ "id": "full_attendance", "name": "全勤之星", "desc": "某月训练出勤率100%", "unlocked": bool(full_attendance_athletes), "athletes": full_attendance_athletes, }) unlocked_ids = {a["id"] for a in achievements if a["unlocked"]} for defn in [ {"id": "first_race", "name": "初出茅庐", "desc": "首次参加正式比赛"}, {"id": "triple_crown", "name": "三冠王", "desc": "累计获得3枚金牌"}, {"id": "veteran", "name": "百战老兵", "desc": "参加10场以上比赛"}, {"id": "five_races", "name": "五战健将", "desc": "参加5场以上比赛"}, {"id": "medal_collector", "name": "奖牌收藏家", "desc": "累计获得10枚奖牌"}, {"id": "ten_golds", "name": "十金传奇", "desc": "累计获得10枚金牌"}, {"id": "iron_training", "name": "铁人训练", "desc": "累计训练30天以上"}, {"id": "full_attendance", "name": "全勤之星", "desc": "某月训练出勤率100%"}, ]: if defn["id"] not in unlocked_ids: achievements.append({"id": defn["id"], "name": defn["name"], "desc": defn["desc"], "unlocked": False, "athletes": []}) return jsonify(achievements) @app.route("/api/athlete//growth") @login_required def api_athlete_growth(name): """运动员成长时间轴 — 自动生成关键里程碑""" conn = get_db() events = [] # Get all results for this athlete chronologically results = conn.execute(""" SELECT r.name, r.place, r.time_result, r.qual, r.team, rg.event, rg.category, rg.gender, rg.round_type, m.title, m.date_range, m.season FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.name = ? ORDER BY m.date_range ASC, rg.event """, (name,)).fetchall() if not results: return jsonify({"events": []}) # First race first = results[0] events.append({ "type": "first_race", "date": first["date_range"], "title": f"首次参赛 · {first['title']}", "detail": f"项目: {first['event']} | {first['category'] or ''}", }) # Track PBs per event pb_map = {} for r in results: evt = r["event"] seconds = parse_time_to_seconds(r["time_result"]) if seconds is None: continue if evt not in pb_map or seconds < pb_map[evt]: old_best = pb_map.get(evt) pb_map[evt] = seconds if old_best is not None: diff = round(old_best - seconds, 2) events.append({ "type": "pb", "date": r["date_range"], "title": f"⚡ {evt} 刷新PB", "detail": f"{r['time_result']} (提升 {diff}秒)", }) # Medals for r in results: place_str = str(r["place"]) if place_str in ("1", "2", "3"): medal = {"1": "🥇", "2": "🥈", "3": "🥉"}.get(place_str, "") events.append({ "type": "medal", "date": r["date_range"], "title": f"{medal} {r['title']} · {r['event']}", "detail": f"第{r['place']}名 | {r['category'] or ''}", }) # Honors honors = conn.execute( "SELECT title, awarded_date, source FROM athlete_honors WHERE athlete_name = ? ORDER BY awarded_date", (name,) ).fetchall() for h in honors: events.append({ "type": "honor", "date": h["awarded_date"] or "", "title": f"🎖️ {h['title']}", "detail": h["source"] or "", }) # Sort by date events.sort(key=lambda e: e.get("date", "") or "") return jsonify({"events": events}) @app.route("/api/sources") @login_required def api_sources(): """获取数据源列表""" conn = get_db() rows = conn.execute("SELECT data_source, COUNT(*) as cnt FROM matches GROUP BY data_source").fetchall() sources = [{"data_source": r["data_source"], "match_count": r["cnt"]} for r in rows] conn.close() return jsonify(sources) # ========== 接力队员关联管理 ========== # ========== 运动员管理 ========== @app.route("/api/admin/athletes", methods=["GET"]) @admin_required def api_get_athletes(): """获取所有手动添加的队员（支持 include_deleted 参数）""" include_deleted = request.args.get("include_deleted", "0") == "1" conn = get_db() if include_deleted: rows = conn.execute("SELECT * FROM athletes ORDER BY name").fetchall() else: rows = conn.execute("SELECT * FROM athletes WHERE deleted_at IS NULL ORDER BY name").fetchall() conn.close() return jsonify([dict(a) for a in rows]) @app.route("/api/timeline") @login_required def api_timeline(): """获取赛季比赛日历""" season = request.args.get("season", "") conn = get_db() params = [] where = "" if season: where = "WHERE m.season = ?" params.append(season) rows = conn.execute(f"SELECT m.id, m.title, m.date_range, m.season, m.venue FROM matches m {where} ORDER BY m.date_range DESC", params).fetchall() conn.close() return jsonify([dict(r) for r in rows]) @app.route("/api/last-update") @login_required def api_last_update(): """获取最后更新时间""" last = get_last_update() return jsonify({"last_update": last}) @app.route("/api/sync/all_athletes_results") @login_required def api_sync_all_athletes_results(): """全量队员成绩同步（移动端离线数据用）""" from datetime import datetime db = get_db() team_id = _get_effective_team_id() athletes, results, _ = _get_team_sync_payload(db, team_id) return jsonify({ "athletes": athletes, "results": results, "sync_time": datetime.now().isoformat() }) @app.route("/api/sync/full") @login_required def api_sync_full(): """完整数据同步（移动端首次打开或完整刷新时使用），返回所有离线所需数据+data_version""" from datetime import datetime db = get_db() team_id = _get_effective_team_id() athletes, results, _ = _get_team_sync_payload(db, team_id) seasons = db.execute("SELECT DISTINCT season FROM matches ORDER BY season DESC").fetchall() current_matches = get_current_matches() training_types = get_training_types() recent_matches = db.execute(""" SELECT m.id, m.title, m.date_range, m.venue, m.season, m.status, m.data_source FROM matches m ORDER BY m.date_range DESC LIMIT 50 """).fetchall() last_update = get_last_update() data_version = int(datetime.now().timestamp() * 1000) return jsonify({ "athletes": athletes, "results": results, "seasons": [dict(s) for s in seasons], "current_matches": current_matches, "training_types": training_types, "recent_matches": [dict(m) for m in recent_matches], "data_version": data_version, "last_update": last_update, "sync_time": datetime.now().isoformat() }) @app.route("/api/sync/preload") @login_required def api_sync_preload(): """预加载数据：一次性返回所有需要离线缓存的数据，减少请求数""" from datetime import datetime db = get_db() team_id = _get_effective_team_id() athletes, results, opponents = _get_team_sync_payload(db, team_id) seasons = db.execute("SELECT DISTINCT season FROM matches ORDER BY season DESC").fetchall() matches = db.execute(""" SELECT m.id, m.title, m.date_range, m.venue, m.season, m.status, m.match_type, m.data_source FROM matches m ORDER BY m.date_range DESC """).fetchall() current_matches = get_current_matches() dashboard_summary = db.execute( "SELECT COUNT(DISTINCT name) as athlete_count FROM athletes WHERE deleted_at IS NULL" ).fetchone() training_types = get_training_types() team_aliases = db.execute( "SELECT id, alias, is_active, team_id FROM team_aliases WHERE team_id = ? ORDER BY alias", (team_id,) ).fetchall() athlete_overrides = db.execute( "SELECT name, is_chaoyang, updated_at, team_id FROM athlete_overrides WHERE team_id = ? ORDER BY name", (team_id,) ).fetchall() athlete_names = [{"name": a["name"]} for a in athletes] nicknames = db.execute("SELECT id, nickname, real_name FROM athlete_nicknames ORDER BY nickname").fetchall() last_update = get_last_update() data_version = int(datetime.now().timestamp() * 1000) return jsonify({ "athletes": athletes, "all_athletes_results": { "athletes": athletes, "results": results, }, "seasons": [dict(s) for s in seasons], "matches": [dict(m) for m in matches], "current_matches": current_matches, "opponents": opponents, "dashboard_summary": { "athlete_count": dashboard_summary["athlete_count"] if dashboard_summary else 0, "medals": [], "season_stats": [], "last_update": last_update, }, "training_types": training_types, "team_aliases": [dict(t) for t in team_aliases], "athlete_overrides": [dict(a) for a in athlete_overrides], "athlete_names": athlete_names, "nicknames": [dict(n) for n in nicknames], "data_version": data_version, "last_update": last_update, "sync_time": datetime.now().isoformat() }) @app.route("/api/sync/since") @login_required def api_sync_since(): """增量同步：仅返回指定data_version之后更新的数据 查询参数: - data_version: 上次同步时返回的data_version值（毫秒时间戳） - 返回 has_updates=True/False 和更新后的 data_version """ from datetime import datetime data_version = request.args.get("data_version", "0") try: since_ts = float(data_version) / 1000.0 since_dt = datetime.fromtimestamp(since_ts) since_str = since_dt.strftime('%Y-%m-%d %H:%M:%S') except (ValueError, OSError): since_str = "2000-01-01 00:00:00" db = get_db() team_id = _get_effective_team_id() has_new_matches = db.execute(""" SELECT 1 FROM matches WHERE last_updated >= ? LIMIT 1 """, (since_str,)).fetchone() has_new_athletes = db.execute(""" SELECT 1 FROM athletes WHERE created_at >= ? AND deleted_at IS NULL LIMIT 1 """, (since_str,)).fetchone() has_new_race_groups = db.execute(""" SELECT 1 FROM race_groups rg JOIN matches m ON rg.match_id = m.id WHERE m.last_updated >= ? LIMIT 1 """, (since_str,)).fetchone() has_updates = bool(has_new_matches or has_new_athletes or has_new_race_groups) if not has_updates and str(data_version) == '0': athletes, results, _ = _get_team_sync_payload(db, team_id) has_updates = bool(athletes or results) else: athletes = results = None new_data_version = int(datetime.now().timestamp() * 1000) if not has_updates: return jsonify({ "has_updates": False, "data_version": new_data_version, "sync_time": datetime.now().isoformat() }) if athletes is None or results is None: athletes, results, _ = _get_team_sync_payload(db, team_id) current_matches = get_current_matches() seasons = db.execute("SELECT DISTINCT season FROM matches ORDER BY season DESC").fetchall() training_types = get_training_types() return jsonify({ "has_updates": True, "athletes": athletes, "results": results, "seasons": [dict(s) for s in seasons], "current_matches": current_matches, "training_types": training_types, "data_version": new_data_version, "sync_time": datetime.now().isoformat() }) # ========== 多队伍管理 ========== @app.route("/api/admin/teams") @superadmin_required def api_get_teams(): """获取所有队伍配置""" user = get_user_by_username(session['username']) return jsonify({"teams": get_all_teams(), "team_context": _build_team_context_payload(user)}) @app.route("/api/admin/active_team", methods=["GET"]) def api_get_active_team(): """获取当前请求实际生效的队伍上下文""" user = get_user_by_username(session['username']) if 'username' in session else None team_context = _build_team_context_payload(user) return jsonify(team_context['effective_team']) @app.route("/api/admin/active_team", methods=["PUT"]) @superadmin_required def api_set_active_team(): """切换超级管理员当前会话的查看队伍""" data = request.get_json() or {} team_id = data.get("team_id") if not team_id: return jsonify({"error": "缺少 team_id"}), 400 teams = get_all_teams() if not any(t["team_id"] == team_id for t in teams): return jsonify({"error": "无效的 team_id"}), 400 session['view_team_id'] = team_id user = get_user_by_username(session['username']) team_context = _build_team_context_payload(user) return jsonify({"status": "success", "team": team_context['effective_team'], "team_context": team_context}) @app.route("/api/app/version") def api_app_version(): """获取移动端最新版本信息""" import os static_dir = os.path.join(os.path.dirname(__file__), 'static') version = os.environ.get("APP_VERSION") or globals().get("APP_VERSION", "1.0.0") android_variants = [] for filename in ('app-debug-latest.apk', 'app-debug.apk'): apk_path = os.path.join(static_dir, filename) if os.path.exists(apk_path): android_variants.append({ "channel": "direct-apk", "url": f"/{filename}", "size": os.path.getsize(apk_path), "label": "下载安装包", "min_os": "Android 5.0+", "filename": '朝阳短道.apk', }) break ios_variants = [{ "channel": "xcode-project", "url": None, "label": "请通过 Xcode / TestFlight 分发", "min_os": "iOS", }] return jsonify({ "version": version, "platforms": { "android": { "available": bool(android_variants), "variants": android_variants, "primary_action": android_variants[0] if android_variants else None, }, "ios": { "available": True, "variants": ios_variants, "primary_action": ios_variants[0], }, "web": { "available": False, "variants": [], "primary_action": None, } } }) @app.route("/app-debug.apk") @app.route("/app-debug-latest.apk") def download_apk(): """下载APK文件""" import os from flask import send_from_directory static_folder = os.path.join(os.path.dirname(__file__), 'static') filename = os.path.basename(request.path) return send_from_directory(static_folder, filename, as_attachment=True, download_name='朝阳短道.apk') def is_match_day(): """判断今天是否是比赛日（有进行中或即将开始的比赛，或日期在比赛范围内）""" from datetime import datetime db = get_db() today_str = datetime.now().strftime('%Y/%m/%d') today_dash = datetime.now().strftime('%Y-%m-%d') # 检查是否有进行中/即将开始的比赛 ongoing = db.execute(""" SELECT 1 FROM matches WHERE status LIKE '%进行%' OR status LIKE '%即将%' LIMIT 1 """).fetchone() if ongoing: return True # 检查今天是否在任何比赛的日期范围内 in_range = db.execute(""" SELECT date_range FROM matches WHERE date_range IS NOT NULL AND date_range != '' """).fetchall() for row in in_range: dr = row['date_range'] or '' parts = dr.split('—') if len(parts) == 2: try: start = _norm_date(parts[0].strip()) end = _norm_date(parts[1].strip()) if start <= today_dash <= end: return True except: pass elif today_str in dr: return True return False def _norm_date(raw): """标准化日期为 YYYY-MM-DD 格式，补齐前导零""" s = raw.replace('/', '-').replace('.', '-').strip() parts = s.split('-') if len(parts) == 3: return "{}-{:02d}-{:02d}".format(int(parts[0]), int(parts[1]), int(parts[2])) return s def crawl_update_all(): """更新所有数据源""" print("开始更新 zhitikeji 数据源...") try: crawl_update_zhitikeji() except Exception as e: print(f"zhitikeji 更新出错: {e}") print("开始更新 chinacsa 数据源...") try: crawl_update_chinacsa() except Exception as e: print(f"chinacsa 更新出错: {e}") _update_running = False @app.route("/api/update", methods=["POST"]) @admin_required def api_trigger_update(): """手动触发数据更新（所有数据源），后台异步执行""" global _update_running if _update_running: return jsonify({"status": "already_running", "message": "更新正在进行中，请稍候"}) import threading def run_update(): global _update_running _update_running = True try: crawl_update_all() except Exception as e: print(f"后台更新出错: {e}") finally: _update_running = False t = threading.Thread(target=run_update, daemon=True) t.start() return jsonify({"status": "started", "message": "数据更新已开始，预计需要几分钟"}) @app.route("/api/update_status") @admin_required def api_update_status(): """查询更新是否正在进行""" return jsonify({"running": _update_running}) @app.route("/api/datasources") @login_required def api_datasources(): """获取数据源信息""" from database import get_db conn = get_db() sources = {} for row in conn.execute(""" SELECT data_source, COUNT(*) as match_count FROM matches GROUP BY data_source """).fetchall(): sources[row["data_source"]] = {"match_count": row["match_count"]} # 统计每个数据源的记录数 for row in conn.execute(""" SELECT rg.data_source, COUNT(DISTINCT rg.rcbh) as group_count, COUNT(r.id) as result_count FROM race_groups rg LEFT JOIN results r ON r.rcbh = rg.rcbh GROUP BY rg.data_source """).fetchall(): if row["data_source"] in sources: sources[row["data_source"]]["group_count"] = row["group_count"] sources[row["data_source"]]["result_count"] = row["result_count"] conn.close() return jsonify(sources) def _is_team_name(name): """判断名字是否为集体名称（接力队名），不应作为运动员显示""" import re if not name: return False # 匹配 U15、U13、U17、U19 等年龄段标记 if re.search(r'U\d{2}', name, re.IGNORECASE): return True # 匹配纯集体名称 team_only = ['集体', '混合接力', '混合', '接力队'] if name in team_only: return True # 含空格的多队组合名（如"东城朝阳"、"朝阳 东城 平谷"、"北京朝阳 蓓蕾"） if ' ' in name or '\u3000' in name: return True # 匹配"区+组"格式（如"朝阳区丁组"、"朝阳区乙组"） if re.search(r'[市区县].*[甲乙丙丁]组$', name): return True # 匹配"组+区名"格式（如"乙组朝阳区"、"丁组朝阳区"） if re.match(r'^[甲乙丙丁]组.*[市区县]', name): return True # 匹配纯地区区名（如"朝阳区"、"海淀区"） relay_team_patterns = ['朝阳区', '北京市朝阳区', '海淀区', '西城区', '东城区', '丰台区', '石景山区', '延庆区'] if name in relay_team_patterns: return True # 匹配含俱乐部/体校/学校/冰雪/体育等关键词的队名 if re.search(r'俱乐部|体校|学校|冰雪|轮滑|体育|蓓蕾|星锐|瑞杰|启鸣|巅峰|飞扬|巩华|昆仑|斗罗|添越|尽责|梦起源|平谷|雪人|点冰|奥通|蓓蕾|星锐', name): return True # 短队名精确匹配（<=6字，不含关键词但已知是队名） short_team_names = ['朝阳三体校', '朝阳蓓蕾', '朝阳启鸣', '东城朝阳', '朝阳三体'] if name in short_team_names: return True return False @app.route("/api/dashboard/summary") @login_required def api_dashboard_summary(): """Dashboard概要信息""" from database import get_db conn = get_db() team_id = _get_effective_team_id() team_names = _get_team_athlete_names(conn, team_id) active_names = set() for name in team_names: if name in active_names or _is_team_name(name): continue if is_athlete_active(conn, name, team_id): active_names.add(name) filter_clause, filter_params = _build_team_filter_clause(conn, team_id, 'r') season_stats = conn.execute(f""" SELECT m.season, COUNT(DISTINCT m.id) as match_count, COUNT(DISTINCT r.name) as athlete_count, COUNT(DISTINCT rg.event) as event_count FROM matches m JOIN race_groups rg ON rg.match_id = m.id JOIN results r ON r.rcbh = rg.rcbh WHERE {filter_clause} GROUP BY m.season ORDER BY m.season """, filter_params).fetchall() medal_stats = conn.execute(f""" SELECT m.id as match_id, m.season, r.name, rg.event, rg.category, r.place, rg.race_date, m.title as match_title FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE {filter_clause} AND r.place IN ('1', '2', '3') AND (rg.round_type = '决赛/Finals' OR rg.round_type = '决赛A' OR rg.round_type = '决赛') AND r.time_result NOT LIKE '%00:00.000%' AND r.qual != '犯规' GROUP BY m.id, r.name, rg.event, rg.category HAVING r.place = MIN(CAST(r.place AS INTEGER)) ORDER BY m.season DESC, rg.race_date DESC """, filter_params).fetchall() medal_list = [dict(m) for m in medal_stats if m["name"] in active_names or _is_team_name(m["name"])] if active_names: relay_name_placeholders = ','.join(['?'] * len(active_names)) relay_filter_clause, relay_filter_params = _build_team_filter_clause(conn, team_id, 'r') relay_member_filter_clause, relay_member_filter_params = _build_team_filter_clause(conn, team_id, 'r3') relay_medal_stats = conn.execute(f""" SELECT m.id as match_id, m.season, rm.athlete_name as name, rg.event, rg.category, r.place, rg.race_date, m.title as match_title FROM relay_members rm JOIN results r ON r.name = rm.relay_team_name JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE {relay_filter_clause} AND r.place IN ('1', '2', '3') AND (rg.round_type = '决赛/Finals' OR rg.round_type = '决赛A' OR rg.round_type = '决赛') AND r.time_result NOT LIKE '%%00:00.000%%' AND r.qual != '犯规' AND (rm.season IS NULL OR rm.season = m.season) AND (rm.event IS NULL OR rm.event = rg.event) AND rm.athlete_name IN ({relay_name_placeholders}) AND EXISTS ( SELECT 1 FROM results r3 JOIN race_groups rg3 ON r3.rcbh = rg3.rcbh WHERE r3.name = rm.athlete_name AND rg3.match_id = m.id AND rg3.event NOT LIKE '%%接力%%' AND {relay_member_filter_clause} ) GROUP BY m.id, rm.athlete_name, rg.event, rg.category HAVING r.place = MIN(CAST(r.place AS INTEGER)) ORDER BY m.season DESC, rg.race_date DESC """, relay_filter_params + list(active_names) + relay_member_filter_params).fetchall() medal_list.extend([dict(m) for m in relay_medal_stats]) relay_team_medals = conn.execute(f""" SELECT m.id as match_id, m.season, r.name, rg.event, rg.category, r.place, rg.race_date, m.title as match_title FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE {filter_clause} AND r.place IN ('1', '2', '3') AND (rg.round_type = '决赛/Finals' OR rg.round_type = '决赛A' OR rg.round_type = '决赛') AND r.time_result NOT LIKE '%00:00.000%' AND r.qual != '犯规' AND rg.event LIKE '%接力%' GROUP BY m.id, r.name, rg.event, rg.category HAVING r.place = MIN(CAST(r.place AS INTEGER)) """, filter_params).fetchall() mapped_teams = set() if active_names: mapped = conn.execute( "SELECT DISTINCT rm.relay_team_name FROM relay_members rm WHERE rm.athlete_name IN ({})".format(relay_name_placeholders), list(active_names) ).fetchall() mapped_teams = {r["relay_team_name"] for r in mapped} existing_keys = {(m.get('match_id'), m.get('name'), m.get('event'), m.get('category')) for m in medal_list} for rm in relay_team_medals: key = (rm['match_id'], rm['name'], rm['event'], rm['category']) if rm["name"] not in mapped_teams and key not in existing_keys: medal_list.append(dict(rm)) medal_list.sort(key=lambda m: ( m.get('season', ''), parse_chinese_date(m.get('race_date', '')) ), reverse=True) current = get_current_matches() conn.close() return jsonify({ "athlete_count": len(active_names), "athletes": list(active_names), "season_stats": [dict(s) for s in season_stats], "medals": medal_list, "current_matches": current, "last_update": get_last_update(), }) def parse_chinese_date(date_str): """将中文日期(如 2024年10月19日)转为可排序的字符串(如 2024-10-19)""" if not date_str: return "" import re m = re.match(r'(\d+)年(\d+)月(\d+)日', date_str.strip()) if m: return f"{int(m.group(1)):04d}-{int(m.group(2)):02d}-{int(m.group(3)):02d}" return date_str def parse_time_to_seconds(time_str): """将时间字符串(如 01:42.619)转为秒数""" if not time_str or time_str == "00:00.000": return None try: parts = time_str.split(":") if len(parts) == 2: minutes = int(parts[0]) seconds = float(parts[1]) return minutes * 60 + seconds elif len(parts) == 3: hours = int(parts[0]) minutes = int(parts[1]) seconds = float(parts[2]) return hours * 3600 + minutes * 60 + seconds else: return float(time_str) except (ValueError, IndexError): return None # ========== 用户认证 API ========== @app.route("/api/auth/athlete_search") def api_auth_athlete_search(): """公开接口：注册时搜索队员（仅返回名字和组别，不暴露详细成绩）""" from database import get_db as _get_db q = request.args.get("q", "").strip() if not q or len(q) < 1: return jsonify([]) conn = _get_db() rows = conn.execute(""" SELECT DISTINCT name, category FROM athletes WHERE name LIKE ? AND deleted_at IS NULL ORDER BY name LIMIT 20 """, (f'%{q}%',)).fetchall() # Also search from race results for names not in athletes table race_rows = conn.execute(""" SELECT DISTINCT r.name, rg.category FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE r.name LIKE ? AND r.is_chaoyang = 1 ORDER BY r.name LIMIT 20 """, (f'%{q}%',)).fetchall() # Merge and deduplicate seen = set() result = [] for r in list(rows) + list(race_rows): key = r["name"] if key not in seen: seen.add(key) result.append({"name": r["name"], "category": r["category"] or ""}) conn.close() return jsonify(result[:20]) @app.route("/api/auth/register", methods=["POST"]) @limiter.limit("5 per minute") def api_register(): """用户注册（可选关联队员和关系）""" data = request.json username = data.get("username", "").strip() password = data.get("password", "") athlete_name = data.get("athlete_name", "").strip() relationship = data.get("relationship", "").strip() identity = data.get("identity", "family").strip() if not username or not password: return jsonify({"error": "用户名和密码不能为空"}), 400 # 支持手机号格式（11位纯数字）或普通用户名（2-20字符） if username.isdigit(): if len(username) != 11: return jsonify({"error": "手机号格式错误，请输入11位数字"}), 400 elif len(username) < 2 or len(username) > 20: return jsonify({"error": "用户名长度2-20个字符"}), 400 if len(password) < 6: return jsonify({"error": "密码至少6个字符"}), 400 existing = get_user_by_username(username) if existing: return jsonify({"error": "用户名已存在"}), 400 # 教练身份：无需关联队员，直接注册 if identity == 'coach': relationship = '教练' team_id = _get_issued_team()['team_id'] password_hash = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') if create_user(username, password_hash, 'user', None, relationship, 1, team_id=team_id): user = get_user_by_username(username) session.permanent = True session['user_id'] = user['id'] session['username'] = user['username'] session['role'] = user['role'] if user['role'] != 'superadmin': session.pop('view_team_id', None) team_context = _build_team_context_payload(user) return jsonify({ "status": "success", "user": _build_user_payload(user), "team_context": team_context, "active_team": team_context['effective_team'] }) return jsonify({"error": "注册失败"}), 400 # 亲属身份：检查同一队员的同一关系是否已被注册（抢注检查） if athlete_name and relationship: conn = get_db() taken = conn.execute(""" SELECT username FROM users WHERE athlete_name = ? AND relationship = ? AND approved = 1 """, (athlete_name, relationship)).fetchone() conn.close() if taken: return jsonify({"error": f"该队员的「{relationship}」关系已被用户「{taken['username']}」注册，如确为您本人请联系管理员确认"}), 400 # 检查关联的队员是否存在于系统中 athlete_found = False if athlete_name: conn = get_db() # 从成绩表中查找 row = conn.execute("SELECT 1 FROM results WHERE name = ? LIMIT 1", (athlete_name,)).fetchone() # 从手动队员表中查找 if not row: row = conn.execute("SELECT 1 FROM athletes WHERE name = ? AND deleted_at IS NULL LIMIT 1", (athlete_name,)).fetchone() conn.close() athlete_found = row is not None # 如果关联了队员但队员不在系统中，需要管理员审核 needs_approval = bool(athlete_name) and not athlete_found password_hash = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') approved = 0 if needs_approval else 1 team_id = _get_issued_team()['team_id'] if create_user(username, password_hash, 'user', athlete_name or None, relationship or None, approved, team_id=team_id): user = get_user_by_username(username) # 亲属用户：同步写入athlete_family表 if athlete_name and relationship: from database import get_db as _get_db from datetime import datetime as _dt try: _conn = _get_db() _conn.execute("INSERT OR IGNORE INTO athlete_family (athlete_name, user_id, relationship, approved, created_at) VALUES (?, ?, ?, ?, ?)", (athlete_name, user['id'], relationship, approved, _dt.now().isoformat())) _conn.commit() _conn.close() except Exception: pass if needs_approval: # 需要管理员审核，不自动登录 return jsonify({ "status": "pending", "message": "注册成功，您关联的队员未在系统中找到，需等待管理员审核批准后才能登录", "username": username }) session.permanent = True session['user_id'] = user['id'] session['username'] = user['username'] session['role'] = user['role'] if user['role'] != 'superadmin': session.pop('view_team_id', None) family = get_user_family(user['id']) team_context = _build_team_context_payload(user) return jsonify({ "status": "success", "user": _build_user_payload(user, family), "team_context": team_context, "active_team": team_context['effective_team'] }) return jsonify({"error": "注册失败"}), 400 @app.route("/api/auth/login", methods=["POST"]) def api_login(): """用户登录""" data = request.json username = data.get("username", "").strip() password = data.get("password", "") if not username or not password: return jsonify({"error": "用户名和密码不能为空"}), 400 user = get_user_by_username(username) if not user: return jsonify({"error": "用户名或密码错误"}), 401 stored_hash = user['password_hash'] password_ok = False if stored_hash.startswith('$2'): # bcrypt hash try: password_ok = bcrypt.checkpw(password.encode('utf-8'), stored_hash.encode('utf-8')) except (ValueError, TypeError): pass else: # 旧格式：明文、MD5 或 SHA-256，兼容验证 import hashlib sha256_hash = hashlib.sha256(password.encode('utf-8')).hexdigest() md5_hash = hashlib.md5(password.encode('utf-8')).hexdigest() if stored_hash == password or stored_hash == md5_hash or stored_hash == sha256_hash: password_ok = True # 自动升级为 bcrypt try: new_hash = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') update_user_password(user['id'], new_hash) except Exception: pass if not password_ok: return jsonify({"error": "用户名或密码错误"}), 401 # 检查用户是否已被批准 if not user.get('approved'): return jsonify({"error": "您的账号尚未通过审核，请等待管理员批准"}), 403 session.permanent = True # 使 PERMANENT_SESSION_LIFETIME 配置生效 session['user_id'] = user['id'] session['username'] = user['username'] session['role'] = user['role'] if user['role'] != 'superadmin': session.pop('view_team_id', None) family = get_user_family(user['id']) team_context = _build_team_context_payload(user) # 生成 token 供小程序 Bearer 鉴权 token = create_user_token(user['id']) return jsonify({ "status": "success", "user": _build_user_payload(user, family), "team_context": team_context, "active_team": team_context['effective_team'], "token": token }) @app.route("/api/auth/logout", methods=["POST"]) def api_logout(): """用户登出""" session.clear() return jsonify({"status": "success"}) # ====== 小程序微信登录 ====== @app.route("/api/auth/wx-login", methods=["POST"]) @limiter.limit("10 per minute") def api_wx_login(): """小程序微信登录 — 用 code 换取 openid 并验证绑定""" data = request.json code = data.get("code", "").strip() athlete_name = data.get("athlete_name", "").strip() relationship = data.get("relationship", "本人").strip() if not code: return jsonify({"error": "缺少登录凭证"}), 400 if not athlete_name: return jsonify({"error": "请选择关联的队员"}), 400 # 从数据库获取小程序的 appsecret（存储在 app_settings 中） appsecret = _get_wx_appsecret() if not appsecret: return jsonify({"error": "服务端未配置微信小程序 AppSecret"}), 500 # 调用微信 API 换取 openid wx_api = "https://api.weixin.qq.com/sns/jscode2session" try: resp = http_requests.get(wx_api, params={ "appid": "wx1379263366089442", "secret": appsecret, "js_code": code, "grant_type": "authorization_code" }, timeout=10) wx_data = resp.json() except Exception as e: return jsonify({"error": f"微信服务调用失败: {str(e)}"}), 502 openid = wx_data.get("openid") if not openid: errmsg = wx_data.get("errmsg", "未知错误") return jsonify({"error": f"微信登录失败: {errmsg}"}), 401 # 检查是否已有绑定 binding = get_wx_binding_by_openid(openid) if binding: if binding["approved"]: # 已审核通过 → 直接登录 user = get_or_create_wx_user(binding["athlete_name"], openid) if not user: return jsonify({"error": "关联账号异常，请联系管理员"}), 500 token = create_user_token(user["id"]) family = get_user_family(user["id"]) return jsonify({ "status": "success", "user": _build_user_payload(user, family), "token": token, "binding_status": "approved" }) else: return jsonify({ "status": "pending", "message": "您的绑定申请正在等待管理员审核，请稍后再试", "athlete_name": binding["athlete_name"] }) # 新绑定 → 创建待审核记录 create_wx_binding(openid, athlete_name, relationship) return jsonify({ "status": "pending", "message": f"已提交绑定申请（{athlete_name}，{relationship}），等待管理员审核", "athlete_name": athlete_name }) @app.route("/api/auth/wx-bindings/pending", methods=["GET"]) @admin_required def api_wx_pending_bindings(): """管理员查看待审核的微信绑定""" bindings = get_pending_wx_bindings() return jsonify(bindings) @app.route("/api/auth/wx-bindings//approve", methods=["POST"]) @admin_required def api_wx_approve_binding(binding_id): """管理员批准微信绑定""" approve_wx_binding(binding_id) return jsonify({"status": "ok"}) @app.route("/api/auth/wx-bindings//reject", methods=["POST"]) @admin_required def api_wx_reject_binding(binding_id): """管理员拒绝微信绑定 — 直接删除记录""" conn = get_db() conn.execute("DELETE FROM wx_bindings WHERE id = ?", (binding_id,)) conn.commit() return jsonify({"status": "ok"}) def _get_wx_appsecret(): """从本地文件读取微信小程序 AppSecret""" # 优先从环境变量读取 from os import environ secret = environ.get("WX_APPSECRET", "") if secret: return secret # 从数据库 app_settings 读取（管理员通过 API 设置） conn = get_db() row = conn.execute( "SELECT value FROM app_settings WHERE key = 'wx_appsecret'" ).fetchone() return row["value"] if row else "" @app.route("/api/auth/change_password", methods=["POST"]) @login_required def api_change_password(): """修改自己的密码""" data = request.json old_password = data.get("old_password", "") new_password = data.get("new_password", "") if not old_password or not new_password: return jsonify({"error": "请输入旧密码和新密码"}), 400 if len(new_password) < 6: return jsonify({"error": "新密码至少6个字符"}), 400 # 验证旧密码 conn = get_db() row = conn.execute("SELECT password_hash FROM users WHERE id = ?", (session['user_id'],)).fetchone() conn.close() if not row: return jsonify({"error": "旧密码不正确"}), 400 old_hash = row['password_hash'] old_ok = False if old_hash.startswith('$2'): try: old_ok = bcrypt.checkpw(old_password.encode('utf-8'), old_hash.encode('utf-8')) except (ValueError, TypeError): pass else: import hashlib if old_hash == hashlib.sha256(old_password.encode('utf-8')).hexdigest() or old_hash == hashlib.md5(old_password.encode('utf-8')).hexdigest() or old_hash == old_password: old_ok = True if not old_ok: return jsonify({"error": "旧密码不正确"}), 400 new_hash = bcrypt.hashpw(new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') update_user_password(session['user_id'], new_hash) return jsonify({"status": "success"}) @app.route("/api/admin/users//reset_password", methods=["POST"]) @superadmin_required def api_reset_user_password(user_id): """重置用户密码（仅超级管理员）""" data = request.json new_password = data.get("new_password", "") if not new_password: return jsonify({"error": "新密码不能为空"}), 400 if len(new_password) < 6: return jsonify({"error": "新密码至少6个字符"}), 400 # 不允许重置超级管理员密码（除了自己） conn = get_db() row = conn.execute("SELECT role FROM users WHERE id = ?", (user_id,)).fetchone() conn.close() if row and row['role'] == 'superadmin' and user_id != session.get('user_id'): return jsonify({"error": "不能重置其他超级管理员的密码"}), 403 new_hash = bcrypt.hashpw(new_password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') update_user_password(user_id, new_hash) return jsonify({"status": "success"}) @app.route("/api/auth/me") def api_auth_me(): """获取当前登录用户信息""" if 'user_id' in session: user = get_user_by_username(session['username']) if not user: session.clear() return jsonify({"logged_in": False}) family = get_user_family(session['user_id']) team_context = _build_team_context_payload(user) return jsonify({ "logged_in": True, "user": _build_user_payload(user, family), "team_context": team_context, "active_team": team_context['effective_team'] }) return jsonify({"logged_in": False, "team_context": _build_team_context_payload(None), "active_team": _get_issued_team()}) @app.route("/api/admin/users") @admin_required def api_users(): """获取用户列表（管理员）""" users = get_all_users() # 标记超级管理员 for u in users: u['is_superadmin'] = u.get('role') == 'superadmin' return jsonify(users) @app.route("/api/admin/users//approve", methods=["POST"]) @admin_required def api_approve_user(user_id): """批准用户（管理员）""" conn = get_db() row = conn.execute("SELECT role FROM users WHERE id = ?", (user_id,)).fetchone() conn.close() if not row: return jsonify({"error": "用户不存在"}), 404 if row['role'] == 'superadmin': return jsonify({"error": "不能操作超级管理员"}), 403 approve_user(user_id) return jsonify({"status": "success"}) @app.route("/api/admin/users//reject", methods=["POST"]) @admin_required def api_reject_user(user_id): """拒绝用户（管理员）""" conn = get_db() row = conn.execute("SELECT role FROM users WHERE id = ?", (user_id,)).fetchone() conn.close() if not row: return jsonify({"error": "用户不存在"}), 404 if row['role'] == 'superadmin': return jsonify({"error": "不能操作超级管理员"}), 403 reject_user(user_id) return jsonify({"status": "success"}) @app.route("/api/admin/users//role", methods=["PUT"]) @superadmin_required def api_update_user_role(user_id): """更新用户角色（仅超级管理员可操作）""" data = request.json role = data.get("role", "") if role not in ('admin', 'user'): return jsonify({"error": "无效角色，仅支持 admin 和 user"}), 400 # 不允许修改超级管理员的角色 conn = get_db() row = conn.execute("SELECT role, username FROM users WHERE id = ?", (user_id,)).fetchone() conn.close() if row and row['role'] == 'superadmin': return jsonify({"error": "不能修改超级管理员的角色"}), 403 update_user_role(user_id, role) return jsonify({"status": "success"}) @app.route("/api/admin/users/", methods=["DELETE"]) @superadmin_required def api_delete_user(user_id): """删除用户（仅超级管理员可操作）""" if user_id == session.get('user_id'): return jsonify({"error": "不能删除自己"}), 400 # 不允许删除超级管理员 conn = get_db() row = conn.execute("SELECT role FROM users WHERE id = ?", (user_id,)).fetchone() conn.close() if row and row['role'] == 'superadmin': return jsonify({"error": "不能删除超级管理员"}), 403 delete_user(user_id) return jsonify({"status": "success"}) # ========== 队员管理 API ========== @app.route("/api/admin/team_aliases") @admin_required def api_team_aliases(): """获取当前队伍的队名别名列表""" team_id = _get_effective_team_id() aliases = get_team_aliases(team_id) return jsonify(aliases) @app.route("/api/admin/team_aliases", methods=["POST"]) @admin_required def api_add_team_alias(): """添加当前队伍的队名别名""" data = request.json alias = data.get("alias", "").strip() if not alias: return jsonify({"error": "别名不能为空"}), 400 team_id = _get_effective_team_id() if add_team_alias(alias, team_id): recalculate_chaoyang_flags() return jsonify({"status": "success"}) return jsonify({"error": "别名已存在"}), 400 @app.route("/api/admin/team_aliases/", methods=["DELETE"]) @admin_required def api_delete_team_alias(alias_id): """删除当前队伍的队名别名""" team_id = _get_effective_team_id() delete_team_alias(alias_id, team_id) recalculate_chaoyang_flags() return jsonify({"status": "success"}) @app.route("/api/admin/athlete_overrides") @admin_required def api_athlete_overrides(): """获取当前队伍的手动管理队员覆盖列表""" team_id = _get_effective_team_id() overrides = get_athlete_overrides(team_id) return jsonify(overrides) @app.route("/api/athlete//registered_relationships") def api_athlete_registered_relationships(name): """查询某队员已被注册的亲属关系列表""" conn = get_db() # 从 users 表查找已批准用户中关联了该队员的关系 rows = conn.execute(""" SELECT relationship FROM users WHERE athlete_name = ? AND approved = 1 AND relationship IS NOT NULL AND relationship != '' """, (name,)).fetchall() conn.close() relationships = [r['relationship'] for r in rows] return jsonify(relationships) @app.route("/api/athletes/search") @login_required def api_athletes_search(): """搜索队员（用于自由对比等场景）""" from database import get_db as _get_db q = request.args.get("q", "").strip() if not q: return jsonify([]) conn = _get_db() # 从比赛成绩中搜索，按姓名或队伍名匹配 rows = conn.execute(""" SELECT DISTINCT name, team, CASE WHEN is_chaoyang = 1 THEN 1 ELSE 0 END as is_chaoyang FROM results WHERE name LIKE ? OR team LIKE ? ORDER BY CASE WHEN name LIKE ? THEN 0 ELSE 1 END, name LIMIT 30 """, (f'%{q}%', f'%{q}%', f'{q}%')).fetchall() # Deduplicate by name: same athlete may appear with different teams seen_names = set() result_rows = [] for r in rows: if r["name"] in seen_names: continue seen_names.add(r["name"]) rd = dict(r) cnt = conn.execute("SELECT COUNT(*) as cnt FROM results WHERE name = ?", (rd["name"],)).fetchone() rd["result_count"] = cnt["cnt"] if cnt else 0 teams = conn.execute("SELECT DISTINCT team FROM results WHERE name = ?", (rd["name"],)).fetchall() rd["all_teams"] = [t["team"] for t in teams] result_rows.append(rd) # 也从手动添加的队员中搜索 manual_rows = conn.execute(""" SELECT name, gender, category FROM athletes WHERE name LIKE ? """, (f'%{q}%',)).fetchall() for m in manual_rows: if m["name"] not in seen_names: result_rows.append({"name": m["name"], "team": "手动添加", "is_chaoyang": 1, "result_count": 0, "all_teams": []}) conn.close() return jsonify(result_rows) @app.route("/api/admin/athlete_overrides", methods=["POST"]) @admin_required def api_set_athlete_override(): """手动设置当前队伍的队员标记""" data = request.json name = data.get("name", "").strip() is_cy = data.get("is_chaoyang", 1) if not name: return jsonify({"error": "队员姓名不能为空"}), 400 team_id = _get_effective_team_id() set_athlete_override(name, is_cy, team_id) recalculate_chaoyang_flags() return jsonify({"status": "success"}) @app.route("/api/admin/athlete_overrides/", methods=["DELETE"]) @admin_required def api_delete_athlete_override(name): """删除当前队伍的队员覆盖标记""" team_id = _get_effective_team_id() delete_athlete_override(name, team_id) recalculate_chaoyang_flags() return jsonify({"status": "success"}) # ========== 接力队员关联管理 ========== @app.route("/api/admin/relay_members", methods=["GET"]) @admin_required def api_get_relay_members(): """获取接力队员关联列表""" team_name = request.args.get("team_name", "").strip() athlete_name = request.args.get("athlete_name", "").strip() members = get_relay_members(relay_team_name=team_name or None, athlete_name=athlete_name or None) return jsonify(members) @app.route("/api/admin/relay_members", methods=["POST"]) @admin_required def api_set_relay_members(): """设置接力队队员列表""" data = request.json relay_team_name = data.get("relay_team_name", "").strip() athlete_names = data.get("athlete_names", []) season = data.get("season", "").strip() or None event = data.get("event", "").strip() or None if not relay_team_name: return jsonify({"error": "接力队名不能为空"}), 400 if not athlete_names: return jsonify({"error": "队员列表不能为空"}), 400 set_relay_members(relay_team_name, athlete_names, season=season, event=event) return jsonify({"status": "success"}) @app.route("/api/admin/relay_members/", methods=["DELETE"]) @admin_required def api_delete_relay_member(member_id): """删除单条接力队员关联""" delete_relay_member(member_id) return jsonify({"status": "success"}) @app.route("/api/admin/relay_members/auto_populate", methods=["POST"]) @admin_required def api_auto_populate_relay_members(): """根据 category + gender 自动推断接力队员关联""" count = auto_populate_relay_members() return jsonify({"status": "success", "inserted": count}) # ========== 运动员照片 API ========== @app.route("/api/athlete//photo") def api_athlete_photo(name): """获取运动员照片（公开访问，img标签需无鉴权）""" photo = get_athlete_photo(name) if not photo or not photo["photo_data"]: return '', 404 from flask import Response return Response(photo["photo_data"], mimetype=photo["content_type"]) @app.route("/api/athlete//photo", methods=["POST"]) @login_required def api_upload_athlete_photo(name): """上传运动员照片（管理员或该队员指定家属）""" # 检查权限：超级管理员或该队员的指定家属 if session.get('role') != 'superadmin' and not is_family_of(session['user_id'], name): return jsonify({"error": "您没有权限修改该队员的照片，只有超级管理员和队员亲属可以修改"}), 403 if 'photo' not in request.files: return jsonify({"error": "未找到照片文件"}), 400 file = request.files['photo'] if not file.filename: return jsonify({"error": "未选择文件"}), 400 # 限制文件大小（2MB） file.seek(0, 2) size = file.tell() file.seek(0) if size > 2 * 1024 * 1024: return jsonify({"error": "照片文件不能超过2MB"}), 400 content_type = file.content_type or 'image/jpeg' photo_data = file.read() set_athlete_photo(name, photo_data, content_type) return jsonify({"status": "success"}) @app.route("/api/athlete//photo", methods=["DELETE"]) @admin_required def api_delete_athlete_photo(name): """删除运动员照片""" delete_athlete_photo(name) return jsonify({"status": "success"}) # ========== 训练记录 API ========== @app.route("/api/admin/training_types") @login_required def api_training_types(): """获取训练类型列表""" types = get_training_types() return jsonify(types) @app.route("/api/admin/training_types", methods=["POST"]) @admin_required def api_add_training_type(): """添加训练类型""" data = request.json name = data.get("name", "").strip() description = data.get("description", "").strip() is_class = data.get("is_class", 0) if not name: return jsonify({"error": "训练类型名称不能为空"}), 400 if add_training_type(name, description, is_class): return jsonify({"status": "success"}) return jsonify({"error": "训练类型已存在"}), 400 @app.route("/api/admin/training_types/", methods=["DELETE"]) @admin_required def api_delete_training_type(type_id): """删除训练类型""" delete_training_type(type_id) return jsonify({"status": "success"}) @app.route("/api/training_records") @login_required def api_training_records(): """获取训练记录（普通用户只能查看关联队员的记录）""" athlete_name = request.args.get("athlete_name", "") # 权限控制：普通用户只能查看关联队员 if session.get('role') not in ('admin', 'superadmin'): accessible = get_accessible_athletes(session['user_id'], session['role']) if athlete_name and athlete_name not in accessible: return jsonify({"error": "无权查看该队员数据"}), 403 if not athlete_name: # 返回所有可访问队员的记录 result = [] for name in accessible: records = get_training_records(name) result.extend(records) return jsonify(result) records = get_training_records(athlete_name if athlete_name else None) return jsonify(records) @app.route("/api/training/calendar_dates") @login_required def api_training_calendar_dates(): """获取训练日期列表（用于日历高亮），返回 {date: {is_class: bool, count: int}} """ athlete_name = request.args.get("athlete_name", "") year = request.args.get("year", "") month = request.args.get("month", "") # 权限控制 if session.get('role') not in ('admin', 'superadmin'): accessible = get_accessible_athletes(session['user_id'], session['role']) if athlete_name and athlete_name not in accessible: return jsonify({"error": "无权查看该队员数据"}), 403 if not athlete_name: athlete_name = None # 将在下面遍历 accessible from database import get_db conn = get_db() if athlete_name: names = [athlete_name] elif session.get('role') not in ('admin', 'superadmin'): names = accessible else: names = None # 全部队员 # 构建查询 conditions = [] params = [] if year: conditions.append("strftime('%Y', training_date) = ?") params.append(year) if month: conditions.append("strftime('%m', training_date) = ?") params.append(month.zfill(2)) if names is not None: placeholders = ','.join(['?'] * len(names)) conditions.append(f"athlete_name IN ({placeholders})") params.extend(names) where_clause = (" AND " + " AND ".join(conditions)) if conditions else "" rows = conn.execute(f""" SELECT training_date, tt.is_class, COUNT(*) as cnt FROM training_records tr JOIN training_types tt ON tr.training_type_id = tt.id WHERE 1=1{where_clause} GROUP BY training_date, tt.is_class """, params).fetchall() conn.close() # 汇总：每个日期有大课/小课标记 result = {} for row in rows: date = row["training_date"] if date not in result: result[date] = {"has_class": False, "has_custom": False, "count": 0} if row["is_class"]: result[date]["has_class"] = True else: result[date]["has_custom"] = True result[date]["count"] += row["cnt"] return jsonify(result) @app.route("/api/training_records", methods=["POST"]) @admin_required def api_add_training_record(): """添加训练记录（管理员/超管可添加任意队员，普通用户只能添加关联队员）""" data = request.json athlete_name = data.get("athlete_name", "").strip() training_type_id = data.get("training_type_id") training_date = data.get("training_date", "").strip() duration_minutes = data.get("duration_minutes") notes = data.get("notes", "").strip() if not all([athlete_name, training_type_id, training_date]): return jsonify({"error": "运动员姓名、训练类型和日期不能为空"}), 400 # 权限控制：普通用户只能为关联队员添加记录 if session.get('role') not in ('admin', 'superadmin'): accessible = get_accessible_athletes(session['user_id'], session.get('role', 'user')) if athlete_name not in accessible: return jsonify({"error": "无权为该队员添加训练记录"}), 403 add_training_record(athlete_name, training_type_id, training_date, duration_minutes, notes) return jsonify({"status": "success"}) @app.route("/api/training_records/", methods=["DELETE"]) @admin_required def api_delete_training_record(record_id): """删除训练记录""" delete_training_record(record_id) return jsonify({"status": "success"}) @app.route("/api/training_records/", methods=["PUT"]) @login_required def api_update_training_record(record_id): """修改训练记录 - 管理员/超级管理员：直接修改 - 普通用户（家属）：提交修改申请，需管理员审批 """ data = request.json training_type_id = data.get("training_type_id") training_date = data.get("training_date") duration_minutes = data.get("duration_minutes") notes = data.get("notes") # 获取该记录的运动员名 conn = get_db() record = conn.execute("SELECT athlete_name FROM training_records WHERE id = ?", (record_id,)).fetchone() conn.close() if not record: return jsonify({"error": "训练记录不存在"}), 404 athlete_name = record["athlete_name"] if session.get('role') in ('admin', 'superadmin'): # 管理员直接修改 update_training_record(record_id, training_type_id, training_date, duration_minutes, notes) return jsonify({"status": "success"}) else: # 普通用户检查权限 if not can_edit_athlete(athlete_name): return jsonify({"error": "没有权限修改该队员训练记录"}), 403 # 提交审批 changes = {} if training_type_id is not None: changes["training_type_id"] = training_type_id if training_date is not None: changes["training_date"] = training_date if duration_minutes is not None: changes["duration_minutes"] = duration_minutes if notes is not None: changes["notes"] = notes if not changes: return jsonify({"error": "没有修改内容"}), 400 submit_training_change(record_id, athlete_name, changes, session['user_id']) return jsonify({"status": "pending"}) @app.route("/api/admin/training_pending_changes") @admin_required def api_get_pending_changes(): """获取待审批的训练记录修改（管理员）""" changes = get_pending_changes() return jsonify(changes) @app.route("/api/admin/training_pending_changes//approve", methods=["POST"]) @admin_required def api_approve_pending_change(change_id): """批准待审批修改""" if approve_pending_change(change_id, session['user_id']): return jsonify({"status": "success"}) return jsonify({"error": "审批失败，可能已被处理"}), 400 @app.route("/api/admin/training_pending_changes//reject", methods=["POST"]) @admin_required def api_reject_pending_change(change_id): """拒绝待审批修改""" if reject_pending_change(change_id, session['user_id']): return jsonify({"status": "success"}) return jsonify({"error": "操作失败"}), 400 @app.route("/api/training/attendance") @login_required def api_attendance_stats(): """获取出勤统计（家长只能查看自己孩子的）""" start_date = request.args.get("start_date", "") end_date = request.args.get("end_date", "") athlete_name = request.args.get("athlete_name", "") show_inactive = request.args.get("show_inactive", "0") == "1" active_only = not show_inactive # 权限控制 if session.get('role') not in ('admin', 'superadmin'): accessible = get_accessible_athletes(session['user_id'], session['role']) if athlete_name and athlete_name not in accessible: return jsonify({"error": "无权查看该队员数据"}), 403 # 普通用户只能查看自己的队员 if not athlete_name: # 返回所有可访问队员的统计 result = [] for name in accessible: stats = get_attendance_stats( start_date if start_date else None, end_date if end_date else None, name, active_only=False # 已按accessible过滤 ) result.extend(stats) return jsonify(result) stats = get_attendance_stats( start_date if start_date else None, end_date if end_date else None, athlete_name if athlete_name else None, active_only=active_only ) return jsonify(stats) @app.route("/api/training/attendance/") @login_required def api_attendance_detail(athlete_name): """获取某个队员的出勤明细（家长只能查看自己孩子的）""" # 权限控制 if session.get('role') not in ('admin', 'superadmin'): accessible = get_accessible_athletes(session['user_id'], session['role']) if athlete_name not in accessible: return jsonify({"error": "无权查看该队员数据"}), 403 start_date = request.args.get("start_date", "") end_date = request.args.get("end_date", "") detail_mode = request.args.get("detail", "") if detail_mode == "1": # 返回逐条出勤明细 records = get_athlete_attendance_detail( athlete_name, start_date if start_date else None, end_date if end_date else None ) return jsonify(records) detail = get_attendance_detail( athlete_name, start_date if start_date else None, end_date if end_date else None ) return jsonify(detail) # ========== 费用管理 API ========== @app.route("/api/admin/fee_config") @admin_required def api_fee_config(): """获取费用配置""" config = get_fee_config() return jsonify(config) @app.route("/api/admin/fee_config", methods=["POST"]) @admin_required def api_set_fee_config(): """设置训练类型单价""" data = request.json training_type_id = data.get("training_type_id") price = data.get("price_per_session", 0) if not training_type_id: return jsonify({"error": "训练类型ID不能为空"}), 400 try: price = float(price) except (ValueError, TypeError): return jsonify({"error": "单价必须为数字"}), 400 if price < 0: return jsonify({"error": "单价不能为负数"}), 400 set_fee_config(training_type_id, price) return jsonify({"status": "success"}) @app.route("/api/admin/athlete_exemptions") @admin_required def api_athlete_exemptions(): """获取队员费用豁免列表""" athlete_name = request.args.get("athlete_name", "") exemptions = get_athlete_exemptions(athlete_name if athlete_name else None) return jsonify(exemptions) @app.route("/api/admin/athlete_exemptions", methods=["POST"]) @admin_required def api_set_athlete_exemption(): """设置队员费用豁免""" data = request.json athlete_name = data.get("athlete_name", "").strip() training_type_id = data.get("training_type_id") is_exempt = data.get("is_exempt", True) if not athlete_name or not training_type_id: return jsonify({"error": "队员姓名和训练类型不能为空"}), 400 set_athlete_exemption(athlete_name, training_type_id, is_exempt) return jsonify({"status": "success"}) @app.route("/api/admin/athlete_exemptions", methods=["DELETE"]) @admin_required def api_delete_athlete_exemption(): """删除队员费用豁免""" data = request.json athlete_name = data.get("athlete_name", "").strip() training_type_id = data.get("training_type_id") if not athlete_name or not training_type_id: return jsonify({"error": "队员姓名和训练类型不能为空"}), 400 delete_athlete_exemption(athlete_name, training_type_id) return jsonify({"status": "success"}) @app.route("/api/training/fees") @login_required def api_calculate_fees(): """计算费用（普通用户只能查看关联队员的，管理员可查看所有）""" start_date = request.args.get("start_date", "") end_date = request.args.get("end_date", "") athlete_name = request.args.get("athlete_name", "") show_inactive = request.args.get("show_inactive", "0") == "1" active_only = not show_inactive # 权限控制：普通用户只能查看自己关联的队员 if session.get('role') not in ('admin', 'superadmin'): accessible = get_accessible_athletes(session['user_id'], session['role']) if athlete_name and athlete_name not in accessible: return jsonify({"error": "无权查看该队员数据"}), 403 # 只查询可访问的队员 if not athlete_name: # 返回所有可访问队员的费用 result = [] for name in accessible: fees = calculate_fees( start_date if start_date else None, end_date if end_date else None, name, active_only=False # 已按accessible过滤 ) result.extend(fees) return jsonify(result) fees = calculate_fees( start_date if start_date else None, end_date if end_date else None, athlete_name if athlete_name else None, active_only=active_only ) return jsonify(fees) # ========== 个性化费用配置 API ========== @app.route("/api/admin/athlete_fee_config") @admin_required def api_athlete_fee_config(): """获取队员个性化费用配置""" athlete_name = request.args.get("athlete_name", "") training_type_id = request.args.get("training_type_id", "") config = get_athlete_fee_config( athlete_name if athlete_name else None, int(training_type_id) if training_type_id else None ) return jsonify(config) @app.route("/api/admin/athlete_fee_config", methods=["POST"]) @admin_required def api_set_athlete_fee_config(): """设置队员个性化费用""" data = request.json athlete_name = data.get("athlete_name", "").strip() training_type_id = data.get("training_type_id") price = data.get("price_per_session", 0) start_date = data.get("start_date", "").strip() or None if not athlete_name or not training_type_id: return jsonify({"error": "队员姓名和训练类型不能为空"}), 400 try: price = float(price) except (ValueError, TypeError): return jsonify({"error": "单价必须为数字"}), 400 set_athlete_fee_config(athlete_name, training_type_id, price, start_date) return jsonify({"status": "success"}) @app.route("/api/admin/athlete_fee_config", methods=["DELETE"]) @admin_required def api_delete_athlete_fee_config(): """删除队员个性化费用（回退到全局配置）""" data = request.json athlete_name = data.get("athlete_name", "").strip() training_type_id = data.get("training_type_id") if not athlete_name or not training_type_id: return jsonify({"error": "队员姓名和训练类型不能为空"}), 400 delete_athlete_fee_config(athlete_name, training_type_id) return jsonify({"status": "success"}) # ========== 预付费管理 API ========== @app.route("/api/admin/prepaid_balances") @admin_required def api_prepaid_balances(): """获取预付费余额""" athlete_name = request.args.get("athlete_name", "") balances = get_prepaid_balances(athlete_name if athlete_name else None) return jsonify(balances) @app.route("/api/admin/prepaid_balances", methods=["POST"]) @admin_required def api_set_prepaid_balance(): """充值/设置预付费""" data = request.json athlete_name = data.get("athlete_name", "").strip() training_type_id = data.get("training_type_id") amount = data.get("amount", 0) start_date = data.get("start_date", "").strip() or None if not athlete_name or not training_type_id: return jsonify({"error": "队员姓名和训练类型不能为空"}), 400 try: amount = float(amount) except (ValueError, TypeError): return jsonify({"error": "金额必须为数字"}), 400 set_prepaid_balance(athlete_name, training_type_id, amount, start_date) return jsonify({"status": "success"}) @app.route("/api/admin/prepaid_balances", methods=["DELETE"]) @admin_required def api_delete_prepaid_balance(): """删除预付费余额""" data = request.json athlete_name = data.get("athlete_name", "").strip() training_type_id = data.get("training_type_id") if not athlete_name or not training_type_id: return jsonify({"error": "队员姓名和训练类型不能为空"}), 400 delete_prepaid_balance(athlete_name, training_type_id) return jsonify({"status": "success"}) @app.route("/api/admin/prepaid_transactions") @admin_required def api_prepaid_transactions(): """获取预付费交易记录""" athlete_name = request.args.get("athlete_name", "") limit = request.args.get("limit", 100, type=int) transactions = get_prepaid_transactions(athlete_name if athlete_name else None, limit) return jsonify(transactions) # ========== 费用结算管理 API ========== @app.route("/api/admin/fee_settlements") @admin_required def api_fee_settlements(): """获取费用结算记录""" athlete_name = request.args.get("athlete_name", "") training_type_id = request.args.get("training_type_id", "") settlements = get_fee_settlements( athlete_name if athlete_name else None, int(training_type_id) if training_type_id else None ) return jsonify(settlements) @app.route("/api/admin/fee_settlements", methods=["POST"]) @admin_required def api_add_fee_settlement(): """添加费用结算记录""" data = request.json athlete_name = data.get("athlete_name", "").strip() training_type_id = data.get("training_type_id") start_date = data.get("start_date", "").strip() end_date = data.get("end_date", "").strip() amount = data.get("amount", 0) note = data.get("note", "").strip() or None if not all([athlete_name, training_type_id, start_date, end_date]): return jsonify({"error": "队员姓名、训练类型、开始日期和结束日期不能为空"}), 400 try: amount = float(amount) except (ValueError, TypeError): return jsonify({"error": "金额必须为数字"}), 400 settled_by = session.get('username', 'unknown') add_fee_settlement(athlete_name, training_type_id, start_date, end_date, amount, note, settled_by) return jsonify({"status": "success"}) @app.route("/api/admin/fee_settlements/", methods=["DELETE"]) @admin_required def api_delete_fee_settlement(settlement_id): """删除费用结算记录""" delete_fee_settlement(settlement_id) return jsonify({"status": "success"}) # ========== 家长查询 API ========== @app.route("/api/my_athletes") @login_required def api_my_athletes(): """获取当前用户可查看的队员列表""" accessible = get_accessible_athletes(session['user_id'], session.get('role', 'user')) return jsonify(accessible) @app.route("/api/my_training_data") @login_required def api_my_training_data(): """获取当前用户关联队员的训练数据""" accessible = get_accessible_athletes(session['user_id'], session.get('role', 'user')) if not accessible: return jsonify([]) start_date = request.args.get("start_date", "") end_date = request.args.get("end_date", "") result = [] for name in accessible: stats = get_attendance_stats( start_date if start_date else None, end_date if end_date else None, name ) detail = get_attendance_detail( name, start_date if start_date else None, end_date if end_date else None ) fees = calculate_fees( start_date if start_date else None, end_date if end_date else None, name ) result.append({ "athlete_name": name, "stats": stats[0] if stats else None, "detail": detail, "fees": fees[0] if fees else None, }) return jsonify(result) # ========== 每周训练计划 API ========== @app.route("/api/admin/weekly_schedule") @admin_required def api_weekly_schedule(): """获取每周训练计划""" schedule = get_weekly_schedule() return jsonify(schedule) @app.route("/api/admin/weekly_schedule", methods=["POST"]) @admin_required def api_set_weekly_schedule(): """设置某天的训练类型""" data = request.json day_of_week = data.get("day_of_week") training_type_id = data.get("training_type_id") if not day_of_week or not training_type_id: return jsonify({"error": "星期几和训练类型不能为空"}), 400 if day_of_week < 1 or day_of_week > 7: return jsonify({"error": "星期几必须在1-7之间"}), 400 if set_weekly_schedule(day_of_week, training_type_id): return jsonify({"status": "success"}) return jsonify({"error": "设置失败"}), 400 @app.route("/api/admin/weekly_schedule/", methods=["DELETE"]) @admin_required def api_delete_weekly_schedule(schedule_id): """删除周训练计划条目""" delete_weekly_schedule(schedule_id) return jsonify({"status": "success"}) @app.route("/api/training/schedule_for_date") @login_required def api_schedule_for_date(): """根据日期获取当天的训练类型""" date_str = request.args.get("date", "") if not date_str: return jsonify([]) schedule = get_schedule_for_date(date_str) return jsonify(schedule) @app.route("/api/training_records/batch", methods=["POST"]) @admin_required def api_add_training_records_batch(): """批量添加训练记录""" data = request.json records = data.get("records", []) if not records: return jsonify({"error": "记录列表不能为空"}), 400 count = add_training_records_batch(records) return jsonify({"status": "success", "count": count}) # ========== 昵称映射 API ========== @app.route("/api/admin/nicknames") @admin_required def api_nicknames(): """获取所有昵称映射""" nicknames = get_all_nicknames() return jsonify(nicknames) @app.route("/api/admin/nicknames", methods=["POST"]) @admin_required def api_add_nickname(): """添加昵称映射""" data = request.json nickname = data.get("nickname", "").strip() real_name = data.get("real_name", "").strip() if not nickname or not real_name: return jsonify({"error": "昵称和真名不能为空"}), 400 if add_nickname(nickname, real_name): return jsonify({"status": "success"}) return jsonify({"error": "添加失败，昵称可能已存在"}), 400 @app.route("/api/admin/nicknames/", methods=["DELETE"]) @admin_required def api_delete_nickname(nickname_id): """删除昵称映射""" delete_nickname(nickname_id) return jsonify({"status": "success"}) # ========== 留言板 API ========== @app.route("/api/messages") @login_required def api_messages(): """获取留言列表""" limit = request.args.get("limit", 50, type=int) messages = get_messages(limit) # Enrich with family info for each message's author from database import get_db as _get_db conn = _get_db() for m in messages: if m.get('username'): row = conn.execute(""" SELECT af.athlete_name, af.relationship FROM athlete_family af JOIN users u ON u.id = af.user_id WHERE u.username = ? AND af.approved = 1 """, (m['username'],)).fetchone() if row: m['family_athlete'] = row['athlete_name'] m['family_relationship'] = row['relationship'] else: m['family_athlete'] = None m['family_relationship'] = None else: m['family_athlete'] = None m['family_relationship'] = None conn.close() # 获取当前用户的点赞状态 msg_ids = [m['id'] for m in messages] liked_set = get_message_likes_by_user(session['user_id'], msg_ids) if 'user_id' in session else set() for m in messages: m['is_liked'] = m['id'] in liked_set return jsonify(messages) @app.route("/api/messages", methods=["POST"]) @login_required def api_add_message(): """添加留言（需登录）""" data = request.json content = data.get("content", "").strip() image_data = data.get("image_data", "") reply_to_id = data.get("reply_to_id") if not content and not image_data: return jsonify({"error": "留言内容不能为空"}), 400 if len(content) > 500: return jsonify({"error": "留言内容不能超过500字"}), 400 # Limit image size (base64 ~1.5MB) if image_data and len(image_data) > 2000000: return jsonify({"error": "图片太大，请压缩后上传"}), 400 if 'user_id' in session: username = session['username'] else: username = data.get("username", "匿名").strip() if not username: username = "匿名" add_message(username, content, image_data if image_data else None, reply_to_id=reply_to_id) return jsonify({"status": "success"}) @app.route("/api/messages//like", methods=["POST"]) @login_required def api_like_message(msg_id): """给留言点赞/取消赞""" likes, liked = like_message(msg_id, session['user_id']) return jsonify({"status": "success", "likes": likes, "liked": liked}) @app.route("/api/messages/", methods=["DELETE"]) @login_required def api_delete_message(msg_id): """删除留言：管理员可删任何留言，普通用户5分钟内可删自己的""" from database import get_db as _get_db conn = _get_db() msg = conn.execute("SELECT username, created_at FROM messages WHERE id = ?", (msg_id,)).fetchone() conn.close() if not msg: return jsonify({"error": "留言不存在"}), 404 is_admin = session.get('role') in ('admin', 'superadmin') is_owner = session.get('username') == msg['username'] if is_admin: delete_message(msg_id) return jsonify({"status": "success"}) if is_owner: # Check 5-minute window from datetime import datetime, timedelta try: created = datetime.fromisoformat(msg['created_at']) if datetime.now() - created <= timedelta(minutes=5): delete_message(msg_id) return jsonify({"status": "success"}) else: return jsonify({"error": "超过5分钟，无法自行删除，请联系管理员"}), 403 except Exception: return jsonify({"error": "无法判断留言时间"}), 500 return jsonify({"error": "无权删除此留言"}), 403 @app.route("/api/messages/", methods=["PUT"]) @login_required def api_edit_message(msg_id): """修改留言：仅作者在5分钟内可修改""" from database import get_db as _get_db conn = _get_db() msg = conn.execute("SELECT username, created_at FROM messages WHERE id = ?", (msg_id,)).fetchone() conn.close() if not msg: return jsonify({"error": "留言不存在"}), 404 is_owner = session.get('username') == msg['username'] if not is_owner: return jsonify({"error": "只能修改自己的留言"}), 403 from datetime import datetime, timedelta try: created = datetime.fromisoformat(msg['created_at']) if datetime.now() - created > timedelta(minutes=5): return jsonify({"error": "超过5分钟，无法修改，请联系管理员"}), 403 except Exception: return jsonify({"error": "无法判断留言时间"}), 500 data = request.json content = data.get("content", "").strip() if not content and not data.get("image_data"): return jsonify({"error": "留言内容不能为空"}), 400 if len(content) > 500: return jsonify({"error": "留言内容不能超过500字"}), 400 image_data = data.get("image_data", "") if image_data and len(image_data) > 2000000: return jsonify({"error": "图片太大，请压缩后上传"}), 400 conn = _get_db() conn.execute("UPDATE messages SET content = ?, image_data = ? WHERE id = ?", (content, image_data if image_data else None, msg_id)) conn.commit() conn.close() return jsonify({"status": "success"}) # ========== 用户头像 API ========== @app.route("/api/user/avatar/") def api_user_avatar(user_id): """获取用户头像""" avatar = get_user_avatar(user_id) if not avatar or not avatar["avatar_data"]: return '', 404 from flask import Response return Response(avatar["avatar_data"], mimetype=avatar["avatar_type"]) @app.route("/api/user/avatar_by_name/") def api_user_avatar_by_name(username): """按用户名获取用户头像""" user = get_user_by_username(username) if not user: return '', 404 avatar = get_user_avatar(user['id']) if not avatar or not avatar["avatar_data"]: return '', 404 from flask import Response return Response(avatar["avatar_data"], mimetype=avatar["avatar_type"]) @app.route("/api/user/avatar", methods=["POST"]) @login_required def api_upload_user_avatar(): """上传当前用户头像""" if 'avatar' not in request.files: return jsonify({"error": "未找到头像文件"}), 400 file = request.files['avatar'] if not file.filename: return jsonify({"error": "未选择文件"}), 400 # 限制 2MB file.seek(0, 2) size = file.tell() file.seek(0) if size > 2 * 1024 * 1024: return jsonify({"error": "头像文件不能超过2MB"}), 400 content_type = file.content_type or 'image/jpeg' avatar_data = file.read() set_user_avatar(session['user_id'], avatar_data, content_type) return jsonify({"status": "success"}) @app.route("/api/user/avatar", methods=["DELETE"]) @login_required def api_delete_user_avatar(): """删除当前用户头像""" set_user_avatar(session['user_id'], None, None) return jsonify({"status": "success"}) @app.route("/api/user/bio", methods=["PUT"]) @login_required def api_update_bio(): """更新当前用户个性签名""" data = request.json bio = (data.get("bio") or "").strip() if len(bio) > 50: return jsonify({"error": "签名不能超过50字"}), 400 update_user_bio(session['user_id'], bio) return jsonify({"status": "success"}) @app.route("/api/user/profile/") @login_required def api_user_profile(username): """获取用户公开信息（头像+签名）""" profile = get_user_public_profile(username) if not profile: return jsonify({"error": "用户不存在"}), 404 result = {"username": profile["username"], "bio": profile.get("bio") or ""} if profile.get("avatar_data"): import base64 result["avatar"] = f"data:{profile.get('avatar_type', 'image/jpeg')};base64,{base64.b64encode(profile['avatar_data']).decode()}" return jsonify(result) # ========== 访问统计 API ========== @app.route("/api/visit_stats") @admin_required def api_visit_stats(): """获取访问统计（管理员）""" try: stats = get_visit_stats() return jsonify(stats) except Exception as e: import traceback traceback.print_exc() return jsonify({"total": 0, "today": 0, "unique_visitors": 0, "daily": [], "monthly": [], "error": str(e)}) # ========== 家属管理 API ========== @app.route("/api/admin/families") @admin_required def api_families(): """获取所有家属关联（管理员）""" families = get_all_families() return jsonify(families) @app.route("/api/admin/families", methods=["POST"]) @admin_required def api_set_family(): """设置队员的指定家属（管理员）""" data = request.json athlete_name = data.get("athlete_name", "").strip() user_id = data.get("user_id") relationship = data.get("relationship", "").strip() if not athlete_name or not user_id: return jsonify({"error": "队员姓名和用户ID不能为空"}), 400 if set_athlete_family(athlete_name, user_id, relationship): return jsonify({"status": "success"}) return jsonify({"error": "设置失败"}), 400 @app.route("/api/admin/families/", methods=["DELETE"]) @admin_required def api_delete_family(athlete_name): """删除队员的指定家属（管理员）""" delete_athlete_family(athlete_name) return jsonify({"status": "success"}) # ========== 多服务器同步 API ========== @app.route("/api/sync/check_hash", methods=["POST"]) def api_sync_check_hash(): """检查对端是否已有此版本的 DB（避免冗余传输）""" token = request.headers.get("Authorization", "").replace("Bearer ", "") if token != SYNC_SECRET: return jsonify({"error": "unauthorized"}), 403 data = request.get_json(silent=True) or {} remote_hash = data.get("hash", "") db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "short_track.db") if not os.path.exists(db_path): return jsonify({"match": False}), 200 sha = hashlib.sha256() with open(db_path, 'rb') as f: while chunk := f.read(8192): sha.update(chunk) local_hash = sha.hexdigest() return jsonify({"match": local_hash == remote_hash}), 200 @app.route("/api/sync/push_db", methods=["POST"]) def api_sync_push_db(): """接收对端推送的完整数据库文件并原子替换（带备份轮转）""" token = request.headers.get("Authorization", "").replace("Bearer ", "") if token != SYNC_SECRET: return jsonify({"error": "unauthorized"}), 403 if "db_file" not in request.files: return jsonify({"error": "missing db_file"}), 400 db_file = request.files["db_file"] db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "short_track.db") data_dir = os.path.dirname(db_path) temp_path = db_path + ".incoming" db_file.save(temp_path) import sqlite3 try: test_conn = sqlite3.connect(temp_path) result = test_conn.execute("PRAGMA integrity_check").fetchone() test_conn.close() if result[0] != "ok": try: os.remove(temp_path) except Exception: pass print(f"[sync] Received DB failed integrity check: {result[0]}, rejected") return jsonify({"error": f"corrupted database: {result[0]}"}), 400 except Exception as e: try: os.remove(temp_path) except Exception: pass return jsonify({"error": f"invalid database: {e}"}), 400 try: close_db() except Exception: pass # 备份轮转：保留最近 10 个备份 if os.path.exists(db_path): ts = datetime.now().strftime("%Y%m%d_%H%M%S") rotated = os.path.join(data_dir, f"short_track.db.bak_{ts}") os.rename(db_path, rotated) # 清理超出 10 个的旧备份 backups = sorted( [f for f in os.listdir(data_dir) if f.startswith("short_track.db.bak_")], reverse=True ) for old in backups[10:]: try: os.remove(os.path.join(data_dir, old)) except Exception: pass os.rename(temp_path, db_path) # 清理旧的 WAL/SHM 文件（新 DB 会创建新的） for suffix in ("-wal", "-shm"): wf = db_path + suffix if os.path.exists(wf): try: os.remove(wf) except Exception: pass print(f"[sync] Database replaced successfully from peer push ({os.path.getsize(db_path)} bytes)") return jsonify({"status": "ok", "size": os.path.getsize(db_path)}), 200 @app.route("/api/sync/pull_db", methods=["GET"]) def api_sync_pull_db(): """供对端拉取完整数据库（用于自动恢复）""" token = request.headers.get("Authorization", "").replace("Bearer ", "") if token != SYNC_SECRET: return jsonify({"error": "unauthorized"}), 403 from database import get_db db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "short_track.db") if not os.path.exists(db_path): return jsonify({"error": "no database"}), 404 # 拉取前执行 checkpoint 确保文件完整 try: db = get_db() db.execute("PRAGMA wal_checkpoint(TRUNCATE)") except Exception: pass return send_file(db_path, mimetype="application/octet-stream", as_attachment=True, download_name="short_track.db") init_db() # 设置定时自动更新（仅在直接运行 app.py 时启动，gunicorn 通过 wsgi.py 管理） # 每5分钟检查一次，比赛日执行爬取，非比赛日跳过（7天也会执行一次兜底） from apscheduler.schedulers.background import BackgroundScheduler _last_non_match_day_run = [0] # track last run time on non-match days def smart_crawl_update(): """智能爬取：比赛日5分钟一次，非比赛日7天一次""" import time try: match_day = is_match_day() except Exception: match_day = False if match_day: print("比赛日 - 执行数据更新") crawl_update_all() else: # 非比赛日：7天跑一次兜底 now = time.time() SEVEN_DAYS = 7 * 24 * 3600 if now - _last_non_match_day_run[0] >= SEVEN_DAYS: print("非比赛日 - 7天兜底更新") crawl_update_all() _last_non_match_day_run[0] = now # else: skip scheduler = BackgroundScheduler() scheduler.add_job( func=smart_crawl_update, trigger="interval", minutes=5, id="auto_update", name="智能自动更新（比赛日5分钟，非比赛日7天）", replace_existing=True, max_instances=1, coalesce=True, misfire_grace_time=120, ) # 周期性 DB 完整性检查 + 自动恢复（每5分钟） scheduler.add_job( func=db_health_check, trigger="interval", minutes=5, id="db_health_check", name="DB完整性检查+自动恢复", replace_existing=True, max_instances=1, coalesce=True, misfire_grace_time=120, ) scheduler.start() print("已启动智能定时更新：比赛日5分钟一次，非比赛日7天一次") print("已启动 DB 完整性检查：每5分钟一次，异常时自动恢复") # 启动时执行一次完整性检查 db_health_check() app.run(debug=False, host="0.0.0.0", port=5000)