from typing import Optional from fastapi import FastAPI, HTTPException, Query, Depends from fastapi.responses import JSONResponse from dotenv import load_dotenv from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from utils import env, resolve_difficulty, resolve_tags, build_query, get_problem, require_admin from db import get_db from workbook_picker import pick_from_workbook from workbook_enricher import enrich_workbook # from workbook_importer import import_workbook load_dotenv() app = FastAPI() @app.get("/") def root(): return {"status": "ok"} @app.post("/admin/workbooks/{workbook_id}/enrich") async def admin_enrich_workbook( workbook_id: int, only_missing: bool = Query(True, description="True면 NULL만 채움 / False면 덮어씀"), commit_every: int = Query(50, ge=1, le=500, description="몇 개마다 commit 할지"), sleep_sec: float = Query(0.12, ge=0.0, le=2.0, description="solved.ac 호출 사이 sleep"), db: AsyncSession = Depends(get_db), _: None = Depends(require_admin), ): result = await enrich_workbook( db, workbook_id=workbook_id, only_missing=only_missing, commit_every=commit_every, sleep_sec=sleep_sec, ) return {"status": "ok", "result": result} @app.delete("/admin/workbooks/{workbook_id}/reset") async def reset_workbook_progress( workbook_id: int, db: AsyncSession = Depends(get_db), _: None = Depends(require_admin), ): try: res = await db.execute( text("DELETE FROM workbook_sends WHERE workbook_id = :wid"), {"wid": workbook_id}, ) await db.commit() # res.rowcount: 삭제된 행 수(=초기화된 문제 수) return { "status": "ok", "workbook_id": workbook_id, "deleted_sends": int(res.rowcount or 0), "message": "workbook progress reset (problems can be picked again)", } except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/today") async def today( source_mode: str = Query(env("SOURCE_MODE_DEFAULT", "search"), description="search|workbook"), workbook_id: Optional[int] = Query(None, description="문제집 모드일 때 workbook id"), workbook_pick: str = Query("level_asc", description="random|level_asc"), difficulty_mode: str = Query(env("DIFFICULTY_MODE_DEFAULT", "easy"), description="easy|hard|all"), tag_mode: str = Query(env("TAG_MODE_DEFAULT", "easy"), description="easy|hard|all"), difficulty: Optional[str] = Query(None, description="예: 6..10 (주면 mode보다 우선)"), tags: Optional[str] = Query(None, description="예: dp,graphs (주면 mode보다 우선)"), lang: str = Query(env("LANG_DEFAULT", "all"), description="ko | en | ko,en | all"), db: AsyncSession = Depends(get_db), ): sm = (source_mode or "").lower().strip() if sm == "workbook": wid = workbook_id or (int(env("WORKBOOK_ID_DEFAULT")) if env("WORKBOOK_ID_DEFAULT") else None) if not wid: return JSONResponse(status_code=400, content={"error": "workbook_id is required for workbook mode"}) pid, title, level, current_idx, total_cnt = await pick_from_workbook(db, wid, pick=workbook_pick) if not pid: return JSONResponse(status_code=409, content={"error": "no_more_problems_in_workbook", "workbook_id": wid}) problem_url = f"https://www.acmicpc.net/problem/{pid}" solved_url = f"https://solved.ac/problems/id/{pid}" level_text = f"Lv. {level}" if level is not None else "Lv. ?" progress_text = f"({current_idx}/{total_cnt})" if current_idx and total_cnt else "" discord_payload = { "embeds": [{ "title": f"🔔 오늘의 백준 추천 문제 (문제집) {progress_text}", "description": ( f"**{pid}번: {title}**\n" f"난이도: **{level_text}**\n" f"source_mode: `workbook` / workbook_id: `{wid}`" ), "fields": [ {"name": "문제 링크", "value": f"[바로가기]({problem_url})", "inline": True}, {"name": "해설/정보", "value": f"[Solved.ac]({solved_url})", "inline": True}, ], }] } return { "source_mode": "workbook", "workbook_id": wid, "problemId": pid, "title": title, "level": level, "problemUrl": problem_url, "solvedUrl": solved_url, "discordPayload": discord_payload, } # 2) 기존 search 모드(네가 쓰던 그대로) dm = (difficulty_mode or "").lower() tm = (tag_mode or "").lower() if dm not in ("easy", "hard", "all"): return JSONResponse(status_code=400, content={"error": "difficulty_mode must be easy|hard|all"}) if tm not in ("easy", "hard", "all"): return JSONResponse(status_code=400, content={"error": "tag_mode must be easy|hard|all"}) chosen_difficulty = resolve_difficulty(difficulty, dm) chosen_tags = resolve_tags(tags if tags is not None else None, tm) query = build_query(chosen_difficulty, chosen_tags, lang) pid, title, level = get_problem(query=query) if not pid: return JSONResponse(status_code=503, content={"error": "failed_to_fetch_problem", "query": query}) problem_url = f"https://www.acmicpc.net/problem/{pid}" solved_url = f"https://solved.ac/problems/id/{pid}" level_text = f"Lv. {level}" if level is not None else "Lv. ?" discord_payload = { "embeds": [{ "title": "🔔 오늘의 백준 추천 문제", "description": ( f"**{pid}번: {title}**\n" f"난이도: **{level_text}**\n" f"difficulty_mode: `{dm}` / tag_mode: `{tm}` / lang: `{lang}`\n" f"filter: `{chosen_difficulty}` / tags: `{', '.join(chosen_tags) if chosen_tags else 'none'}`" ), "fields": [ {"name": "문제 링크", "value": f"[바로가기]({problem_url})", "inline": True}, {"name": "해설/정보", "value": f"[Solved.ac]({solved_url})", "inline": True}, ], }] } return { "source_mode": "search", "difficulty_mode": dm, "tag_mode": tm, "lang": lang, "difficulty": chosen_difficulty, "tags": chosen_tags, "query": query, "problemId": pid, "title": title, "level": level, "problemUrl": problem_url, "solvedUrl": solved_url, "discordPayload": discord_payload, } # 백준 사이트는 beautifulsoap 크롤링이 안됨 # @app.post("/admin/workbooks/{workbook_id}/import") # async def admin_import_workbook( # workbook_id: int, # title: Optional[str] = Query(None, description="문제집 제목(옵션)"), # db: AsyncSession = Depends(get_db), # ): # try: # result = await import_workbook(db, workbook_id=workbook_id, title=title) # return {"status": "ok", "result": result} # except Exception as e: # return JSONResponse(status_code=500, content={"error": str(e)})