| """肆带贰足球俱乐部 · 球队管理系统（Shiny for Python）。 |
| |
| 运行：uv run shiny run app.py --reload |
| 默认本地 SQLite（data/team.db）；设置环境变量 DATABASE_URL 可切换到云端 Postgres。 |
| """ |
| from __future__ import annotations |
|
|
| import datetime as dt |
| import logging |
|
|
| from shiny import App, Inputs, Outputs, Session, reactive, render, ui |
|
|
| from src import config, export_img, repo, views |
| from src.seed import ensure_seed |
| from src.theme import CSS |
|
|
| logging.basicConfig(level=logging.INFO) |
|
|
| |
| ensure_seed() |
|
|
| app_ui = ui.page_fluid( |
| ui.head_content( |
| ui.tags.title(f"{config.TEAM_NAME} · 球队系统"), |
| ui.tags.meta(name="viewport", content="width=device-width, initial-scale=1"), |
| ui.tags.link(rel="icon", href=config.CREST_URL), |
| ui.tags.style(ui.HTML(CSS)), |
| ), |
| ui.output_ui("app_body"), |
| ) |
|
|
|
|
| def server(input: Inputs, output: Outputs, session: Session) -> None: |
| current_user: reactive.Value = reactive.value(None) |
| data_version: reactive.Value = reactive.value(0) |
|
|
| def bump() -> None: |
| data_version.set(data_version.get() + 1) |
|
|
| def is_admin() -> bool: |
| u = current_user.get() |
| return bool(u and u["is_admin"]) |
|
|
| def user_is_gk(uid: int) -> bool: |
| """实时查询某队员当前是否门将（避免依赖登录时的旧状态）。""" |
| return any(x["id"] == uid and x["is_goalkeeper"] for x in repo.list_users()) |
|
|
| |
| @render.ui |
| def app_body(): |
| u = current_user.get() |
| if u is None: |
| return views.login_view() |
| choices = { |
| str(x["id"]): f'{x["display_name"]} ({x["username"]})' |
| for x in repo.list_users() |
| } |
| return views.app_shell(u, choices, repo.available_seasons()) |
|
|
| |
| @reactive.effect |
| @reactive.event(input.btn_login, ignore_init=True) |
| def _login(): |
| if not input.btn_login(): |
| return |
| u = repo.authenticate((input.in_user() or "").strip(), input.in_pw() or "") |
| if u is None: |
| ui.notification_show("用户名或密码错误", type="error") |
| return |
| current_user.set(u) |
|
|
| @reactive.effect |
| @reactive.event(input.btn_logout, ignore_init=True) |
| def _logout(): |
| if not input.btn_logout(): |
| return |
| current_user.set(None) |
|
|
| |
| @render.ui |
| def home_content(): |
| data_version.get() |
| u = current_user.get() |
| sv = input.home_season() |
| season = int(sv) if (sv and sv != "all" and str(sv).isdigit()) else None |
| return views.home_view( |
| repo.team_summary(season), |
| repo.list_matches(season), |
| repo.player_aggregate(season), |
| u["id"] if u else None, |
| repo.match_contributions(season), |
| repo.mvp_ranking(season), |
| repo.settled_mvps(), |
| repo.dmvp_ranking(season), |
| repo.all_fifa_cards(), |
| ) |
|
|
| |
| @render.download(filename="肆带贰_榜单.png") |
| def dl_leaderboard(): |
| sv = input.home_season() |
| season = int(sv) if (sv and sv != "all" and str(sv).isdigit()) else None |
| label = "全部赛季" if season is None else f"{season} 赛季" |
| agg = repo.player_aggregate(season) |
| scorers = sorted(agg, key=lambda x: (-x["goals"], -x["assists"])) |
| assisters = sorted(agg, key=lambda x: (-x["assists"], -x["goals"])) |
| attend = sorted(agg, key=lambda x: (-x["attended"], -x["goals"])) |
| keepers = sorted( |
| [r for r in agg if r.get("is_goalkeeper")], |
| key=lambda x: (-x["saves"], -x["attended"]), |
| ) |
| mvp = repo.mvp_ranking(season) |
| dmvp = repo.dmvp_ranking(season) |
|
|
| def rowsify(items, valuefn): |
| return [(str(i), it["name"], valuefn(it)) for i, it in enumerate(items, 1)] |
|
|
| boards = [ |
| ("射手榜", rowsify(scorers, lambda x: f'{x["goals"]} 球')), |
| ("助攻榜", rowsify(assisters, lambda x: f'{x["assists"]} 助')), |
| ("出勤榜", rowsify(attend, lambda x: f'{x["attended"]}/{x["total"]}')), |
| ("最佳球员 MVP", rowsify(mvp, lambda x: f'{x["mvps"]} 次')), |
| ("铁血防守 MVP", rowsify(dmvp, lambda x: f'{x["dmvps"]} 次')), |
| ] |
| if keepers: |
| boards.append(("扑救榜", rowsify(keepers, lambda x: f'{x["saves"]} 救'))) |
| yield export_img.render_leaderboard_png( |
| config.TEAM_NAME, label, dt.date.today().isoformat(), boards |
| ) |
|
|
| |
| @render.ui |
| def my_fifa_card(): |
| data_version.get() |
| u = current_user.get() |
| if not u: |
| return ui.div() |
| |
| return views.my_fifa_card_view( |
| repo.player_fifa_card(u["id"]), diamond=config.is_diamond_user(u) |
| ) |
|
|
| @render.ui |
| def my_totals(): |
| data_version.get() |
| u = current_user.get() |
| if not u: |
| return ui.div() |
| return views.my_totals_view(repo.player_rank(u["id"])) |
|
|
| @render.ui |
| def my_form(): |
| data_version.get() |
| u = current_user.get() |
| if not u: |
| return ui.div() |
| return views.my_form_view(repo.player_matchlog(u["id"]), user_is_gk(u["id"])) |
|
|
| @render.ui |
| def my_trend(): |
| data_version.get() |
| u = current_user.get() |
| if not u: |
| return ui.div() |
| return views.my_trend_view(repo.player_matchlog(u["id"])) |
|
|
| @reactive.effect |
| @reactive.event(input.btn_save_stats, ignore_init=True) |
| def _save_stats(): |
| if not input.btn_save_stats(): |
| return |
| u = current_user.get() |
| if not u: |
| return |
| is_gk = user_is_gk(u["id"]) |
| for m in repo.player_matchlog(u["id"]): |
| mid = m["id"] |
| saves = input[f"sv_{mid}"]() if is_gk else 0 |
| repo.upsert_stat( |
| u["id"], mid, |
| input[f"g_{mid}"](), input[f"a_{mid}"](), input[f"att_{mid}"](), saves, |
| ) |
| bump() |
| ui.notification_show("已保存你的数据 ✅", type="message") |
|
|
| |
| @render.ui |
| def my_mvp_form(): |
| data_version.get() |
| u = current_user.get() |
| if not u: |
| return ui.div() |
| players = [x for x in repo.list_users() if x["is_player"] and x["active"]] |
| return views.my_mvp_form_view( |
| repo.player_matchlog(u["id"]), players, repo.my_votes(u["id"]), u["id"], |
| repo.settled_mvps(), |
| ) |
|
|
| @reactive.effect |
| @reactive.event(input.btn_save_mvp, ignore_init=True) |
| def _save_mvp(): |
| if not input.btn_save_mvp(): |
| return |
| u = current_user.get() |
| if not u: |
| return |
| for m in repo.player_matchlog(u["id"]): |
| if m.get("mvp_locked"): |
| continue |
| mid = m["id"] |
| raw = input[f"mvp_{mid}"]() |
| vid = int(raw) if (raw and raw != "none" and str(raw).isdigit()) else None |
| repo.set_vote(u["id"], mid, vid) |
| bump() |
| ui.notification_show("已保存你的 MVP 投票 🏅", type="message") |
|
|
| |
| @reactive.effect |
| @reactive.event(input.btn_change_pw, ignore_init=True) |
| def _change_pw(): |
| if not input.btn_change_pw(): |
| return |
| u = current_user.get() |
| if not u: |
| return |
| new = (input.cp_new() or "").strip() |
| if len(new) < 6: |
| ui.notification_show("新密码至少 6 位", type="error") |
| return |
| if new != (input.cp_new2() or "").strip(): |
| ui.notification_show("两次输入的新密码不一致", type="error") |
| return |
| if not repo.change_password(u["id"], input.cp_old() or "", new): |
| ui.notification_show("当前密码不正确", type="error") |
| return |
| ui.notification_show("密码已修改 ✅ 下次请用新密码登录", type="message") |
|
|
| |
| @render.ui |
| def admin_matches(): |
| data_version.get() |
| if not is_admin(): |
| return ui.div() |
| return views.admin_match_editor(repo.list_matches()) |
|
|
| @render.ui |
| def admin_players(): |
| data_version.get() |
| if not is_admin(): |
| return ui.div() |
| return views.admin_players_view(repo.list_users()) |
|
|
| |
| @render.ui |
| def admin_edit_player(): |
| data_version.get() |
| if not is_admin(): |
| return ui.div() |
| uid_raw = input.ae_user() |
| if not uid_raw: |
| return ui.div("请选择要编辑的队员。", class_="sde-empty") |
| uid = int(uid_raw) |
| info = next((x for x in repo.list_users() if x["id"] == uid), None) |
| return views.admin_edit_player_view( |
| repo.player_matchlog(uid), |
| user_is_gk(uid), |
| info["display_name"] if info else "", |
| info["jersey"] if info else None, |
| ) |
|
|
| @reactive.effect |
| @reactive.event(input.btn_ae_save, ignore_init=True) |
| def _ae_save(): |
| if not input.btn_ae_save() or not is_admin(): |
| return |
| uid_raw = input.ae_user() |
| if not uid_raw: |
| ui.notification_show("请先选择队员", type="warning") |
| return |
| uid = int(uid_raw) |
| |
| jraw = (input.ae_jersey() or "").strip() |
| jersey = int(jraw) if jraw.isdigit() else None |
| repo.update_player_profile(uid, input.ae_name() or "", jersey) |
| repo.set_goalkeeper(uid, bool(input.ae_gk())) |
| for m in repo.player_matchlog(uid): |
| mid = m["id"] |
| repo.upsert_stat( |
| uid, mid, |
| input[f"ag_{mid}"](), input[f"aa_{mid}"](), |
| input[f"aatt_{mid}"](), input[f"asv_{mid}"](), |
| ) |
| bump() |
| ui.notification_show("已保存该队员资料与数据 ✅", type="message") |
|
|
| |
| @render.ui |
| def admin_mvp_settle(): |
| data_version.get() |
| if not is_admin(): |
| return ui.div() |
| player_choices = { |
| str(x["id"]): f'{x["display_name"]} ({x["username"]})' |
| for x in repo.list_users() |
| if x["is_player"] and x["active"] |
| } |
| return views.admin_mvp_settle_view( |
| repo.list_matches(), repo.match_mvp_summary(), player_choices |
| ) |
|
|
| @reactive.effect |
| @reactive.event(input.btn_save_mvp_lock, ignore_init=True) |
| def _save_mvp_lock(): |
| if not input.btn_save_mvp_lock() or not is_admin(): |
| return |
| def _sel(mid: int, prefix: str = "mvpsel") -> int | None: |
| raw = input[f"{prefix}_{mid}"]() |
| return int(raw) if (raw and raw != "none" and str(raw).isdigit()) else None |
| settled = opened = 0 |
| for m in repo.list_matches(): |
| mid = m["id"] |
| w1, w2, wd = _sel(mid), _sel(mid, "mvpsel2"), _sel(mid, "dmvpsel") |
| cur = (m.get("mvp_user_id"), m.get("mvp2_user_id"), m.get("dmvp_user_id")) |
| if (w1, w2, wd) != cur: |
| repo.settle_mvp(mid, w1, w2, wd) |
| settled += 1 if (w1 or w2 or wd) else 0 |
| opened += 1 if (w1 is None and w2 is None and wd is None) else 0 |
| bump() |
| ui.notification_show( |
| f"已结算 {settled} 场、重新开放 {opened} 场 🏅", type="message" |
| ) |
|
|
| @reactive.effect |
| @reactive.event(input.btn_add_match, ignore_init=True) |
| def _add_match(): |
| if not input.btn_add_match() or not is_admin(): |
| return |
| opp = (input.m_opp() or "").strip() |
| if not opp: |
| ui.notification_show("请填写对手名称", type="error") |
| return |
| try: |
| repo.add_match( |
| input.m_date(), opp, input.m_ha(), |
| input.m_our(), input.m_oppscore(), input.m_note(), |
| ) |
| except ValueError as e: |
| ui.notification_show(str(e), type="error") |
| return |
| bump() |
| ui.update_text("m_opp", value="") |
| ui.update_numeric("m_our", value=0) |
| ui.update_numeric("m_oppscore", value=0) |
| ui.update_text("m_note", value="") |
| ui.notification_show("已添加比赛 ✅", type="message") |
|
|
| @reactive.effect |
| @reactive.event(input.btn_save_matches, ignore_init=True) |
| def _save_matches(): |
| if not input.btn_save_matches() or not is_admin(): |
| return |
| updated = deleted = 0 |
| for m in repo.list_matches(): |
| mid = m["id"] |
| if input[f"mdel_{mid}"](): |
| repo.delete_match(mid) |
| deleted += 1 |
| continue |
| repo.update_match( |
| mid, input[f"mopp_{mid}"](), input[f"mha_{mid}"](), |
| input[f"mo_{mid}"](), input[f"mp_{mid}"](), |
| ) |
| updated += 1 |
| bump() |
| ui.notification_show(f"已更新 {updated} 场，删除 {deleted} 场 ✅", type="message") |
|
|
| |
| @reactive.effect |
| @reactive.event(input.btn_add_player, ignore_init=True) |
| def _add_player(): |
| if not input.btn_add_player() or not is_admin(): |
| return |
| jersey_raw = (input.p_jersey() or "").strip() |
| jersey = int(jersey_raw) if jersey_raw.isdigit() else None |
| try: |
| repo.add_user( |
| input.p_user(), input.p_name(), |
| input.p_pw() or config.DEFAULT_PASSWORD, |
| jersey, is_admin=bool(input.p_admin()), is_player=True, |
| is_goalkeeper=bool(input.p_gk()), |
| ) |
| except ValueError as e: |
| ui.notification_show(str(e), type="error") |
| return |
| bump() |
| ui.update_text("p_name", value="") |
| ui.update_text("p_user", value="") |
| ui.update_text("p_jersey", value="") |
| ui.notification_show("已添加队员 ✅", type="message") |
|
|
| @reactive.effect |
| @reactive.event(input.btn_reset_pw, ignore_init=True) |
| def _reset_pw(): |
| if not input.btn_reset_pw() or not is_admin(): |
| return |
| uid = input.rp_user() |
| pw = input.rp_pw() |
| if not uid or not pw: |
| ui.notification_show("请选择队员并填写新密码", type="error") |
| return |
| repo.reset_password(int(uid), pw) |
| bump() |
| ui.notification_show("密码已重置 ✅", type="message") |
|
|
| @reactive.effect |
| @reactive.event(input.btn_clear_demo, ignore_init=True) |
| def _clear_demo(): |
| if not input.btn_clear_demo() or not is_admin(): |
| return |
| if not input.confirm_clear(): |
| ui.notification_show("请先勾选「我确认」开关再清空", type="warning") |
| return |
| n = repo.clear_matches_and_stats() |
| bump() |
| ui.notification_show(f"已清空 {n} 场比赛及相关数据 ✅", type="message") |
|
|
| |
| |
| |
| @reactive.effect |
| def _refresh_reset_pw_choices(): |
| data_version.get() |
| if not is_admin(): |
| return |
| choices = { |
| str(x["id"]): f'{x["display_name"]} ({x["username"]})' |
| for x in repo.list_users() |
| } |
| |
| ui.update_select("rp_user", choices=choices) |
| ui.update_select("ae_user", choices=choices) |
|
|
|
|
| app = App(app_ui, server, static_assets={"/static": str(config.WWW_DIR)}) |
|
|
