Feat: [sm/auth] admin api 인증 헤더 추가

This commit is contained in:
sm4640
2026-01-14 17:24:47 +09:00
parent 122b367bed
commit 1a67844e41
2 changed files with 17 additions and 2 deletions

7
app.py
View File

@@ -6,11 +6,11 @@ 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
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_importer import import_workbook
from workbook_enricher import enrich_workbook
# from workbook_importer import import_workbook
load_dotenv()
@@ -30,6 +30,7 @@ async def admin_enrich_workbook(
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,
@@ -45,6 +46,7 @@ async def admin_enrich_workbook(
async def reset_workbook_progress(
workbook_id: int,
db: AsyncSession = Depends(get_db),
_: None = Depends(require_admin),
):
try:
res = await db.execute(
@@ -174,6 +176,7 @@ async def today(
}
# 백준 사이트는 beautifulsoap 크롤링이 안됨
# @app.post("/admin/workbooks/{workbook_id}/import")
# async def admin_import_workbook(
# workbook_id: int,

View File

@@ -5,6 +5,8 @@ from typing import Optional, Tuple, List
import requests
from fastapi import HTTPException, Header
# ====== HTTP Session ======
SESSION = requests.Session()
SESSION.headers.update({"User-Agent": "baekjoon-n8n-bot/1.0"})
@@ -13,6 +15,16 @@ SESSION.headers.update({"User-Agent": "baekjoon-n8n-bot/1.0"})
KNOWN_LANGS = ["ko", "en", "ja", "ru", "zh", "de", "fr", "es", "pt", "it"]
def require_admin(x_admin_password: str | None = Header(default=None, alias="X-Admin-Password")):
expected = env("ADMIN_PASSWORD", "")
if not expected:
raise HTTPException(status_code=500, detail="ADMIN_PASSWORD is not configured")
if not x_admin_password or x_admin_password != expected:
raise HTTPException(status_code=403, detail="invalid admin password")
def fetch_json_with_retry(url: str, params: dict, retries: int = 3, timeout=(3.05, 10)) -> dict:
last_err = None
for i in range(retries):