""" 해커톤 투표 앱 - 참가자: 이름 선택 → 본인 팀 자동 → 본인 팀 제외 3분야 투표 - 진행자: ?mode=admin&token=XXX 로 결과 확인 """ import json import os import sqlite3 from datetime import datetime from pathlib import Path import streamlit as st DB_PATH = os.environ.get("VOTE_DB", "votes.db") ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "change-me") PARTICIPANTS_PATH = os.environ.get( "PARTICIPANTS", str(Path(__file__).parent / "participants.json") ) def load_participants(): """이름→팀 매핑. 파일 없으면 빈 dict.""" p = Path(PARTICIPANTS_PATH) if not p.exists(): return {} return json.loads(p.read_text(encoding="utf-8")) PARTICIPANTS = load_participants() TEAMS = sorted(set(PARTICIPANTS.values())) if PARTICIPANTS else [ f"팀{i}" for i in range(1, 8) ] CATEGORIES = [ ("fun_team", "🎉 재미상", "손선풍기 5개"), ("polish_team", "🏆 완성도상", "양우산 5개"), ("utility_team", "🛠 실용성상", "팜레스트 5개"), ] # 수상 결정 우선순위 (높을수록 먼저 결정, 후순위 상에서 그 팀 제외) # 팜레스트(실용성) > 양우산(완성도) > 손선풍기(재미) PRIZE_PRIORITY = ["utility_team", "polish_team", "fun_team"] def get_conn(): conn = sqlite3.connect(DB_PATH, timeout=10) # WAL 모드: 동시 read/write 충돌 ↓ conn.execute("PRAGMA journal_mode = WAL") conn.execute("PRAGMA synchronous = NORMAL") conn.execute( """ CREATE TABLE IF NOT EXISTS votes ( id INTEGER PRIMARY KEY AUTOINCREMENT, voter_name TEXT NOT NULL UNIQUE, employee_id TEXT, voter_team TEXT NOT NULL, fun_team TEXT NOT NULL, polish_team TEXT NOT NULL, utility_team TEXT NOT NULL, created_at TEXT NOT NULL ) """ ) # 기존 DB 마이그레이션 cols = [r[1] for r in conn.execute("PRAGMA table_info(votes)").fetchall()] if "employee_id" not in cols: conn.execute("ALTER TABLE votes ADD COLUMN employee_id TEXT") conn.execute( """ CREATE TABLE IF NOT EXISTS tie_breaks ( category TEXT PRIMARY KEY, winner_team TEXT NOT NULL, method TEXT NOT NULL, -- 'random' or 'manual' decided_at TEXT NOT NULL ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS team_titles ( team_name TEXT PRIMARY KEY, title TEXT NOT NULL DEFAULT '' ) """ ) conn.commit() return conn def get_titles(): """팀명 → 결과물 제목 dict. 매 호출 DB 조회 (라이브 반영).""" conn = get_conn() rows = conn.execute("SELECT team_name, title FROM team_titles").fetchall() conn.close() return dict(rows) def set_title(team, title): conn = get_conn() conn.execute( "INSERT INTO team_titles (team_name, title) VALUES (?, ?) " "ON CONFLICT(team_name) DO UPDATE SET title = excluded.title", (team, title.strip()), ) conn.commit() conn.close() def fmt_team(team, titles): """팀 라벨 — 제목 있으면 'N팀 — 제목' 없으면 'N팀'.""" t = titles.get(team, "") return f"{team} — {t}" if t else team def is_voting_open(): conn = get_conn() row = conn.execute( "SELECT value FROM settings WHERE key = 'voting_open'" ).fetchone() conn.close() return row is None or row[0] == "1" def set_voting_open(flag): conn = get_conn() conn.execute( "INSERT INTO settings (key, value) VALUES ('voting_open', ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", ("1" if flag else "0",), ) conn.commit() conn.close() def get_tie_breaks(): conn = get_conn() rows = conn.execute("SELECT category, winner_team FROM tie_breaks").fetchall() conn.close() return dict(rows) def save_tie_break(category, winner, method): conn = get_conn() conn.execute( "INSERT INTO tie_breaks (category, winner_team, method, decided_at) " "VALUES (?, ?, ?, ?) " "ON CONFLICT(category) DO UPDATE SET " "winner_team = excluded.winner_team, " "method = excluded.method, " "decided_at = excluded.decided_at", (category, winner, method, datetime.now().isoformat(timespec="seconds")), ) conn.commit() conn.close() def clear_tie_break(category): conn = get_conn() conn.execute("DELETE FROM tie_breaks WHERE category = ?", (category,)) conn.commit() conn.close() def compute_winners(): """ 우선순위 기반 1팀 1상 + 동률 처리. 동률 발견 시 tie_breaks 저장 결정 사용. 미결정이면 status='tie' 반환. return: (winners dict, rankings dict) winners[col] = { "status": "ok"|"tie"|"empty", "team": str, # ok 시 "votes": int, "diff": int, # ok 시 2위와 차이 "tied": [str], # tie 시 동률 후보들 "method": str|None, # tie_break 적용 방식 } """ conn = get_conn() rankings = {} for col, _, _ in CATEGORIES: rankings[col] = conn.execute( f"SELECT {col} AS team, COUNT(*) AS c FROM votes " f"GROUP BY {col} ORDER BY c DESC, team ASC" ).fetchall() tie_decisions = dict( conn.execute( "SELECT category, winner_team || '||' || method FROM tie_breaks" ).fetchall() ) conn.close() winners = {} excluded = set() for col in PRIZE_PRIORITY: rows = rankings[col] filtered = [(t, c) for t, c in rows if t not in excluded] if not filtered: winners[col] = {"status": "empty"} continue top_votes = filtered[0][1] tied = [t for t, c in filtered if c == top_votes] if len(tied) > 1: decision = tie_decisions.get(col, "") chosen, _, method = decision.partition("||") if chosen and chosen in tied: winner = chosen else: winners[col] = { "status": "tie", "tied": tied, "votes": top_votes, } continue else: winner = filtered[0][0] method = "" runner_votes = next((c for t, c in filtered if t != winner), 0) winners[col] = { "status": "ok", "team": winner, "votes": top_votes, "diff": top_votes - runner_votes, "method": method, } excluded.add(winner) return winners, rankings def render_voter(): st.title("🗳 해커톤 투표") if not is_voting_open(): st.error("⏹️ 투표가 마감되었습니다. 시상식을 기다려주세요.") return with st.expander("📖 투표 방식 / 수상 결정 (꼭 읽어주세요)", expanded=True): st.markdown( """ **투표 방식** - 본인 이름 선택 + **본인 사번 입력** - 본인 팀 자동 매핑 - 본인 팀 **제외**, 다른 6팀 중 각 분야별 1팀씩 투표 (총 3표) - **한 번만 제출 가능** (이름 unique) - 사번은 사칭 의심 시 추적용으로 기록됨 (본인 이름이 아닌데 투표한 경우 사번 주인에게 확인) **수상 결정 — 1팀 1상 한정** - 우선순위: 🛠 **실용성상 (팜레스트)** > 🏆 **완성도상 (양우산)** > 🎉 **재미상 (손선풍기)** - 실용성 1위 팀이 팜레스트 수상 → 그 팀은 다른 상 후보에서 자동 제외 - 완성도상도 같은 방식, 마지막 재미상까지 결정 - → 한 팀이 모든 분야 1위여도 **가장 비싼 상 1개**만 받음 **시상 발표 순서** - 🎉 재미상 → 🏆 완성도상 → 🛠 실용성상 (긴장감 build-up, 최고가 마무리) """ ) st.caption("이름 선택 → 본인 팀 자동 매핑 → 본인 팀 제외 3분야 투표. 한 번만 제출 가능.") if not PARTICIPANTS: st.error("참가자 명단이 없습니다. `assign_teams.py` 먼저 실행하세요.") return # 이미 투표한 사람 미리 조회 → selectbox에 ✅ 표시 conn = get_conn() voted_set = { r[0] for r in conn.execute("SELECT voter_name FROM votes").fetchall() } conn.close() def fmt_name(n): return f"{n} ✅ (이미 투표함)" if n in voted_set else n name = st.selectbox( "본인 이름", options=sorted(PARTICIPANTS.keys()), index=None, placeholder="이름 선택", format_func=fmt_name, ) employee_id = st.text_input( "사번", placeholder="본인 사번만 (전화번호/주민번호 입력 금지)", help="사칭 추적용으로 기록됨. 본인 사번 외 입력하지 마세요.", ) if not name: st.info("이름을 선택하면 본인 팀이 자동 표시됩니다.") return if name in voted_set: st.error( f"❌ {name}님은 이미 투표하셨습니다. 한 번만 가능합니다. " "다른 사람으로 잘못 누른 경우 진행자에게 문의하세요." ) return if not employee_id.strip(): st.warning("사번 입력 후 진행하세요.") return my_team = PARTICIPANTS[name] titles = get_titles() st.info(f"본인 팀: **{fmt_team(my_team, titles)}**") candidates = [t for t in TEAMS if t != my_team] with st.form("vote", clear_on_submit=False): st.divider() picks = {} for col, label, _ in CATEGORIES: picks[col] = st.radio( label, candidates, index=None, key=col, format_func=lambda t: fmt_team(t, titles), ) submitted = st.form_submit_button("제출") if submitted: if any(picks.get(col) is None for col, _, _ in CATEGORIES): st.error("3분야 모두 선택하세요.") return conn = get_conn() try: conn.execute( """ INSERT INTO votes (voter_name, employee_id, voter_team, fun_team, polish_team, utility_team, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( name, employee_id.strip(), my_team, picks["fun_team"], picks["polish_team"], picks["utility_team"], datetime.now().isoformat(timespec="seconds"), ), ) conn.commit() st.success(f"✅ {name}님 ({my_team}) 투표 완료. 창 닫아도 됩니다.") st.balloons() except sqlite3.IntegrityError: st.error(f"❌ '{name}' 이미 투표함. 진행자에게 문의하세요.") finally: conn.close() def render_admin(): token = st.query_params.get("token", "") if token != ADMIN_TOKEN: st.error("권한 없음. ?mode=admin&token=... 형식 필요.") return st.title("🔐 진행자 콘솔") # 투표 마감 토글 voting_open = is_voting_open() cur_label = "🟢 투표 진행 중" if voting_open else "🔴 투표 마감됨" st.markdown(f"### 투표 상태: {cur_label}") cc1, cc2 = st.columns(2) with cc1: if voting_open and st.button("🛑 투표 마감 (시상 시작 전 필수)", type="primary"): set_voting_open(False) st.rerun() with cc2: if not voting_open and st.button("🔓 투표 재개"): set_voting_open(True) st.rerun() st.divider() conn = get_conn() total = conn.execute("SELECT COUNT(*) FROM votes").fetchone()[0] expected = len(PARTICIPANTS) if PARTICIPANTS else None col_a, col_b, col_c = st.columns(3) col_a.metric("투표 참여자", f"{total}명") col_b.metric("팀 수", f"{len(TEAMS)}팀") if expected: pct = int(100 * total / expected) col_c.metric("참여율", f"{pct}% ({total}/{expected})") if expected and total < expected: voted = {row[0] for row in conn.execute("SELECT voter_name FROM votes").fetchall()} not_voted = [n for n in PARTICIPANTS.keys() if n not in voted] with st.expander(f"⏳ 미투표자 ({len(not_voted)}명)"): for n in sorted(not_voted): st.write(f"- {n} ({PARTICIPANTS[n]})") st.divider() st.subheader("📝 팀별 결과물 제목 입력") st.caption("발표 직후 입력하면 투표 페이지에 즉시 반영됩니다.") titles = get_titles() with st.form("titles_form"): new_titles = {} for team in TEAMS: new_titles[team] = st.text_input( team, value=titles.get(team, ""), placeholder="예: 슬랙 멘션 자동 분류기", key=f"title_{team}", ) if st.form_submit_button("제목 저장"): for team, title in new_titles.items(): set_title(team, title) st.success("제목 저장 완료. 투표 페이지에 반영됨.") st.rerun() st.divider() st.subheader("📊 분야별 집계 (우선순위 적용 — 1팀 1상)") st.caption( "수상 결정 순서: 팜레스트(실용성) → 양우산(완성도) → 손선풍기(재미). " "이미 받은 팀은 후순위 상에서 제외." ) winners, rankings = compute_winners() awarded_teams = { w["team"] for w in winners.values() if w.get("status") == "ok" } public_lines = [] # 동률 미해결 부문 먼저 표시 (진행자 액션 필요) pending_ties = [ (col, label, prize) for col, label, prize in CATEGORIES if winners.get(col, {}).get("status") == "tie" ] if pending_ties: st.error( f"⚠️ 동률 미해결 {len(pending_ties)}건. 시상 전 추첨/선택 필요." ) for col, label, prize in CATEGORIES: rows = rankings[col] st.markdown(f"### {label} ({prize})") if not rows: st.caption("표 없음") continue result = winners.get(col, {}) status = result.get("status") if status == "empty": st.warning("후보 없음 (모두 우선순위 상 수상)") continue if status == "tie": tied = result["tied"] votes = result["votes"] tied_labels = ", ".join(fmt_team(t, titles) for t in tied) st.warning(f"🟰 동률 ({votes}표): {tied_labels}") ca, cb = st.columns([1, 2]) with ca: if st.button("🎲 즉석 추첨", key=f"draw_{col}"): import random as _r chosen = _r.choice(tied) save_tie_break(col, chosen, "random") st.rerun() with cb: manual = st.selectbox( "수동 선택", tied, index=None, key=f"manual_{col}", format_func=lambda t: fmt_team(t, titles), placeholder="팀 선택", ) if manual and st.button("확정", key=f"manual_btn_{col}"): save_tie_break(col, manual, "manual") st.rerun() continue # status == "ok" winner_team = result["team"] winner_votes = result["votes"] diff = result["diff"] method = result.get("method", "") winner_label = fmt_team(winner_team, titles) method_tag = "" if method == "random": method_tag = " 🎲(추첨)" elif method == "manual": method_tag = " 🖊️(수동)" st.success( f"**우승: {winner_label}**{method_tag} — {winner_votes}표 " f"(2위와 {diff}표 차이)" ) if method: if st.button(f"동률 결정 취소 (재추첨)", key=f"clear_{col}"): clear_tie_break(col) st.rerun() public_lines.append( f"- {label} 우승: **{winner_label}** " f"({winner_votes}표, 2위와 {diff}표 차이)" ) with st.expander("전체 분포 — raw (제외 적용 전)"): for team, c in rows: marker = "" if team in awarded_teams and team != winner_team: marker = " 🚫상위상수상으로 제외" st.write(f"- {fmt_team(team, titles)}: {c}표{marker}") st.divider() st.subheader("🎤 시상식 발표용 (복사해서 화면 공유)") st.code("\n".join(public_lines), language="markdown") st.divider() with st.expander("🔍 감사 로그 (사칭 추적용)"): st.caption("이름, 사번, 시각, 투표 내역. 의심 시 사번 주인에게 확인.") rows = conn.execute( "SELECT created_at, voter_name, employee_id, voter_team, " "fun_team, polish_team, utility_team FROM votes ORDER BY created_at" ).fetchall() if not rows: st.caption("투표 없음") else: import io import csv buf = io.StringIO() w = csv.writer(buf) w.writerow(["시각", "이름", "사번", "본인팀", "재미", "완성도", "실용성"]) for r in rows: w.writerow(r) # UTF-8 BOM → Excel 한글 호환 csv_data = "" + buf.getvalue() st.download_button( "CSV 내려받기", csv_data, file_name="audit.csv", mime="text/csv; charset=utf-8", ) st.dataframe( { "시각": [r[0] for r in rows], "이름": [r[1] for r in rows], "사번": [r[2] or "" for r in rows], "본인팀": [r[3] for r in rows], "재미": [r[4] for r in rows], "완성도": [r[5] for r in rows], "실용성": [r[6] for r in rows], }, use_container_width=True, hide_index=True, ) # 사번 중복 의심 (같은 사번이 여러 이름으로 투표) from collections import defaultdict by_emp = defaultdict(list) for r in rows: if r[2]: by_emp[r[2]].append(r[1]) dups = {emp: names for emp, names in by_emp.items() if len(set(names)) > 1} if dups: st.error("⚠️ 같은 사번이 여러 이름으로 투표한 케이스:") for emp, names in dups.items(): st.write(f"- 사번 `{emp}`: {', '.join(names)}") st.divider() with st.expander("⚠️ 위험 작업"): if st.button("모든 투표 삭제 (되돌릴 수 없음)"): conn.execute("DELETE FROM votes") conn.commit() st.warning("전체 삭제됨. 새로고침하세요.") conn.close() def archive_results(): """결과를 timestamped JSON 파일로 저장 (DB 손실 보험).""" winners, _ = compute_winners() if any(w.get("status") != "ok" for w in winners.values()): return None titles = get_titles() data = { "timestamp": datetime.now().isoformat(timespec="seconds"), "results": [], } for col, label, prize in CATEGORIES: w = winners.get(col) if w and w.get("status") == "ok": data["results"].append( { "category": label, "prize": prize, "team": w["team"], "title": titles.get(w["team"], ""), "votes": w["votes"], "diff_2nd": w["diff"], "method": w.get("method") or "majority", } ) archive_dir = os.path.dirname(DB_PATH) or "." ts = datetime.now().strftime("%Y%m%d_%H%M%S") path = os.path.join(archive_dir, f"results_{ts}.json") Path(path).write_text( json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8" ) return path def render_ceremony(): """시상식 reveal 페이지. 진행자가 클릭으로 단계별 공개.""" token = st.query_params.get("token", "") if token != ADMIN_TOKEN: st.error("권한 없음. ?mode=ceremony&token=... 형식 필요.") return titles = get_titles() # 투표 0건이면 ceremony 진입 불가 conn = get_conn() total_votes = conn.execute("SELECT COUNT(*) FROM votes").fetchone()[0] conn.close() if total_votes == 0: st.warning("📊 아직 투표가 없습니다. 투표 종료 후 시상식을 시작하세요.") return # 투표가 아직 열려 있으면 경고 (시상 도중 결과 변경 위험) if is_voting_open(): st.error( "⚠️ 투표가 아직 진행 중입니다. 어드민에서 '투표 마감' 후 시상하세요. " "(현재 결과가 시상 중 바뀔 수 있음)" ) return winners, _ = compute_winners() # 동률 미해결 있으면 ceremony 차단 (부문/후보 정보는 노출 X — 발표 spoiler 방지) pending_count = sum( 1 for col, _, _ in CATEGORIES if winners.get(col, {}).get("status") == "tie" ) if pending_count > 0: st.warning("⏳ 시상 준비 중입니다. 잠시만 기다려주세요.") return # 진입 시 1회 archive (DB 손실 보험) if not st.session_state.get("ceremony_archived"): archived_path = archive_results() if archived_path: st.session_state.ceremony_archived = archived_path # CATEGORIES 순서로 reveal (손선풍기 → 양우산 → 팜레스트) results = [] for col, label, prize in CATEGORIES: result = winners.get(col, {}) if result.get("status") == "ok": results.append( (label, prize, result["team"], result["votes"], result["diff"]) ) if "ceremony_step" not in st.session_state: st.session_state.ceremony_step = 0 if "ceremony_revealed" not in st.session_state: st.session_state.ceremony_revealed = False st.markdown( """ """, unsafe_allow_html=True, ) step = st.session_state.ceremony_step if step == 0: st.markdown('