""" app.py — ETP Distributed Scraper: Control Plane v5.1 (Item-Level Dashboard + Hash Status + Refresher Lock) FastAPI + HTML/JS. KEY CHANGES FROM v5: 1. DASHBOARD: Uses fixed get_dashboard_item_stats() RPC - success_items and failed_items counted for ALL chunk statuses (not just completed/failed) - This ensures: done + pending = total ALWAYS HOLDS - No more 1,000-row .select() limit - Mathematically perfect item-level progress 2. HASH STATUS: Shows current auth_hash status from system_tokens - When it was last refreshed - Whether it's active 3. REFRESHER LOCK STATUS: Shows which hash_refresher instance is active - Instance ID - Last heartbeat time - If no refresher is running, shows warning 4. Removed direct sub_tasks .select() — all stats via RPC """ import os, time, threading, logging from datetime import datetime, timezone from supabase import create_client, Client from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse import uvicorn logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") log = logging.getLogger("ControlPlane") SUPABASE_URL = os.environ["SUPABASE_URL"] SUPABASE_KEY = os.environ["SUPABASE_SERVICE_KEY"] SURGEON_TIMEOUT = int(os.environ.get("SURGEON_TIMEOUT_MINUTES", "5")) WATCHDOG_INTERVAL = 300 supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY) app = FastAPI() SERIES_MAP = {"0": "QL", "1": "ST", "2": "ML"} # ─── BACKGROUND SURGEON (Auto-Cleaner) ────────────────────────────────── def watchdog_loop(): log.info("Watchdog started. Will permanently delete dead workers every 5 mins.") while True: try: res = supabase.rpc("surgeon_recover_dead_tasks", {"p_timeout_minutes": SURGEON_TIMEOUT}).execute() recovered = res.data or 0 if recovered > 0: log.info(f"Surgeon recovered {recovered} stuck chunks and wiped dead workers.") except Exception as e: log.error(f"Surgeon error: {e}") time.sleep(WATCHDOG_INTERVAL) threading.Thread(target=watchdog_loop, daemon=True).start() # ─── API ROUTES ──────────────────────────────────────────────── @app.get("/api/dashboard") def api_dashboard(): # 1. Fetch tasks (few rows, no limit issue) tasks = supabase.table("main_tasks").select("*").order("priority").order("created_at").execute().data or [] # 2. Fetch item-level stats via RPC (server-side aggregation, no row limit) # Fixed RPC: success_items + failed_items counted for ALL statuses # This ensures: done + pending = total ALWAYS stats_data = supabase.rpc("get_dashboard_item_stats").execute().data or [] stats_map = {} for s in stats_data: stats_map[s["main_task_id"]] = s # 3. Compute global item-level stats global_stats = {"total_items": 0, "pending_items": 0, "in_progress_items": 0, "success_items": 0, "failed_items": 0} for t in tasks: s = stats_map.get(t["id"], {}) t["item_stats"] = { "total_items": s.get("total_items", 0) or 0, "success_items": s.get("success_items", 0) or 0, "failed_items": s.get("failed_items", 0) or 0, "in_progress_items": s.get("in_progress_items", 0) or 0, "pending_items": s.get("pending_items", 0) or 0, } for key in global_stats: global_stats[key] += t["item_stats"].get(key, 0) # 4. Fetch and filter live workers (HIDE IF DEAD > 5 MINS) raw_workers = supabase.table("workers").select("*").order("started_at", desc=True).execute().data or [] live_workers = [] now_utc = datetime.now(timezone.utc) for w in raw_workers: try: hb_str = w["last_heartbeat"].replace("Z", "+00:00") hb_time = datetime.fromisoformat(hb_str) seconds_dead = (now_utc - hb_time).total_seconds() if seconds_dead <= 300: w["seconds_dead"] = int(seconds_dead) live_workers.append(w) except Exception: pass # 5. Fetch System State sys_config = supabase.table("system_config").select("*").eq("id", 1).execute().data global_paused = sys_config[0]["is_globally_paused"] if sys_config else False # 6. Fetch Auth Hash Status hash_info = None try: hash_res = supabase.rpc("get_active_auth_hash").execute() if hash_res.data and len(hash_res.data) > 0: hash_info = hash_res.data[0] except Exception: pass # 7. Fetch Refresher Lock Status refresher_info = None if sys_config: sc = sys_config[0] if sc.get("active_refresher_id"): refresher_info = { "instance_id": sc["active_refresher_id"], "last_heartbeat": sc.get("refresher_last_heartbeat"), } return JSONResponse({ "tasks": tasks, "workers": live_workers, "global_stats": global_stats, "hash_info": hash_info, "refresher_info": refresher_info, "global_paused": global_paused }) @app.post("/api/global-pause") async def api_global_pause(request: Request): b = await request.json() is_paused = bool(b.get("paused", False)) supabase.rpc("set_global_pause", {"p_paused": is_paused}).execute() return JSONResponse({"ok": True, "paused": is_paused}) @app.post("/api/create-task") async def api_create_task(request: Request): try: b = await request.json() start_id = str(int(b["start_id"])).zfill(10) end_id = str(int(b["end_id"])).zfill(10) priority = int(b.get("priority", 100)) if start_id[:3] != end_id[:3]: return JSONResponse({"ok": False, "error": "First 3 digits must match (year+series)."}, status_code=400) year = f"20{start_id[:2]}" series_digit = start_id[2] series_label = SERIES_MAP.get(series_digit, series_digit) series_name = f"{series_digit} - {series_label}" supabase.table("main_tasks").insert({ "target_url_type": "eKhanij", "series_name": series_name, "start_etp_id": int(start_id), "end_etp_id": int(end_id), "chunk_size": int(b.get("chunk_size", 5000)), "concurrency": int(b.get("concurrency", 300)), "priority": priority }).execute() return JSONResponse({"ok": True}) except Exception as e: return JSONResponse({"ok": False, "error": str(e)}, status_code=400) @app.post("/api/tasks/{task_id}/action") async def api_task_action(task_id: str, request: Request): b = await request.json() action = b.get("action") try: if action == "pause": supabase.rpc("pause_main_task", {"p_main_id": task_id}).execute() elif action == "resume": supabase.rpc("resume_main_task", {"p_main_id": task_id}).execute() elif action == "delete": supabase.table("main_tasks").delete().eq("id", task_id).execute() elif action == "priority": supabase.rpc("reorder_task_priority", {"p_main_id": task_id, "p_new_priority": int(b.get("priority", 100))}).execute() elif action == "retry": supabase.table("sub_tasks").update({"status": "pending", "assigned_worker_id": None})\ .eq("status", "failed").eq("main_task_id", task_id).execute() return JSONResponse({"ok": True}) except Exception as e: return JSONResponse({"ok": False, "error": str(e)}, status_code=400) @app.post("/api/surgeon") def api_surgeon(): res = supabase.rpc("surgeon_recover_dead_tasks", {"p_timeout_minutes": 5}).execute() return JSONResponse({"ok": True, "recovered": res.data or 0}) # ─── FRONTEND HTML (PREMIUM UI — ITEM-LEVEL STATS + REFRESHER LOCK) ──────── HTML_CONTENT = r""" ETP Command Center

⚒ ETP Control Plane

Refresher: Loading...
Hash: Loading...
Syncing...
SYSTEM ACTIVE
📊 Live Dashboard
➕ Manage Tasks
🤖 Launch Workers
0%
Global Progress
0
Active Workers
0
Total Items
0
Done (Success)
0
In Progress
0
Failed Items
Integrity Check: Done: 0 + Pending: 0 + Active: 0 = Total: 0

📋 Task Priority Queue (Item-Level)

Loading system data...

🤖 Live Worker Fleet

Loading workers...

➕ Enqueue New Task

Tasks process strictly in order of Priority (Priority 1 processes before Priority 100).
If a chunk was already processed in the past, workers will automatically skip it (Delta Recovery).

🔒 Hash Refresher (Run ONE Instance in Colab)

This MUST be running before workers can operate. It refreshes the auth_hash every 5 minutes and syncs the Data Lake.
Distributed Lock: If you accidentally start multiple instances, only the LATEST one runs — all older ones self-terminate.

# 1. Install required packages silently !pip install -q supabase requests huggingface_hub urllib3 lxml import os, urllib.request # 2. Pipeline Credentials os.environ["SUPABASE_URL"] = "YOUR_SUPABASE_URL" os.environ["SUPABASE_SERVICE_KEY"] = "YOUR_SUPABASE_KEY" os.environ["HF_TOKEN"] = "YOUR_HF_TOKEN" os.environ["HF_REPO_ID"] = "YOUR_REPO" # 3. Download and run hash refresher REPO_NAME = "YOUR_REPO_NAME" urllib.request.urlretrieve(f"https://raw.githubusercontent.com/YOUR_GITHUB/{REPO_NAME}/main/hash_refresher.py", "hash_refresher.py") !python hash_refresher.py

🌎 Launch Worker Nodes (Run as Many as You Want)

Workers are now thin and stateless. They read the auth_hash from the database and perform 1-step GETs. No more heavy boot sync. No more race conditions. Launch as many as you want simultaneously.

# 1. Install required packages silently !pip install -q supabase requests beautifulsoup4 huggingface_hub urllib3 lxml import os, urllib.request # 2. Pipeline Credentials os.environ["SUPABASE_URL"] = "YOUR_SUPABASE_URL" os.environ["SUPABASE_SERVICE_KEY"] = "YOUR_SUPABASE_KEY" os.environ["HF_TOKEN"] = "YOUR_HF_TOKEN" os.environ["HF_REPO_ID"] = "YOUR_REPO" # 3. Download and run thin worker REPO_NAME = "YOUR_REPO_NAME" urllib.request.urlretrieve(f"https://raw.githubusercontent.com/YOUR_GITHUB/{REPO_NAME}/main/worker.py", "worker.py") !python worker.py
""" @app.get("/", response_class=HTMLResponse) def root(): return HTML_CONTENT if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)