""" cron_generate_links.py - keep iDEAL links stocked for every price tier ======================================================================= Run this from cron every few minutes (see CRON_SETUP.md). On every tick it makes ONE link per price tier in PRICING by running g2a_ideal_link.py once per tier. Each run creates its own GoLogin profile, takes its own panel email and saves its own link to the panel DB. The GoLogin plan only allows MAX_PARALLEL (2) cloud profiles at the same time, so runs go through a small worker pool: two run at once and the next tier starts the moment one finishes. 3 tiers, 2 slots -> tier A + tier B together, then tier C 5 tiers, 2 slots -> A + B, then C + D, then E Fairness: tiers are queued with the fewest fresh unused links first; ties go to the tier whose last link was generated longest ago. Over time every tier gets the same number of links. Adding a tier to PRICING in g2a_ideal_link.py is all it takes. Safety: * a lock file stops two cron ticks from overlapping (a tick with 3 tiers takes two rounds, so a 5-minute cron simply skips while one is running) * tiers are dropped when there are fewer available emails than tiers * MAX_READY_PER_TIER (optional) skips a tier that already has enough fresh unused links, so emails are not burnt while nobody is buying Usage: python cron_generate_links.py normal cron tick python cron_generate_links.py --dry-run show what would run, run nothing """ import datetime import os import pathlib import subprocess import sys import time HERE = pathlib.Path(__file__).resolve().parent SCRIPT = HERE / "g2a_ideal_link.py" LOG_DIR = HERE / "logs" LOCK_FILE = HERE / "cron_generate.lock" # ----------------------------------------------------------------- settings # How long a generated link is considered fresh. MUST match # IDEAL_LINK_MAX_AGE in the checkout config.php (there it is in seconds) # and LINK_MAX_AGE_MINUTES in panel/config.php. LINK_MAX_AGE_MINUTES = int(os.environ.get("LINK_MAX_AGE_MINUTES", "15")) # 0 = unlimited: generate one link per tier on every tick, no matter how # many fresh links that tier already has. Set e.g. 2 to stop generating for # a tier once it has 2 fresh unused links. MAX_READY_PER_TIER = int(os.environ.get("MAX_READY_PER_TIER", "0")) # How many g2a_ideal_link.py runs may be active at the same time. Limited by # the GoLogin plan: number of cloud profiles allowed to run concurrently. MAX_PARALLEL = max(1, int(os.environ.get("MAX_PARALLEL", "2"))) # A full run is browser start-up + 5 checkout steps (25 s timeout each) + # LINK_WAIT_MINUTES (3) for the iDEAL page. Kill a run still going after this. RUN_TIMEOUT_MINUTES = int(os.environ.get("RUN_TIMEOUT_MINUTES", "10")) # Extra arguments passed to every g2a_ideal_link.py run. RUN_MODE = os.environ.get("G2A_MODE", "gologin") # -------------------------------------------------------------------------- DRY_RUN = "--dry-run" in sys.argv def log(msg: str) -> None: line = f"[{datetime.datetime.now():%Y-%m-%d %H:%M:%S}] {msg}" print(line, flush=True) try: LOG_DIR.mkdir(exist_ok=True) with (LOG_DIR / "cron_generate.log").open("a", encoding="utf-8") as f: f.write(line + "\n") except Exception: pass # ----------------------------------------------------------------- lock def acquire_lock() -> bool: """Create the lock file atomically. Stale locks are taken over.""" # A tick may need several rounds of MAX_PARALLEL runs. stale_after = (RUN_TIMEOUT_MINUTES * 3 + 5) * 60 for _ in range(2): try: fd = os.open(str(LOCK_FILE), os.O_CREAT | os.O_EXCL | os.O_WRONLY) with os.fdopen(fd, "w") as f: f.write(f"{os.getpid()} {int(time.time())}\n") return True except FileExistsError: try: age = time.time() - LOCK_FILE.stat().st_mtime except FileNotFoundError: continue if age > stale_after: log(f"Stale lock ({age / 60:.0f} min old) - taking over") try: LOCK_FILE.unlink() except FileNotFoundError: pass continue log(f"Previous tick still running ({age / 60:.0f} min) - skipping") return False return False def release_lock() -> None: try: LOCK_FILE.unlink() except FileNotFoundError: pass # ----------------------------------------------------------------- DB def count_available_emails(g) -> int: conn = g.panel_db() if not conn: return 0 cur = conn.cursor() cur.execute("SELECT COUNT(*) FROM emails WHERE status = 'available'") n = int(cur.fetchone()[0]) cur.close() conn.close() return n def fresh_links_per_tier(g, tiers) -> dict: """How many unused, still-fresh links each tier has right now.""" counts = {t: 0 for t in tiers} conn = g.panel_db() if not conn: return counts cutoff = int(time.time()) - LINK_MAX_AGE_MINUTES * 60 cur = conn.cursor() cur.execute( "SELECT price, COUNT(*) FROM orders " "WHERE status = 'completed' AND ideal_link <> '' " "AND completed_at >= %s GROUP BY price", (cutoff,), ) for price, n in cur.fetchall(): if price in counts: counts[price] = int(n) cur.close() conn.close() return counts def last_generated_per_tier(g, tiers) -> dict: """When each tier last got a run started (0 = never). Used as the tie-breaker so tiers take turns instead of PRICING order always winning.""" last = {t: 0 for t in tiers} conn = g.panel_db() if not conn: return last cur = conn.cursor() cur.execute("SELECT price, MAX(created_at) FROM orders GROUP BY price") for price, ts in cur.fetchall(): if price in last and ts: last[price] = int(ts) cur.close() conn.close() return last # ----------------------------------------------------------------- pool def start_run(tier: str, stamp: str): out = LOG_DIR / f"run_{stamp}_{tier.replace('.', '_')}.log" fh = out.open("w", encoding="utf-8") cmd = [sys.executable, "-u", str(SCRIPT), RUN_MODE, "--db", "--price", tier] p = subprocess.Popen(cmd, cwd=str(HERE), stdout=fh, stderr=subprocess.STDOUT) log(f" started {tier} EUR -> pid {p.pid} -> {out.name}") return {"tier": tier, "proc": p, "fh": fh, "started": time.time()} def run_pool(queue) -> dict: """Run every tier in `queue`, at most MAX_PARALLEL at the same time. The next tier starts as soon as a slot frees up.""" LOG_DIR.mkdir(exist_ok=True) stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") pending = list(queue) active = [] results = {} while pending or active: # fill free slots while pending and len(active) < MAX_PARALLEL: active.append(start_run(pending.pop(0), stamp)) if pending and len(active) < MAX_PARALLEL: time.sleep(3) # stagger GoLogin profile creation a little time.sleep(5) # collect finished / timed-out runs still = [] for r in active: rc = r["proc"].poll() ran = time.time() - r["started"] if rc is None and ran > RUN_TIMEOUT_MINUTES * 60: log( f" TIMEOUT {r['tier']} EUR (pid {r['proc'].pid}) " f"after {RUN_TIMEOUT_MINUTES} min - killing" ) try: r["proc"].kill() r["proc"].wait(timeout=10) except Exception: pass rc = -1 if rc is None: still.append(r) continue r["fh"].close() results[r["tier"]] = rc log( f" finished {r['tier']} EUR -> exit {rc} " f"({'OK' if rc == 0 else 'FAILED'}, {ran / 60:.1f} min)" ) active = still return results # ----------------------------------------------------------------- main def main() -> int: if not SCRIPT.exists(): log(f"ERROR: {SCRIPT} not found") return 1 # Importing the generator gives us PRICING and the DB helper. The module # does nothing on import (tier selection happens only under __main__). sys.path.insert(0, str(HERE)) import g2a_ideal_link as g # noqa: E402 tiers = list(g.PRICING) if not tiers: log("PRICING is empty - nothing to do") return 0 if not DRY_RUN and not acquire_lock(): return 0 try: available = count_available_emails(g) fresh = fresh_links_per_tier(g, tiers) last = last_generated_per_tier(g, tiers) log( f"Tick | tiers={tiers} | available emails={available} " f"| fresh links={fresh} | parallel={MAX_PARALLEL} " f"| max per tier={MAX_READY_PER_TIER or 'unlimited'}" ) # Fair queue: fewest fresh links first; ties go to the tier that was # generated longest ago; PRICING order only as the final tie-break. order = sorted(tiers, key=lambda t: (fresh[t], last[t], tiers.index(t))) todo = [] for t in order: if MAX_READY_PER_TIER and fresh[t] >= MAX_READY_PER_TIER: log(f" skip {t} EUR: already has {fresh[t]} fresh link(s)") continue todo.append(t) if len(todo) > available: log( f" only {available} email(s) available for {len(todo)} tier(s) " f"- dropping: {todo[available:]}" ) todo = todo[:available] if not todo: log("Nothing to run this tick") return 0 rounds = -(-len(todo) // MAX_PARALLEL) # ceil log( f"Queue ({len(todo)} run(s), {MAX_PARALLEL} at a time, " f"{rounds} round(s)): {todo}" ) if DRY_RUN: return 0 results = run_pool(todo) ok = sum(1 for rc in results.values() if rc == 0) log(f"Tick done: {ok}/{len(results)} link(s) generated") return 0 finally: if not DRY_RUN: release_lock() if __name__ == "__main__": sys.exit(main())