Spaces:
Runtime error
Runtime error
| """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()) |