Harden infra validation flow

backend/app/main.py

@@ -4,6 +4,7 @@ from datetime import datetime
 from pathlib import Path
 import csv
 from io import BytesIO, StringIO
+import re
 import shutil
 import uuid
 
@@ -57,20 +58,41 @@ class MemberBulkPayload(BaseModel):
 
 LEGACY_HEADER_MAP = {
     "이름": "name",
+    "name": "name",
     "소속회사": "company",
+    "co": "company",
+    "company": "company",
     "직급": "rank",
+    "rank": "rank",
     "직책": "role",
+    "pos": "role",
+    "role": "role",
     "부서": "department",
+    "part": "department",
+    "department": "department",
     "그룹": "grp",
+    "gr": "grp",
+    "grp": "grp",
     "디비전": "division",
+    "div": "division",
+    "division": "division",
     "팀": "team",
+    "team": "team",
+    "teal": "team",
     "셀": "cell",
+    "cell": "cell",
     "근무상태": "work_status",
+    "work_status": "work_status",
     "근무시간": "work_time",
+    "work_time": "work_time",
     "전화번호": "phone",
+    "phone": "phone",
     "이메일": "email",
+    "email": "email",
     "자리위치": "seat_label",
+    "seat_label": "seat_label",
     "사진": "photo_url",
+    "photo_url": "photo_url",
 }
 
 
@@ -110,6 +132,20 @@ def fetch_members() -> list[dict[str, object]]:
             return cur.fetchall()
 
 
+def get_member_count() -> int:
+    with get_conn() as conn:
+        with conn.cursor() as cur:
+            cur.execute("SELECT COUNT(*) AS count FROM members")
+            return int(cur.fetchone()["count"])
+
+
+def ensure_snapshot_month(snapshot_month: str) -> str:
+    value = snapshot_month.strip()
+    if not re.fullmatch(r"\d{4}-\d{2}", value):
+        raise HTTPException(status_code=400, detail="snapshot_month 형식은 YYYY-MM 이어야 합니다.")
+    return value
+
+
 def replace_members(items: list[MemberPayload]) -> list[dict[str, object]]:
     with get_conn() as conn:
         with conn.cursor() as cur:
@@ -130,18 +166,22 @@ def replace_members(items: list[MemberPayload]) -> list[dict[str, object]]:
 
 
 def rows_to_member_payloads(rows: list[list[object]]) -> list[MemberPayload]:
+    def normalize_header(value: object) -> str:
+        return str(value or "").strip().lower()
+
     header_idx = next(
         (
             idx
             for idx, row in enumerate(rows)
-            if "이름" in row and "부서" in row
+            if {"이름", "부서"}.issubset({str(value).strip() for value in row})
+            or {"name", "part"}.issubset({normalize_header(value) for value in row})
         ),
         -1,
     )
     if header_idx < 0:
-        raise HTTPException(status_code=400, detail="지원하지 않는 파일 형식입니다. '이름'과 '부서' 헤더를 찾지 못했습니다.")
+        raise HTTPException(status_code=400, detail="지원하지 않는 파일 형식입니다. 필수 헤더(이름/부서 또는 name/part)를 찾지 못했습니다.")
 
-    headers = [str(value).strip() for value in rows[header_idx]]
+    headers = [normalize_header(value) for value in rows[header_idx]]
     payloads: list[MemberPayload] = []
 
     for row in rows[header_idx + 1 :]:
@@ -185,8 +225,26 @@ app.mount("/legacy/static", StaticFiles(directory=LEGACY_STATIC_DIR, check_dir=F
 
 
 @app.get("/api/health")
-def health() -> dict[str, str]:
-    return {"status": "ok"}
+def health() -> dict[str, object]:
+    checks = {
+        "upload_dir": UPLOAD_DIR.exists(),
+        "snapshot_dir": SNAPSHOT_DIR.exists(),
+    }
+
+    try:
+        member_count = get_member_count()
+        checks["database"] = True
+    except Exception:
+        member_count = None
+        checks["database"] = False
+
+    status = "ok" if all(checks.values()) else "degraded"
+    return {
+        "status": status,
+        "checks": checks,
+        "member_count": member_count,
+        "timestamp": datetime.utcnow().isoformat() + "Z",
+    }
 
 
 @app.post("/api/mock-login")
@@ -311,11 +369,18 @@ def upload_profile_photo(file: UploadFile = File(...), member_name: str = Form("
 
 @app.post("/api/snapshots/monthly")
 def create_monthly_snapshot(snapshot_month: str = Form(...)) -> dict[str, str]:
+    snapshot_month = ensure_snapshot_month(snapshot_month)
     filename = f"organization-snapshot-{snapshot_month}.csv"
     target = SNAPSHOT_DIR / filename
 
     with get_conn() as conn:
         with conn.cursor() as cur:
+            cur.execute(
+                "SELECT id FROM snapshots WHERE snapshot_month = %s ORDER BY created_at DESC LIMIT 1",
+                (snapshot_month,),
+            )
+            if cur.fetchone() is not None:
+                raise HTTPException(status_code=409, detail="해당 월의 스냅샷이 이미 존재합니다.")
             cur.execute(
                 """
                 SELECT name, company, rank, role, department, grp, division, team, cell,