"""Configure free wake cron: HF Space Scheduler (Playwright) or cron-job.org API.""" from __future__ import annotations import json import os import sys from pathlib import Path import requests from dotenv import dotenv_values from huggingface_hub import get_token SPACE = "cramamoorthy/stock-prediction-api" SPACE_URL = "https://cramamoorthy-stock-prediction-api.hf.space" SETTINGS = f"https://huggingface.co/spaces/{SPACE}/settings" CRON_API = "https://api.cron-job.org" def _secret() -> str: v = dotenv_values(Path(__file__).resolve().parent.parent / ".env").get("CRON_SECRET") if not v: raise SystemExit("CRON_SECRET missing in backend/.env — run configure_hf_space.py first") return v def _job_schedule_daily() -> dict: return { "timezone": "UTC", "expiresAt": 0, "hours": [21], "minutes": [0], "mdays": [-1], "months": [-1], "wdays": [1, 2, 3, 4, 5], } def _job_schedule_weekly() -> dict: return { "timezone": "UTC", "expiresAt": 0, "hours": [2], "minutes": [0], "mdays": [-1], "months": [-1], "wdays": [0], } def setup_cron_job_org(secret: str) -> bool: api_key = os.getenv("CRON_JOB_ORG_API_KEY") or dotenv_values( Path(__file__).resolve().parent.parent / ".env" ).get("CRON_JOB_ORG_API_KEY") if not api_key: return False headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } existing = requests.get(f"{CRON_API}/jobs", headers=headers, timeout=30) existing.raise_for_status() titles = {j.get("title") for j in existing.json().get("jobs", [])} jobs = [ ( "stock-prediction-daily", f"{SPACE_URL}/api/v1/cron/daily", _job_schedule_daily(), ), ( "stock-prediction-weekly", f"{SPACE_URL}/api/v1/cron/metadata/weekly", _job_schedule_weekly(), ), ] for title, url, schedule in jobs: if title in titles: print(f"cron-job.org: {title} already exists") continue payload = { "job": { "title": title, "url": url, "enabled": True, "requestMethod": 1, "schedule": schedule, "extendedData": { "headers": {"X-Cron-Secret": secret}, }, } } r = requests.put(f"{CRON_API}/jobs", headers=headers, json=payload, timeout=30) r.raise_for_status() print(f"cron-job.org: created {title} (id={r.json().get('jobId')})") return True def setup_hf_scheduler_playwright(secret: str) -> bool: try: from playwright.sync_api import sync_playwright except ImportError: return False token = get_token() if not token: return False captured: list[dict] = [] with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context( user_agent=( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" ) ) def on_request(req): u = req.url if "huggingface.co" in u and req.method in ("POST", "PUT", "PATCH", "DELETE"): if any(x in u.lower() for x in ("schedul", "cron", "webhook", "request")): captured.append({"method": req.method, "url": u, "post": req.post_data}) page = context.new_page() page.on("request", on_request) # Token login flow used by HF hub page.goto("https://huggingface.co/login", wait_until="domcontentloaded", timeout=60_000) page.goto( f"https://huggingface.co/login?token={token}&next=/spaces/{SPACE}/settings", wait_until="domcontentloaded", timeout=120_000, ) page.wait_for_timeout(3000) if "login" in page.url.lower() and "settings" not in page.url: page.goto(SETTINGS, wait_until="domcontentloaded", timeout=120_000) page.wait_for_timeout(2000) if "403" in (page.title() or "") or "ERROR" in (page.inner_text("body")[:200] if page.locator("body").count() else ""): browser.close() return False # Open Scheduler tab for sel in [ "a[href*='scheduler']", "button:has-text('Scheduler')", "text=Scheduler", ]: loc = page.locator(sel) if loc.count(): loc.first.click() page.wait_for_timeout(2000) break def add_job(title: str, cron_expr: str, path: str) -> None: add_btn = page.get_by_role("button", name=lambda n: n and any( w in (n or "").lower() for w in ("add", "new", "create") )) if add_btn.count(): add_btn.first.click() page.wait_for_timeout(1000) page.locator("input").filter(has_text="").all() # noop wake for inp in page.locator("input:visible").all(): ph = (inp.get_attribute("placeholder") or "").lower() name = (inp.get_attribute("name") or "").lower() if "cron" in ph or "schedule" in ph or "cron" in name: inp.fill(cron_expr) elif "path" in ph or "path" in name or "url" in ph: inp.fill(path) page.locator("select:visible").first.select_option("POST") if page.locator("select:visible").count() else None # headers page.get_by_text("Header", exact=False).first.click() if page.get_by_text("Header").count() else None header_inputs = page.locator("input:visible") for i, inp in enumerate(header_inputs.all()): val = inp.input_value() if not val and i == 0: inp.fill("X-Cron-Secret") elif val == "X-Cron-Secret" or (not val and i == 1): inp.fill(secret) save = page.get_by_role("button", name=lambda n: n and any( w in (n or "").lower() for w in ("save", "create", "add", "submit") )) if save.count(): save.first.click() page.wait_for_timeout(3000) print(f"HF Scheduler UI: attempted {title}") add_job("daily", "0 21 * * 1-5", "/api/v1/cron/daily") add_job("weekly", "0 2 * * 0", "/api/v1/cron/metadata/weekly") out = Path(__file__).parent / "hf_scheduler_capture.json" out.write_text(json.dumps(captured, indent=2), encoding="utf-8") # Verify jobs listed in UI body = page.inner_text("body") if page.locator("body").count() else "" ok = "/api/v1/cron/daily" in body or "cron/daily" in body browser.close() return ok return False def setup_hf_scheduler_api(secret: str) -> bool: """Try discovered/internal HF endpoints.""" from huggingface_hub.utils import build_hf_headers token = get_token() if not token: return False h = build_hf_headers(token=token) h["Content-Type"] = "application/json" base = f"https://huggingface.co/api/spaces/{SPACE}" payloads = [ { "schedule": "0 21 * * 1-5", "method": "POST", "path": "/api/v1/cron/daily", "headers": [{"name": "X-Cron-Secret", "value": secret}], }, { "schedule": "0 2 * * 0", "method": "POST", "path": "/api/v1/cron/metadata/weekly", "headers": [{"name": "X-Cron-Secret", "value": secret}], }, ] for path_suffix in ("scheduled-requests", "scheduler/requests", "schedule"): for p in payloads: r = requests.post(f"{base}/{path_suffix}", headers=h, json=p, timeout=30) if r.status_code in (200, 201): print(f"HF API {path_suffix}: OK") return True return False def main() -> int: import subprocess root = Path(__file__).resolve().parent.parent.parent script = root / "scripts" / "deploy-hf-cron-worker.ps1" if script.exists(): subprocess.run( ["powershell", "-NoProfile", "-File", str(script)], cwd=str(root), check=True, ) print("Done: Cloudflare Worker cron (free).") return 0 print(f"Missing {script}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())