hackerthon-vote/app.py

"""
해커톤 투표 앱
- 참가자: 이름 선택 → 본인 팀 자동 → 본인 팀 제외 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)
    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 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 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("🗳 해커톤 투표")

    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

    name = st.selectbox(
        "본인 이름",
        options=sorted(PARTICIPANTS.keys()),
        index=None,
        placeholder="이름 선택",
    )
    employee_id = st.text_input(
        "사번",
        placeholder="본인 사번 입력 (사칭 추적용으로 기록됨)",
    )

    if not name:
        st.info("이름을 선택하면 본인 팀이 자동 표시됩니다.")
        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("🔐 진행자 콘솔")

    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)
            st.download_button("CSV 내려받기", buf.getvalue(), file_name="audit.csv")
            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 render_ceremony():
    """시상식 reveal 페이지. 진행자가 클릭으로 단계별 공개."""
    token = st.query_params.get("token", "")
    if token != ADMIN_TOKEN:
        st.error("권한 없음. ?mode=ceremony&token=... 형식 필요.")
        return

    titles = get_titles()
    winners, _ = compute_winners()

    # 동률 미해결 있으면 ceremony 차단
    pending = [
        (col, label) for col, label, _ in CATEGORIES
        if winners.get(col, {}).get("status") == "tie"
    ]
    if pending:
        st.error("⚠️ 동률 미해결 — 어드민 페이지에서 추첨/선택 후 재진입")
        for col, label in pending:
            tied = winners[col]["tied"]
            tied_labels = ", ".join(fmt_team(t, titles) for t in tied)
            st.write(f"- {label}: {tied_labels}")
        return

    # 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(
        """
        <style>
        .stage-title { font-size: 70px; text-align: center; padding: 30px 0; }
        .stage-prize { font-size: 32px; text-align: center; color: #888; }
        .stage-drum { font-size: 90px; text-align: center; padding: 60px 0; }
        .winner-name {
            font-size: 110px; text-align: center; font-weight: bold;
            background: linear-gradient(90deg, #ffd700, #ff8c00);
            -webkit-background-clip: text;
            -webkit-text-fill-color: transparent;
            padding: 40px 0;
        }
        .winner-meta { font-size: 36px; text-align: center; padding: 20px 0; }
        .stReveal { animation: fadeIn 1.5s ease-in; }
        @keyframes fadeIn {
            from { opacity: 0; transform: scale(0.5); }
            to { opacity: 1; transform: scale(1); }
        }
        </style>
        """,
        unsafe_allow_html=True,
    )

    step = st.session_state.ceremony_step

    if step == 0:
        st.markdown('<div class="stage-title">🎉 해커톤 시상식 🎉</div>', unsafe_allow_html=True)
        st.markdown('<div class="winner-meta">준비됐습니다</div>', unsafe_allow_html=True)
        if st.button("시작 →", use_container_width=True, type="primary"):
            st.session_state.ceremony_step = 1
            st.session_state.ceremony_revealed = False
            st.rerun()

    elif step > len(results):
        st.markdown('<div class="stage-title">수고하셨습니다!</div>', unsafe_allow_html=True)
        st.balloons()
        st.snow()
        if st.button("처음으로", use_container_width=True):
            st.session_state.ceremony_step = 0
            st.session_state.ceremony_revealed = False
            st.rerun()

    else:
        label, prize, winner, votes, diff = results[step - 1]
        st.markdown(f'<div class="stage-title">{label}</div>', unsafe_allow_html=True)
        st.markdown(f'<div class="stage-prize">🎁 상품: {prize}</div>', unsafe_allow_html=True)

        if not st.session_state.ceremony_revealed:
            st.markdown('<div class="stage-drum">🥁🥁🥁</div>', unsafe_allow_html=True)
            if st.button("우승팀 공개 →", use_container_width=True, type="primary"):
                st.session_state.ceremony_revealed = True
                st.rerun()
        else:
            st.balloons()
            winner_label = fmt_team(winner, titles)
            st.markdown(
                f'<div class="winner-name stReveal">🏆<br>{winner_label}</div>',
                unsafe_allow_html=True,
            )
            st.markdown(
                f'<div class="winner-meta">{votes}표 (2위와 {diff}표 차이)</div>',
                unsafe_allow_html=True,
            )
            next_label = "다음 부문 →" if step < len(results) else "마무리 →"
            if st.button(next_label, use_container_width=True, type="primary"):
                st.session_state.ceremony_step = step + 1
                st.session_state.ceremony_revealed = False
                st.rerun()


def main():
    st.set_page_config(page_title="해커톤 투표", page_icon="🗳", layout="wide")
    mode = st.query_params.get("mode", "vote")
    if mode == "admin":
        render_admin()
    elif mode == "ceremony":
        render_ceremony()
    else:
        render_voter()


if __name__ == "__main__":
    main()