278 lines
9.9 KiB
Python
278 lines
9.9 KiB
Python
"""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()
|