""" 短道速滑竞赛成绩数据库模块 """ import sqlite3 import os import json import threading import time from datetime import datetime, timedelta DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "short_track.db") # 线程局部连接池 _local = threading.local() # DB 重建信号路径（与 app.py 中 DB_REBUILD_SIGNAL 保持一致） _DB_REBUILD_SIGNAL = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", ".db_rebuild_signal") def get_db(): """获取数据库连接（线程局部缓存，同一请求复用连接） 检测 DB_REBUILD_SIGNAL：当 db_health_check 替换了底层 DB 文件后， 所有 worker 的下一次 get_db() 会自动关闭旧连接并重建。 """ conn = getattr(_local, 'conn', None) # 检测重建信号：DB 文件已被替换，需要强制重建连接 _check_rebuild_signal() if conn is not None: try: conn.execute("SELECT 1") return conn except (sqlite3.Error, Exception): try: conn.close() except Exception: pass _local.conn = None os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) conn = sqlite3.connect(DB_PATH, timeout=30) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") conn.execute("PRAGMA temp_store=MEMORY") conn.execute("PRAGMA busy_timeout=10000") conn.execute("PRAGMA wal_autocheckpoint=1000") _local.conn = conn return conn def _check_rebuild_signal(): """检测 DB 重建信号文件，若存在且时间戳新于上次重建则关闭旧连接""" if not os.path.exists(_DB_REBUILD_SIGNAL): return try: with open(_DB_REBUILD_SIGNAL, 'r') as f: sig_time = float(f.read().strip()) except Exception: sig_time = 0.0 last_rebuild = getattr(_local, 'last_rebuild_time', 0.0) if sig_time <= last_rebuild: return # 已处理过此信号 conn = getattr(_local, 'conn', None) if conn is not None: try: conn.close() except Exception: pass _local.conn = None _local.last_rebuild_time = sig_time def close_db(): """关闭当前线程的数据库连接（请求结束时调用）""" conn = getattr(_local, 'conn', None) if conn is not None: try: conn.close() except Exception: pass _local.conn = None def init_db(): """初始化数据库表结构""" conn = get_db() cursor = conn.cursor() # 迁移：为已有表添加 data_source 列（必须在 CREATE TABLE IF NOT EXISTS 之前） for table in ["matches", "race_groups", "results"]: try: cols = [row[1] for row in cursor.execute(f"PRAGMA table_info({table})").fetchall()] if "data_source" not in cols: cursor.execute(f"ALTER TABLE {table} ADD COLUMN data_source TEXT DEFAULT 'zhitikeji'") print(f"已为 {table} 表添加 data_source 列") except Exception as e: # 表不存在，不需要迁移 pass # 迁移：为 users 表添加 athlete_name 和 relationship 列 try: user_cols = [row[1] for row in cursor.execute("PRAGMA table_info(users)").fetchall()] if "athlete_name" not in user_cols: cursor.execute("ALTER TABLE users ADD COLUMN athlete_name TEXT") print("已为 users 表添加 athlete_name 列") if "relationship" not in user_cols: cursor.execute("ALTER TABLE users ADD COLUMN relationship TEXT") print("已为 users 表添加 relationship 列") if "approved" not in user_cols: cursor.execute("ALTER TABLE users ADD COLUMN approved INTEGER DEFAULT 1") print("已为 users 表添加 approved 列") # 已有用户默认为已批准 cursor.execute("UPDATE users SET approved = 1 WHERE approved IS NULL") except Exception: pass # 迁移：为 athletes 表添加 deleted_at 列 try: athlete_cols = [row[1] for row in cursor.execute("PRAGMA table_info(athletes)").fetchall()] if "deleted_at" not in athlete_cols: cursor.execute("ALTER TABLE athletes ADD COLUMN deleted_at TEXT") print("已为 athletes 表添加 deleted_at 列") except Exception: pass # 迁移：为 messages 表添加 image_data 和 likes 列 try: msg_cols = [row[1] for row in cursor.execute("PRAGMA table_info(messages)").fetchall()] if "image_data" not in msg_cols: cursor.execute("ALTER TABLE messages ADD COLUMN image_data TEXT") print("已为 messages 表添加 image_data 列") if "likes" not in msg_cols: cursor.execute("ALTER TABLE messages ADD COLUMN likes INTEGER DEFAULT 0") print("已为 messages 表添加 likes 列") if "reply_to_id" not in msg_cols: cursor.execute("ALTER TABLE messages ADD COLUMN reply_to_id INTEGER") print("已为 messages 表添加 reply_to_id 列") except Exception: pass # 迁移：创建 message_likes 表（记录谁赞了哪条留言） try: cursor.execute(""" CREATE TABLE IF NOT EXISTS message_likes ( message_id INTEGER NOT NULL, user_id INTEGER NOT NULL, created_at TEXT, PRIMARY KEY (message_id, user_id), FOREIGN KEY (message_id) REFERENCES messages(id), FOREIGN KEY (user_id) REFERENCES users(id) ) """) except Exception: pass # 迁移：为 users 表添加 avatar_data 和 avatar_type 列 try: user_avatar_cols = [row[1] for row in cursor.execute("PRAGMA table_info(users)").fetchall()] if "avatar_data" not in user_avatar_cols: cursor.execute("ALTER TABLE users ADD COLUMN avatar_data BLOB") print("已为 users 表添加 avatar_data 列") if "avatar_type" not in user_avatar_cols: cursor.execute("ALTER TABLE users ADD COLUMN avatar_type TEXT DEFAULT 'image/jpeg'") print("已为 users 表添加 avatar_type 列") except Exception: pass # 迁移：为 users 表添加 bio 列 try: user_bio_cols = [row[1] for row in cursor.execute("PRAGMA table_info(users)").fetchall()] if "bio" not in user_bio_cols: cursor.execute("ALTER TABLE users ADD COLUMN bio TEXT DEFAULT ''") print("已为 users 表添加 bio 列") except Exception: pass # 迁移：创建新表（如果不存在） # 运动员个人资料表 try: cursor.execute(""" CREATE TABLE IF NOT EXISTS athlete_profiles ( athlete_name TEXT PRIMARY KEY, bio TEXT DEFAULT '', updated_at TEXT, updated_by TEXT ) """) except Exception: pass # 运动员自定义成绩表（未统计的比赛成绩） try: cursor.execute(""" CREATE TABLE IF NOT EXISTS athlete_custom_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, athlete_name TEXT NOT NULL, competition_date TEXT, competition_name TEXT NOT NULL, event TEXT, round_type TEXT, team TEXT, time_result TEXT, place INTEGER, result_text TEXT, season TEXT, created_at TEXT, created_by TEXT ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_custom_results_athlete ON athlete_custom_results(athlete_name)") # 迁移：为旧表添加新字段 for col, coldef in [('event', 'TEXT'), ('round_type', 'TEXT'), ('team', 'TEXT'), ('time_result', 'TEXT'), ('place', 'INTEGER'), ('season', 'TEXT')]: try: cursor.execute(f"ALTER TABLE athlete_custom_results ADD COLUMN {col} {coldef}") except Exception: pass # 列已存在 except Exception: pass try: cursor.execute(""" CREATE TABLE IF NOT EXISTS athlete_fee_config ( athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, price_per_session REAL NOT NULL, start_date TEXT, updated_at TEXT, PRIMARY KEY (athlete_name, training_type_id), FOREIGN KEY (training_type_id) REFERENCES training_types(id) ) """) except Exception: pass try: cursor.execute(""" CREATE TABLE IF NOT EXISTS prepaid_balance ( id INTEGER PRIMARY KEY AUTOINCREMENT, athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, balance REAL NOT NULL DEFAULT 0, start_date TEXT, updated_at TEXT, UNIQUE(athlete_name, training_type_id), FOREIGN KEY (training_type_id) REFERENCES training_types(id) ) """) except Exception: pass try: cursor.execute(""" CREATE TABLE IF NOT EXISTS prepaid_transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, amount REAL NOT NULL, transaction_type TEXT NOT NULL, description TEXT, related_date TEXT, created_at TEXT, FOREIGN KEY (training_type_id) REFERENCES training_types(id) ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_prepaid_transactions_athlete ON prepaid_transactions(athlete_name)") except Exception: pass try: cursor.execute(""" CREATE TABLE IF NOT EXISTS fee_settlements ( id INTEGER PRIMARY KEY AUTOINCREMENT, athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, start_date TEXT NOT NULL, end_date TEXT NOT NULL, amount REAL NOT NULL DEFAULT 0, note TEXT, settled_by TEXT, created_at TEXT, FOREIGN KEY (training_type_id) REFERENCES training_types(id) ) """) cursor.execute("CREATE INDEX IF NOT EXISTS idx_fee_settlements_athlete ON fee_settlements(athlete_name)") except Exception: pass cursor.executescript(""" CREATE TABLE IF NOT EXISTS matches ( id INTEGER PRIMARY KEY, season TEXT NOT NULL, title TEXT, match_type TEXT, round_info TEXT, date_range TEXT, venue TEXT, status TEXT, last_updated TEXT, data_source TEXT DEFAULT 'zhitikeji' ); CREATE TABLE IF NOT EXISTS race_groups ( rcbh INTEGER PRIMARY KEY, match_id INTEGER NOT NULL, race_date TEXT, race_time TEXT, event TEXT, category TEXT, gender TEXT, round_type TEXT, bsbh INTEGER, race_id INTEGER, data_source TEXT DEFAULT 'zhitikeji', FOREIGN KEY (match_id) REFERENCES matches(id) ); CREATE TABLE IF NOT EXISTS results ( id INTEGER PRIMARY KEY AUTOINCREMENT, rcbh INTEGER NOT NULL, heat_number INTEGER DEFAULT 1, place TEXT, lane TEXT, helmet TEXT, name TEXT NOT NULL, team TEXT, time_result TEXT, qual TEXT, finished TEXT, is_chaoyang INTEGER DEFAULT 0, data_source TEXT DEFAULT 'zhitikeji', FOREIGN KEY (rcbh) REFERENCES race_groups(rcbh) ); CREATE INDEX IF NOT EXISTS idx_results_name ON results(name); CREATE INDEX IF NOT EXISTS idx_results_team ON results(team); CREATE INDEX IF NOT EXISTS idx_results_chaoyang ON results(is_chaoyang); CREATE INDEX IF NOT EXISTS idx_race_groups_match ON race_groups(match_id); CREATE INDEX IF NOT EXISTS idx_race_groups_event ON race_groups(event); CREATE INDEX IF NOT EXISTS idx_matches_season ON matches(season); CREATE INDEX IF NOT EXISTS idx_matches_source ON matches(data_source); CREATE INDEX IF NOT EXISTS idx_race_groups_source ON race_groups(data_source); CREATE INDEX IF NOT EXISTS idx_results_source ON results(data_source); CREATE TABLE IF NOT EXISTS update_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, update_time TEXT NOT NULL, season TEXT, match_id INTEGER, status TEXT, message TEXT ); CREATE TABLE IF NOT EXISTS team_aliases ( id INTEGER PRIMARY KEY AUTOINCREMENT, alias TEXT NOT NULL UNIQUE, is_active INTEGER DEFAULT 1 ); CREATE TABLE IF NOT EXISTS athlete_overrides ( name TEXT PRIMARY KEY, is_chaoyang INTEGER NOT NULL, updated_at TEXT ); CREATE TABLE IF NOT EXISTS athletes ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, gender TEXT, category TEXT, notes TEXT, created_at TEXT, deleted_at TEXT ); CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'user', athlete_name TEXT, relationship TEXT, approved INTEGER DEFAULT 1, team_id TEXT DEFAULT 'chaoyang', avatar_data BLOB, avatar_type TEXT DEFAULT 'image/jpeg', bio TEXT DEFAULT '', created_at TEXT ); CREATE TABLE IF NOT EXISTS athlete_family ( athlete_name TEXT NOT NULL, user_id INTEGER NOT NULL, relationship TEXT, approved INTEGER DEFAULT 0, created_at TEXT, PRIMARY KEY (athlete_name), FOREIGN KEY (user_id) REFERENCES users(id) ); CREATE TABLE IF NOT EXISTS athlete_photos ( name TEXT PRIMARY KEY, photo_data BLOB, content_type TEXT DEFAULT 'image/jpeg', updated_at TEXT ); CREATE TABLE IF NOT EXISTS training_types ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, description TEXT, is_class INTEGER DEFAULT 0 ); CREATE TABLE IF NOT EXISTS training_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, training_date TEXT NOT NULL, duration_minutes INTEGER, notes TEXT, created_at TEXT, FOREIGN KEY (training_type_id) REFERENCES training_types(id) ); CREATE INDEX IF NOT EXISTS idx_training_records_athlete ON training_records(athlete_name); CREATE INDEX IF NOT EXISTS idx_training_records_date ON training_records(training_date); CREATE TABLE IF NOT EXISTS training_pending_changes ( id INTEGER PRIMARY KEY AUTOINCREMENT, record_id INTEGER NOT NULL, athlete_name TEXT NOT NULL, changes TEXT NOT NULL, submitted_by INTEGER NOT NULL, submitted_at TEXT NOT NULL, status TEXT DEFAULT 'pending', reviewed_by INTEGER, reviewed_at TEXT, FOREIGN KEY (record_id) REFERENCES training_records(id), FOREIGN KEY (submitted_by) REFERENCES users(id) ); CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT, content TEXT NOT NULL, image_data TEXT, likes INTEGER DEFAULT 0, created_at TEXT, reply_to_id INTEGER ); CREATE TABLE IF NOT EXISTS visit_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, ip TEXT, user_agent TEXT, page TEXT, visited_at TEXT ); CREATE INDEX IF NOT EXISTS idx_messages_created ON messages(created_at); CREATE INDEX IF NOT EXISTS idx_visit_logs_date ON visit_logs(visited_at); CREATE TABLE IF NOT EXISTS fee_config ( training_type_id INTEGER NOT NULL, price_per_session REAL DEFAULT 0, updated_at TEXT, PRIMARY KEY (training_type_id), FOREIGN KEY (training_type_id) REFERENCES training_types(id) ); CREATE TABLE IF NOT EXISTS athlete_fee_exemptions ( athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, is_exempt INTEGER DEFAULT 1, updated_at TEXT, PRIMARY KEY (athlete_name, training_type_id), FOREIGN KEY (training_type_id) REFERENCES training_types(id) ); -- 每个队员每个训练类型的个性化费用（覆盖全局fee_config） CREATE TABLE IF NOT EXISTS athlete_fee_config ( athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, price_per_session REAL NOT NULL, start_date TEXT, updated_at TEXT, PRIMARY KEY (athlete_name, training_type_id), FOREIGN KEY (training_type_id) REFERENCES training_types(id) ); -- 预付费余额管理 CREATE TABLE IF NOT EXISTS prepaid_balance ( id INTEGER PRIMARY KEY AUTOINCREMENT, athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, balance REAL NOT NULL DEFAULT 0, start_date TEXT, updated_at TEXT, UNIQUE(athlete_name, training_type_id), FOREIGN KEY (training_type_id) REFERENCES training_types(id) ); -- 预付费充值/扣款记录 CREATE TABLE IF NOT EXISTS prepaid_transactions ( id INTEGER PRIMARY KEY AUTOINCREMENT, athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, amount REAL NOT NULL, transaction_type TEXT NOT NULL, description TEXT, related_date TEXT, created_at TEXT, FOREIGN KEY (training_type_id) REFERENCES training_types(id) ); -- 费用结算标记（按时间段标记某队员某类型费用已结清） CREATE TABLE IF NOT EXISTS fee_settlements ( id INTEGER PRIMARY KEY AUTOINCREMENT, athlete_name TEXT NOT NULL, training_type_id INTEGER NOT NULL, start_date TEXT NOT NULL, end_date TEXT NOT NULL, amount REAL NOT NULL DEFAULT 0, note TEXT, settled_by TEXT, created_at TEXT, FOREIGN KEY (training_type_id) REFERENCES training_types(id) ); CREATE INDEX IF NOT EXISTS idx_fee_settlements_athlete ON fee_settlements(athlete_name); CREATE INDEX IF NOT EXISTS idx_prepaid_transactions_athlete ON prepaid_transactions(athlete_name); CREATE TABLE IF NOT EXISTS athlete_honors ( id INTEGER PRIMARY KEY AUTOINCREMENT, athlete_name TEXT NOT NULL, title TEXT NOT NULL, awarded_date TEXT, source TEXT, created_at TEXT, updated_at TEXT ); CREATE INDEX IF NOT EXISTS idx_athlete_honors_name ON athlete_honors(athlete_name); CREATE TABLE IF NOT EXISTS weekly_training_schedule ( id INTEGER PRIMARY KEY AUTOINCREMENT, day_of_week INTEGER NOT NULL, training_type_id INTEGER NOT NULL, UNIQUE(day_of_week, training_type_id), FOREIGN KEY (training_type_id) REFERENCES training_types(id) ); CREATE TABLE IF NOT EXISTS athlete_nicknames ( id INTEGER PRIMARY KEY AUTOINCREMENT, nickname TEXT NOT NULL UNIQUE, real_name TEXT NOT NULL, created_at TEXT ); CREATE TABLE IF NOT EXISTS relay_members ( id INTEGER PRIMARY KEY AUTOINCREMENT, relay_team_name TEXT NOT NULL, athlete_name TEXT NOT NULL, season TEXT, event TEXT, created_at TEXT, updated_at TEXT, UNIQUE(relay_team_name, athlete_name, season, event) ); CREATE INDEX IF NOT EXISTS idx_relay_members_team ON relay_members(relay_team_name); CREATE INDEX IF NOT EXISTS idx_relay_members_athlete ON relay_members(athlete_name); -- 性能优化：补充索引 CREATE INDEX IF NOT EXISTS idx_results_rcbh ON results(rcbh); CREATE INDEX IF NOT EXISTS idx_training_records_type_date ON training_records(training_type_id, training_date); CREATE INDEX IF NOT EXISTS idx_athlete_family_user ON athlete_family(user_id); CREATE INDEX IF NOT EXISTS idx_athlete_family_athlete ON athlete_family(athlete_name); CREATE INDEX IF NOT EXISTS idx_users_username ON users(username); CREATE INDEX IF NOT EXISTS idx_matches_last_updated ON matches(last_updated); -- 复合索引：加速 is_athlete_active 和运动员详情查询 CREATE INDEX IF NOT EXISTS idx_results_name_chaoyang ON results(name, is_chaoyang); CREATE INDEX IF NOT EXISTS idx_results_rcbh_name ON results(rcbh, name); CREATE INDEX IF NOT EXISTS idx_training_records_athlete_date ON training_records(athlete_name, training_date); CREATE INDEX IF NOT EXISTS idx_athlete_overrides_name ON athlete_overrides(name); CREATE INDEX IF NOT EXISTS idx_athletes_name_active ON athletes(name, deleted_at); CREATE INDEX IF NOT EXISTS idx_race_groups_rcbh ON race_groups(rcbh); CREATE INDEX IF NOT EXISTS idx_race_groups_event_category ON race_groups(event, category); """) # 多队伍配置表 cursor.executescript(""" CREATE TABLE IF NOT EXISTS team_configs ( id INTEGER PRIMARY KEY AUTOINCREMENT, team_id TEXT UNIQUE NOT NULL, team_name TEXT NOT NULL, color_primary TEXT NOT NULL, color_accent TEXT NOT NULL, color_light TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS app_settings ( key TEXT PRIMARY KEY, value TEXT ); """) # 小程序微信登录绑定 cursor.execute(""" CREATE TABLE IF NOT EXISTS wx_bindings ( id INTEGER PRIMARY KEY AUTOINCREMENT, openid TEXT NOT NULL, athlete_name TEXT NOT NULL, relationship TEXT DEFAULT '本人', approved INTEGER DEFAULT 0, created_at TEXT, user_id INTEGER ); """) # 迁移：为 wx_bindings 添加 user_id 列 try: wxb_cols = [row[1] for row in cursor.execute("PRAGMA table_info(wx_bindings)").fetchall()] if "user_id" not in wxb_cols: cursor.execute("ALTER TABLE wx_bindings ADD COLUMN user_id INTEGER") print("已为 wx_bindings 表添加 user_id 列") except Exception: pass # 用户 Token 表（供小程序 Bearer 鉴权） cursor.execute(""" CREATE TABLE IF NOT EXISTS user_tokens ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, token TEXT NOT NULL UNIQUE, expires_at TEXT, created_at TEXT ); """) # 初始化默认队伍配置 team_presets = [ ('chaoyang', '朝阳', '#0F4C81', '#4A90D9', '#E8F0FE'), ('haidian', '海淀', '#8B1A1A', '#D94444', '#FEE8E8'), ('yanqing', '延庆', '#166534', '#3E8E5E', '#E8F5E9'), ] for team_id, team_name, primary, accent, light in team_presets: try: cursor.execute( "INSERT OR IGNORE INTO team_configs (team_id, team_name, color_primary, color_accent, color_light) VALUES (?, ?, ?, ?, ?)", (team_id, team_name, primary, accent, light) ) except Exception: pass # 初始化默认活跃队伍 try: cursor.execute("INSERT OR IGNORE INTO app_settings (key, value) VALUES ('active_team', 'chaoyang')") except Exception: pass # 初始化默认超级管理员（密码: admin123） import bcrypt admin_hash = bcrypt.hashpw('admin123'.encode('utf-8'), bcrypt.gensalt()).decode('utf-8') try: cursor.execute("INSERT OR IGNORE INTO users (username, password_hash, role, created_at) VALUES (?, ?, ?, ?)", ('chui', admin_hash, 'superadmin', datetime.now().isoformat())) except Exception: pass # 迁移：将默认超级管理员用户名从 admin 改为 chui（如果还是 admin 的话） try: # 如果 chui 不存在且 admin 存在且是 superadmin，则改名 existing_chui = cursor.execute("SELECT id FROM users WHERE username = 'chui'").fetchone() existing_admin = cursor.execute("SELECT id FROM users WHERE username = 'admin' AND role = 'superadmin'").fetchone() if not existing_chui and existing_admin: cursor.execute("UPDATE users SET username = 'chui' WHERE username = 'admin' AND role = 'superadmin'") except Exception: pass # 迁移：为 team_aliases 和 athlete_overrides 添加 team_id 列 try: alias_cols = [row[1] for row in cursor.execute("PRAGMA table_info(team_aliases)").fetchall()] if "team_id" not in alias_cols: cursor.execute("ALTER TABLE team_aliases ADD COLUMN team_id TEXT DEFAULT 'chaoyang'") print("已为 team_aliases 表添加 team_id 列") except Exception: pass try: override_cols = [row[1] for row in cursor.execute("PRAGMA table_info(athlete_overrides)").fetchall()] if "team_id" not in override_cols: cursor.execute("ALTER TABLE athlete_overrides ADD COLUMN team_id TEXT DEFAULT 'chaoyang'") print("已为 athlete_overrides 表添加 team_id 列") except Exception: pass try: user_cols = [row[1] for row in cursor.execute("PRAGMA table_info(users)").fetchall()] if "team_id" not in user_cols: cursor.execute("ALTER TABLE users ADD COLUMN team_id TEXT DEFAULT 'chaoyang'") print("已为 users 表添加 team_id 列") cursor.execute("UPDATE users SET team_id = 'chaoyang' WHERE team_id IS NULL OR team_id = ''") except Exception: pass # 迁移：为 matches 表添加 summary 列 try: match_cols = [row[1] for row in cursor.execute("PRAGMA table_info(matches)").fetchall()] if "summary" not in match_cols: cursor.execute("ALTER TABLE matches ADD COLUMN summary TEXT DEFAULT ''") print("已为 matches 表添加 summary 列") except Exception: pass # 初始化默认队名别名 default_aliases = [ "朝阳区", "朝阳", "北京市朝阳区第三少儿业余体校", "北京市朝阳区蓓蕾青少年体育俱乐部", "第三少儿体校", "北京市朝阳区第三少儿业余体校U15北京市朝阳区第三少儿业余体校", "朝阳区U15朝阳区", "朝阳区乙组朝阳区" ] for alias in default_aliases: try: cursor.execute("INSERT OR IGNORE INTO team_aliases (alias, team_id) VALUES (?, 'chaoyang')", (alias,)) except Exception: pass # 初始化默认昵称映射（接龙中常用的小名） default_nicknames = [ ("六六", "潘宏阳"), ("七七", "潘宏明"), ("VV", "闫昱烨"), ("蛋蛋", "周煜博"), ] for nickname, real_name in default_nicknames: try: cursor.execute("INSERT OR IGNORE INTO athlete_nicknames (nickname, real_name, created_at) VALUES (?, ?, ?)", (nickname, real_name, datetime.now().isoformat())) except Exception: pass conn.commit() # 如果 athletes 表意外清空，自动从 results 重建 auto_repair_athletes() def upsert_match(conn, match_id, season, title, match_type, round_info, date_range, venue, status, data_source="zhitikeji"): """插入或更新比赛信息""" conn.execute(""" INSERT INTO matches (id, season, title, match_type, round_info, date_range, venue, status, last_updated, data_source) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET season=excluded.season, title=excluded.title, match_type=excluded.match_type, round_info=excluded.round_info, date_range=excluded.date_range, venue=excluded.venue, status=excluded.status, last_updated=excluded.last_updated, data_source=excluded.data_source """, (match_id, season, title, match_type, round_info, date_range, venue, status, datetime.now().isoformat(), data_source)) def upsert_race_group(conn, rcbh, match_id, race_date, race_time, event, category, gender, round_type, bsbh, race_id, data_source="zhitikeji"): """插入或更新比赛分组""" conn.execute(""" INSERT INTO race_groups (rcbh, match_id, race_date, race_time, event, category, gender, round_type, bsbh, race_id, data_source) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(rcbh) DO UPDATE SET match_id=excluded.match_id, race_date=excluded.race_date, race_time=excluded.race_time, event=excluded.event, category=excluded.category, gender=excluded.gender, round_type=excluded.round_type, bsbh=excluded.bsbh, race_id=excluded.race_id, data_source=excluded.data_source """, (rcbh, match_id, race_date, race_time, event, category, gender, round_type, bsbh, race_id, data_source)) def upsert_result(conn, rcbh, place, lane, helmet, name, team, time_result, qual, finished, is_chaoyang, heat_number=1, data_source="zhitikeji"): """插入或更新比赛结果""" # 检查是否已存在相同记录 existing = conn.execute( "SELECT id FROM results WHERE rcbh=? AND name=? AND helmet=?", (rcbh, name, helmet) ).fetchone() if existing: conn.execute(""" UPDATE results SET place=?, lane=?, time_result=?, qual=?, finished=?, is_chaoyang=?, team=?, heat_number=?, data_source=? WHERE id=? """, (place, lane, time_result, qual, finished, is_chaoyang, team, heat_number, data_source, existing["id"])) else: conn.execute(""" INSERT INTO results (rcbh, heat_number, place, lane, helmet, name, team, time_result, qual, finished, is_chaoyang, data_source) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (rcbh, heat_number, place, lane, helmet, name, team, time_result, qual, finished, is_chaoyang, data_source)) def log_update(conn, season, match_id, status, message): """记录更新日志""" conn.execute(""" INSERT INTO update_log (update_time, season, match_id, status, message) VALUES (?, ?, ?, ?, ?) """, (datetime.now().isoformat(), season, match_id, status, message)) def get_chaoyang_athletes(): """获取所有朝阳队运动员列表""" conn = get_db() rows = conn.execute(""" SELECT DISTINCT name, team, category, gender FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE r.is_chaoyang = 1 ORDER BY name """).fetchall() return [dict(r) for r in rows] def get_athlete_results(name, event=None): """获取运动员的比赛成绩""" conn = get_db() query = """ SELECT r.name, r.team, r.place, r.lane, r.helmet, r.time_result, r.qual, r.finished, rg.event, rg.category, rg.gender, rg.round_type, rg.race_date, rg.race_time, m.title as match_title, m.season, m.match_type, m.round_info, m.date_range, m.venue, m.status FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE r.name = ? """ params = [name] if event: query += " AND rg.event = ?" params.append(event) query += " ORDER BY m.season DESC, rg.race_date DESC, rg.race_time DESC, rg.event, rg.round_type" rows = conn.execute(query, params).fetchall() return [dict(r) for r in rows] def get_athlete_events(name): """获取运动员参加的所有项目""" conn = get_db() rows = conn.execute(""" SELECT DISTINCT rg.event FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE r.name = ? ORDER BY rg.event """, (name,)).fetchall() return [r["event"] for r in rows] 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 get_event_timeseries(name, event): """获取运动员某个项目的时间序列数据（同一比赛同一项目只取最好成绩）""" 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, m.date_range, m.status 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 """, (name, event, name, event)).fetchall() result = [dict(r) for r in rows] # 按赛季+真实日期排序（中文日期字符串排序不对） result.sort(key=lambda x: (x.get("season", ""), parse_chinese_date(x.get("race_date", "")))) return result def get_opponent_results(event, category, gender, season=None, exclude_name=None): """获取对手的成绩数据""" conn = get_db() query = """ SELECT r.name, r.team, r.time_result, r.place, r.qual, rg.race_date, rg.round_type, 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 WHERE rg.event = ? AND rg.category = ? AND rg.gender = ? AND r.time_result NOT LIKE '%00:00.000%' AND r.qual != '犯规' """ params = [event, category, gender] if season: query += " AND m.season = ?" params.append(season) if exclude_name: query += " AND r.name != ?" params.append(exclude_name) query += " ORDER BY m.season, rg.race_date, r.time_result" rows = conn.execute(query, params).fetchall() return [dict(r) for r in rows] def get_current_matches(): """获取有分组数据的比赛（状态文字不可靠，以实际数据为准）""" conn = get_db() rows = conn.execute(""" SELECT DISTINCT m.* FROM matches m INNER JOIN race_groups rg ON rg.match_id = m.id WHERE m.status NOT LIKE '%结束%' ORDER BY m.id DESC """).fetchall() return [dict(r) for r in rows] def get_all_matches(season=None, chaoyang_only=False): """获取所有比赛 Args: season: 指定赛季，为空则返回全部 chaoyang_only: 为True时只返回朝阳队有人参加的比赛 """ conn = get_db() if chaoyang_only: # 只返回朝阳队有人参加的比赛 if season: rows = conn.execute(""" SELECT DISTINCT m.* FROM matches m JOIN race_groups rg ON rg.match_id = m.id JOIN results r ON r.rcbh = rg.rcbh WHERE m.season = ? AND r.is_chaoyang = 1 ORDER BY m.season DESC, m.date_range DESC """, (season,)).fetchall() else: rows = conn.execute(""" SELECT DISTINCT m.* FROM matches m JOIN race_groups rg ON rg.match_id = m.id JOIN results r ON r.rcbh = rg.rcbh WHERE r.is_chaoyang = 1 ORDER BY m.season DESC, m.date_range DESC """).fetchall() else: if season: rows = conn.execute("SELECT * FROM matches WHERE season = ? ORDER BY date_range DESC", (season,)).fetchall() else: rows = conn.execute("SELECT * FROM matches ORDER BY season DESC, date_range DESC").fetchall() matches = [dict(r) for r in rows] # 按解析后的日期从新到旧排序 matches.sort(key=lambda m: (m.get("season", ""), parse_chinese_date(m.get("date_range", ""))), reverse=True) return matches def get_seasons(): """获取所有赛季""" conn = get_db() rows = conn.execute("SELECT DISTINCT season FROM matches ORDER BY season DESC").fetchall() return [r["season"] for r in rows] def get_last_update(): """获取最后更新时间""" conn = get_db() row = conn.execute("SELECT MAX(update_time) as last_update FROM update_log").fetchone() return row["last_update"] if row and row["last_update"] else None def get_match_race_groups(match_id): """获取比赛的所有分组""" conn = get_db() rows = conn.execute(""" SELECT rg.*, m.title, m.season, m.status FROM race_groups rg JOIN matches m ON rg.match_id = m.id WHERE rg.match_id = ? ORDER BY rg.race_date, rg.race_time """, (match_id,)).fetchall() return [dict(r) for r in rows] def get_race_group_results(rcbh): """获取分组的比赛结果（rcbh 已为小组级，直接查询即可）""" conn = get_db() rows = conn.execute(""" SELECT r.*, rg.event, rg.category, rg.gender, rg.round_type, rg.race_date FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh WHERE r.rcbh = ? ORDER BY CAST(r.place AS INTEGER) """, (rcbh,)).fetchall() return [dict(r) for r in rows] def get_athlete_categories(name): """获取运动员参加的组别信息（按最新比赛排序）""" conn = get_db() rows = conn.execute(""" 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 rg.category IS NOT NULL """, (name,)).fetchall() # 解析date_range结束日期用于排序（格式如 "2026/4/11—2026/4/12"） 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 rows_sorted = sorted(rows, key=lambda r: parse_end_date(dict(r)['date_range']), reverse=True) # 去重，保留最新的在前 seen = set() result = [] for r in rows_sorted: rd = dict(r) key = (rd['category'], rd['gender']) if key not in seen: seen.add(key) result.append(rd) return result def get_team_aliases(team_id=None): """获取队名别名""" conn = get_db() if team_id: rows = conn.execute("SELECT * FROM team_aliases WHERE team_id = ? ORDER BY id", (team_id,)).fetchall() else: rows = conn.execute("SELECT * FROM team_aliases ORDER BY id").fetchall() return [dict(r) for r in rows] def add_team_alias(alias, team_id='chaoyang'): """添加队名别名""" conn = get_db() try: conn.execute("INSERT INTO team_aliases (alias, team_id) VALUES (?, ?)", (alias, team_id or 'chaoyang')) conn.commit() return True except Exception: return False def delete_team_alias(alias_id, team_id=None): """删除队名别名""" conn = get_db() if team_id: conn.execute("DELETE FROM team_aliases WHERE id = ? AND team_id = ?", (alias_id, team_id)) else: conn.execute("DELETE FROM team_aliases WHERE id = ?", (alias_id,)) conn.commit() def get_athlete_overrides(team_id=None): """获取手动管理的队员覆盖""" conn = get_db() if team_id: rows = conn.execute("SELECT * FROM athlete_overrides WHERE team_id = ? ORDER BY name", (team_id,)).fetchall() else: rows = conn.execute("SELECT * FROM athlete_overrides ORDER BY name").fetchall() return [dict(r) for r in rows] def set_athlete_override(name, is_chaoyang, team_id='chaoyang'): """设置队员的队伍覆盖标记""" conn = get_db() now = datetime.now().isoformat() conn.execute(""" INSERT INTO athlete_overrides (name, is_chaoyang, updated_at, team_id) VALUES (?, ?, ?, ?) ON CONFLICT(name) DO UPDATE SET is_chaoyang=?, updated_at=?, team_id=? """, (name, is_chaoyang, now, team_id or 'chaoyang', is_chaoyang, now, team_id or 'chaoyang')) conn.commit() def delete_athlete_override(name, team_id=None): """删除队员的覆盖标记""" conn = get_db() if team_id: conn.execute("DELETE FROM athlete_overrides WHERE name = ? AND team_id = ?", (name, team_id)) else: conn.execute("DELETE FROM athlete_overrides WHERE name = ?", (name,)) conn.commit() def is_chaoyang_team(team_name): """根据配置表判断是否为朝阳队""" conn = get_db() aliases = conn.execute("SELECT alias FROM team_aliases WHERE is_active = 1").fetchall() return any(row["alias"] in team_name for row in aliases) # ========== 队员管理 ========== def add_athlete(name, gender=None, category=None, notes=None): """手动添加队员（不依赖比赛成绩），如果已软删除则恢复""" conn = get_db() try: # 检查是否已有该队员（包括已软删除的） existing = conn.execute("SELECT id, deleted_at FROM athletes WHERE name = ?", (name,)).fetchone() if existing: if existing["deleted_at"]: conn.execute("UPDATE athletes SET deleted_at = NULL WHERE id = ?", (existing["id"],)) return True else: conn.execute("INSERT INTO athletes (name, gender, category, notes) VALUES (?, ?, ?, ?)", (name, gender, category, notes)) return True except Exception: return False def delete_athlete(name): """软删除队员""" conn = get_db() try: conn.execute("UPDATE athletes SET deleted_at = ? WHERE name = ?", (datetime.now().isoformat(), name)) return True except Exception: return False def get_athletes_with_details(athlete_names=None): """ 批量获取运动员详细信息，避免N+1查询问题 Args: athlete_names: 运动员姓名列表，如果为None则获取所有运动员 Returns: list of athlete details with all related data """ conn = get_db() # 构建基础查询条件 if athlete_names: name_placeholders = ','.join('?' * len(athlete_names)) base_condition = f"WHERE r.name IN ({name_placeholders})" query_params = athlete_names else: base_condition = "" query_params = [] # 获取所有相关数据的联合查询 results_query = f""" SELECT r.name, rg.race_date as date, m.season, m.title as competition, rg.event, rg.round_type as round, r.place, r.time_result as time, r.team, r.is_chaoyang, r.data_source, a.gender, a.category, a.notes FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id LEFT JOIN athletes a ON r.name = a.name AND a.deleted_at IS NULL {base_condition} ORDER BY rg.race_date DESC, m.title, rg.event, rg.round_type """ results_rows = conn.execute(results_query, query_params).fetchall() # 获取荣誉信息 honors_condition = base_condition.replace('r.name', 'h.athlete_name') if base_condition else '' honors_query = f""" SELECT h.athlete_name, h.title, h.awarded_date, h.source FROM athlete_honors h {honors_condition} ORDER BY h.awarded_date DESC """ honors_rows = conn.execute(honors_query, query_params).fetchall() # 获取队伍信息 teams_query = f""" SELECT r.name, r.team, r.is_chaoyang, COUNT(*) as race_count FROM results r {base_condition} GROUP BY r.name, r.team, r.is_chaoyang ORDER BY r.name, race_count DESC """ teams_rows = conn.execute(teams_query, query_params).fetchall() # 组织数据结构 athletes_data = {} # 初始化运动员数据 names_to_process = athlete_names or set(row['name'] for row in results_rows) for name in names_to_process: athletes_data[name] = { 'name': name, 'results': [], 'honors': [], 'teams': [], 'gender': None, 'category': None, 'notes': None } # 填充比赛结果 for row in results_rows: name = row['name'] if name in athletes_data: athletes_data[name]['results'].append({ 'date': row['date'], 'season': row['season'], 'competition': row['competition'], 'event': row['event'], 'round': row['round'], 'place': row['place'], 'time': row['time'], 'team': row['team'], 'is_chaoyang': row['is_chaoyang'], 'data_source': row['data_source'] }) # 保存手动添加的信息 if row['gender']: athletes_data[name]['gender'] = row['gender'] if row['category']: athletes_data[name]['category'] = row['category'] if row['notes']: athletes_data[name]['notes'] = row['notes'] # 填充荣誉信息 for row in honors_rows: name = row['athlete_name'] if name in athletes_data: athletes_data[name]['honors'].append({ 'title': row['title'], 'awarded_date': row['awarded_date'], 'source': row['source'], }) # 填充队伍信息 current_name = None current_teams = [] for row in teams_rows: name = row['name'] if name != current_name: if current_name and current_name in athletes_data: athletes_data[current_name]['teams'] = current_teams current_name = name current_teams = [] current_teams.append({ 'team': row['team'], 'is_chaoyang': row['is_chaoyang'], 'race_count': row['race_count'] }) # 处理最后一个运动员 if current_name and current_name in athletes_data: athletes_data[current_name]['teams'] = current_teams return list(athletes_data.values()) def get_medal_records_batch(): """ 批量获取奖牌记录，优化性能 """ conn = get_db() query = """ SELECT r.name, m.title as competition, rg.event, rg.race_date as date, r.place, r.team, r.is_chaoyang, a.gender, a.category FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id LEFT JOIN athletes a ON r.name = a.name AND a.deleted_at IS NULL WHERE r.place IN (1, 2, 3) ORDER BY rg.race_date DESC, r.place ASC """ rows = conn.execute(query).fetchall() medals = [] for row in rows: medal_type = "🥇" if str(row['place']) == '1' else "🥈" if str(row['place']) == '2' else "🥉" medals.append({ 'name': row['name'], 'competition': row['competition'], 'event': row['event'], 'date': row['date'], 'place': row['place'], 'medal_type': medal_type, 'team': row['team'], 'is_chaoyang': row['is_chaoyang'], 'gender': row['gender'], 'category': row['category'] }) return medals def update_athlete(name, gender=None, category=None, notes=None): """ 更新队员信息 """ conn = get_db() existing = conn.execute("SELECT * FROM athletes WHERE name=? AND deleted_at IS NULL", (name,)).fetchone() try: if existing: if existing['deleted_at']: # 已软删除，恢复 conn.execute("UPDATE athletes SET gender=?, category=?, notes=?, deleted_at=NULL WHERE id=?", (gender, category, notes, existing["id"])) conn.commit() return True else: # 已存在且未删除 return False conn.execute(""" INSERT INTO athletes (name, gender, category, notes, created_at, deleted_at) VALUES (?, ?, ?, ?, ?, NULL) """, (name, gender, category, notes, datetime.now().isoformat())) conn.commit() return True except Exception: return False def get_all_athletes(include_deleted=False): """获取所有手动添加的队员""" conn = get_db() if include_deleted: rows = conn.execute("SELECT * FROM athletes ORDER BY deleted_at ASC, name").fetchall() else: rows = conn.execute("SELECT * FROM athletes WHERE deleted_at IS NULL ORDER BY name").fetchall() return [dict(r) for r in rows] def soft_delete_athlete(athlete_id): """软删除队员""" conn = get_db() conn.execute("UPDATE athletes SET deleted_at = ? WHERE id = ?", (datetime.now().isoformat(), athlete_id)) conn.commit() def restore_athlete(athlete_id): """恢复被软删除的队员""" conn = get_db() conn.execute("UPDATE athletes SET deleted_at = NULL WHERE id = ?", (athlete_id,)) conn.commit() def delete_athlete(athlete_id): """永久删除队员""" conn = get_db() conn.execute("DELETE FROM athletes WHERE id = ?", (athlete_id,)) conn.commit() def auto_repair_athletes(): """athletes 表为空是合法状态，队员列表主要来自 results + profiles + training。 此函数仅做一次诊断日志，不自动重建，避免导入非朝阳队员或复合队名。""" conn = get_db() count = conn.execute("SELECT COUNT(*) FROM athletes").fetchone()[0] if count == 0: print("[auto_repair] athletes 表为空，这是合法状态，不自动重建。手动添加的队员请通过 Web UI 操作。") return 0 def get_athlete_names(): """获取所有队员名称（比赛成绩 + 手动添加 + 训练记录），去重，排除已软删除，昵称解析为真名""" conn = get_db() # 从比赛成绩中获取 result_names = conn.execute("SELECT DISTINCT name FROM results").fetchall() # 从手动添加中获取（排除已软删除的） manual_names = conn.execute("SELECT name, gender, category FROM athletes WHERE deleted_at IS NULL").fetchall() # 从训练记录中获取（只训练不比赛的队员），同时解析昵称 training_names = conn.execute(""" SELECT DISTINCT tr.athlete_name, an.real_name FROM training_records tr LEFT JOIN athlete_nicknames an ON tr.athlete_name = an.nickname """).fetchall() names = set() for r in result_names: names.add(r["name"]) manual_info = {} for m in manual_names: names.add(m["name"]) manual_info[m["name"]] = {"gender": m["gender"], "category": m["category"]} nickname_map = get_nickname_map() for t in training_names: # 优先用映射表解析，否则用LEFT JOIN的结果 resolved = nickname_map.get(t["athlete_name"], t["real_name"] or t["athlete_name"]) names.add(resolved) return sorted(names), manual_info def batch_get_active_athletes(conn): """批量获取所有活跃队员名称（单次查询替代N次 is_athlete_active 调用） 活跃条件（满足任一）: 1. 手动添加到 athletes 表且未删除 2. 手动覆盖为朝阳队 3. 最近半年有训练记录 4. 最近一次出场代表朝阳队且在最近2个赛季内 """ from datetime import datetime, timedelta active = set() # 1. 手动添加的队员 manual = conn.execute("SELECT name FROM athletes WHERE deleted_at IS NULL").fetchall() for r in manual: active.add(r["name"]) # 2. 手动覆盖为朝阳队的 overrides = conn.execute("SELECT name FROM athlete_overrides WHERE is_chaoyang = 1").fetchall() for r in overrides: active.add(r["name"]) # 3. 最近半年有训练记录的（含昵称解析） cutoff = (datetime.now() - timedelta(days=180)).strftime('%Y-%m-%d') nickname_map = get_nickname_map() training = conn.execute( "SELECT DISTINCT athlete_name FROM training_records WHERE training_date >= ?", (cutoff,) ).fetchall() for r in training: name = nickname_map.get(r["athlete_name"], r["athlete_name"]) active.add(name) # 4. 最近2个赛季内代表朝阳队出场的（排除已转队的） # 获取所有赛季按时间倒序 seasons = [s["season"] for s in conn.execute( "SELECT DISTINCT season FROM matches ORDER BY date_range DESC" ).fetchall()] recent_seasons = set(seasons[:2]) if seasons else set() # 获取最近一次出场非朝阳的（已转队） transferred = conn.execute(""" SELECT r.name 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 = 0 GROUP BY r.name HAVING m.date_range = MAX( (SELECT MAX(m2.date_range) FROM results r2 JOIN race_groups rg2 ON r2.rcbh = rg2.rcbh JOIN matches m2 ON rg2.match_id = m2.id WHERE r2.name = r.name) ) """).fetchall() transferred_names = {r["name"] for r in transferred} # 获取最近2个赛季朝阳队出场过的 if recent_seasons: placeholders = ','.join(['?'] * len(recent_seasons)) cy_recent = conn.execute(f""" SELECT DISTINCT r.name 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 AND m.season IN ({placeholders}) """, list(recent_seasons)).fetchall() for r in cy_recent: if r["name"] not in transferred_names: active.add(r["name"]) return active def get_active_athlete_names(months=6): """获取最近N个月还有训练记录的活跃队员名称 Args: months: 月数，默认6个月（半年） Returns: set of active athlete names（已解析昵称为真名） """ from datetime import datetime, timedelta conn = get_db() cutoff = (datetime.now() - timedelta(days=months * 30)).strftime('%Y-%m-%d') rows = conn.execute( "SELECT DISTINCT athlete_name FROM training_records WHERE training_date >= ?", (cutoff,) ).fetchall() # 解析昵称为真名 names = set() nickname_map = get_nickname_map() for r in rows: name = r["athlete_name"] names.add(nickname_map.get(name, name)) return names # ========== 用户管理 ========== def create_user(username, password_hash, role='user', athlete_name=None, relationship=None, approved=1, team_id='chaoyang'): """创建用户""" conn = get_db() try: conn.execute("INSERT INTO users (username, password_hash, role, athlete_name, relationship, approved, team_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (username, password_hash, role, athlete_name, relationship, approved, team_id or 'chaoyang', datetime.now().isoformat())) conn.commit() return True except Exception: return False finally: pass def get_user_by_username(username): """根据用户名获取用户""" conn = get_db() row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone() return dict(row) if row else None def get_all_users(): """获取所有用户""" conn = get_db() rows = conn.execute("SELECT id, username, role, athlete_name, relationship, approved, team_id, created_at FROM users ORDER BY id").fetchall() return [dict(r) for r in rows] def update_user_role(user_id, role): """更新用户角色""" conn = get_db() conn.execute("UPDATE users SET role = ? WHERE id = ?", (role, user_id)) conn.commit() def update_user_password(user_id, password_hash): """更新用户密码""" conn = get_db() conn.execute("UPDATE users SET password_hash = ? WHERE id = ?", (password_hash, user_id)) conn.commit() def update_user_bio(user_id, bio): """更新用户个性签名""" conn = get_db() conn.execute("UPDATE users SET bio = ? WHERE id = ?", (bio, user_id)) conn.commit() def get_user_public_profile(username): """获取用户公开信息（头像+签名）""" conn = get_db() row = conn.execute("SELECT username, bio, avatar_data, avatar_type FROM users WHERE username = ?", (username,)).fetchone() return dict(row) if row else None def delete_user(user_id): """删除用户""" conn = get_db() conn.execute("DELETE FROM users WHERE id = ?", (user_id,)) conn.commit() def get_user_avatar(user_id): """获取用户头像""" conn = get_db() row = conn.execute("SELECT avatar_data, avatar_type FROM users WHERE id = ?", (user_id,)).fetchone() if row and row["avatar_data"]: return {"avatar_data": row["avatar_data"], "avatar_type": row["avatar_type"] or "image/jpeg"} return None def set_user_avatar(user_id, avatar_data, avatar_type="image/jpeg"): """设置用户头像""" conn = get_db() conn.execute("UPDATE users SET avatar_data = ?, avatar_type = ? WHERE id = ?", (avatar_data, avatar_type, user_id)) conn.commit() def approve_user(user_id): """批准用户""" conn = get_db() conn.execute("UPDATE users SET approved = 1 WHERE id = ?", (user_id,)) # 同步更新athlete_family表 user = conn.execute("SELECT athlete_name, relationship FROM users WHERE id = ?", (user_id,)).fetchone() if user and user['athlete_name']: conn.execute("INSERT OR IGNORE INTO athlete_family (athlete_name, user_id, relationship, approved, created_at) VALUES (?, ?, ?, 1, ?)", (user['athlete_name'], user_id, user['relationship'], datetime.now().isoformat())) conn.execute("UPDATE athlete_family SET approved = 1 WHERE user_id = ?", (user_id,)) conn.commit() def reject_user(user_id): """拒绝用户（删除）""" conn = get_db() conn.execute("DELETE FROM users WHERE id = ? AND role != 'superadmin'", (user_id,)) conn.commit() def get_athlete_photo(name): """获取运动员照片""" conn = get_db() row = conn.execute("SELECT photo_data, content_type FROM athlete_photos WHERE name = ?", (name,)).fetchone() if row: return {"photo_data": row["photo_data"], "content_type": row["content_type"]} return None def set_athlete_photo(name, photo_data, content_type="image/jpeg"): """设置运动员照片""" conn = get_db() conn.execute(""" INSERT INTO athlete_photos (name, photo_data, content_type, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(name) DO UPDATE SET photo_data=?, content_type=?, updated_at=? """, (name, photo_data, content_type, datetime.now().isoformat(), photo_data, content_type, datetime.now().isoformat())) conn.commit() def delete_athlete_photo(name): """删除运动员照片""" conn = get_db() conn.execute("DELETE FROM athlete_photos WHERE name = ?", (name,)) conn.commit() def get_training_types(): """获取所有训练类型""" conn = get_db() rows = conn.execute("SELECT * FROM training_types ORDER BY id").fetchall() return [dict(r) for r in rows] def add_training_type(name, description="", is_class=0): """添加训练类型""" conn = get_db() try: conn.execute("INSERT INTO training_types (name, description, is_class) VALUES (?, ?, ?)", (name, description, is_class)) conn.commit() return True except Exception: return False finally: pass def delete_training_type(type_id): """删除训练类型""" conn = get_db() conn.execute("DELETE FROM training_types WHERE id = ?", (type_id,)) conn.commit() def get_training_records(athlete_name=None): """获取训练记录，支持昵称查询，返回real_name字段""" conn = get_db() nickname_map = get_nickname_map() if athlete_name: # 获取该队员的所有昵称 nick_rows = conn.execute("SELECT nickname FROM athlete_nicknames WHERE real_name = ?", (athlete_name,)).fetchall() all_names = [athlete_name] + [r["nickname"] for r in nick_rows] placeholders = ','.join(['?'] * len(all_names)) rows = conn.execute(f""" SELECT tr.*, tt.name as type_name, tt.is_class FROM training_records tr JOIN training_types tt ON tr.training_type_id = tt.id WHERE tr.athlete_name IN ({placeholders}) ORDER BY tr.training_date DESC """, list(all_names)).fetchall() else: rows = conn.execute(""" SELECT tr.*, tt.name as type_name, tt.is_class FROM training_records tr JOIN training_types tt ON tr.training_type_id = tt.id ORDER BY tr.training_date DESC """).fetchall() result = [] for r in rows: d = dict(r) # 附加真名（解析昵称） d['real_name'] = nickname_map.get(d['athlete_name'], d['athlete_name']) result.append(d) return result def get_athlete_attendance_detail(athlete_name, start_date=None, end_date=None): """获取单个队员在指定时间段的出勤明细记录，昵称合并""" conn = get_db() # 获取该队员的所有昵称 nick_rows = conn.execute("SELECT nickname FROM athlete_nicknames WHERE real_name = ?", (athlete_name,)).fetchall() all_names = [athlete_name] + [r["nickname"] for r in nick_rows] placeholders = ','.join(['?'] * len(all_names)) query = f""" SELECT tr.id, tr.training_date, tr.training_type_id, tr.duration_minutes, tr.notes, tt.name as type_name, tt.is_class FROM training_records tr JOIN training_types tt ON tr.training_type_id = tt.id WHERE tr.athlete_name IN ({placeholders}) """ params = list(all_names) if start_date: query += " AND tr.training_date >= ?" params.append(start_date) if end_date: query += " AND tr.training_date <= ?" params.append(end_date) query += " ORDER BY tr.training_date ASC, tt.is_class DESC" rows = conn.execute(query, params).fetchall() return [dict(r) for r in rows] def add_training_record(athlete_name, training_type_id, training_date, duration_minutes=None, notes=""): """添加训练记录""" conn = get_db() conn.execute(""" INSERT INTO training_records (athlete_name, training_type_id, training_date, duration_minutes, notes, created_at) VALUES (?, ?, ?, ?, ?, ?) """, (athlete_name, training_type_id, training_date, duration_minutes, notes, datetime.now().isoformat())) conn.commit() def delete_training_record(record_id): """删除训练记录""" conn = get_db() conn.execute("DELETE FROM training_records WHERE id = ?", (record_id,)) conn.commit() # ========== 出勤统计 ========== def get_attendance_stats(start_date=None, end_date=None, athlete_name=None, active_only=True): """获取出勤统计，按队员+大课/小课分组，昵称合并到真名 Args: active_only: 只统计最近半年还活跃的队员（默认True） 返回: [ { "athlete_name": "张三", "class_count": 10, # 大课次数 "class_duration": 900, # 大课总时长(分钟) "custom_count": 5, # 小课次数 "custom_duration": 450, # 小课总时长(分钟) "total_count": 15, # 总次数 "total_duration": 1350 # 总时长(分钟) }, ... ] """ conn = get_db() # 获取活跃队员列表 active_names = None if active_only: active_names = get_active_athlete_names() query = """ SELECT COALESCE(an.real_name, tr.athlete_name) as athlete_name, SUM(CASE WHEN tt.is_class = 1 THEN 1 ELSE 0 END) as class_count, SUM(CASE WHEN tt.is_class = 1 THEN COALESCE(tr.duration_minutes, 0) ELSE 0 END) as class_duration, SUM(CASE WHEN tt.is_class = 0 OR tt.is_class IS NULL THEN 1 ELSE 0 END) as custom_count, SUM(CASE WHEN tt.is_class = 0 OR tt.is_class IS NULL THEN COALESCE(tr.duration_minutes, 0) ELSE 0 END) as custom_duration, COUNT(*) as total_count, SUM(COALESCE(tr.duration_minutes, 0)) as total_duration FROM training_records tr JOIN training_types tt ON tr.training_type_id = tt.id LEFT JOIN athlete_nicknames an ON tr.athlete_name = an.nickname WHERE 1=1 """ params = [] if start_date: query += " AND tr.training_date >= ?" params.append(start_date) if end_date: query += " AND tr.training_date <= ?" params.append(end_date) if athlete_name: query += " AND (COALESCE(an.real_name, tr.athlete_name) = ? OR tr.athlete_name = ?)" params.append(athlete_name) params.append(athlete_name) query += " GROUP BY COALESCE(an.real_name, tr.athlete_name) ORDER BY total_count DESC" rows = conn.execute(query, params).fetchall() result = [dict(r) for r in rows] # 过滤非活跃队员 if active_only and active_names is not None: result = [r for r in result if r['athlete_name'] in active_names] return result def get_attendance_detail(athlete_name, start_date=None, end_date=None): """获取某个队员的出勤明细，按训练类型分组，昵称合并 返回: [ { "training_type_id": 1, "type_name": "区队大课", "is_class": 1, "count": 10, "total_duration": 900, "dates": ["2026-04-01", "2026-04-02", ...] }, ... ] """ conn = get_db() # 获取该队员的所有昵称 nick_rows = conn.execute("SELECT nickname FROM athlete_nicknames WHERE real_name = ?", (athlete_name,)).fetchall() all_names = [athlete_name] + [r["nickname"] for r in nick_rows] placeholders = ','.join(['?'] * len(all_names)) query = f""" SELECT tt.id as training_type_id, tt.name as type_name, tt.is_class, COUNT(*) as count, SUM(COALESCE(tr.duration_minutes, 0)) as total_duration FROM training_records tr JOIN training_types tt ON tr.training_type_id = tt.id WHERE tr.athlete_name IN ({placeholders}) """ params = list(all_names) if start_date: query += " AND tr.training_date >= ?" params.append(start_date) if end_date: query += " AND tr.training_date <= ?" params.append(end_date) query += " GROUP BY tt.id ORDER BY tt.is_class DESC, tt.name" rows = conn.execute(query, params).fetchall() result = [] for r in rows: d = dict(r) # 获取具体日期列表 date_query = f""" SELECT tr.training_date FROM training_records tr WHERE tr.athlete_name IN ({placeholders}) AND tr.training_type_id = ? """ date_params = list(all_names) + [r["training_type_id"]] if start_date: date_query += " AND tr.training_date >= ?" date_params.append(start_date) if end_date: date_query += " AND tr.training_date <= ?" date_params.append(end_date) date_query += " ORDER BY tr.training_date" dates = conn.execute(date_query, date_params).fetchall() d["dates"] = [row["training_date"] for row in dates] result.append(d) return result # ========== 费用配置 ========== def get_fee_config(): """获取所有训练类型的费用配置""" conn = get_db() rows = conn.execute(""" SELECT tt.id as training_type_id, tt.name as type_name, tt.is_class, COALESCE(fc.price_per_session, 0) as price_per_session, fc.updated_at FROM training_types tt LEFT JOIN fee_config fc ON tt.id = fc.training_type_id ORDER BY tt.is_class DESC, tt.name """).fetchall() return [dict(r) for r in rows] def set_fee_config(training_type_id, price_per_session): """设置训练类型的单价""" conn = get_db() conn.execute(""" INSERT INTO fee_config (training_type_id, price_per_session, updated_at) VALUES (?, ?, ?) ON CONFLICT(training_type_id) DO UPDATE SET price_per_session=?, updated_at=? """, (training_type_id, price_per_session, datetime.now().isoformat(), price_per_session, datetime.now().isoformat())) conn.commit() def get_athlete_exemptions(athlete_name=None): """获取队员的费用豁免配置""" conn = get_db() if athlete_name: rows = conn.execute(""" SELECT afe.athlete_name, afe.training_type_id, afe.is_exempt, afe.updated_at, tt.name as type_name, tt.is_class FROM athlete_fee_exemptions afe JOIN training_types tt ON afe.training_type_id = tt.id WHERE afe.athlete_name = ? ORDER BY tt.is_class DESC, tt.name """, (athlete_name,)).fetchall() else: rows = conn.execute(""" SELECT afe.athlete_name, afe.training_type_id, afe.is_exempt, afe.updated_at, tt.name as type_name, tt.is_class FROM athlete_fee_exemptions afe JOIN training_types tt ON afe.training_type_id = tt.id ORDER BY afe.athlete_name, tt.is_class DESC, tt.name """).fetchall() return [dict(r) for r in rows] def set_athlete_exemption(athlete_name, training_type_id, is_exempt=True): """设置队员某训练类型的费用豁免""" conn = get_db() conn.execute(""" INSERT INTO athlete_fee_exemptions (athlete_name, training_type_id, is_exempt, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(athlete_name, training_type_id) DO UPDATE SET is_exempt=?, updated_at=? """, (athlete_name, training_type_id, 1 if is_exempt else 0, datetime.now().isoformat(), 1 if is_exempt else 0, datetime.now().isoformat())) conn.commit() def delete_athlete_exemption(athlete_name, training_type_id): """删除队员某训练类型的费用豁免""" conn = get_db() conn.execute( "DELETE FROM athlete_fee_exemptions WHERE athlete_name = ? AND training_type_id = ?", (athlete_name, training_type_id) ) conn.commit() def calculate_fees(start_date=None, end_date=None, athlete_name=None, active_only=True): """计算费用，根据出勤 × 单价 - 豁免 - 结算，昵称合并到真名 Args: active_only: 只统计最近半年还活跃的队员（默认True） 费用计算规则: 1. 每个队员每个训练类型的单价：优先使用athlete_fee_config，否则使用全局fee_config 2. 预付费大课：从start_date开始，每次上课扣一次费，从prepaid_balance扣减 3. 小课：按次收费，直接计算次数×单价 4. 豁免：免费 5. 已结算的费用不再重复计算 返回: [ { "athlete_name": "张三", "items": [ { "training_type_id": 1, "type_name": "区队大课", "is_class": 1, "count": 10, "price_per_session": 100, "is_exempt": 0, "is_prepaid": 1, "prepaid_deducted": 10, "settled_count": 3, "unsettled_count": 7, "subtotal": 700, "settled_amount": 300 }, ... ], "total_fee": 1000, "total_settled": 300, "total_unsettled": 700 } ] """ conn = get_db() # 获取出勤明细，昵称合并到真名 query = """ SELECT COALESCE(an.real_name, tr.athlete_name) as athlete_name, tt.id as training_type_id, tt.name as type_name, tt.is_class, COUNT(*) as count FROM training_records tr JOIN training_types tt ON tr.training_type_id = tt.id LEFT JOIN athlete_nicknames an ON tr.athlete_name = an.nickname WHERE 1=1 """ params = [] if start_date: query += " AND tr.training_date >= ?" params.append(start_date) if end_date: query += " AND tr.training_date <= ?" params.append(end_date) if athlete_name: query += " AND (COALESCE(an.real_name, tr.athlete_name) = ? OR tr.athlete_name = ?)" params.append(athlete_name) params.append(athlete_name) query += " GROUP BY COALESCE(an.real_name, tr.athlete_name), tt.id ORDER BY COALESCE(an.real_name, tr.athlete_name), tt.is_class DESC, tt.name" rows = conn.execute(query, params).fetchall() # 获取全局费用配置 fee_rows = conn.execute("SELECT training_type_id, price_per_session FROM fee_config").fetchall() fee_map = {r["training_type_id"]: r["price_per_session"] for r in fee_rows} # 获取个性化费用配置 athlete_fee_rows = conn.execute("SELECT athlete_name, training_type_id, price_per_session FROM athlete_fee_config").fetchall() athlete_fee_map = {} for r in athlete_fee_rows: key = (r["athlete_name"], r["training_type_id"]) athlete_fee_map[key] = r["price_per_session"] # 获取豁免配置 exempt_rows = conn.execute("SELECT athlete_name, training_type_id, is_exempt FROM athlete_fee_exemptions").fetchall() exempt_map = {} for r in exempt_rows: key = (r["athlete_name"], r["training_type_id"]) exempt_map[key] = r["is_exempt"] # 获取预付费配置 prepaid_rows = conn.execute("SELECT athlete_name, training_type_id, balance, start_date FROM prepaid_balance").fetchall() prepaid_map = {} for r in prepaid_rows: key = (r["athlete_name"], r["training_type_id"]) prepaid_map[key] = {"balance": r["balance"], "start_date": r["start_date"]} # 获取已结算记录 settled_rows = conn.execute(""" SELECT athlete_name, training_type_id, start_date, end_date, amount FROM fee_settlements """).fetchall() # 按队员分组 athlete_map = {} for r in rows: name = r["athlete_name"] if name not in athlete_map: athlete_map[name] = [] tid = r["training_type_id"] # 单价：优先使用个性化配置 price = athlete_fee_map.get((name, tid), fee_map.get(tid, 0)) is_exempt = exempt_map.get((name, tid), 0) # 预付费 prepaid_info = prepaid_map.get((name, tid)) is_prepaid = 1 if prepaid_info else 0 prepaid_deducted = 0 if is_prepaid and not is_exempt: # 预付费：从start_date开始，每次上课扣一次 prepaid_start = prepaid_info["start_date"] prepaid_balance = prepaid_info["balance"] # 计算从start_date起的上课次数 # 简化：假设所有count都扣预付费 prepaid_deducted = min(r["count"], int(prepaid_balance / price)) if price > 0 else 0 # 计算已结算的费用 settled_amount = 0 settled_count = 0 for sr in settled_rows: if sr["athlete_name"] == name and sr["training_type_id"] == tid: if start_date and sr["end_date"] < start_date: continue if end_date and sr["start_date"] > end_date: continue # 结算区间与查询区间有重叠 settled_amount += sr["amount"] # 估算已结算的次数 if price > 0: settled_count += int(sr["amount"] / price) if is_exempt: subtotal = 0 else: unsettled_count = max(0, r["count"] - settled_count - prepaid_deducted) subtotal = unsettled_count * price athlete_map[name].append({ "training_type_id": tid, "type_name": r["type_name"], "is_class": r["is_class"], "count": r["count"], "price_per_session": price, "is_exempt": is_exempt, "is_prepaid": is_prepaid, "prepaid_deducted": prepaid_deducted, "settled_count": settled_count, "unsettled_count": max(0, r["count"] - settled_count), "subtotal": subtotal, "settled_amount": settled_amount, }) # 计算每人的总费用 # 获取活跃队员列表 active_names = get_active_athlete_names() if active_only else None result = [] for name, items in athlete_map.items(): # 过滤非活跃队员 if active_only and active_names is not None and name not in active_names: continue total_fee = sum(item["subtotal"] for item in items) total_settled = sum(item["settled_amount"] for item in items) total_unsettled = total_fee result.append({ "athlete_name": name, "items": items, "total_fee": total_fee, "total_settled": total_settled, "total_unsettled": total_unsettled, }) result.sort(key=lambda x: x["total_fee"], reverse=True) return result # ========== 留言板 ========== def get_messages(limit=50): """获取留言列表""" conn = get_db() rows = conn.execute( "SELECT id, username, content, image_data, likes, created_at, reply_to_id FROM messages ORDER BY created_at DESC LIMIT ?", (limit,) ).fetchall() return [dict(r) for r in rows] def add_message(username, content, image_data=None, reply_to_id=None): """添加留言""" conn = get_db() conn.execute("INSERT INTO messages (username, content, image_data, likes, created_at, reply_to_id) VALUES (?, ?, ?, 0, ?, ?)", (username, content, image_data, datetime.now().isoformat(), reply_to_id)) conn.commit() def like_message(msg_id, user_id): """给留言点赞（取消赞），返回点赞数和是否已赞""" conn = get_db() # 检查是否已赞 existing = conn.execute( "SELECT 1 FROM message_likes WHERE message_id = ? AND user_id = ?", (msg_id, user_id) ).fetchone() if existing: # 取消赞 conn.execute("DELETE FROM message_likes WHERE message_id = ? AND user_id = ?", (msg_id, user_id)) conn.execute("UPDATE messages SET likes = MAX(0, likes - 1) WHERE id = ?", (msg_id,)) liked = False else: # 点赞 conn.execute("INSERT INTO message_likes (message_id, user_id, created_at) VALUES (?, ?, ?)", (msg_id, user_id, datetime.now().isoformat())) conn.execute("UPDATE messages SET likes = likes + 1 WHERE id = ?", (msg_id,)) liked = True conn.commit() row = conn.execute("SELECT likes FROM messages WHERE id = ?", (msg_id,)).fetchone() return row["likes"] if row else 0, liked def get_message_likes_by_user(user_id, message_ids): """获取用户对指定留言的点赞状态""" if not message_ids: return set() conn = get_db() placeholders = ','.join('?' * len(message_ids)) rows = conn.execute( f"SELECT message_id FROM message_likes WHERE user_id = ? AND message_id IN ({placeholders})", [user_id] + list(message_ids) ).fetchall() return {row["message_id"] for row in rows} def delete_message(msg_id): """删除留言""" conn = get_db() conn.execute("DELETE FROM messages WHERE id = ?", (msg_id,)) conn.commit() # ========== 家属管理 ========== def set_athlete_family(athlete_name, user_id, relationship="", approved=True): """设置队员的指定家属（每个队员只能有一个指定家属）""" conn = get_db() try: conn.execute(""" INSERT INTO athlete_family (athlete_name, user_id, relationship, approved, created_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(athlete_name) DO UPDATE SET user_id=?, relationship=?, approved=? """, (athlete_name, user_id, relationship, 1 if approved else 0, datetime.now().isoformat(), user_id, relationship, 1 if approved else 0)) conn.commit() return True except Exception: return False finally: pass def delete_athlete_family(athlete_name): """删除队员的指定家属""" conn = get_db() conn.execute("DELETE FROM athlete_family WHERE athlete_name = ?", (athlete_name,)) conn.commit() def get_athlete_family(athlete_name): """获取队员的指定家属""" conn = get_db() row = conn.execute(""" SELECT af.*, u.username FROM athlete_family af JOIN users u ON af.user_id = u.id WHERE af.athlete_name = ? """, (athlete_name,)).fetchone() return dict(row) if row else None def get_user_family(user_id): """获取用户关联的队员信息（优先从athlete_family表，回退到users表）""" conn = get_db() row = conn.execute(""" SELECT af.athlete_name, af.relationship, af.approved FROM athlete_family af WHERE af.user_id = ? """, (user_id,)).fetchone() if row: return dict(row) # 回退：从users表获取家庭关联信息，并自动补录到athlete_family表 user = conn.execute("SELECT athlete_name, relationship, approved FROM users WHERE id = ?", (user_id,)).fetchone() if user and user['athlete_name']: try: conn.execute("INSERT OR IGNORE INTO athlete_family (athlete_name, user_id, relationship, approved, created_at) VALUES (?, ?, ?, ?, ?)", (user['athlete_name'], user_id, user['relationship'], user['approved'], datetime.now().isoformat())) conn.commit() result = {"athlete_name": user['athlete_name'], "relationship": user['relationship'], "approved": user['approved']} except Exception: result = {"athlete_name": user['athlete_name'], "relationship": user['relationship'], "approved": user['approved']} else: result = None return result def get_all_families(): """获取所有家属关联""" conn = get_db() rows = conn.execute(""" SELECT af.athlete_name, af.user_id, af.relationship, af.approved, af.created_at, u.username FROM athlete_family af JOIN users u ON af.user_id = u.id ORDER BY af.athlete_name """).fetchall() return [dict(r) for r in rows] def is_family_of(user_id, athlete_name): """判断用户是否是指定队员的授权家属""" conn = get_db() row = conn.execute( "SELECT approved FROM athlete_family WHERE user_id = ? AND athlete_name = ? AND approved = 1", (user_id, athlete_name) ).fetchone() return row is not None # ========== 访问统计 ========== def log_visit(ip, user_agent, page="/"): """记录访问""" conn = get_db() conn.execute("INSERT INTO visit_logs (ip, user_agent, page, visited_at) VALUES (?, ?, ?, ?)", (ip, user_agent, page, datetime.now().isoformat())) conn.commit() def get_visit_stats(): """获取访问统计""" conn = get_db() try: # 检查 visit_logs 表是否存在 table_exists = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='visit_logs'").fetchone() if not table_exists: return {"total": 0, "today": 0, "unique_visitors": 0, "daily": [], "monthly": []} # 总访问量 total = conn.execute("SELECT COUNT(*) as cnt FROM visit_logs").fetchone()["cnt"] # 今日访问量 today = datetime.now().strftime("%Y-%m-%d") today_count = conn.execute( "SELECT COUNT(*) as cnt FROM visit_logs WHERE visited_at LIKE ?", (f'{today}%',) ).fetchone()["cnt"] # 最近7天每日访问量 daily = conn.execute(""" SELECT DATE(visited_at) as date, COUNT(*) as count FROM visit_logs WHERE visited_at >= DATE('now', '-7 days') GROUP BY DATE(visited_at) ORDER BY date """).fetchall() # 最近30天每日访问量（用于图表） monthly = conn.execute(""" SELECT DATE(visited_at) as date, COUNT(*) as count FROM visit_logs WHERE visited_at >= DATE('now', '-30 days') GROUP BY DATE(visited_at) ORDER BY date """).fetchall() # 唯一IP数 unique_ips = conn.execute("SELECT COUNT(DISTINCT ip) as cnt FROM visit_logs").fetchone()["cnt"] return { "total": total, "today": today_count, "unique_visitors": unique_ips, "daily": [dict(r) for r in daily], "monthly": [dict(r) for r in monthly], } finally: pass def recalculate_chaoyang_flags(team_id=None): """根据配置表和覆盖规则重新计算所有记录的 is_chaoyang 标记""" conn = get_db() if team_id is None: row = conn.execute("SELECT value FROM app_settings WHERE key = 'active_team'").fetchone() team_id = row["value"] if row else "chaoyang" # 获取当前队伍的队名别名 aliases = [row["alias"] for row in conn.execute( "SELECT alias FROM team_aliases WHERE is_active = 1 AND (team_id = ? OR team_id IS NULL)", (team_id,) ).fetchall()] # 获取当前队伍的手动覆盖 overrides = {} for row in conn.execute( "SELECT name, is_chaoyang FROM athlete_overrides WHERE team_id = ? OR team_id IS NULL", (team_id,) ).fetchall(): overrides[row["name"]] = row["is_chaoyang"] # 先清零所有 is_chaoyang conn.execute("UPDATE results SET is_chaoyang = 0") # 按队名自动识别（匹配 team 列） for alias in aliases: conn.execute("UPDATE results SET is_chaoyang = 1 WHERE team LIKE ?", (f'%{alias}%',)) # 同时匹配 name 列（接力队名可能在 name 里，team 列是数据源名称） for alias in aliases: conn.execute("UPDATE results SET is_chaoyang = 1 WHERE name LIKE ? AND is_chaoyang = 0", (f'%{alias}%',)) # 排除长春市朝阳区（不是北京朝阳） if team_id == 'chaoyang': conn.execute("UPDATE results SET is_chaoyang = 0 WHERE team LIKE '%长春%朝阳%'") # 应用手动覆盖（添加或移除） for name, flag in overrides.items(): conn.execute("UPDATE results SET is_chaoyang = ? WHERE name = ?", (flag, name)) conn.commit() # ========== 多队伍管理 ========== def get_team_config(team_id): """根据 team_id 获取队伍配置""" conn = get_db() if not team_id: team_id = 'chaoyang' team = conn.execute("SELECT * FROM team_configs WHERE team_id = ?", (team_id,)).fetchone() return dict(team) if team else {"team_id": "chaoyang", "team_name": "朝阳", "color_primary": "#0F4C81", "color_accent": "#4A90D9", "color_light": "#E8F0FE"} def get_active_team(): """获取当前活跃队伍的完整配置""" conn = get_db() row = conn.execute("SELECT value FROM app_settings WHERE key = 'active_team'").fetchone() team_id = row["value"] if row else "chaoyang" return get_team_config(team_id) def get_all_teams(): """获取所有队伍配置""" conn = get_db() return [dict(r) for r in conn.execute("SELECT * FROM team_configs ORDER BY id").fetchall()] def get_default_issued_team_id(): """获取当前发行环境默认队伍""" return os.environ.get('ISSUED_TEAM_ID', 'chaoyang').strip() or 'chaoyang' def set_active_team(team_id): """切换活跃队伍并重新计算标记""" conn = get_db() conn.execute("INSERT OR REPLACE INTO app_settings (key, value) VALUES ('active_team', ?)", (team_id,)) conn.commit() recalculate_chaoyang_flags(team_id) return get_active_team() # ========== 运动员荣誉 ========== def get_athlete_honors(athlete_name=None): """获取运动员荣誉列表""" conn = get_db() if athlete_name: rows = conn.execute( "SELECT * FROM athlete_honors WHERE athlete_name = ? ORDER BY awarded_date DESC, id DESC", (athlete_name,) ).fetchall() else: rows = conn.execute("SELECT * FROM athlete_honors ORDER BY athlete_name, awarded_date DESC").fetchall() return [dict(r) for r in rows] def add_athlete_honor(athlete_name, title, awarded_date=None, source=None): """添加运动员荣誉""" conn = get_db() try: conn.execute( "INSERT INTO athlete_honors (athlete_name, title, awarded_date, source, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)", (athlete_name, title, awarded_date, source, datetime.now().isoformat(), datetime.now().isoformat()) ) conn.commit() return True except Exception: return False finally: pass def update_athlete_honor(honor_id, title=None, awarded_date=None, source=None): """更新运动员荣誉""" conn = get_db() updates = [] params = [] if title is not None: updates.append("title = ?") params.append(title) if awarded_date is not None: updates.append("awarded_date = ?") params.append(awarded_date) if source is not None: updates.append("source = ?") params.append(source) if not updates: return updates.append("updated_at = ?") params.append(datetime.now().isoformat()) params.append(honor_id) conn.execute(f"UPDATE athlete_honors SET {', '.join(updates)} WHERE id = ?", params) conn.commit() def delete_athlete_honor(honor_id): """删除运动员荣誉""" conn = get_db() conn.execute("DELETE FROM athlete_honors WHERE id = ?", (honor_id,)) conn.commit() # ========== 运动员个人资料 ========== def get_athlete_profile(athlete_name): """获取运动员个人资料""" conn = get_db() row = conn.execute("SELECT * FROM athlete_profiles WHERE athlete_name = ?", (athlete_name,)).fetchone() return dict(row) if row else None def update_athlete_profile(athlete_name, bio, updated_by=None): """更新运动员个人资料（简介）""" conn = get_db() conn.execute(""" INSERT INTO athlete_profiles (athlete_name, bio, updated_at, updated_by) VALUES (?, ?, ?, ?) ON CONFLICT(athlete_name) DO UPDATE SET bio=?, updated_at=?, updated_by=? """, (athlete_name, bio, datetime.now().isoformat(), updated_by, bio, datetime.now().isoformat(), updated_by)) conn.commit() def get_athlete_custom_results(athlete_name): """获取运动员自定义成绩""" conn = get_db() rows = conn.execute( "SELECT * FROM athlete_custom_results WHERE athlete_name = ? ORDER BY competition_date DESC, id DESC", (athlete_name,) ).fetchall() return [dict(r) for r in rows] def add_athlete_custom_result(athlete_name, competition_date, competition_name, result_text, created_by=None, event=None, round_type=None, team=None, time_result=None, place=None, season=None): """添加运动员自定义成绩""" conn = get_db() conn.execute(""" INSERT INTO athlete_custom_results (athlete_name, competition_date, competition_name, result_text, event, round_type, team, time_result, place, season, created_at, created_by) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (athlete_name, competition_date, competition_name, result_text, event, round_type, team, time_result, place, season, datetime.now().isoformat(), created_by)) conn.commit() return True def delete_athlete_custom_result(result_id, athlete_name=None): """删除运动员自定义成绩""" conn = get_db() if athlete_name: conn.execute("DELETE FROM athlete_custom_results WHERE id = ? AND athlete_name = ?", (result_id, athlete_name)) else: conn.execute("DELETE FROM athlete_custom_results WHERE id = ?", (result_id,)) conn.commit() def update_athlete_custom_result(result_id, athlete_name, competition_date=None, competition_name=None, result_text=None, event=None, round_type=None, team=None, time_result=None, place=None, season=None): """更新运动员自定义成绩""" conn = get_db() updates = [] params = [] if competition_date is not None: updates.append("competition_date = ?") params.append(competition_date) if competition_name is not None: updates.append("competition_name = ?") params.append(competition_name) if result_text is not None: updates.append("result_text = ?") params.append(result_text) if event is not None: updates.append("event = ?") params.append(event) if round_type is not None: updates.append("round_type = ?") params.append(round_type) if team is not None: updates.append("team = ?") params.append(team) if time_result is not None: updates.append("time_result = ?") params.append(time_result) if place is not None: updates.append("place = ?") params.append(place) if season is not None: updates.append("season = ?") params.append(season) if not updates: return False params.extend([result_id, athlete_name]) conn.execute(f"UPDATE athlete_custom_results SET {', '.join(updates)} WHERE id = ? AND athlete_name = ?", params) conn.commit() return True # ========== 接力队员关联 ========== def get_relay_members(relay_team_name=None, athlete_name=None): """获取接力队员关联""" conn = get_db() if relay_team_name: rows = conn.execute("SELECT * FROM relay_members WHERE relay_team_name = ? ORDER BY id", (relay_team_name,)).fetchall() elif athlete_name: rows = conn.execute("SELECT * FROM relay_members WHERE athlete_name = ? ORDER BY id", (athlete_name,)).fetchall() else: rows = conn.execute("SELECT * FROM relay_members ORDER BY relay_team_name, id").fetchall() return [dict(r) for r in rows] def set_relay_members(relay_team_name, athlete_names, season=None, event=None): """设置接力队的队员列表（替换模式）""" conn = get_db() # 删除旧的 if season and event: conn.execute("DELETE FROM relay_members WHERE relay_team_name = ? AND season = ? AND event = ?", (relay_team_name, season, event)) elif season: conn.execute("DELETE FROM relay_members WHERE relay_team_name = ? AND season = ?", (relay_team_name, season)) else: conn.execute("DELETE FROM relay_members WHERE relay_team_name = ?", (relay_team_name,)) # 插入新的 now = datetime.now().isoformat() for name in athlete_names: conn.execute("""INSERT OR IGNORE INTO relay_members (relay_team_name, athlete_name, season, event, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)""", (relay_team_name, name, season, event, now, now)) conn.commit() return True def delete_relay_member(member_id): """删除单条接力队员关联""" conn = get_db() conn.execute("DELETE FROM relay_members WHERE id = ?", (member_id,)) conn.commit() return True def auto_populate_relay_members(): """根据同一比赛(m.id)同一场接力赛的参赛队员自动推断接力队员关联 只匹配同一比赛中、同样category+gender的个人运动员（有非接力项目成绩）， 并且要求该运动员在同比赛中有非接力成绩记录（证明确实参加了该比赛）。 注意：此方法为估算，可能不准确。建议管理员手动确认。 """ conn = get_db() now = datetime.now().isoformat() # 先清除旧的自动填充数据（保留手动添加的——有 updated_at 不同的） conn.execute("DELETE FROM relay_members WHERE created_at = updated_at") inserted = 0 # 获取所有朝阳接力队名及其比赛信息 relay_teams = conn.execute(""" SELECT DISTINCT r.name as relay_team_name, rg.category, rg.gender, rg.event, rg.race_date, m.id as match_id, m.season FROM results r JOIN race_groups rg ON r.rcbh = rg.rcbh JOIN matches m ON rg.match_id = m.id WHERE rg.event LIKE '%接力%' AND r.is_chaoyang = 1 """).fetchall() for rt in relay_teams: team_name = rt["relay_team_name"] match_id = rt["match_id"] event = rt["event"] season = rt["season"] # 只查找同一比赛中有非接力项目成绩的运动员 athletes = conn.execute(""" SELECT DISTINCT r2.name FROM results r2 JOIN race_groups rg2 ON r2.rcbh = rg2.rcbh WHERE rg2.match_id = ? AND rg2.category = ? AND rg2.gender = ? AND r2.is_chaoyang = 1 AND length(r2.name) <= 4 AND r2.name NOT IN ('朝阳区', '乙组朝阳区', '北京市朝阳区', '集体', '混合接力', '混合', '接力队') AND rg2.event NOT LIKE '%接力%' ORDER BY r2.name """, (match_id, rt["category"], rt["gender"])).fetchall() for a in athletes: try: conn.execute("""INSERT OR IGNORE INTO relay_members (relay_team_name, athlete_name, season, event, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)""", (team_name, a["name"], season, event, now, now)) inserted += 1 except Exception: pass conn.commit() return inserted def get_athlete_relay_results(athlete_name): """获取运动员关联的接力成绩（通过 relay_members 表）""" conn = get_db() rows = conn.execute(""" SELECT r.name as relay_team_name, r.team, r.place, r.lane, r.helmet, r.time_result, r.qual, r.finished, rg.event, rg.category, rg.gender, rg.round_type, rg.race_date, rg.race_time, m.title as match_title, m.season, m.match_type, m.round_info, m.date_range, m.venue, m.status, 'relay' as result_type 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 rm.athlete_name = ? AND (rm.season IS NULL OR rm.season = m.season) AND (rm.event IS NULL OR rm.event = rg.event) ORDER BY m.season DESC, rg.race_date DESC, rg.race_time DESC, rg.event, rg.round_type """, (athlete_name,)).fetchall() return [dict(r) for r in rows] # ========== 每周训练计划 ========== DAY_NAMES = {1: '周一', 2: '周二', 3: '周三', 4: '周四', 5: '周五', 6: '周六', 7: '周日'} def get_weekly_schedule(): """获取每周训练计划""" conn = get_db() rows = conn.execute(""" SELECT ws.id, ws.day_of_week, ws.training_type_id, tt.name as type_name, tt.is_class FROM weekly_training_schedule ws JOIN training_types tt ON ws.training_type_id = tt.id ORDER BY ws.day_of_week, tt.is_class DESC, tt.name """).fetchall() result = [] for r in rows: d = dict(r) d['day_name'] = DAY_NAMES.get(r['day_of_week'], '') result.append(d) return result def set_weekly_schedule(day_of_week, training_type_id): """设置某天的训练类型""" if day_of_week < 1 or day_of_week > 7: return False conn = get_db() try: conn.execute(""" INSERT INTO weekly_training_schedule (day_of_week, training_type_id) VALUES (?, ?) ON CONFLICT(day_of_week, training_type_id) DO UPDATE SET training_type_id = excluded.training_type_id """, (day_of_week, training_type_id)) conn.commit() return True except Exception: return False finally: pass def delete_weekly_schedule(schedule_id): """删除周训练计划条目""" conn = get_db() conn.execute("DELETE FROM weekly_training_schedule WHERE id = ?", (schedule_id,)) conn.commit() def get_schedule_for_date(date_str): """根据日期获取当天的训练类型（根据星期几）""" if not date_str: return [] from datetime import datetime as dt try: d = dt.strptime(date_str, '%Y-%m-%d') day_of_week = d.isoweekday() # 1=Monday, 7=Sunday except ValueError: return [] conn = get_db() rows = conn.execute(""" SELECT ws.id, ws.day_of_week, ws.training_type_id, tt.name as type_name, tt.is_class FROM weekly_training_schedule ws JOIN training_types tt ON ws.training_type_id = tt.id WHERE ws.day_of_week = ? ORDER BY tt.is_class DESC, tt.name """, (day_of_week,)).fetchall() result = [] for r in rows: d = dict(r) d['day_name'] = DAY_NAMES.get(r['day_of_week'], '') result.append(d) return result def add_training_records_batch(records): """批量添加训练记录 records: list of {athlete_name, training_type_id, training_date, duration_minutes, notes} """ conn = get_db() now = datetime.now().isoformat() for r in records: conn.execute(""" INSERT INTO training_records (athlete_name, training_type_id, training_date, duration_minutes, notes, created_at) VALUES (?, ?, ?, ?, ?, ?) """, (r['athlete_name'], r['training_type_id'], r['training_date'], r.get('duration_minutes'), r.get('notes', ''), now)) conn.commit() return len(records) def update_training_record(record_id, training_type_id=None, training_date=None, duration_minutes=None, notes=None): """直接更新训练记录（管理员使用）""" conn = get_db() sets = [] params = [] if training_type_id is not None: sets.append("training_type_id = ?") params.append(training_type_id) if training_date is not None: sets.append("training_date = ?") params.append(training_date) if duration_minutes is not None: sets.append("duration_minutes = ?") params.append(duration_minutes) if notes is not None: sets.append("notes = ?") params.append(notes) if not sets: return False params.append(record_id) conn.execute(f"UPDATE training_records SET {', '.join(sets)} WHERE id = ?", params) conn.commit() return True def submit_training_change(record_id, athlete_name, changes, submitted_by): """提交训练记录修改申请（普通用户使用，需审批）""" import json conn = get_db() now = datetime.now().isoformat() conn.execute(""" INSERT INTO training_pending_changes (record_id, athlete_name, changes, submitted_by, submitted_at, status) VALUES (?, ?, ?, ?, ?, 'pending') """, (record_id, athlete_name, json.dumps(changes, ensure_ascii=False), submitted_by, now)) conn.commit() return True def get_pending_changes(): """获取所有待审批的训练记录修改""" conn = get_db() rows = conn.execute(""" SELECT pc.*, u.username as submitted_by_name FROM training_pending_changes pc LEFT JOIN users u ON pc.submitted_by = u.id WHERE pc.status = 'pending' ORDER BY pc.submitted_at ASC """).fetchall() return [dict(r) for r in rows] def approve_pending_change(change_id, reviewer_id): """批准待审批修改，将变更应用到训练记录""" import json conn = get_db() row = conn.execute("SELECT * FROM training_pending_changes WHERE id = ? AND status = 'pending'", (change_id,)).fetchone() if not row: return False changes = json.loads(row['changes']) record_id = row['record_id'] # Apply changes to training_records sets = [] params = [] for k, v in changes.items(): if k in ('training_type_id', 'training_date', 'duration_minutes', 'notes'): sets.append(f"{k} = ?") params.append(v) if sets: params.append(record_id) conn.execute(f"UPDATE training_records SET {', '.join(sets)} WHERE id = ?", params) # Mark as approved now = datetime.now().isoformat() conn.execute("UPDATE training_pending_changes SET status = 'approved', reviewed_by = ?, reviewed_at = ? WHERE id = ?", (reviewer_id, now, change_id)) conn.commit() return True def reject_pending_change(change_id, reviewer_id): """拒绝待审批修改""" conn = get_db() now = datetime.now().isoformat() conn.execute("UPDATE training_pending_changes SET status = 'rejected', reviewed_by = ?, reviewed_at = ? WHERE id = ?", (reviewer_id, now, change_id)) conn.commit() return True # ========== 个性化费用配置 ========== def get_athlete_fee_config(athlete_name=None, training_type_id=None): """获取队员个性化费用配置""" conn = get_db() query = """ SELECT afc.athlete_name, afc.training_type_id, afc.price_per_session, afc.start_date, afc.updated_at, tt.name as type_name, tt.is_class FROM athlete_fee_config afc JOIN training_types tt ON afc.training_type_id = tt.id WHERE 1=1 """ params = [] if athlete_name: query += " AND afc.athlete_name = ?" params.append(athlete_name) if training_type_id: query += " AND afc.training_type_id = ?" params.append(training_type_id) query += " ORDER BY afc.athlete_name, tt.is_class DESC, tt.name" rows = conn.execute(query, params).fetchall() return [dict(r) for r in rows] def set_athlete_fee_config(athlete_name, training_type_id, price_per_session, start_date=None): """设置队员某训练类型的个性化单价""" conn = get_db() conn.execute(""" INSERT INTO athlete_fee_config (athlete_name, training_type_id, price_per_session, start_date, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(athlete_name, training_type_id) DO UPDATE SET price_per_session=?, start_date=COALESCE(?, start_date), updated_at=? """, (athlete_name, training_type_id, price_per_session, start_date, datetime.now().isoformat(), price_per_session, start_date, datetime.now().isoformat())) conn.commit() def delete_athlete_fee_config(athlete_name, training_type_id): """删除队员个性化费用配置（回退到全局配置）""" conn = get_db() conn.execute("DELETE FROM athlete_fee_config WHERE athlete_name = ? AND training_type_id = ?", (athlete_name, training_type_id)) conn.commit() # ========== 预付费管理 ========== def get_prepaid_balances(athlete_name=None): """获取预付费余额""" conn = get_db() if athlete_name: rows = conn.execute(""" SELECT pb.athlete_name, pb.training_type_id, pb.balance, pb.start_date, pb.updated_at, tt.name as type_name, tt.is_class FROM prepaid_balance pb JOIN training_types tt ON pb.training_type_id = tt.id WHERE pb.athlete_name = ? ORDER BY tt.is_class DESC, tt.name """, (athlete_name,)).fetchall() else: rows = conn.execute(""" SELECT pb.athlete_name, pb.training_type_id, pb.balance, pb.start_date, pb.updated_at, tt.name as type_name, tt.is_class FROM prepaid_balance pb JOIN training_types tt ON pb.training_type_id = tt.id ORDER BY pb.athlete_name, tt.is_class DESC, tt.name """).fetchall() return [dict(r) for r in rows] def set_prepaid_balance(athlete_name, training_type_id, balance, start_date=None): """设置预付费余额（充值或调整） 如果已有记录，增加余额；如果不存在则创建 start_date: 费用开始扣的起始时间 """ conn = get_db() now = datetime.now().isoformat() existing = conn.execute( "SELECT balance, start_date FROM prepaid_balance WHERE athlete_name = ? AND training_type_id = ?", (athlete_name, training_type_id) ).fetchone() if existing: new_balance = existing["balance"] + balance new_start = start_date or existing["start_date"] conn.execute(""" UPDATE prepaid_balance SET balance = ?, start_date = ?, updated_at = ? WHERE athlete_name = ? AND training_type_id = ? """, (new_balance, new_start, now, athlete_name, training_type_id)) else: conn.execute(""" INSERT INTO prepaid_balance (athlete_name, training_type_id, balance, start_date, updated_at) VALUES (?, ?, ?, ?, ?) """, (athlete_name, training_type_id, balance, start_date, now)) # 记录充值交易 conn.execute(""" INSERT INTO prepaid_transactions (athlete_name, training_type_id, amount, transaction_type, description, created_at) VALUES (?, ?, ?, 'recharge', ?, ?) """, (athlete_name, training_type_id, balance, f'充值{balance}元', now)) conn.commit() def delete_prepaid_balance(athlete_name, training_type_id): """删除预付费余额""" conn = get_db() conn.execute("DELETE FROM prepaid_balance WHERE athlete_name = ? AND training_type_id = ?", (athlete_name, training_type_id)) conn.commit() def get_prepaid_transactions(athlete_name=None, limit=100): """获取预付费交易记录""" conn = get_db() if athlete_name: rows = conn.execute(""" SELECT pt.*, tt.name as type_name FROM prepaid_transactions pt JOIN training_types tt ON pt.training_type_id = tt.id WHERE pt.athlete_name = ? ORDER BY pt.created_at DESC LIMIT ? """, (athlete_name, limit)).fetchall() else: rows = conn.execute(""" SELECT pt.*, tt.name as type_name FROM prepaid_transactions pt JOIN training_types tt ON pt.training_type_id = tt.id ORDER BY pt.created_at DESC LIMIT ? """, (limit,)).fetchall() return [dict(r) for r in rows] # ========== 费用结算管理 ========== def get_fee_settlements(athlete_name=None, training_type_id=None): """获取费用结算记录""" conn = get_db() query = """ SELECT fs.*, tt.name as type_name, tt.is_class FROM fee_settlements fs JOIN training_types tt ON fs.training_type_id = tt.id WHERE 1=1 """ params = [] if athlete_name: query += " AND fs.athlete_name = ?" params.append(athlete_name) if training_type_id: query += " AND fs.training_type_id = ?" params.append(training_type_id) query += " ORDER BY fs.created_at DESC" rows = conn.execute(query, params).fetchall() return [dict(r) for r in rows] def add_fee_settlement(athlete_name, training_type_id, start_date, end_date, amount, note=None, settled_by=None): """添加费用结算记录（标记某时间段某类型的费用已结清）""" conn = get_db() conn.execute(""" INSERT INTO fee_settlements (athlete_name, training_type_id, start_date, end_date, amount, note, settled_by, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (athlete_name, training_type_id, start_date, end_date, amount, note, settled_by, datetime.now().isoformat())) conn.commit() def delete_fee_settlement(settlement_id): """删除费用结算记录""" conn = get_db() conn.execute("DELETE FROM fee_settlements WHERE id = ?", (settlement_id,)) conn.commit() # ========== 家长权限查询 ========== def get_accessible_athletes(user_id, role): """获取用户可查看的队员列表 - superadmin/admin: 可查看所有队员 - 普通用户: 只能查看自己关联的队员 """ if role in ('admin', 'superadmin'): # 返回所有活跃队员 names, _ = get_athlete_names() return names # 普通用户：通过athlete_family表查找关联的队员 conn = get_db() rows = conn.execute(""" SELECT athlete_name FROM athlete_family WHERE user_id = ? AND approved = 1 """, (user_id,)).fetchall() # 也检查users表中的athlete_name user = conn.execute("SELECT athlete_name FROM users WHERE id = ?", (user_id,)).fetchone() result = [r["athlete_name"] for r in rows] if user and user["athlete_name"] and user["athlete_name"] not in result: result.append(user["athlete_name"]) return result # ========== 昵称映射 ========== def get_nickname_map(): """获取昵称到真名的映射字典 {nickname: real_name}""" conn = get_db() rows = conn.execute("SELECT nickname, real_name FROM athlete_nicknames").fetchall() return {r["nickname"]: r["real_name"] for r in rows} def get_all_nicknames(): """获取所有昵称映射""" conn = get_db() rows = conn.execute("SELECT * FROM athlete_nicknames ORDER BY real_name, nickname").fetchall() return [dict(r) for r in rows] def add_nickname(nickname, real_name): """添加昵称映射""" conn = get_db() try: conn.execute("INSERT INTO athlete_nicknames (nickname, real_name, created_at) VALUES (?, ?, ?)", (nickname, real_name, datetime.now().isoformat())) conn.commit() return True except Exception: return False finally: pass def delete_nickname(nickname_id): """删除昵称映射""" conn = get_db() conn.execute("DELETE FROM athlete_nicknames WHERE id = ?", (nickname_id,)) conn.commit() def resolve_nickname(name): """将昵称解析为真名，如果不是昵称则返回原名""" nickname_map = get_nickname_map() return nickname_map.get(name, name) # ====== 小程序微信登录相关 ====== def get_wx_binding_by_openid(openid): """查询 openid 是否已绑定""" conn = get_db() row = conn.execute( "SELECT * FROM wx_bindings WHERE openid = ?", (openid,) ).fetchone() return dict(row) if row else None def create_wx_binding(openid, athlete_name, relationship="本人", user_id=None, approved=0): """创建微信绑定""" conn = get_db() conn.execute( "INSERT OR REPLACE INTO wx_bindings (openid, athlete_name, relationship, approved, created_at, user_id) VALUES (?, ?, ?, ?, ?, ?)", (openid, athlete_name, relationship, approved, datetime.now().isoformat(), user_id) ) conn.commit() def approve_wx_binding(binding_id): """管理员批准微信绑定""" conn = get_db() conn.execute("UPDATE wx_bindings SET approved = 1 WHERE id = ?", (binding_id,)) conn.commit() def get_pending_wx_bindings(): """获取待审批的微信绑定列表""" conn = get_db() rows = conn.execute( "SELECT * FROM wx_bindings WHERE approved = 0 ORDER BY created_at DESC" ).fetchall() return [dict(r) for r in rows] def create_user_token(user_id): """创建用户 Token，返回 token 字符串""" import secrets token = secrets.token_hex(32) conn = get_db() # 30天过期 expires = (datetime.now() + timedelta(days=30)).isoformat() conn.execute( "INSERT INTO user_tokens (user_id, token, expires_at, created_at) VALUES (?, ?, ?, ?)", (user_id, token, expires, datetime.now().isoformat()) ) conn.commit() return token def get_user_by_token(token): """通过 token 获取用户信息""" conn = get_db() row = conn.execute( "SELECT u.* FROM users u JOIN user_tokens t ON u.id = t.user_id WHERE t.token = ? AND t.expires_at > ?", (token, datetime.now().isoformat()) ).fetchone() return dict(row) if row else None def get_or_create_wx_user(athlete_name, openid): """为微信登录的运动员创建或查找用户账号""" conn = get_db() # 先看是否已绑定并通过审核 binding = get_wx_binding_by_openid(openid) if not binding or not binding['approved']: return None # 如果绑定时指定了 user_id，直接返回该用户 if binding.get('user_id'): user = conn.execute( "SELECT * FROM users WHERE id = ?", (binding['user_id'],) ).fetchone() if user: return dict(user) # 查找该运动员对应的用户 user = conn.execute( "SELECT * FROM users WHERE athlete_name = ? AND role = 'user'", (athlete_name,) ).fetchone() if not user: # 自动创建用户（无密码，仅微信登录） username = f"wx_{athlete_name}" existing = conn.execute("SELECT id FROM users WHERE username = ?", (username,)).fetchone() if existing: username = f"wx_{athlete_name}_{openid[-6:]}" conn.execute( "INSERT INTO users (username, password_hash, role, athlete_name, relationship, approved, created_at) VALUES (?, '', 'user', ?, '本人', 1, ?)", (username, athlete_name, datetime.now().isoformat()) ) conn.commit() user = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone() return dict(user) if user else None