"""MASTER FLEET — super-parser orchestrator. One process that runs the ENTIRE key farm: 1. All 41 *_eternal.py lanes (each = one public source walked forever) 2. validator_loop.py (validates NEW -> WORKING/DEAD every 15 min) 3. universal_hunt.py from ../keyfarm-actions (GitHub code-search dork hunt) 4. dashscope_hunt.py from ../keyfarm-actions (targeted DS/ZAI sweep) 5. Every MERGE_EVERY sec: pull new keys from keyfarm-actions/data/keys.db into the main data/keys.db (dedupe by hash, never overwrite better status) 6. Auto-restart any lane that dies (3s grace). Logs everything. Usage: python -u master_fleet.py # run forever python -u master_fleet.py --status # one-shot: what's alive + DB stats python -u master_fleet.py --stop # kill all children, exit Env: KF_DB override main DB path MF_LANES comma-separated subset of lane scripts (debug) """ import os import signal import sqlite3 import subprocess import sys import time BASE = os.path.dirname(os.path.abspath(__file__)) PY = sys.executable DB = os.environ.get("KF_DB") or os.path.join(BASE, "data", "keys.db") ACTIONS_DIR = os.path.normpath(os.path.join(BASE, "..", "keyfarm-actions")) ACTIONS_DB = os.path.join(ACTIONS_DIR, "data", "keys.db") LOG = os.path.join(BASE, "data", "master_fleet.log") STATE = os.path.join(BASE, "data", "master_fleet.pid") MERGE_EVERY = 900 # 15 min RESTART_GRACE = 3 # seconds before restart check os.makedirs(os.path.join(BASE, "data"), exist_ok=True) STOP = False def log(msg): line = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}" print(line, flush=True) try: with open(LOG, "a", encoding="utf-8", errors="replace") as f: f.write(line + "\n") except Exception: pass # ---------------------------------------------------------------- lanes ---- def discover_lanes(): lanes = [] sub = os.environ.get("MF_LANES") for fn in sorted(os.listdir(BASE)): if not fn.endswith("_eternal.py") or fn.startswith(("test_", "dbg_")): continue if fn == "dockerhub_eternal.py": # disabled upstream continue if sub and fn not in sub.split(","): continue lanes.append((fn, os.path.join(BASE, fn), BASE)) # external hunts from keyfarm-actions (lean remote versions) for fn in ("universal_hunt.py", "dashscope_hunt.py"): p = os.path.join(ACTIONS_DIR, fn) if os.path.exists(p): lanes.append((fn, p, ACTIONS_DIR)) # validator (must see main DB; actions hunts write their own DB) lanes.append(("validator_loop.py", os.path.join(BASE, "validator_loop.py"), BASE)) return lanes # ------------------------------------------------------------- processes --- class Child: def __init__(self, name, path, cwd): self.name = name self.path = path self.cwd = cwd self.proc = None self.starts = 0 self.last_start = 0.0 def alive(self): return self.proc is not None and self.proc.poll() is None def start(self): env = dict(os.environ) env["PYTHONUNBUFFERED"] = "1" env["PYTHONIOENCODING"] = "utf-8" logfile = os.path.join(self.cwd, "data", f"mf_{self.name.replace('.py','')}.out") os.makedirs(os.path.dirname(logfile), exist_ok=True) fh = open(logfile, "ab") kwargs = {} if os.name == "nt": kwargs["creationflags"] = ( subprocess.CREATE_NO_WINDOW | subprocess.CREATE_NEW_PROCESS_GROUP) self.proc = subprocess.Popen( [PY, "-u", self.path], cwd=self.cwd, env=env, stdout=fh, stderr=subprocess.STDOUT, **kwargs) self.starts += 1 self.last_start = time.time() log(f"START {self.name} pid={self.proc.pid} (starts={self.starts})") def stop(self): if self.proc and self.proc.poll() is None: try: if os.name == "nt": self.proc.send_signal(signal.CTRL_BREAK_EVENT) try: self.proc.wait(timeout=8) except subprocess.TimeoutExpired: self.proc.kill() else: self.proc.terminate() self.proc.wait(timeout=8) except Exception: try: self.proc.kill() except Exception: pass log(f"STOP {self.name}") def list_children(): """Existing python children matching our lane names (from previous run).""" out = subprocess.run( ["powershell", "-NoProfile", "-Command", "Get-CimInstance Win32_Process -Filter \"name like 'python%'\" | " "ForEach-Object { $_.ProcessId.ToString() + ' ' + $_.CommandLine }"], capture_output=True, text=True, timeout=30).stdout if os.name == "nt" else "" return out # ---------------------------------------------------------------- merge ---- STATUS_RANK = {"WORKING": 5, "FREE": 4, "LIMITED": 3, "NEW": 2, "ERR": 1, "DEAD": 0} def merge_actions_db(): if not os.path.exists(ACTIONS_DB) or not os.path.exists(DB): return 0 dst = sqlite3.connect(DB, timeout=120) src = sqlite3.connect(f"file:{ACTIONS_DB}?mode=ro", uri=True, timeout=60) n = 0 try: rows = src.execute( "SELECT hash,val,prov,status,found,repo,plan,remaining FROM keys " "WHERE status IN ('WORKING','FREE','LIMITED','NEW')").fetchall() have = {r[0]: r[1] for r in dst.execute( "SELECT hash,status FROM keys").fetchall()} for h, val, prov, st, found, repo, plan, rem in rows: if h not in have: dst.execute( "INSERT OR IGNORE INTO keys " "(hash,val,prov,status,found,repo,plan,remaining) " "VALUES (?,?,?,?,?,?,?,?)", (h, val, prov, st, found, repo, plan, rem)) n += 1 elif STATUS_RANK.get(st, 0) > STATUS_RANK.get(have[h], 0): dst.execute( "UPDATE keys SET status=?,plan=?,remaining=? WHERE hash=?", (st, plan, rem, h)) n += 1 dst.commit() finally: src.close() dst.close() return n # --------------------------------------------------------------- status ---- def db_stats(): try: c = sqlite3.connect(f"file:{DB}?mode=ro", uri=True, timeout=60) total = c.execute("SELECT count(*) FROM keys").fetchone()[0] live = c.execute( "SELECT count(*) FROM keys WHERE status IN " "('WORKING','FREE','LIMITED')").fetchone()[0] new = c.execute( "SELECT count(*) FROM keys WHERE status='NEW'").fetchone()[0] c.close() return total, live, new except Exception as e: return (0, 0, str(e)) # ---------------------------------------------------------------- main ----- def run(): with open(STATE, "w") as f: f.write(str(os.getpid())) lanes = discover_lanes() log(f"=== MASTER FLEET: {len(lanes)} lanes, DB={DB} ===") children = [Child(n, p, cwd) for n, p, cwd in lanes] for ch in children: ch.start() time.sleep(0.4) # stagger DB open storm last_merge = 0.0 try: while not STOP: time.sleep(10) now = time.time() for ch in children: if not ch.alive(): rc = ch.proc.poll() if ch.proc else "?" # restart unless flapping >20 times in 10 min if now - ch.last_start < 30 and ch.starts > 20: continue log(f"DEAD {ch.name} rc={rc} -> restart") ch.start() if now - last_merge > MERGE_EVERY: last_merge = now try: n = merge_actions_db() if n: log(f"MERGE actions->main: +{n} rows") t, l, new = db_stats() log(f"DB total={t} live={l} new={new}") except Exception as e: log(f"merge error: {e!r}") except KeyboardInterrupt: pass finally: for ch in children: ch.stop() log("=== MASTER FLEET down ===") def status(): total, live, new = db_stats() log(f"STATUS db: total={total} live={live} new={new}") out = list_children() names = {} for line in out.splitlines(): for fn in ("eternal.py", "validator_loop.py", "universal_hunt.py", "dashscope_hunt.py", "master_fleet.py"): if fn in line: pid = line.split(" ", 1)[0] script = "?" if ".py" in line: parts = line.split() script = next((p for p in reversed(parts) if p.endswith(".py")), "?") names.setdefault(script, []).append(pid) for s, pids in sorted(names.items()): log(f" ALIVE {s}: {','.join(pids)}") try: print(open(STATE).read().strip(), "<- master pid") except Exception: pass def stop_all(): out = list_children() killed = 0 for line in out.splitlines(): if any(k in line for k in ("_eternal.py", "validator_loop.py", "universal_hunt.py", "dashscope_hunt.py", "master_fleet.py")): pid = line.split(" ", 1)[0].strip() if pid.isdigit() and int(pid) != os.getpid(): subprocess.run(["taskkill", "/PID", pid, "/F"] if os.name == "nt" else ["kill", pid], capture_output=True) killed += 1 log(f"STOP-ALL killed {killed} processes") if __name__ == "__main__": if "--status" in sys.argv: status() elif "--stop" in sys.argv: stop_all() else: run()