131 lines
3.8 KiB
Python
131 lines
3.8 KiB
Python
"""도메인 헬퍼 — 투표 단계(phase) 계산 + ORM→응답 스키마 직렬화.
|
|
|
|
phase 는 now 기준 동적 계산 (lib/schedule.ts matchPhase 와 동일 규칙):
|
|
result 존재 → finished
|
|
now >= lockAt → locked
|
|
now < opensAt → scheduled
|
|
그 외 → open
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from .config import settings
|
|
from .models import AIPrediction, CrowdStats, Match
|
|
from .schemas import (
|
|
AIPredictionOut,
|
|
CrowdStatsOut,
|
|
MatchOut,
|
|
MatchResult,
|
|
Team,
|
|
)
|
|
|
|
|
|
def now_utc() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
def ensure_aware(dt: datetime) -> datetime:
|
|
"""naive datetime(일부 DB 드라이버 반환)을 UTC aware 로 보정."""
|
|
return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt
|
|
|
|
|
|
def compute_phase(m: Match, now: datetime | None = None) -> str:
|
|
now = now or now_utc()
|
|
if m.result_outcome is not None:
|
|
return "finished"
|
|
if now >= ensure_aware(m.lock_at):
|
|
return "locked"
|
|
if now < ensure_aware(m.opens_at):
|
|
return "scheduled"
|
|
return "open"
|
|
|
|
|
|
def is_open_for_voting(m: Match, now: datetime | None = None) -> bool:
|
|
"""제출 허용 여부. demo_force_open 이면 마감 전까지 항상 오픈."""
|
|
now = now or now_utc()
|
|
if m.result_outcome is not None:
|
|
return False
|
|
if now >= ensure_aware(m.lock_at):
|
|
return False
|
|
if settings.demo_force_open:
|
|
return True
|
|
return now >= ensure_aware(m.opens_at)
|
|
|
|
|
|
def _team_a(m: Match) -> Team:
|
|
return Team(
|
|
name=m.team_a_name, shortName=m.team_a_short, code=m.team_a_code, flag=m.team_a_flag
|
|
)
|
|
|
|
|
|
def _team_b(m: Match) -> Team:
|
|
return Team(
|
|
name=m.team_b_name, shortName=m.team_b_short, code=m.team_b_code, flag=m.team_b_flag
|
|
)
|
|
|
|
|
|
def prediction_out(p: AIPrediction, lang: str = "ko") -> AIPredictionOut:
|
|
reason = p.reason_en if lang == "en" and p.reason_en else p.reason_ko
|
|
return AIPredictionOut(
|
|
matchId=p.match_id,
|
|
model=p.model, # type: ignore[arg-type]
|
|
outcome=p.outcome, # type: ignore[arg-type]
|
|
scoreA=p.score_a,
|
|
scoreB=p.score_b,
|
|
confidencePct=p.confidence_pct,
|
|
reasonShort=reason,
|
|
generatedAt=p.generated_at.date().isoformat() if p.generated_at else "",
|
|
)
|
|
|
|
|
|
def crowd_out(c: CrowdStats | None, match_id: str) -> CrowdStatsOut:
|
|
if c is None:
|
|
return CrowdStatsOut(matchId=match_id, total=0, teamAWin=0, draw=0, teamBWin=0)
|
|
return CrowdStatsOut(
|
|
matchId=match_id,
|
|
total=c.total,
|
|
teamAWin=c.team_a_win,
|
|
draw=c.draw,
|
|
teamBWin=c.team_b_win,
|
|
)
|
|
|
|
|
|
def match_out(
|
|
m: Match,
|
|
lang: str = "ko",
|
|
include_predictions: bool = True,
|
|
now: datetime | None = None,
|
|
) -> MatchOut:
|
|
result = None
|
|
if m.result_outcome is not None:
|
|
result = MatchResult(
|
|
scoreA=m.result_score_a or 0,
|
|
scoreB=m.result_score_b or 0,
|
|
outcome=m.result_outcome, # type: ignore[arg-type]
|
|
)
|
|
preds: list[AIPredictionOut] = []
|
|
if include_predictions:
|
|
# 모델 순서 고정: GPT, Claude, Gemini
|
|
order = {"GPT": 0, "Claude": 1, "Gemini": 2}
|
|
for p in sorted(m.predictions, key=lambda x: order.get(x.model, 9)):
|
|
preds.append(prediction_out(p, lang))
|
|
return MatchOut(
|
|
matchId=m.match_id,
|
|
roundLabel=m.round_label,
|
|
group=m.group,
|
|
teamA=_team_a(m),
|
|
teamB=_team_b(m),
|
|
kickoffKst=m.kickoff_at.isoformat(),
|
|
venue=m.venue,
|
|
opensAt=m.opens_at.isoformat(),
|
|
lockAt=m.lock_at.isoformat(),
|
|
status=m.status,
|
|
phase=compute_phase(m, now),
|
|
votingOpen=is_open_for_voting(m, now),
|
|
hookText=m.hook_text,
|
|
result=result,
|
|
predictions=preds,
|
|
crowd=crowd_out(m.crowd, m.match_id),
|
|
)
|