from __future__ import annotations from typing import List, Dict from fastapi import APIRouter, Request, Query from fastapi.responses import HTMLResponse, JSONResponse from app.web.deps import templates, sb router = APIRouter() @router.get("/gov/me/heads", response_class=HTMLResponse) async def gov_me_heads(request: Request, q: str = Query(default="")): """환경부(ME) 부서별 책임자 표. 1) me_heads_cur 뷰가 있으면 그것을 우선 사용 2) 없으면 me_org_cur에서 부서별 리더를 단순 휴리스틱으로 선택 """ # 1) 뷰 우선 try: v = ( sb.table("me_heads_cur") .select("unit,position,name,phone,task,indent,ord") .order("ord", desc=False) .limit(5000) .execute() .data ) or [] except Exception: v = [] items: List[dict] = [] key = (q or "").strip().lower() if v: for r in v: row = { "unit": r.get("unit"), "indent": r.get("indent") or 0, "name": (r.get("name") or "(공석)").strip() or "(공석)", "position": r.get("position"), "phone": (r.get("phone") or "-").strip() or "-", "task": (r.get("task") or "-").strip() or "-", } hay = " ".join([str(x or "") for x in row.values()]).lower() if not key or key in hay: items.append(row) return templates.TemplateResponse( "gov/me/heads.html", {"request": request, "ministry": "환경부(ME)", "q": q, "items": items}, ) # 2) 폴백: me_org_cur에서 간이 리더 추출 try: rows = ( sb.table("me_org_cur") .select("department,position,name,phone,task") .limit(50000) .execute() .data ) or [] except Exception: rows = [] # 부서별 후보 그룹 by_dep: Dict[str, List[dict]] = {} for r in rows: dep = (r.get("department") or "").strip() if dep: by_dep.setdefault(dep, []).append(r) def guess_rank_score(pos: str) -> int: p = (pos or "").strip() if p.endswith("장관"): return 100 if p.endswith("차관"): return 95 if p.endswith("실장"): return 90 if p.endswith("국장"): return 85 if p.endswith("관리관") or p.endswith("정책관") or p.endswith("기획관"): return 80 if p.endswith("과장"): return 70 if p.endswith("팀장"): return 60 return 10 # 알파벳/한글 순서로 정렬 후 표시 for dep in sorted(by_dep.keys()): cands = by_dep[dep] # 가장 높은 직위 점수, 전화/이름 유무 가점 def score(r: dict) -> tuple: s = guess_rank_score(r.get("position") or "") s += 5 if (r.get("phone") or "").strip() not in ("", "-") else 0 s += 1 if (r.get("name") or "").strip() else 0 # 최근 정렬 정보가 없으니 이름순 보조 return (s, (r.get("name") or "")) rec = sorted(cands, key=score, reverse=True)[0] row = { "unit": dep, "indent": 1 if dep.endswith("실") else (2 if dep.endswith("국") else (3 if dep.endswith(("과","팀")) else 0)), "name": (rec.get("name") or "(공석)").strip() or "(공석)", "position": (rec.get("position") or "-").strip() or "-", "phone": (rec.get("phone") or "-").strip() or "-", "task": (rec.get("task") or "-").strip() or "-", } if not key or key in " ".join([str(x or "") for x in row.values()]).lower(): items.append(row) return templates.TemplateResponse( "gov/me/heads.html", {"request": request, "ministry": "환경부(ME)", "q": q, "items": items}, ) @router.get("/gov/me/heads/debug", response_class=JSONResponse) async def gov_me_heads_debug(dep: str | None = Query(default=None), pos: str | None = Query(default=None), limit: int = 100): try: rows = ( sb.table("me_org_cur").select("department,position,name,phone,task").limit(50000).execute().data ) or [] except Exception: rows = [] out = [] for r in rows: d = (r.get("department") or "").strip(); p = (r.get("position") or "").strip() if dep and dep not in d: continue if pos and pos not in p: continue out.append(r) if len(out) >= limit: break return {"count": len(out), "rows": out} @router.get("/gov/me/heads/target_sync", response_class=JSONResponse) async def gov_me_heads_target_sync(apply: int = 0, update_existing: int = 0): """me_org_cur 기반으로 me_heads_target 자동 보강(테스트용).""" def guess_expected_pos(unit: str) -> str: u = (unit or "").strip() if u.endswith("본부장실"): return "본부장" if u.endswith("위원회"): return "위원장" if u.endswith("사무국"): return "사무국장" if u.endswith("실"): return "실장" if u.endswith("국"): return "국장" if u.endswith("관리관") or u.endswith("정책관") or u.endswith("기획관"): return "국장" if u.endswith("과"): return "과장" if u.endswith("팀"): return "팀장" if u == "대변인": return "대변인" return "-" def guess_indent(unit: str) -> int: u = (unit or "").strip() if u.endswith(("본부장실",)): return 0 if u.endswith(("실",)) or u in ("대변인",): return 1 if u.endswith(("정책관","기획관","국","위원회")): return 2 if u.endswith(("과","팀","사무국")): return 3 return 1 # deps try: deps = sb.table("me_org_cur").select("department").limit(50000).execute().data or [] except Exception: deps = [] dep_names = sorted({(r.get("department") or "").strip() for r in deps if (r.get("department") or "").strip()}) try: tgt = sb.table("me_heads_target").select("ord,unit,expected_pos,indent").order("ord", desc=False).limit(5000).execute().data or [] except Exception: tgt = [] by_unit = { (r.get("unit") or "").strip(): r for r in tgt } max_ord = max([r.get("ord") or 0 for r in tgt] + [0]) inserts, updates = [], [] for d in dep_names: if d not in by_unit: max_ord += 1 inserts.append({ "ord": max_ord, "unit": d, "expected_pos": guess_expected_pos(d), "indent": guess_indent(d), }) elif update_existing: cur = by_unit[d] new_ep = guess_expected_pos(d); new_ind = guess_indent(d) if (cur.get("expected_pos") != new_ep) or (cur.get("indent") != new_ind): updates.append({"ord": cur.get("ord"), "unit": d, "expected_pos": new_ep, "indent": new_ind}) preview = {"total_deps": len(dep_names), "existing_targets": len(tgt), "missing_units": len(inserts), "will_update": len(updates)} if not apply: preview["sample_inserts"] = inserts[:20] preview["sample_updates"] = updates[:20] return preview for i in range(0, len(inserts), 500): part = inserts[i:i+500] if part: sb.table("me_heads_target").upsert(part).execute() if update_existing and updates: for u in updates: sb.table("me_heads_target").update({"unit": u["unit"], "expected_pos": u["expected_pos"], "indent": u["indent"]}).eq("ord", u["ord"]).execute() return {"applied_inserts": len(inserts), "applied_updates": len(updates)}