from __future__ import annotations from typing import Optional, Dict, List, Set from datetime import datetime, timezone, timedelta import hashlib import json from fastapi import APIRouter, Request, Query, Form from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from starlette import status from app.web.deps import templates, sb, logger, motie_person_keys from postgrest.exceptions import APIError router = APIRouter() KST = timezone(timedelta(hours=9)) def _is_admin(request: Request) -> bool: user = request.session.get("user") or {} return user.get("role") == "admin" def _to_kst(dt: Optional[str]) -> Optional[str]: if not dt: return None try: d = datetime.fromisoformat(dt.replace("Z", "+00:00")) return d.astimezone(KST).strftime("%Y-%m-%d %H:%M") except Exception: return dt def _person_key(name: Optional[str], phone: Optional[str]) -> str: base = f"{(name or '').strip()}|{(phone or '').strip()}" return hashlib.sha1(base.encode("utf-8")).hexdigest() def _event_category(event_type: Optional[str], changed_dep: bool, changed_pos: bool) -> str: evt = (event_type or "").lower() if evt == "added": return "신규" if evt in {"removed", "closed"}: return "종료" if changed_dep and changed_pos: return "부서+직위" if changed_dep: return "부서 이동" if changed_pos: return "직위 변경" return "기타" # /gov/motie/changes (또는 changes_raw) 단순화 버전 @router.get("/gov/motie/changes", response_class=HTMLResponse) async def motie_changes( request: Request, days: int = Query(default=14, ge=1, le=365), limit: int = Query(default=200, ge=10, le=2000), page: int = Query(default=1, ge=1), ): since_dt = datetime.now(timezone.utc) - timedelta(days=days) since_iso = since_dt.isoformat() offset = (page - 1) * limit rows_raw: List[Dict] = [] error: Optional[str] = None try: fetch_size = min(5000, max(limit * page * 4, limit * 6, 400)) resp = sb.rpc( "motie_org_events_list", { "p_since": since_iso, "p_limit": fetch_size, "p_offset": 0, "p_core_only": True, }, ).execute() rows_raw = resp.data or [] except APIError as exc: logger.warning("gov motie changes: rpc failed: %s", exc) error = "timeout" if getattr(exc, "code", None) == "57014" else "api" rows_raw = [] except Exception as exc: logger.warning("gov motie changes: rpc unexpected error: %s", exc) error = "unknown" rows_raw = [] # 3) 템플릿용 후처리 meta_rows: List[Dict] = [] person_keys_set: Set[str] = set() name_clusters: Dict[str, Set[str]] = {} for r in rows_raw: keys = motie_person_keys(r.get("name"), r.get("phone"), r.get("department"), r.get("position")) or {} person_key = keys.get("display") or keys.get("status") or _person_key(r.get("name"), r.get("phone")) meta_rows.append({"row": r, "person_key": person_key, "keys": keys}) if person_key: person_keys_set.add(person_key) name_norm = (r.get("name") or "").strip() if name_norm and person_key: name_clusters.setdefault(name_norm, set()).add(person_key) decisions_map: Dict[tuple, str] = {} if person_keys_set: try: dec_rows = ( sb.table("admin_motie_change_status") .select("person_key,latest_valid_from,decision") .in_("person_key", list(person_keys_set)) .gte("latest_valid_from", since_iso) .limit(20000) .execute() .data or [] ) for d in dec_rows: pkey = (d or {}).get("person_key") or "" ts = (d or {}).get("latest_valid_from") or "" decision_val = ((d or {}).get("decision") or "").strip().lower() if pkey and ts and decision_val: decisions_map[(pkey, ts)] = decision_val except Exception as exc: logger.warning("gov motie changes: decision fetch failed: %s", exc) processed: List[Dict] = [] for item in meta_rows: r = item["row"] person_key = item["person_key"] if not person_key: continue decision_val = decisions_map.get((person_key, r.get("valid_from"))) or "" if decision_val != "accept": continue changed = r.get("changed_fields") or [] if isinstance(changed, str): try: changed = json.loads(changed) if not isinstance(changed, list): changed = [] except Exception: changed = [] dep_before = r.get("dep_before") or "-" pos_before = r.get("pos_before") or "-" dep_after = r.get("department") or "-" pos_after = r.get("position") or "-" changed_dep = "department" in changed changed_pos = "position" in changed phone_before = r.get("phone_before") or "" phone_after = r.get("phone") or "" if mode_norm == "ambiguous": name_norm = (r.get("name") or "").strip() cluster_size = len(name_clusters.get(name_norm, set())) if cluster_size <= 1 and not (changed_dep or changed_pos or phone_before != phone_after): continue if mode_norm == "unique": name_norm = (r.get("name") or "").strip() if len(name_clusters.get(name_norm, set())) > 1: continue category = _event_category(r.get("event_type"), changed_dep, changed_pos) is_current = bool(r.get("is_current")) processed.append({ "valid_from": r.get("valid_from"), "latest_valid_from": r.get("valid_from"), "name": r.get("name"), "phone": phone_after, "dep_before": dep_before, "pos_before": pos_before, "dep_after": dep_after, "pos_after": pos_after, "changed_dep": changed_dep, "changed_pos": changed_pos, "category": category, "is_current": is_current, "person_key": person_key, "phone_before": phone_before, "task_before": r.get("task_before"), "task_after": r.get("task"), "run_id": r.get("run_id"), "event_type": r.get("event_type"), }) processed.sort(key=lambda x: x.get("valid_from") or "", reverse=True) total = len(processed) pages = (total + limit - 1) // limit if total else 1 has_prev = page > 1 has_next = page < pages page_rows = processed[offset:offset + limit] return templates.TemplateResponse( "gov/motie/changes.html", { "request": request, "rows": page_rows, "days": days, "limit": limit, "page": page, "pages": pages, "has_prev": has_prev, "has_next": has_next, "total": total, "error": error, "fallback_days": None, }, ) @router.get("/admin/motie/changes", response_class=HTMLResponse) async def admin_motie_changes( request: Request, days: int = Query(default=30, ge=1, le=365), event: Optional[str] = Query(default=None), department: Optional[str] = Query(default=None), position: Optional[str] = Query(default=None), name: Optional[str] = Query(default=None), focus: str = Query(default="core"), mode: str = Query(default="ambiguous"), status_value: Optional[str] = Query(alias="status", default=None), limit: int = Query(default=100, ge=10, le=500), page: int = Query(default=1, ge=1), ): """관리자 MOTIE 변경 검토 페이지.""" if not _is_admin(request): return RedirectResponse("/auth/login", status_code=status.HTTP_303_SEE_OTHER) since_dt = datetime.now(timezone.utc) - timedelta(days=days) since_iso = since_dt.isoformat() # 1) Supabase RPC 호출 (필터 가능한 항목만 서버에서 처리) raw_rows: List[Dict] = [] event_norm = (event or "").strip().lower() rpc_event = None if event_norm in {"added", "removed", "updated", "closed"}: if event_norm == "removed": rpc_event = "closed" else: rpc_event = event_norm widen_factor = max(1, min(8, page)) fetch_size = min(2000, max(limit * widen_factor * 4, 400)) try: params = { "p_since": since_iso, "p_limit": fetch_size, "p_offset": 0, "p_event": rpc_event, } if department: params["p_department"] = department.strip() if position: params["p_position"] = position.strip() if name: params["p_name"] = name.strip() resp = sb.rpc("motie_org_events_list", params).execute() raw_rows = resp.data or [] except APIError as exc: logger.warning("admin motie changes: rpc failed: %s", exc) raw_rows = [] except Exception as exc: logger.warning("admin motie changes: rpc unexpected error: %s", exc) raw_rows = [] # 2) 관리자 결정 상태 조회 decisions: Dict[tuple, str] = {} try: dec_rows = ( sb.table("admin_motie_change_status") .select("person_key,latest_valid_from,decision") .gte("latest_valid_from", since_iso) .limit(20000) .execute() .data or [] ) for d in dec_rows: pkey = (d or {}).get("person_key") or "" ts = (d or {}).get("latest_valid_from") or "" decision_val = ((d or {}).get("decision") or "").strip().lower() if pkey and ts and decision_val: decisions[(pkey, ts)] = decision_val except Exception as exc: logger.warning("admin motie changes: fetch decisions failed: %s", exc) decisions = {} mode_norm = (mode or "").strip().lower() focus_norm = (focus or "").strip().lower() status_norm = (status_value or "").strip().lower() name_clusters: Dict[str, Set[str]] = {} prepared_rows: List[Dict] = [] for r in raw_rows: keys = motie_person_keys( r.get("name"), r.get("phone"), r.get("department"), r.get("position"), ) or {} person_key_display = keys.get("display") or keys.get("status") or _person_key(r.get("name"), r.get("phone")) person_key_status = keys.get("status") or person_key_display name_norm = (r.get("name") or "").strip() if name_norm: name_clusters.setdefault(name_norm, set()).add(person_key_display) prepared_rows.append( { "row": r, "display": person_key_display, "status": person_key_status, "keys": keys, } ) processed: List[Dict] = [] for entry in prepared_rows: r = entry["row"] person_key_display = entry["display"] person_key_status = entry["status"] changed_fields = r.get("changed_fields") or [] if isinstance(changed_fields, str): try: data = json.loads(changed_fields) changed_fields = data if isinstance(data, list) else [] except Exception: changed_fields = [] if focus_norm != "all": core_hit = False evt_lower = (r.get("event_type") or "").lower() if evt_lower in {"added", "removed", "closed"}: core_hit = True else: core_hit = any(f in {"department", "position"} for f in changed_fields) if not core_hit: continue dep_before = r.get("dep_before") or "-" pos_before = r.get("pos_before") or "-" dep_after = r.get("department") or "-" pos_after = r.get("position") or "-" phone_before = (r.get("phone_before") or "").strip() phone_after = (r.get("phone") or "").strip() changed_dep = "department" in changed_fields changed_pos = "position" in changed_fields if not (changed_dep or changed_pos): continue if mode_norm == "ambiguous": name_norm = (r.get("name") or "").strip() cluster_size = len(name_clusters.get(name_norm, set())) if cluster_size <= 1 and not (changed_dep or changed_pos or phone_before != phone_after): continue if mode_norm == "unique": name_norm = (r.get("name") or "").strip() if len(name_clusters.get(name_norm, set())) > 1: continue valid_from = r.get("valid_from") or "" decision_val = decisions.get((person_key_display, valid_from)) or "" decision_val = (decision_val.strip().lower() if decision_val else "") event_type = (r.get("event_type") or "").lower() category = _event_category(event_type, changed_dep, changed_pos) processed.append( { "person_key": person_key_display, "status_key": person_key_status, "latest_valid_from": valid_from, "latest_valid_from_kst": _to_kst(valid_from) or "", "name": r.get("name"), "department": r.get("department"), "position": r.get("position"), "phone": r.get("phone"), "task": r.get("task"), "dep_before": dep_before, "pos_before": pos_before, "dep_after": dep_after, "pos_after": pos_after, "changed_fields": changed_fields, "changed_dep": changed_dep, "changed_pos": changed_pos, "category": category, "is_current": bool(r.get("is_current")), "event_type": event_type, "_decision": decision_val, } ) processed.sort(key=lambda x: x.get("latest_valid_from") or "", reverse=True) counters = {"open": 0, "accept": 0, "skip": 0} for row in processed: dval = (row.get("_decision") or "").strip().lower() if dval in {"accept", "skip"}: counters[dval] += 1 else: counters["open"] += 1 filtered = processed if status_norm == "accept": filtered = [r for r in processed if r.get("_decision") == "accept"] elif status_norm == "skip": filtered = [r for r in processed if r.get("_decision") == "skip"] elif status_norm == "open": filtered = [r for r in processed if not r.get("_decision")] total = len(filtered) pages = max(1, (total + limit - 1) // limit) if total else 1 start = (page - 1) * limit if start >= total and total: page = max(1, min(page, pages)) start = (page - 1) * limit end = start + limit page_rows = filtered[start:end] return templates.TemplateResponse( "admin/motie_changes.html", { "request": request, "rows": page_rows, "days": days, "event": event or "", "department": department or "", "position": position or "", "name": name or "", "focus": focus_norm if focus_norm in {"core", "all"} else "core", "mode": mode_norm if mode_norm in {"ambiguous", "all"} else "ambiguous", "status": status_norm, "limit": limit, "page": page, "pages": pages, "total": total, "counters": counters, }, ) def _append_audit(message: str): try: from pathlib import Path log_dir = Path(__file__).resolve().parents[3] / "logs" log_dir.mkdir(parents=True, exist_ok=True) from datetime import datetime as _dt ts = _dt.utcnow().isoformat() with (log_dir / "admin_audit.log").open("a", encoding="utf-8") as f: f.write(f"{ts}\t{message}\n") except Exception: pass def _store_transition_record( person_key: str, latest_valid_from: Optional[str], decision: str, payload: Optional[Dict], approved_by: str, ): ts = (latest_valid_from or "").strip() if not person_key or not ts: return if decision != "accept": try: sb.table("motie_org_transitions").delete().eq("person_key", person_key).eq("latest_valid_from", ts).execute() except Exception: pass return payload = payload or {} name = (payload.get("name") or "").strip() if not name: return def norm(val: Optional[str]) -> Optional[str]: val = (val or "").strip() return val or None dep_before = norm(payload.get("department_before")) pos_before = norm(payload.get("position_before")) phone_before = norm(payload.get("phone_before")) task_before = norm(payload.get("task_before")) dep_after = norm(payload.get("department_after")) pos_after = norm(payload.get("position_after")) phone_after = norm(payload.get("phone_after")) task_after = norm(payload.get("task_after")) run_id = norm(payload.get("run_id")) event_type = norm(payload.get("event_type")) before_keys = motie_person_keys(name, phone_before, dep_before, pos_before) or {} before_person_key = before_keys.get("display") or before_keys.get("status") if not before_person_key and phone_before: before_person_key = _person_key(name, phone_before) data = { "person_key": person_key, "before_person_key": before_person_key, "name": name, "latest_valid_from": ts, "department_before": dep_before, "position_before": pos_before, "phone_before": phone_before, "task_before": task_before, "department_after": dep_after, "position_after": pos_after, "phone_after": phone_after, "task_after": task_after, "run_id": run_id, "event_type": event_type, "approved_by": approved_by, "approved_at": datetime.utcnow().isoformat(), } try: sb.table("motie_org_transitions").upsert(data, on_conflict="person_key,latest_valid_from").execute() except Exception as exc: logger.warning("motie transitions upsert failed: %s", exc) @router.post("/admin/motie/changes/decision") async def admin_motie_changes_decision( request: Request, person_key: str = Form(...), latest_valid_from: str = Form(default=""), decision: str = Form(...), # 'accept' | 'skip' payload: str = Form(default=""), ): if not _is_admin(request): return JSONResponse({"error": "unauthorized"}, status_code=401) decision = (decision or "").strip().lower() if decision not in ("accept", "skip"): return JSONResponse({"error": "invalid decision"}, status_code=400) user = request.session.get("user") or {} email = user.get("email") or "" # f-string 내부 표현식에 백슬래시(\t) 리터럴이 있으면 SyntaxError가 발생하므로 사전 정제 _append_audit( f"motie_changes_decision\tuser={email}\tperson_key={person_key}\tts={latest_valid_from}\tdecision={decision}" ) # Optional: persist to Supabase if table exists try: data = { "person_key": person_key, "latest_valid_from": latest_valid_from or None, "decision": decision, "updated_by": email, "updated_at": datetime.utcnow().isoformat(), } # upsert best-effort sb.table("admin_motie_change_status").upsert(data, on_conflict="person_key,latest_valid_from").execute() except Exception: # ignore if table not present pass payload_obj: Dict = {} if payload: try: payload_obj = json.loads(payload) if not isinstance(payload_obj, dict): payload_obj = {} except Exception: payload_obj = {} _store_transition_record(person_key, latest_valid_from, decision, payload_obj, email) return JSONResponse({"ok": True}) @router.post("/admin/motie/changes/bulk") async def admin_motie_changes_bulk(request: Request): if not _is_admin(request): return JSONResponse({"error": "unauthorized"}, status_code=401) try: payload = await request.json() except Exception: return JSONResponse({"error": "invalid json"}, status_code=400) items = payload.get("items") or [] decision = (payload.get("decision") or "").strip().lower() note = payload.get("note") or "" if decision not in ("accept", "skip"): return JSONResponse({"error": "invalid decision"}, status_code=400) user = request.session.get("user") or {} email = user.get("email") or "" safe_note = (note or "").replace("\t", " ").strip() ok = 0 for it in items: try: person_key = it.get("person_key") latest_valid_from = it.get("latest_valid_from") or "" if not person_key: continue _append_audit( f"motie_changes_bulk\tuser={email}\tperson_key={person_key}\tts={latest_valid_from}\tdecision={decision}" ) try: data = { "person_key": person_key, "latest_valid_from": latest_valid_from or None, "decision": decision, "updated_by": email, "updated_at": datetime.utcnow().isoformat(), } sb.table("admin_motie_change_status").upsert(data, on_conflict="person_key,latest_valid_from").execute() except Exception: pass payload_obj = it.get("payload") or {} if not isinstance(payload_obj, dict): payload_obj = {} _store_transition_record(person_key, latest_valid_from, decision, payload_obj, email) ok += 1 except Exception: continue return JSONResponse({"ok": True, "count": ok}) @router.get("/admin/motie/changes/diff") async def admin_motie_changes_diff(request: Request, person_key: str): if not _is_admin(request): return JSONResponse({"error": "unauthorized"}, status_code=401) # resolve name/phone from events name, phone = None, None try: evs = ( sb.table("motie_org_events").select("name,phone").order("valid_from", desc=True).limit(10000).execute().data or [] ) for e in evs: if _person_key(e.get("name"), e.get("phone")) == person_key: name, phone = e.get("name"), e.get("phone") break except Exception: pass if not name: return JSONResponse({"diff": [], "valid_from_kst": None}) # fetch latest two from coalesced hist = ( sb.table("motie_org_hist_coalesced") .select("*") .eq("name", name) .eq("phone", phone if phone is not None else "") .order("valid_from", desc=True) .limit(2) .execute() .data or [] ) cur = hist[0] if hist else None prev = hist[1] if len(hist) > 1 else None def _to_kst_local(dt: Optional[str]) -> Optional[str]: return _to_kst(dt) # 핵심: 부서/직위만 비교 fields = ["department", "position"] diff = [] if cur and prev: for f in fields: if (cur.get(f) or "") != (prev.get(f) or ""): diff.append({"field": f, "before": prev.get(f), "after": cur.get(f)}) elif cur: for f in fields: if cur.get(f): diff.append({"field": f, "before": None, "after": cur.get(f)}) return JSONResponse({ "diff": diff, "valid_from_kst": _to_kst_local(cur.get("valid_from") if cur else None), "name": name, "phone": phone, }) @router.get("/admin/motie/changes/hints") async def admin_motie_changes_hints( request: Request, person_key: str, name: Optional[str] = Query(default=None), phone: Optional[str] = Query(default=None), limit: int = Query(default=30, ge=1, le=200), ): if not _is_admin(request): return JSONResponse({"error": "unauthorized"}, status_code=401) # Resolve name/phone via events if not provided if not name: try: evs = ( sb.table("motie_org_events") .select("name,phone") .order("valid_from", desc=True) .limit(3000) .execute() .data or [] ) for e in evs: if _person_key(e.get("name"), e.get("phone")) == person_key: name, phone = e.get("name"), e.get("phone") break except Exception: pass name = (name or "").strip() phone = (phone or "").strip() out = { "name": name, "phone": phone, "same_phone_diff_names": [], "same_name_diff_phones": [], } # Same phone, different names if phone: try: r1 = ( sb.table("motie_org_events") .select("name") .eq("phone", phone) .not_.eq("name", name) if name else sb.table("motie_org_events").select("name").eq("phone", phone) ) if not name: # fix fallback table when name is empty r1 = sb.table("motie_org_events").select("name").eq("phone", phone) r1 = r1.limit(5000).execute().data or [] names = sorted({(x.get("name") or "").strip() for x in r1 if (x.get("name") or "").strip()}) out["same_phone_diff_names"] = names[:limit] except Exception: pass # Same name, different phones (non-empty) if name: try: r2 = ( sb.table("motie_org_events") .select("phone") .eq("name", name) .limit(5000) .execute() .data or [] ) phones = set() for x in r2: p = (x.get("phone") or "").strip() if p and p != phone: phones.add(p) out["same_name_diff_phones"] = sorted(list(phones))[:limit] except Exception: pass return JSONResponse(out)