304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""백그라운드 워커 — APScheduler (별도 컨테이너).
|
|
|
|
조사된 '필요한 시간들'을 모두 자동 처리하는 단일 프로세스:
|
|
|
|
1) 상태 전이 (status_tick_seconds 주기, 기본 60초)
|
|
scheduled → open(킥오프 D-2) → locked(킥오프 정각) — now 기준 자동 갱신.
|
|
|
|
2) AI 예측 생성 (매일 KST ai_generate_hour:ai_generate_minute, 기본 00:05)
|
|
남은(미종료) 경기에 대해 GPT/Claude/Gemini 실 API 호출 → ai_predictions 갱신.
|
|
모델별 독립 처리(키 없거나 실패해도 나머지 모델은 진행).
|
|
|
|
3) 결과 메일 발송 (status_tick 와 함께 점검)
|
|
경기 종료(finished_at) 후 result_email_delay_minutes(기본 180=3시간) 경과 시,
|
|
구독자(notify=True+email)에게 개인화 결과 메일 발송 → notified 표시.
|
|
|
|
실행: python -m app.worker
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from .config import settings
|
|
from .database import SessionLocal, init_db
|
|
from .domain import compute_phase, ensure_aware, now_utc
|
|
from .models import AIPrediction, Match, UserPrediction
|
|
from .scoring import load_scoring_data
|
|
from .services.ai import MatchContext, PROVIDERS, ProviderUnavailable
|
|
from .services.email import EmailUnavailable, build_result_email, send_email
|
|
from .services.grading import apply_result
|
|
from .services.schedule_fetch import fetch_results, fetch_schedule
|
|
from .services.schedule_sync import sync_schedule
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
log = logging.getLogger("triplepick.worker")
|
|
|
|
|
|
# ── 0) 경기 일정 동기화 (외부 크롤링) ──────────────────────
|
|
async def sync_schedule_job() -> None:
|
|
records = await fetch_schedule()
|
|
if not records:
|
|
return
|
|
async with SessionLocal() as db:
|
|
result = await sync_schedule(db, records)
|
|
# 일정 변경 후 즉시 상태/투표시간 재평가
|
|
await tick_status()
|
|
# 새 경기가 삽입됐으면 그 경기 AI 예측을 바로 생성(다음 주기까지 비어있지 않게)
|
|
if result.get("inserted"):
|
|
log.info("schedule: 신규 %d경기 → AI 예측 즉시 생성", result["inserted"])
|
|
await generate_ai_predictions()
|
|
|
|
|
|
# ── 1) 상태 전이 ────────────────────────────────────────────
|
|
async def tick_status() -> None:
|
|
async with SessionLocal() as db:
|
|
now = now_utc()
|
|
matches = (await db.execute(select(Match))).scalars().all()
|
|
changed = 0
|
|
for m in matches:
|
|
# 화면용 phase 와 동일 기준으로 status 저장(단일 기준):
|
|
# scheduled → open → locked(투표종료) → live(경기중) → finished
|
|
target = compute_phase(m, now)
|
|
if m.status != target:
|
|
m.status = target
|
|
changed += 1
|
|
if changed:
|
|
await db.commit()
|
|
log.info("status_tick: %d matches updated", changed)
|
|
await maybe_send_result_emails()
|
|
|
|
|
|
# ── 2) AI 예측 생성 (실연동) ────────────────────────────────
|
|
async def generate_ai_predictions(only_missing: bool = True) -> None:
|
|
"""미종료 경기의 AI 예측 생성.
|
|
|
|
only_missing=True(기본): 이미 있는 (경기×모델) 예측은 보존하고 **비어 있는 것만**
|
|
생성한다(직전에 실패/누락된 모델만 채워짐 — API 비용↓, 예측 안정).
|
|
only_missing=False: 전부 재생성(덮어쓰기) — 수동 재생성 스크립트용.
|
|
"""
|
|
async with SessionLocal() as db:
|
|
matches = (
|
|
await db.execute(
|
|
select(Match)
|
|
.where(Match.result_outcome.is_(None))
|
|
.options(selectinload(Match.predictions))
|
|
)
|
|
).scalars().all()
|
|
|
|
for m in matches:
|
|
ctx = MatchContext(
|
|
team_a=m.team_a_name,
|
|
team_b=m.team_b_name,
|
|
venue=m.venue,
|
|
kickoff=m.kickoff_at.isoformat(),
|
|
)
|
|
for model, fn in PROVIDERS.items():
|
|
pred = next((p for p in m.predictions if p.model == model), None)
|
|
if only_missing and pred is not None:
|
|
continue # 이미 작성됨 — 보존, API 호출 안 함
|
|
try:
|
|
data = await fn(ctx)
|
|
except ProviderUnavailable as e:
|
|
log.warning("AI %s skipped (%s)", model, e)
|
|
continue
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("AI %s failed for %s: %s", model, m.match_id, e)
|
|
continue
|
|
|
|
if pred is None: # 신규 — 위에서 못 찾았으면 생성
|
|
pred = AIPrediction(match_id=m.match_id, model=model)
|
|
db.add(pred)
|
|
m.predictions.append(pred)
|
|
pred.outcome = data["outcome"]
|
|
pred.score_a = data["scoreA"]
|
|
pred.score_b = data["scoreB"]
|
|
pred.confidence_pct = data["confidencePct"]
|
|
pred.reason_ko = data["reasonKo"]
|
|
pred.reason_en = data["reasonEn"]
|
|
pred.generated_at = now_utc()
|
|
pred.source = "llm"
|
|
log.info("AI %s → %s %d-%d", model, m.match_id, data["scoreA"], data["scoreB"])
|
|
await db.commit()
|
|
|
|
|
|
# ── 2.5) 결과 자동 정산 (관리자 입력 불필요) ────────────────
|
|
async def settle_matches() -> None:
|
|
"""킥오프 + result_settle_hours 경과한 미정산 경기를 외부 스코어로 자동 채점.
|
|
|
|
외부 소스(openfootball/football-data)에서 확정 스코어를 수집해 apply_result
|
|
호출 → 채점·유저 포인트 집계·finished_at 기록. 직후 결과 메일을 발송한다.
|
|
소스에 스코어가 아직 없으면 다음 틱에 재시도.
|
|
"""
|
|
now = now_utc()
|
|
cutoff = now - timedelta(hours=settings.result_settle_hours)
|
|
async with SessionLocal() as db:
|
|
pending = (
|
|
await db.execute(select(Match).where(Match.result_outcome.is_(None)))
|
|
).scalars().all()
|
|
due = [m.match_id for m in pending if ensure_aware(m.kickoff_at) <= cutoff]
|
|
|
|
if due:
|
|
results = await fetch_results()
|
|
by_pair = {(r["teamA"], r["teamB"]): r for r in results}
|
|
settled = 0
|
|
async with SessionLocal() as db:
|
|
for mid in due:
|
|
m = await db.get(Match, mid)
|
|
if m is None or m.result_outcome is not None:
|
|
continue
|
|
r = by_pair.get((m.team_a_code, m.team_b_code))
|
|
if not r:
|
|
continue
|
|
await apply_result(db, m, r["scoreA"], r["scoreB"]) # 채점+집계+commit
|
|
settled += 1
|
|
log.info("settle: %s 자동 정산 %d-%d", mid, r["scoreA"], r["scoreB"])
|
|
if settled == 0:
|
|
log.info("settle: %d경기 정산 대기 — 외부 스코어 아직 없음", len(due))
|
|
|
|
await maybe_send_result_emails()
|
|
|
|
|
|
# ── 3) 결과 메일 ────────────────────────────────────────────
|
|
async def maybe_send_result_emails() -> None:
|
|
async with SessionLocal() as db:
|
|
cutoff = now_utc() - timedelta(minutes=settings.result_email_delay_minutes)
|
|
matches = (
|
|
await db.execute(
|
|
select(Match)
|
|
.where(
|
|
Match.result_outcome.is_not(None),
|
|
Match.finished_at.is_not(None),
|
|
Match.results_emailed_at.is_(None),
|
|
)
|
|
.options(selectinload(Match.predictions))
|
|
)
|
|
).scalars().all()
|
|
|
|
for m in matches:
|
|
if m.finished_at and ensure_aware(m.finished_at) > cutoff:
|
|
continue # 아직 지연시간(3시간) 미경과
|
|
|
|
conds = [
|
|
UserPrediction.match_id == m.match_id,
|
|
UserPrediction.notify.is_(True),
|
|
UserPrediction.email.is_not(None),
|
|
UserPrediction.notified.is_(False),
|
|
]
|
|
# '맞춘 사람만' 옵션: 승패(outcome) 적중자에게만 발송
|
|
if settings.result_email_correct_only:
|
|
conds.append(UserPrediction.outcome == m.result_outcome)
|
|
picks = (
|
|
await db.execute(select(UserPrediction).where(*conds))
|
|
).scalars().all()
|
|
|
|
ai_lines = [(p.model, p.outcome == m.result_outcome) for p in m.predictions]
|
|
match_url = f"{settings.public_origin}/match/{m.match_id}"
|
|
sent_any = False
|
|
for pk in picks:
|
|
subject, html, text = build_result_email(
|
|
team_a=m.team_a_short,
|
|
team_b=m.team_b_short,
|
|
result_a=m.result_score_a or 0,
|
|
result_b=m.result_score_b or 0,
|
|
my_a=pk.score_a,
|
|
my_b=pk.score_b,
|
|
my_points=pk.points or 0,
|
|
ai_lines=ai_lines,
|
|
match_url=match_url,
|
|
)
|
|
try:
|
|
await send_email(pk.email, subject, html, text) # type: ignore[arg-type]
|
|
pk.notified = True
|
|
sent_any = True
|
|
except EmailUnavailable as e:
|
|
log.warning("result email skipped (%s)", e)
|
|
break # SMTP 미설정 — 이 경기는 다음 틱에 재시도
|
|
except Exception as e: # noqa: BLE001
|
|
log.error("result email failed → %s: %s", pk.email, e)
|
|
|
|
# 모든 구독자에게 시도 완료(또는 구독자 없음)면 발송완료 표시
|
|
still_pending = any(
|
|
(not pk.notified) for pk in picks
|
|
)
|
|
if not still_pending:
|
|
m.results_emailed_at = now_utc()
|
|
await db.commit()
|
|
|
|
|
|
def build_scheduler() -> AsyncIOScheduler:
|
|
sched = AsyncIOScheduler(timezone=settings.timezone)
|
|
# 경기 일정: 매일 KST 09:00 외부 크롤링 → 경기·투표시간 갱신
|
|
sched.add_job(
|
|
sync_schedule_job,
|
|
"cron",
|
|
hour=settings.schedule_sync_hour,
|
|
minute=settings.schedule_sync_minute,
|
|
id="schedule_sync",
|
|
)
|
|
# 투표시간 상태 전이 (open/locked)
|
|
sched.add_job(
|
|
tick_status,
|
|
"interval",
|
|
seconds=settings.status_tick_seconds,
|
|
id="status_tick",
|
|
next_run_time=now_utc(),
|
|
)
|
|
# AI 예측 생성: 매일 KST 00:05 (실 API)
|
|
sched.add_job(
|
|
generate_ai_predictions,
|
|
"cron",
|
|
hour=settings.ai_generate_hour,
|
|
minute=settings.ai_generate_minute,
|
|
id="ai_generate",
|
|
)
|
|
# 결과 자동 정산 + 메일: 킥오프+Nh 지난 경기 스코어 수집·채점·발송
|
|
sched.add_job(
|
|
settle_matches,
|
|
"interval",
|
|
seconds=settings.settle_tick_seconds,
|
|
id="settle",
|
|
next_run_time=now_utc(),
|
|
)
|
|
return sched
|
|
|
|
|
|
async def main() -> None:
|
|
load_scoring_data() # data/scoring.json → 배점·배제 대상 (채점·결과메일에 사용)
|
|
await init_db() # 테이블 보장 (idempotent). 시드는 API 가 담당.
|
|
# 기동 시 1회: 일정 동기화 → AI 예측 생성 → 밀린 경기 자동 정산
|
|
await sync_schedule_job()
|
|
await generate_ai_predictions()
|
|
await settle_matches()
|
|
|
|
sched = build_scheduler()
|
|
sched.start()
|
|
log.info(
|
|
"scheduling-server started — schedule sync daily %02d:%02d · status every %ss · "
|
|
"AI gen daily %02d:%02d %s · auto-settle 킥오프+%dh (every %ss) · 맞춘사람만=%s",
|
|
settings.schedule_sync_hour,
|
|
settings.schedule_sync_minute,
|
|
settings.status_tick_seconds,
|
|
settings.ai_generate_hour,
|
|
settings.ai_generate_minute,
|
|
settings.timezone,
|
|
settings.result_settle_hours,
|
|
settings.settle_tick_seconds,
|
|
settings.result_email_correct_only,
|
|
)
|
|
# 영구 대기
|
|
stop = asyncio.Event()
|
|
try:
|
|
await stop.wait()
|
|
finally:
|
|
sched.shutdown()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|