from __future__ import annotations from fastapi import APIRouter, Request from fastapi.responses import HTMLResponse, RedirectResponse from starlette import status from app.web.deps import templates from app.services import supabase_service as sps from datetime import datetime from pathlib import Path import subprocess from fastapi import Form from fastapi.responses import JSONResponse from app.web.scheduler import list_jobs, upsert_interval_job, upsert_daily_cron, start_scheduler router = APIRouter() def _is_admin(request: Request) -> bool: user = request.session.get("user") or {} return user.get("role") == "admin" @router.get("/admin/wip/transfer", response_class=HTMLResponse) async def admin_wip_transfer(request: Request): if not _is_admin(request): return RedirectResponse("/auth/login", status_code=status.HTTP_303_SEE_OTHER) return templates.TemplateResponse( "admin/wip_transfer.html", {"request": request}, ) @router.get("/admin/wip/person", response_class=HTMLResponse) async def admin_wip_person(request: Request): if not _is_admin(request): return RedirectResponse("/auth/login", status_code=status.HTTP_303_SEE_OTHER) return templates.TemplateResponse( "admin/wip_person.html", {"request": request}, ) @router.get("/admin/users", response_class=HTMLResponse) async def admin_users(request: Request): if not _is_admin(request): return RedirectResponse("/auth/login", status_code=status.HTTP_303_SEE_OTHER) # 실제 사용자 목록 연동 (가능한 경우) users = [] try: client = sps.get_admin_client() # supabase-py v2: list_users supports optional pagination # 검색/페이징 파라미터 q = (request.query_params.get('q') or '').strip().lower() try: page = max(1, int(request.query_params.get('page') or 1)) per_page = max(10, min(200, int(request.query_params.get('per_page') or 50))) except Exception: page, per_page = 1, 50 # list_users 페이지네이션 (라이브러리별 시그니처 차이를 고려한 폴백) try: res = client.auth.admin.list_users(page=page, per_page=per_page) except Exception: res = client.auth.admin.list_users() data = getattr(res, 'data', None) or getattr(res, 'users', None) or [] for u in data: try: # handle both dict and model obj = u if isinstance(u, dict) else u.model_dump() except Exception: obj = dict(u) if q: em = (obj.get('email') or '').lower() nm = ((obj.get('user_metadata') or {}).get('name') or '').lower() if q not in em and q not in nm: continue users.append({ 'id': obj.get('id'), 'email': obj.get('email'), 'created_at': obj.get('created_at'), 'name': (obj.get('user_metadata') or {}).get('name'), 'org': (obj.get('user_metadata') or {}).get('org'), 'role': (obj.get('user_metadata') or {}).get('role', 'user'), 'email_confirmed_at': obj.get('email_confirmed_at'), 'last_sign_in_at': obj.get('last_sign_in_at'), }) except Exception: users = [] return templates.TemplateResponse( "admin/users.html", {"request": request, "users": users}, ) @router.post("/admin/users/role") async def admin_users_set_role(request: Request, user_id: str = Form(...), role: str = Form(...)): if not _is_admin(request): return RedirectResponse("/auth/login", status_code=status.HTTP_303_SEE_OTHER) try: role_norm = role.lower().strip() if role_norm not in ("admin", "user"): raise ValueError("invalid role") client = sps.get_admin_client() client.auth.admin.update_user_by_id(user_id, {"user_metadata": {"role": role_norm}}) admin_email = (request.session.get("user") or {}).get("email") _append_audit(f"role_change\tadmin={admin_email}\tuser_id={user_id}\trole={role_norm}") return RedirectResponse("/admin/users", status_code=status.HTTP_303_SEE_OTHER) except Exception: _append_audit(f"role_change_failed\tuser_id={user_id}\trole={role}") return RedirectResponse("/admin/users?e=role_failed", status_code=status.HTTP_303_SEE_OTHER) @router.get("/admin/logs", response_class=HTMLResponse) async def admin_logs(request: Request): if not _is_admin(request): return RedirectResponse("/auth/login", status_code=status.HTTP_303_SEE_OTHER) return templates.TemplateResponse( "admin/logs.html", {"request": request}, ) @router.get("/admin/api/logs") async def api_logs(request: Request, source: str = "journal", unit: str = "govbot-web.service", lines: int = 200): if not _is_admin(request): return JSONResponse({"error": "unauthorized"}, status_code=401) allowed_units = sps.get_log_units() if source == "journal" and unit not in allowed_units: return JSONResponse({"error": "unit not allowed", "allowed": sorted(list(allowed_units))}, status_code=400) try: lines = max(10, min(2000, int(lines))) except Exception: lines = 200 try: if source == "file": path = sps.get_app_log_path() if not path: return JSONResponse({"error": "app log path not configured"}, status_code=400) p = Path(path) if not p.exists(): return JSONResponse({"error": f"log file not found: {path}"}, status_code=404) data = p.read_text(encoding="utf-8", errors="ignore").splitlines() text = "\n".join(data[-lines:]) return JSONResponse({"source": "file", "path": path, "lines": lines, "text": text}) # journal cp = subprocess.run( ["journalctl", "-q", "-u", unit, "-n", str(lines), "--no-pager"], capture_output=True, text=True, timeout=8 ) text = (cp.stdout or "") err = (cp.stderr or "") if cp.returncode != 0 or ("insufficient permissions" in err.lower()): return JSONResponse({ "error": "insufficient permissions", "hint": "Add service user to 'systemd-journal' group and restart: sudo usermod -aG systemd-journal ", "stderr": err, }, status_code=500) return JSONResponse({"source": "journal", "unit": unit, "lines": lines, "text": text}) except Exception as e: return JSONResponse({"error": str(e)}, status_code=500) def _append_audit(message: str): try: log_dir = Path(__file__).resolve().parents[3] / "logs" log_dir.mkdir(parents=True, exist_ok=True) with (log_dir / "admin_audit.log").open("a", encoding="utf-8") as f: ts = datetime.utcnow().isoformat() f.write(f"{ts}\t{message}\n") except Exception: pass # ---------------- Scheduler 관리 ---------------- @router.get("/admin/scheduler", response_class=HTMLResponse) async def admin_scheduler(request: Request): if not _is_admin(request): return RedirectResponse("/auth/login", status_code=status.HTTP_303_SEE_OTHER) # show jobs jobs = list_jobs() return templates.TemplateResponse("admin/scheduler.html", {"request": request, "jobs": jobs}) @router.post("/admin/scheduler") async def admin_scheduler_update( request: Request, job: str = Form(...), enabled: str = Form(default="on"), seconds: int | None = Form(default=None), kind: str = Form(default="interval"), hour: int | None = Form(default=None), minute: int | None = Form(default=None), ): if not _is_admin(request): return RedirectResponse("/auth/login", status_code=status.HTTP_303_SEE_OTHER) start_scheduler() enable = (enabled == "on") # Lazy-resolve job function to avoid heavy imports at startup def _resolve(job_id: str): try: if job_id == "motie_id": from app.crawler.motie_n8n import run_once as fn; return fn if job_id == "moef_id": from app.crawler.moef_n8n import run_once as fn; return fn if job_id == "me_id": from app.crawler.me_n8n import run_once as fn; return fn if job_id == "motie_org": from app.crawler.motie_org_sync_n8n import run_once as fn; return fn if job_id == "moef_org": from app.crawler.moef_org_sync_n8n import run_once as fn; return fn if job_id == "group_n8n": from app.crawler.group_n8n import run_once as fn; return fn if job_id == "daily_digest": from app.services.announce_service import send_daily_digest as fn; return fn except Exception: return None return None fn = _resolve(job) if not fn: return RedirectResponse("/admin/scheduler", status_code=status.HTTP_303_SEE_OTHER) if kind == "daily": upsert_daily_cron(job, fn, hour=(hour or 8), minute=(minute or 0), enabled=enable) else: upsert_interval_job(job, fn, seconds=(seconds or 3600), enabled=enable) return RedirectResponse("/admin/scheduler", status_code=status.HTTP_303_SEE_OTHER)