from __future__ import annotations from typing import Optional, List, Dict, Any from datetime import datetime, date from collections import defaultdict from fastapi import APIRouter, Request, Query, Form, HTTPException from fastapi.responses import HTMLResponse, RedirectResponse from collections import OrderedDict import re from .core import ( templates, sb, logger, COMPANY_ORDER, POS_ORDER, TODAY, _norm_company_label, _normalize_position, _parse_date_parts, _parts_to_date, _fmt_date_display, _date_display_and_compare, _is_female, _is_kepco_alum_text, _today_items_from, _ping_status, _count_last_24h, _current_map_by_person_source, _normalize_name, _to_int_safe, _normalize_multiline, _MOTIE_ORDER_POS, ) from app.sync2.pipeline import apply_decisions router = APIRouter() # -------------------- 홈/헬스 -------------------- @router.get("/", response_class=HTMLResponse) async def index(request: Request): moef_today = _today_items_from("moef_id", "MOEF", title_key="title", url_key="url", tag_key="tag") motie_today = _today_items_from("motie_id", "MOTIE", title_key="title", url_key="url", tag_key=None) gov_today = moef_today + motie_today try: res_k = sb.table("kepco_id").select("department,posted_at,pdf_url").order("id", desc=True).limit(200).execute() k_rows = res_k.data or [] except Exception: k_rows = [] group_today = [] today_str = date.today().isoformat() for r in k_rows: d = str(r.get("posted_at") or r.get("created_at") or "")[:10] if d == today_str: group_today.append({"department": r.get("department") or "-", "url": r.get("pdf_url"), "date": d}) status = { "MOEF": {"ping": _ping_status("https://www.moef.go.kr"), "count24": _count_last_24h("moef_id")}, "MOTIE": {"ping": _ping_status("https://www.motie.go.kr"), "count24": _count_last_24h("motie_id")}, "ALIO": {"ping": _ping_status("https://www.alio.go.kr"), "count24": _count_last_24h("kepco_id")}, } return templates.TemplateResponse("index.html", { "request": request, "gov_today": gov_today, "group_today": group_today, "status": status, }) @router.get("/health") async def health(): return {"ok": True} # -------------------- 대정부 검색/타임라인 -------------------- @router.get("/gov/search", response_class=HTMLResponse) async def gov_search(request: Request, src: str = Query(default="all"), q: str = Query(default="")): q_norm = (q or "").strip() rows: List[Dict[str, Any]] = [] if not q_norm: return templates.TemplateResponse("gov_search.html", {"request": request, "src": src, "q": q_norm, "rows": rows}) try: query = sb.table("gov_staff").select("source,department,position,name,phone,task") if src in ("MOTIE", "MOEF"): query = query.eq("source", src) kw = f"%{q_norm}%" query = query.or_(f"name.ilike.{kw},department.ilike.{kw},position.ilike.{kw},task.ilike.{kw},phone.ilike.{kw}") res = query.limit(1000).execute() rows = res.data or [] except Exception as e: logger.warning(f"gov_staff search fallback (reason: {e})") try: q2 = sb.table("gov_staff").select("source,department,position,name,phone,task") if src in ("MOTIE", "MOEF"): q2 = q2.eq("source", src) data = (q2.limit(5000).execute().data) or [] key = q_norm.lower() def hit(r): return any(key in (r.get(f, "") or "").lower() for f in ("name","department","position","task","phone")) rows = [r for r in data if hit(r)] except Exception as e2: logger.error(f"gov_staff fallback failed: {e2}") rows = [] rows.sort(key=lambda r: (r.get("department", "") or "", r.get("name", "") or "")) return templates.TemplateResponse("gov_search.html", {"request": request, "src": src, "q": q_norm, "rows": rows}) def _ensure_moef_url(row: Dict[str, Any]) -> Optional[str]: DETAIL_URL = "https://www.moef.go.kr/nw/notice/hrDetail.do" menuNo = "4050300" bbsId = row.get("bbsId"); postId = row.get("postId") if not (bbsId and postId): rid = str(row.get("id") or "") if "-" in rid: parts = rid.split("-", 1) if len(parts) == 2: bbsId, postId = parts if bbsId and postId: return f"{DETAIL_URL}?searchBbsId1={bbsId}&searchNttId1={postId}&menuNo={menuNo}" return None def _ensure_motie_url(row: Dict[str, Any]) -> Optional[str]: BASE = "https://www.motie.go.kr" aid = row.get("id") if aid is None: return None return f"{BASE}/kor/article/ATCL6e90bb9de/{aid}/view?" @router.get("/gov/timeline", response_class=HTMLResponse) async def gov_timeline( request: Request, src: str = Query(default="all"), q: Optional[str] = Query(default=None), start: Optional[str] = Query(default=None), end: Optional[str] = Query(default=None), page: Optional[int] = Query(default=1), page_size: int = Query(default=20, ge=5, le=100), ): q_norm = (q or "").strip().lower() def _to_date(s: Optional[str]) -> Optional[date]: if not s: return None try: return datetime.strptime(str(s)[:10], "%Y-%m-%d").date() except Exception: return None s_date = _to_date(start); e_date = _to_date(end) page = page or 1 items: List[Dict[str, Any]] = [] def fetch_motie() -> List[Dict[str, Any]]: try: res = sb.table("motie_id").select("*").order("id", desc=True).limit(1000).execute() data = res.data or [] except Exception as e: logger.warning(f"motie_id fetch failed: {e}") data = [] for r in data: r["source"] = "MOTIE" r["url"] = _ensure_motie_url(r) return data def fetch_moef() -> List[Dict[str, Any]]: try: res = sb.table("moef_id").select("*").order("id", desc=True).limit(1000).execute() data = res.data or [] except Exception as e: logger.warning(f"moef_id fetch failed: {e}") data = [] for r in data: r["source"] = "MOEF" r["url"] = _ensure_moef_url(r) return data if src in ("MOTIE","all"): items.extend(fetch_motie()) if src in ("MOEF","all"): items.extend(fetch_moef()) if q_norm: def hit_text(r): title = (r.get("title") or "").lower() tag = (r.get("tag") or "").lower() return q_norm in title or q_norm in tag items = [r for r in items if hit_text(r)] if s_date or e_date: def d_of(r) -> Optional[date]: p = r.get("posted_at") or r.get("created_at") try: return datetime.strptime(str(p)[:10], "%Y-%m-%d").date() if p else None except Exception: return None def in_range(r): d = d_of(r) if not d: return False if s_date and d < s_date: return False if e_date and d > e_date: return False return True items = [r for r in items if in_range(r)] def sort_key(r): def d_try(k): try: return datetime.strptime(str(r.get(k) or "")[:10], "%Y-%m-%d").date() except Exception: return date.min return (d_try("posted_at"), d_try("created_at"), str(r.get("id") or "")) items.sort(key=sort_key, reverse=True) total = len(items) start_idx = (page-1)*page_size end_idx = start_idx + page_size page_rows = items[start_idx:end_idx] has_more = end_idx < total return templates.TemplateResponse("gov_timeline.html", { "request": request, "src": src, "q": q or "", "start": start or "", "end": end or "", "page": page, "rows": page_rows, "has_more": has_more, }) # -------------------- 그룹 타임라인/임원 -------------------- @router.get("/group/timeline", response_class=HTMLResponse) async def group_timeline(request: Request): try: res = sb.table("kepco_id").select("department,posted_at,pdf_url,created_at").limit(3000).execute() rows = res.data or [] except Exception as e: logger.warning(f"kepco_id fetch failed: {e}") rows = [] order_map = {name: i for i, name in enumerate(COMPANY_ORDER)} def dept_order(dep: str | None) -> int: label = _norm_company_label(dep) return order_map.get(label or "", 999) def parse_dt(any_s: Any) -> date: y, m, d = _parse_date_parts(str(any_s) if any_s is not None else "") return _parts_to_date(y, m, d, assume_last=True) or date.min rows.sort(key=lambda r: (dept_order(r.get("department")), -parse_dt(r.get("posted_at")).toordinal())) return templates.TemplateResponse("group_timeline.html", {"request": request, "rows": rows}) @router.get("/group/executive", response_class=HTMLResponse) async def group_executive(request: Request): try: res = sb.table("kepco_org").select("*").limit(5000).execute() raw = res.data or [] except Exception as e: logger.error(f"kepco_org fetch failed: {e}") raw = [] grid: dict[str, dict[str, list[dict]]] = {c:{p:[] for p in POS_ORDER} for c in COMPANY_ORDER} female_counts = {c:0 for c in COMPANY_ORDER} total_counts = {c:0 for c in COMPANY_ORDER} for r in raw: dep_raw = (r.get("department") or "").strip() dep = _norm_company_label(dep_raw) if dep not in grid: continue total_counts[dep] += 1 if _is_female(r.get("gender")): female_counts[dep] += 1 pos_norm = _normalize_position(r.get("position")) if pos_norm is None: continue y1, m1, d1 = _parse_date_parts(r.get("start")) y2, m2, d2 = _parse_date_parts(r.get("end")) end_dt = _parts_to_date(y2, m2, d2, assume_last=True) expired = bool(end_dt and end_dt < TODAY) person = dict(r) person["_expired"] = expired person["_is_kepco_alum"] = _is_kepco_alum_text(r.get("career")) person["_start_disp"] = _fmt_date_display(y1, m1, d1) person["_end_disp"] = _fmt_date_display(y2, m2, d2) gender = "여" if _is_female(r.get("gender")) else ("남" if r.get("gender") else "-") name = (r.get("name") or "-").strip() person["_name_line"] = f"{name}({gender})" if gender != "-" else name grid[dep][pos_norm].append(person) # def start_key(p: dict) -> date: # y, m, d = _parse_date_parts(p.get("start")) # return _parts_to_date(y, m, d, assume_last=False) or date.min # for dep in COMPANY_ORDER: # for p in ("상임이사","비상임이사"): # grid[dep][p].sort(key=start_key) gender_stats = [{"female": female_counts[c], "total": total_counts[c]} for c in COMPANY_ORDER] columns = COMPANY_ORDER rows_for_tpl = [{"position": p, "cells": [grid[c][p] for c in columns]} for p in POS_ORDER] return templates.TemplateResponse("group_executive.html", { "request": request, "columns": columns, "rows": rows_for_tpl, "gender_stats": gender_stats, }) @router.get("/group/executives", response_class=HTMLResponse) async def group_executives_detail(request: Request, company: str | None = Query(default=None)): try: res = sb.table("kepco_org").select("*").limit(10000).execute() raw: list[dict] = res.data or [] except Exception as e: logger.error(f"kepco_org fetch failed: {e}") raw = [] tabs = COMPANY_ORDER[:] current = company if company in COMPANY_ORDER else tabs[0] def _key(s: str | None) -> str: return (s or "").replace(" ","").strip() cur_key = _key(current) def _matches(dep: str | None) -> bool: if not dep: return False dep_key = _key(dep) if dep_key == cur_key: return True norm = _norm_company_label(dep) if norm and _key(norm) == cur_key: return True return cur_key in dep_key or dep_key in cur_key cur_rows = [r for r in raw if _matches(r.get("department"))] if not cur_rows and raw: cur_rows = [r for r in raw if _key(current) in _key(r.get("department"))] enriched: list[dict] = [] for r in cur_rows: task = (r.get("task") or r.get("title") or r.get("position") or "-").strip() or "-" start_disp, start_cmp = _date_display_and_compare(r.get("start")) end_disp, end_cmp = _date_display_and_compare(r.get("end")) expired = bool(end_cmp and end_cmp < TODAY) person = {**r, "_task": task, "_start_str": start_disp, "_end_str": end_disp, "_expired": expired, "_career_str": _normalize_multiline(r.get("career"))} enriched.append(person) enriched.sort(key=lambda x: _to_int_safe(x.get("id"))) return templates.TemplateResponse("group_executive_detail.html", { "request": request, "tabs": tabs, "current": current, "rows": enriched, }) # -------------------- SYNC 대시보드/배치 -------------------- @router.get("/sync", response_class=HTMLResponse) async def sync_index(request: Request): batches = sb.table("ingest_batches").select("*").order("started_at", desc=True).limit(20).execute().data or [] if not batches: return templates.TemplateResponse("sync_index.html", { "request": request, "rows": [], "event_stats": {}, "pending_stats": {} }) batch_ids = [b["batch_id"] for b in batches if b.get("batch_id")] ev = sb.table("gov_staff_events").select("batch_id,event_type").in_("batch_id", batch_ids).execute().data or [] event_stats = defaultdict(lambda: defaultdict(int)) for e in ev: event_stats[e["batch_id"]][e["event_type"]] += 1 dec = sb.table("match_decisions").select("batch_id,decision").in_("batch_id", batch_ids).execute().data or [] pending_stats = defaultdict(int) for d in dec: if d.get("decision") == "PENDING": pending_stats[d["batch_id"]] += 1 return templates.TemplateResponse("sync_index.html", { "request": request, "rows": batches, "event_stats": dict(event_stats), "pending_stats": dict(pending_stats), }) @router.get("/sync/batch/{batch_id}", response_class=HTMLResponse) async def sync_batch(request: Request, batch_id: str): st = sb.table("staging_gov_staff").select("*").eq("batch_id", batch_id).order("staging_id").execute().data or [] cand_rows = sb.table("match_candidates").select("*")\ .eq("batch_id", batch_id).order("staging_id").order("score", desc=True).limit(10000).execute().data or [] dec_rows = sb.table("match_decisions").select("*").eq("batch_id", batch_id).execute().data or [] dec_by_staging = {r["staging_id"]: r for r in dec_rows} cur_map = _current_map_by_person_source() cands_by_st: Dict[int, list] = {} for c in cand_rows: cands_by_st.setdefault(c["staging_id"], []).append(c) rows_pending, rows_auto = [], [] for s in st: staging_id = s["staging_id"] src = s.get("source") dec = dec_by_staging.get(staging_id) same_name = [] for c in cands_by_st.get(staging_id, []): pid = c.get("candidate_person_id") prev = cur_map.get((pid, src)) if _normalize_name((prev or {}).get("name")) == _normalize_name(s.get("name")): same_name.append({ "person_id": pid, "score": c.get("score"), "prev_ctx": prev, "name": (prev or {}).get("name") or "-" }) item = {"staging": s, "candidates": same_name, "decision": dec} if dec and dec.get("decision") == "AUTO": rows_auto.append(item) else: rows_pending.append(item) confirmed_events = sb.table("gov_staff_events").select("*").eq("batch_id", batch_id).order("event_id").execute().data or [] total = len(st); cnt_pending = len(rows_pending); cnt_auto = len(rows_auto) return templates.TemplateResponse("sync_batch.html", { "request": request, "batch_id": batch_id, "cards": {"total": total, "pending": cnt_pending, "auto": cnt_auto}, "rows_pending": rows_pending, "rows_auto": rows_auto, "events": confirmed_events, }) @router.post("/sync/decide", response_class=HTMLResponse) async def sync_decide( request: Request, batch_id: str = Form(...), staging_id: int = Form(...), choice: str = Form(...), # "__NEW__" | "__SKIP__" | "" ): decision = "MANUAL"; person_id = None; score = 1.0; rationale = None if choice == "__NEW__": rationale = "manual-new" elif choice == "__SKIP__": decision = "SKIP"; score = 0.0; rationale = "manual-skip" else: try: person_id = int(choice) except Exception: person_id = None payload = { "batch_id": batch_id, "staging_id": staging_id, "decision": decision, "person_id": person_id, "score": score, "rationale": rationale, } sb.table("match_decisions").upsert(payload, on_conflict="batch_id,staging_id").execute() return RedirectResponse(f"/sync/batch/{batch_id}?saved=1", status_code=302) @router.post("/sync/batch/{batch_id}/apply", response_class=RedirectResponse) @router.post("/sync/apply/{batch_id}", response_class=RedirectResponse) async def sync_apply_batch(request: Request, batch_id: str): apply_decisions(batch_id) return RedirectResponse(url=f"/sync/batch/{batch_id}?applied=1", status_code=303) # -------------------- MOTIE 부서별 최고 책임자 -------------------- @router.get("/gov/motie/heads", response_class=HTMLResponse) async def org_motie_heads_ordered(request: Request, q: str = Query(default="")): """ SQL(motie_head) 결과만 신뢰. (department, position) '정확 일치'로 부서장 매칭. """ try: # ✅ SQL에서 이미 머릿수만 추출된 뷰 사용 (motie_head) head_rows = ( sb.table("motie_head") .select("department,position,name,phone,task") .limit(50000) .execute() .data or [] ) except Exception: head_rows = [] # (department, position) → 후보 리스트 by_key: dict[tuple[str, str], list[dict]] = {} for r in head_rows: dep = (r.get("department") or "").strip() pos = (r.get("position") or "").strip() if not dep or not pos: continue by_key.setdefault((dep, pos), []).append(r) # 여러 명일 경우: 전화 있는 사람 우선 → 이름 가나다 def rank(r: dict) -> tuple: has_phone = 0 ph = (r.get("phone") or "").strip() if ph and ph != "-": has_phone = 1 name = (r.get("name") or "").strip() return (has_phone, name) key = (q or "").strip().lower() items: list[dict] = [] for unit, indent, expected_pos in _MOTIE_ORDER_POS: if unit == "장관정책보좌관": # 제외 continue cands = sorted(by_key.get((unit, expected_pos), []), key=rank, reverse=True) picked = cands[0] if cands else None name = (picked.get("name") if picked else None) or "(공석)" pos = (picked.get("position") if picked else None) or expected_pos phone = (picked.get("phone") if picked else None) or "-" task = (picked.get("task") if picked else None) or "-" row = { "unit": unit, "indent": indent, "name": name, "position": pos, "phone": phone, "task": task, "_expected": expected_pos, "_src": ("motie_head" if picked else "none"), } if not key or key in (" ".join([unit, name, pos, phone, task]).lower()): items.append(row) return templates.TemplateResponse( "gov_motie_heads.html", {"request": request, "ministry": "산업통상자원부(MOTIE)", "q": q, "items": items}, ) # --- 고정 부서 순서(요청 그대로) --- DEPT_ORDER = [ "장관실","제1차관","기획조정실","정책기획관","혁신행정담당관", "제2차관","에너지정책실","전력정책관","전력산업정책과", "원전산업정책국","원전산업정책과","원전환경과","원전지역협력과", "원전전략기획관","원전수출진흥과","원전수출협력과", ] # --- 부서별 리더(예상 직위) --- EXPECTED_POS: Dict[str, str] = { "장관실":"장관", "제1차관":"1차관", "기획조정실":"실장", "정책기획관":"국장", "혁신행정담당관":"과장", "제2차관":"2차관", "에너지정책실":"실장", "전력정책관":"국장", "전력산업정책과":"과장", "원전산업정책국":"국장", "원전산업정책과":"과장", "원전환경과":"과장", "원전지역협력과":"과장", "원전전략기획관":"국장", "원전수출진흥과":"과장", "원전수출협력과":"과장", } # --- 전화 끝4자리 → 역할(부서별). 정의된 항목은 반드시 한 줄 생성 --- PHONE_ROLE_LAST4 = [ ("장관실","5003","비서실장"), ("장관실","5005","수행비서"), ("장관실","5000","주무관"), ("제1차관","5016","주무관"), ("기획조정실","5201","주무관"), ("제2차관","5022","주무관"), ("에너지정책실","5701","주무관"), ("전력산업정책과","3925","사무관"), ("전력산업정책과","3890","사무관"), ] # 리더 먼저, 그 다음은 정의된 순서대로 def _last4(phone: str | None) -> str | None: if not phone: return None m = re.search(r"(\d{4})\s*$", str(phone)) return m.group(1) if m else None @router.get("/gov/motie/kps", response_class=HTMLResponse) async def gov_motie_heads_kps(request: Request): # 1) 조회 (task 포함) try: res = ( sb.table("motie_kps") .select("department,position,name,phone,task") .in_("department", DEPT_ORDER) .limit(10000) .execute() ) raw: List[Dict] = res.data or [] except Exception: raw = [] # 2) 인덱스 by_dep_pos: Dict[tuple, List[Dict]] = {} by_dep_l4: Dict[tuple, List[Dict]] = {} for r in raw: dep = (r.get("department") or "").strip() pos = (r.get("position") or "").strip() l4 = _last4((r.get("phone") or "").strip()) if dep: if pos: by_dep_pos.setdefault((dep, pos), []).append(r) if l4: by_dep_l4.setdefault((dep, l4), []).append(r) def pick_best(rows: List[Dict]) -> Dict: rows = rows or [] rows.sort(key=lambda x: (0 if (x.get("name") or "").strip() else 1, (x.get("name") or ""))) return rows[0] if rows else {} # 3) 그룹 구성 (리더 + 전화 라벨) groups: List[Dict] = [] for dep in DEPT_ORDER: rows_for_dep: List[Dict] = [] # (a) 리더 leader_pos = EXPECTED_POS.get(dep, "-") leader_row = pick_best(by_dep_pos.get((dep, leader_pos), [])) if leader_row: rows_for_dep.append({ "pos": leader_pos, "name": (leader_row.get("name") or "-").strip() or "-", "phone": (leader_row.get("phone") or "-").strip() or "-", "task": (leader_row.get("task") or "-").strip() or "-", }) else: rows_for_dep.append({"pos": leader_pos, "name": "(공석)", "phone": "-", "task": "-"}) # (b) 전화 라벨 for d, l4, role in PHONE_ROLE_LAST4: if d != dep: continue matches = list(by_dep_l4.get((dep, l4), [])) # 장관실 5000: 리더(장관) 제외 후 모두 표기 if dep == "장관실" and l4 == "5000": def _is_leader_row(x: dict) -> bool: pos = re.sub(r"\s+", "", (x.get("position") or "")) return pos == "장관" matches = [m for m in matches if not _is_leader_row(m)] if matches: matches.sort(key=lambda x: ((x.get("name") or "") == "", (x.get("name") or ""))) for m in matches: rows_for_dep.append({ "pos": role, "name": (m.get("name") or "-").strip() or "-", "phone": (m.get("phone") or "-").strip() or "-", "task": (m.get("task") or "-").strip() or "-", }) else: rows_for_dep.append({"pos": role, "name": "(공석)", "phone": "-", "task": "-"}) continue # 일반: 한 줄만 m = pick_best(matches) if m: rows_for_dep.append({ "pos": role, "name": (m.get("name") or "-").strip() or "-", "phone": (m.get("phone") or "-").strip() or "-", "task": (m.get("task") or "-").strip() or "-", }) else: rows_for_dep.append({"pos": role, "name": "(공석)", "phone": "-", "task": "-"}) groups.append({"department": dep, "rows": rows_for_dep}) return templates.TemplateResponse("gov_motie_kps.html", { "request": request, "groups": groups, }) # === 추가: MOEF 고정 순서/기대 직위 === MOEF_ORDER_POS: list[tuple[str, int, str]] = [ ("부총리", 0, "부총리"), ("대변인", 1, "대변인"), ("홍보담당관", 2, "담당관"), ("감사관", 1, "감사관"), ("감사담당관", 2, "담당관"), ("제1차관", 1, "차관"), ("인사과", 2, "과장"), ("운영지원과", 2, "과장"), ("경제공급망기획관", 2, "기획관"), ("공급망정책담당관", 3, "담당관"), ("공급망대응담당관", 3, "담당관"), ("세제실", 2, "실장"), ("조세총괄정책관", 3, "정책관"), ("조세정책과", 4, "과장"), ("조세분석과", 4, "과장"), ("조세특례제도과", 4, "과장"), ("소득법인세정책관", 3, "정책관"), ("소득세제과", 4, "과장"), ("금융세제과", 4, "과장"), ("법인세제과", 4, "과장"), ("재산소비세정책관", 3, "정책관"), ("재산세제과", 4, "과장"), ("환경에너지세제과", 4, "과장"), ("부가가치세제과", 4, "과장"), ("국제조세정책관", 3, "정책관"), ("국제조세제도과", 4, "과장"), ("신국제조세규범과", 4, "과장"), ("관세정책관", 3, "정책관"), ("관세제도과", 4, "과장"), ("산업관세과", 4, "과장"), ("관세협력과", 4, "과장"), ("자유무역협정관세이행과", 4, "과장"), ("차관보", 2, "차관보"), ("경제정책국", 3, "국장"), ("민생경제정책관", 4, "정책관"), # (민생경제정책관 산하 과들은 과장인데, 화면 들여쓰기는 4단까지만 쓰자) ("종합정책과", 4, "과장"), ("물가정책과", 4, "과장"), ("경제분석과", 4, "과장"), ("정책기획과", 4, "과장"), ("자금시장과", 4, "과장"), ("거시정책과", 4, "과장"), ("정책조정국", 3, "국장"), ("정책조정기획관", 4, "기획관"), ("정책조정총괄과", 4, "과장"), ("서비스경제과", 4, "과장"), ("산업경제과", 4, "과장"), ("지역경제정책과", 4, "과장"), ("신성장정책과", 4, "과장"), ("기업환경과", 4, "과장"), ("경제구조개혁국", 3, "국장"), ("경제구조개혁총괄과", 4, "과장"), ("복지경제과", 4, "과장"), ("인력정책과", 4, "과장"), ("연금보건경제과", 4, "과장"), ("노동시장경제과", 4, "과장"), ("청년정책과", 4, "과장"), ("미래전략국", 3, "국장"), ("미래전략과", 4, "과장"), ("지속가능경제과", 4, "과장"), ("인구경제과", 4, "과장"), ("기후대응전략과", 4, "과장"), ("국제경제관리관", 2, "관리관"), ("국제금융국", 3, "국장"), ("국제금융심의관", 4, "심의관"), ("국제금융과", 4, "과장"), ("금융협력과", 4, "과장"), ("외화자금과", 4, "과장"), ("다자금융과", 4, "과장"), ("외환제도과", 4, "과장"), ("대외경제국", 3, "국장"), ("대외경제총괄과", 4, "과장"), ("통상조정과", 4, "과장"), ("국제경제과", 4, "과장"), ("경제협력기획과", 4, "과장"), ("통상정책과", 4, "과장"), ("남북경제과", 4, "과장"), ("개발금융국", 3, "국장"), ("개발금융총괄과", 4, "과장"), ("개발사업협력과", 4, "과장"), ("국제기구과", 4, "과장"), ("녹색기후기획과", 4, "과장"), ("개발전략과", 4, "과장"), ("신성장전략기획추진단", 2, "단장"), ("전략기획팀", 3, "팀장"), ("미래산업팀", 3, "팀장"), ("디지털전환팀", 3, "팀장"), ("조세개혁추진단", 2, "단장"), ("상속세개편팀", 3, "팀장"), ("보유세개편팀", 3, "팀장"), ("원스톱수출수주지원단", 2, "단장"), ("수출총괄팀", 3, "팀장"), ("수주인프라지원팀", 3, "팀장"), ("금융재정지원팀", 3, "팀장"), ("서비스수출지원팀", 3, "팀장"), ("국제협력대사지원단", 2, "단장"), ("민생안정지원단", 2, "단장"), ("APEC 재무·구조개혁 장관회의 추진단", 2, "팀장"), ("제2차관", 1, "차관"), ("기획조정실", 2, "실장"), ("정책기획관", 3, "기획관"), ("기획재정담당관", 4, "담당관"), ("정보화담당관", 4, "담당관"), ("혁신정책담당관", 4, "담당관"), ("규제개혁법무담당관", 4, "담당관"), ("비상안전기획관", 3, "국장"), ("비상안전기획팀", 4, "팀장"), ("예산실", 2, "실장"), ("예산총괄심의관", 3, "심의관"), ("예산총괄과", 4, "과장"), ("기금운용계획과", 4, "과장"), ("예산정책과", 4, "과장"), ("예산관리과", 4, "과장"), ("예산기준과", 4, "과장"), ("사회예산심의관", 3, "심의관"), ("고용예산과", 4, "과장"), ("기후환경예산과", 4, "과장"), ("교육예산과", 4, "과장"), ("총사업비관리과", 4, "과장"), ("문화예산과", 4, "과장"), ("경제예산심의관", 3, "심의관"), ("산업중소벤처예산과", 4, "과장"), ("연구개발예산과", 4, "과장"), ("국토교통예산과", 4, "과장"), ("정보통신예산과", 4, "과장"), ("농림해양예산과", 4, "과장"), ("복지안전예산심의관", 3, "심의관"), ("복지예산과", 4, "과장"), ("안전예산과", 4, "과장"), ("연금보건예산과", 4, "과장"), ("지역예산과", 4, "과장"), ("행정국방예산심의관", 3, "심의관"), ("법사예산과", 4, "과장"), ("방위사업예산과", 4, "과장"), ("행정예산과", 4, "과장"), ("국방예산과", 4, "과장"), ("재정관리관", 2, "관리관"), ("재정정책국", 3, "국장"), ("재정건전성심의관", 4, "심의관"), ("재정정책총괄과", 4, "과장"), ("재정제도과", 4, "과장"), ("재정건전성과", 4, "과장"), ("재정정책협력과", 4, "과장"), ("재정분석과", 4, "과장"), ("재정정보과", 4, "과장"), ("국고국", 3, "국장"), ("국유재산심의관", 4, "심의관"), ("국고과", 4, "과장"), ("출자관리과", 4, "과장"), ("국유재산정책과", 4, "과장"), ("공공조달정책과", 4, "과장"), ("계약정책과", 4, "과장"), ("국유재산협력과", 4, "과장"), ("국채과", 4, "과장"), ("계약분쟁조정과", 4, "과장"), ("국유재산조정과", 4, "과장"), ("재정관리국", 3, "국장"), ("재정성과심의관", 4, "심의관"), ("재정관리총괄과", 4, "과장"), ("민간투자정책과", 4, "과장"), ("재정성과평가과", 4, "과장"), ("회계결산과", 4, "과장"), ("타당성심사과", 4, "과장"), ("재정지출관리과", 4, "과장"), ("공공정책국", 3, "국장"), ("공공혁신심의관", 4, "심의관"), ("공공정책총괄과", 4, "과장"), ("인재경영과", 4, "과장"), ("공공제도기획과", 4, "과장"), ("공공윤리정책과", 4, "과장"), ("재무경영과", 4, "과장"), ("공공혁신기획과", 4, "과장"), ("평가분석과", 4, "과장"), ("경영관리과", 4, "과장"), ("복권위원회사무처", 2, "사무처장"), ("복권총괄과", 3, "과장"), ("기금사업과", 3, "과장"), ("발행관리과", 3, "과장"), ("국고보조금부정수급관리단", 2, "단장"), ] # === 추가: 기재부 부서장 페이지 === @router.get("/gov/moef/heads", response_class=HTMLResponse) async def gov_moef_heads_ordered(request: Request, q: str = Query(default="")): # 1) 조회: moef_head에서 필요한 컬럼만, 지정 부서만 dept_whitelist = list({d for (d, _, _) in MOEF_ORDER_POS}) try: res = ( sb.table("moef_head") .select("department,position,name,phone,task") .in_("department", dept_whitelist) .limit(50000) .execute() ) data = res.data or [] except Exception: data = [] # 2) (부서,직위) -> 후보 목록 (중복 시 한 명 고르기) by_key: dict[tuple[str, str], list[dict]] = {} for r in data: dep = (r.get("department") or "").strip() pos = (r.get("position") or "").strip() if not dep or not pos: continue by_key.setdefault((dep, pos), []).append(r) def pick_best(cands: list[dict]) -> dict | None: if not cands: return None # 전화 있는 사람 우선, 그 다음 이름 오름차순 (항상 같은 결과) def score(x): has_phone = 1 if (x.get("phone") and str(x.get("phone")).strip()) else 0 return (has_phone, (x.get("name") or "")) return sorted(cands, key=score, reverse=True)[0] # 3) 기대목록 순회 → 일치 레코드 or (공석) key = (q or "").strip().lower() items = [] for unit, indent, expected_pos in MOEF_ORDER_POS: cands = by_key.get((unit, expected_pos), []) rec = pick_best(cands) name = (rec.get("name") if rec else None) or "(공석)" pos = (rec.get("position") if rec else None) or expected_pos phone = (rec.get("phone") if rec else None) or "-" task = (rec.get("task") if rec else None) or "-" row = { "unit": unit, "indent": indent, "name": name, "position": pos, "phone": phone, "task": task, "_expected": expected_pos, "_src": ("moef_head" if rec else "none"), } if not key or key in (" ".join([unit, name, pos, phone, task]).lower()): items.append(row) # 4) 렌더 (motie와 같은 표 템플릿 재사용) return templates.TemplateResponse("gov_moef_heads.html", { "request": request, "ministry": "기획재정부(MOEF)", "q": q, "items": items, }) MOEF_KPS_DEPT_ORDER = [ "부총리실", "제1차관", "제2차관", "공공정책국", "공공혁신심의관", "공공정책총괄과", "공공제도기획과", "재무경영과", "평가분석과", "인재경영과", "공공윤리정책과", "공공혁신기획과", "경영관리과", ] MOEF_KPS_DEPT_WHITELIST = MOEF_KPS_DEPT_ORDER + ["부총리", "장관실"] MOEF_KPS_EXPECTED_POS: Dict[str, str] = { "부총리실": "부총리", "제1차관": "차관", "제2차관": "차관", "공공정책국": "국장", "공공혁신심의관": "심의관", "공공정책총괄과": "과장", "공공제도기획과": "과장", "재무경영과": "과장", "평가분석과": "과장", "인재경영과": "과장", "공공윤리정책과": "과장", "공공혁신기획과": "과장", "경영관리과": "과장", } MOEF_KPS_PHONE_ROLE_LAST4 = [ ("제1차관", "2028", "담당"), ("제2차관", "2022", "담당"), ("재무경영과", "5631", "담당"), ] def _moef_kps_canon_dep(dep: str, pos: str) -> str: d = (dep or "").strip() p = (pos or "").strip() if d == "부총리": return "부총리실" if d == "장관실" and p == "비서실장": return "부총리실" return d def _moef_kps_last4(phone: str | None) -> str | None: if not phone: return None m = re.search(r"(\d{4})\s*$", str(phone)) return m.group(1) if m else None def _pick_best(rows: List[Dict]) -> Dict: rows = rows or [] rows.sort(key=lambda x: ( 0 if (x.get("name") or "").strip() else 1, 0 if (x.get("phone") or "").strip() else 1, (x.get("name") or "") )) return rows[0] if rows else {} @router.get("/gov/moef/kps", response_class=HTMLResponse) async def gov_moef_kps(request: Request): try: res = ( sb.table("moef_kps") .select("id,department,position,name,phone,task") .in_("department", MOEF_KPS_DEPT_WHITELIST) .limit(50000) .execute() ) raw: List[Dict] = res.data or [] except Exception: raw = [] by_dep: Dict[str, List[Dict]] = {} by_dep_pos: Dict[tuple, List[Dict]] = {} by_dep_l4: Dict[tuple, List[Dict]] = {} for r in raw: raw_dep = (r.get("department") or "").strip() pos = (r.get("position") or "").strip() dep = _moef_kps_canon_dep(raw_dep, pos) l4 = _moef_kps_last4((r.get("phone") or "").strip()) r["_dep"] = dep by_dep.setdefault(dep, []).append(r) if pos: by_dep_pos.setdefault((dep, pos), []).append(r) if l4: by_dep_l4.setdefault((dep, l4), []).append(r) groups: List[Dict] = [] for dep in MOEF_KPS_DEPT_ORDER: used_ids: set[int] = set() rows_for_dep: List[Dict] = [] leader_pos = MOEF_KPS_EXPECTED_POS.get(dep, "-") leader_row = _pick_best(by_dep_pos.get((dep, leader_pos), [])) if leader_row: used_ids.add(leader_row.get("id")) rows_for_dep.append({ "pos": leader_pos, "name": (leader_row.get("name") or "-").strip() or "-", "phone": (leader_row.get("phone") or "-").strip() or "-", "task": (leader_row.get("task") or "-").strip() or "-", }) else: rows_for_dep.append({"pos": leader_pos, "name": "(공석)", "phone": "-", "task": "-"}) # (b) 전화 라벨 매핑 for d, l4, role in MOEF_KPS_PHONE_ROLE_LAST4: if d != dep: continue cands = [x for x in by_dep_l4.get((dep, l4), []) if x.get("id") not in used_ids] pick = _pick_best(cands) if pick: used_ids.add(pick.get("id")) rows_for_dep.append({ "pos": (pick.get("position") or "").strip() or role, # ← 직책 우선 "name": (pick.get("name") or "-").strip() or "-", "phone": (pick.get("phone") or "-").strip() or "-", "task": (pick.get("task") or "-").strip() or "-", }) else: rows_for_dep.append({"pos": role, "name": "(공석)", "phone": "-", "task": "-"}) remaining = [x for x in by_dep.get(dep, []) if x.get("id") not in used_ids] remaining.sort(key=lambda x: ((x.get("position") or ""), (x.get("name") or ""))) for r in remaining: used_ids.add(r.get("id")) rows_for_dep.append({ "pos": (r.get("position") or "-").strip() or "-", "name": (r.get("name") or "-").strip() or "-", "phone": (r.get("phone") or "-").strip() or "-", "task": (r.get("task") or "-").strip() or "-", }) groups.append({"department": dep, "rows": rows_for_dep}) return templates.TemplateResponse("gov_moef_kps.html", { "request": request, "groups": groups, }) # -------------------- 도구 -------------------- @router.get("/tools/auto-lecture", response_class=HTMLResponse) async def tools_auto_lecture(request: Request): return templates.TemplateResponse("tools_auto_lecture.html", {"request": request})