#!/usr/bin/env python3 """Refresh and validate the official Devpost event snapshot.""" from __future__ import annotations import argparse import html import json import re import subprocess from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen from zoneinfo import ZoneInfo ROOT = Path(__file__).resolve().parents[1] DEFAULT_JSON = ROOT / "docs" / "assets" / "devpost-event-snapshot.json" DEFAULT_MD = ROOT / "docs" / "assets" / "devpost-event-snapshot.md" OVERVIEW_URL = "https://backblaze-generative-media.devpost.com/" RULES_URL = "https://backblaze-generative-media.devpost.com/rules" SCHEMA = "proofframe.devpost_event_snapshot.v1" TIMEOUT_SECONDS = 20 BROWSER_USER_AGENT = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36" ) MONTHS = { "jan": 1, "feb": 2, "mar": 3, "apr": 4, "may": 5, "jun": 6, "jul": 7, "aug": 8, "sep": 9, "oct": 10, "nov": 11, "dec": 12, } REQUIRED_RULE_MARKERS = { "working_app_url": "Provide a URL to a working application", "github_repo_url": "Provide a URL to your public or private GitHub code repository", "demo_video": "Include a demonstration video", "video_under_three_minutes": "less than three (3) minutes", "public_video_host": "YouTube, Vimeo, or Youku", "b2_usage": "Backblaze B2", "genblaze_usage": "Genblaze", } JUDGING_CRITERIA = [ {"name": "Real-world Utility", "marker": "Real-world Utility"}, {"name": "Production Readiness", "marker": "Production Readiness"}, {"name": "B2 Storage + Data Orchestration", "marker": "B2 Storage + Data Orchestration"}, {"name": "Use of Genblaze", "marker": "Use of Genblaz"}, ] FetchResult = dict[str, Any] Fetcher = Callable[[str, int], FetchResult] def utc_now() -> datetime: return datetime.now(timezone.utc).replace(microsecond=0) def format_utc_now() -> str: return utc_now().isoformat().replace("+00:00", "Z") def beijing_date(now: datetime) -> str: return now.astimezone(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d Asia/Shanghai") def fetch_text(url: str, timeout: int = TIMEOUT_SECONDS) -> FetchResult: request = Request( url, headers={ "User-Agent": BROWSER_USER_AGENT, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.9", "Accept-Encoding": "identity", }, ) try: with urlopen(request, timeout=timeout) as response: body = response.read().decode("utf-8", errors="replace") if response.status == 202 or len(body) < 5000: return fetch_text_with_curl(url, timeout) return {"ok": True, "status": response.status, "body": body, "error": None, "transport": "urllib"} except HTTPError as error: body = error.read().decode("utf-8", errors="replace") return {"ok": False, "status": error.code, "body": body, "error": str(error)} except URLError as error: return {"ok": False, "status": None, "body": "", "error": str(error.reason)} except TimeoutError as error: return {"ok": False, "status": None, "body": "", "error": str(error)} def fetch_text_with_curl(url: str, timeout: int = TIMEOUT_SECONDS) -> FetchResult: marker = "\n__PROOFFRAME_HTTP_STATUS__:" try: completed = subprocess.run( [ "curl", "-L", "--compressed", "-sS", "-A", BROWSER_USER_AGENT, "-w", f"{marker}%{{http_code}}", url, ], check=False, capture_output=True, text=True, timeout=timeout, ) except (OSError, subprocess.TimeoutExpired) as error: return {"ok": False, "status": None, "body": "", "error": str(error), "transport": "curl"} body, _, status_text = completed.stdout.rpartition(marker) status = parse_int(status_text) return { "ok": completed.returncode == 0 and status is not None and 200 <= status < 400, "status": status, "body": body, "error": completed.stderr.strip() or None, "transport": "curl", } def text_from_html(raw_html: str) -> str: raw_text = re.sub(r"<(script|style).*?", " ", raw_html, flags=re.IGNORECASE | re.DOTALL) raw_text = re.sub(r"<[^>]+>", "\n", raw_text) raw_text = html.unescape(raw_text) raw_text = re.sub(r"\s+", " ", raw_text) return raw_text.strip() def first_match(pattern: str, text: str, default: str | None = None) -> str | None: match = re.search(pattern, text, flags=re.IGNORECASE) if not match: return default return " ".join(match.group(1).split()) def parse_int(value: str | None) -> int | None: if not value: return None digits = re.sub(r"[^\d]", "", value) return int(digits) if digits else None def parse_deadline_display(text: str) -> str | None: return first_match(r"Deadline:\s*([A-Z][a-z]{2}\s+\d{1,2},\s+\d{4}\s+@\s+\d{1,2}:\d{2}[ap]m\s+EDT)", text) def parse_edt_deadline(deadline_display: str | None) -> datetime | None: if not deadline_display: return None match = re.search( r"([A-Z][a-z]{2})\s+(\d{1,2}),\s+(\d{4})\s+@\s+(\d{1,2}):(\d{2})(am|pm)\s+EDT", deadline_display, flags=re.IGNORECASE, ) if not match: return None month = MONTHS.get(match.group(1).lower()) if not month: return None hour = int(match.group(4)) minute = int(match.group(5)) am_pm = match.group(6).lower() if am_pm == "pm" and hour != 12: hour += 12 if am_pm == "am" and hour == 12: hour = 0 return datetime( int(match.group(3)), month, int(match.group(2)), hour, minute, tzinfo=ZoneInfo("America/New_York"), ) def parse_event_window(label: str, text: str) -> str | None: stop_labels = "|".join( [ "Judging Period:", "Winners Announced:", "Winner Announced:", "2\\. Sponsor", "Requirements", "Submission Requirements", ] ) pattern = rf"{re.escape(label)}:\s*(.+?)(?:{stop_labels}|$)" return first_match(pattern, text) def parse_overview(text: str) -> dict[str, Any]: deadline_display = parse_deadline_display(text) deadline_dt = parse_edt_deadline(deadline_display) participant_count = parse_int( first_match(r"Participants\s*\(([\d,]+)\)", text) or first_match(r"([\d,]+)\s+participants", text) ) prize_total = parse_int( first_match(r"\$\s*([\d,]+)\s+in\s+cash", text) or first_match(r"\$\s*([\d,]+)\s+in\s+prizes", text) ) return { "name": first_match(r"(Backblaze Generative Media Hackathon)", text, "Backblaze Generative Media Hackathon"), "deadline_display": deadline_display, "deadline_utc": deadline_dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") if deadline_dt else None, "deadline_et": deadline_display, "deadline_beijing": ( deadline_dt.astimezone(ZoneInfo("Asia/Shanghai")).strftime("%Y-%m-%d %H:%M Asia/Shanghai") if deadline_dt else None ), "participant_count_observed": participant_count, "prize_total_usd": prize_total, "sponsor_tech": ["Backblaze B2", "Genblaze"], } def parse_rules(text: str) -> dict[str, Any]: rules_participants = parse_int( first_match(r"Participants\s*\(([\d,]+)\)", text) or first_match(r"([\d,]+)\s+participants", text) ) requirements = { requirement_id: marker.lower() in text.lower() for requirement_id, marker in REQUIRED_RULE_MARKERS.items() } criteria = [ {"name": criterion["name"], "present": criterion["marker"].lower() in text.lower()} for criterion in JUDGING_CRITERIA ] return { "rules_participant_count_observed": rules_participants, "registration_period": parse_event_window("Registration and Submission Period", text), "judging_period": parse_event_window("Judging Period", text), "winner_announced": parse_event_window("Winners Announced", text) or parse_event_window("Winner Announced", text), "requirements": requirements, "judging_criteria": criteria, "feedback_prize_present": "Feedback Prize".lower() in text.lower(), } def source_record(source_id: str, url: str, result: FetchResult) -> dict[str, Any]: return { "id": source_id, "url": url, "ok": bool(result.get("ok")), "status": result.get("status"), "bytes": len(str(result.get("body") or "").encode("utf-8")), "error": result.get("error"), "transport": result.get("transport"), } def build_snapshot_from_text( overview_html: str, rules_html: str, *, fetched_at: str | None = None, overview_status: int | None = 200, rules_status: int | None = 200, ) -> dict[str, Any]: now = utc_now() fetched_at = fetched_at or now.isoformat().replace("+00:00", "Z") overview_text = text_from_html(overview_html) rules_text = text_from_html(rules_html) overview = parse_overview(overview_text) rules = parse_rules(rules_text) participant_note = "Dynamic Devpost count; recheck before final public claims." if ( overview.get("participant_count_observed") and rules.get("rules_participant_count_observed") and overview["participant_count_observed"] != rules["rules_participant_count_observed"] ): participant_note = "Overview and rules pages showed different dynamic participant counts; use overview count for the snapshot." event = { **overview, "devpost_url": OVERVIEW_URL, "rules_url": RULES_URL, "participant_count_checked_at": beijing_date(now), "participant_count_note": participant_note, "registration_period": rules.get("registration_period"), "judging_period": rules.get("judging_period"), "winner_announced": rules.get("winner_announced"), } snapshot = { "schema": SCHEMA, "checked_at": fetched_at, "mode": "live_official_snapshot", "event": event, "rules": { "requirements": rules["requirements"], "judging_criteria": rules["judging_criteria"], "feedback_prize_present": rules["feedback_prize_present"], "rules_participant_count_observed": rules["rules_participant_count_observed"], }, "sources": [ { "id": "overview", "url": OVERVIEW_URL, "ok": overview_status is not None and 200 <= overview_status < 400, "status": overview_status, "bytes": len(overview_html.encode("utf-8")), "error": None, }, { "id": "rules", "url": RULES_URL, "ok": rules_status is not None and 200 <= rules_status < 400, "status": rules_status, "bytes": len(rules_html.encode("utf-8")), "error": None, }, ], } snapshot["validation"] = validate_snapshot(snapshot) return snapshot def build_live_snapshot(fetcher: Fetcher = fetch_text) -> dict[str, Any]: overview = fetcher(OVERVIEW_URL, TIMEOUT_SECONDS) rules = fetcher(RULES_URL, TIMEOUT_SECONDS) snapshot = build_snapshot_from_text( str(overview.get("body") or ""), str(rules.get("body") or ""), fetched_at=format_utc_now(), overview_status=overview.get("status"), rules_status=rules.get("status"), ) snapshot["sources"] = [ source_record("overview", OVERVIEW_URL, overview), source_record("rules", RULES_URL, rules), ] snapshot["validation"] = validate_snapshot(snapshot) return snapshot def load_json(path: Path) -> dict[str, Any] | None: try: return json.loads(path.read_text(encoding="utf-8")) except (FileNotFoundError, json.JSONDecodeError): return None def checked_at_age_days(snapshot: dict[str, Any], now: datetime | None = None) -> int | None: checked_at = snapshot.get("checked_at") if not checked_at: return None try: checked = datetime.fromisoformat(str(checked_at).replace("Z", "+00:00")) except ValueError: return None now = now or utc_now() return max(0, (now - checked.astimezone(timezone.utc)).days) def submission_is_open(snapshot: dict[str, Any], now: datetime | None = None) -> bool: deadline = snapshot.get("event", {}).get("deadline_utc") if not deadline: return False try: deadline_dt = datetime.fromisoformat(str(deadline).replace("Z", "+00:00")) except ValueError: return False now = now or utc_now() return now < deadline_dt.astimezone(timezone.utc) def validate_snapshot( snapshot: dict[str, Any] | None, *, max_age_days: int = 14, now: datetime | None = None, ) -> dict[str, Any]: findings: list[str] = [] if not snapshot: return {"ok": False, "findings": ["Snapshot JSON is missing or invalid."]} if snapshot.get("schema") != SCHEMA: findings.append(f"Expected schema {SCHEMA}, got {snapshot.get('schema')!r}.") sources = snapshot.get("sources", []) if len(sources) < 2 or not all(source.get("ok") for source in sources): findings.append("Official overview and rules sources must both fetch successfully.") event = snapshot.get("event", {}) rules = snapshot.get("rules", {}) if event.get("participant_count_observed") is None: findings.append("Participant count is missing.") if event.get("deadline_utc") is None: findings.append("Deadline is missing.") if not submission_is_open(snapshot, now=now): findings.append("Submission deadline appears closed or could not be parsed.") missing_requirements = [ requirement_id for requirement_id, present in rules.get("requirements", {}).items() if present is not True ] if missing_requirements: findings.append(f"Submission requirement markers missing: {', '.join(missing_requirements)}.") missing_criteria = [ criterion.get("name") for criterion in rules.get("judging_criteria", []) if criterion.get("present") is not True ] if missing_criteria: findings.append(f"Judging criteria markers missing: {', '.join(str(item) for item in missing_criteria)}.") age_days = checked_at_age_days(snapshot, now=now) if age_days is None: findings.append("checked_at is missing or invalid.") elif age_days > max_age_days: findings.append(f"Snapshot is {age_days} days old, above {max_age_days} day limit.") return { "ok": not findings, "findings": findings, "age_days": age_days, "max_age_days": max_age_days, "submission_open": submission_is_open(snapshot, now=now), } def render_markdown(snapshot: dict[str, Any]) -> str: event = snapshot["event"] validation = snapshot.get("validation", {}) lines = [ "# ProofFrame Devpost Event Snapshot", "", f"Mode: `{snapshot['mode']}`", f"Checked: `{snapshot['checked_at']}`", f"Validation ok: `{str(validation.get('ok')).lower()}`", f"Submission open: `{str(validation.get('submission_open')).lower()}`", "", "## Official Event Facts", "", f"- Event: {event.get('name')}", f"- Devpost: {event.get('devpost_url')}", f"- Rules: {event.get('rules_url')}", f"- Deadline: {event.get('deadline_et')} / {event.get('deadline_beijing')}", f"- Prize total: `${event.get('prize_total_usd')}`", f"- Participants observed: `{event.get('participant_count_observed')}` checked `{event.get('participant_count_checked_at')}`", f"- Participant note: {event.get('participant_count_note')}", f"- Registration window: {event.get('registration_period')}", f"- Judging window: {event.get('judging_period')}", f"- Winner announced: {event.get('winner_announced')}", "", "## Submission Requirements", "", ] for requirement_id, present in snapshot["rules"]["requirements"].items(): marker = "OK" if present else "MISSING" lines.append(f"- {marker} `{requirement_id}`") lines.extend(["", "## Judging Criteria", ""]) for criterion in snapshot["rules"]["judging_criteria"]: marker = "OK" if criterion["present"] else "MISSING" lines.append(f"- {marker} {criterion['name']}") lines.extend(["", "## Sources", ""]) for source in snapshot["sources"]: marker = "OK" if source["ok"] else "FAILED" lines.append(f"- {marker} `{source['id']}` {source['url']} status `{source['status']}`") lines.extend(["", "## Validation Findings", ""]) if validation.get("findings"): lines.extend(f"- {finding}" for finding in validation["findings"]) else: lines.append("- None.") return "\n".join(lines) + "\n" def write_outputs(snapshot: dict[str, Any], json_path: Path, markdown_path: Path) -> None: json_path.parent.mkdir(parents=True, exist_ok=True) markdown_path.parent.mkdir(parents=True, exist_ok=True) json_path.write_text(json.dumps(snapshot, indent=2) + "\n", encoding="utf-8") markdown_path.write_text(render_markdown(snapshot), encoding="utf-8") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Refresh or validate the official Devpost event snapshot.") parser.add_argument("--json-out", type=Path, default=DEFAULT_JSON) parser.add_argument("--markdown-out", type=Path, default=DEFAULT_MD) parser.add_argument("--fetch-live", action="store_true", help="Fetch official Devpost overview and rules pages.") parser.add_argument("--validate-committed", action="store_true", help="Validate the committed JSON without network.") parser.add_argument("--max-age-days", type=int, default=14) return parser def main() -> None: args = build_parser().parse_args() if args.fetch_live: snapshot = build_live_snapshot() snapshot["validation"] = validate_snapshot(snapshot, max_age_days=args.max_age_days) write_outputs(snapshot, args.json_out, args.markdown_out) validation = snapshot["validation"] else: snapshot = load_json(args.json_out) validation = validate_snapshot(snapshot, max_age_days=args.max_age_days) if snapshot and not args.validate_committed: snapshot["validation"] = validation write_outputs(snapshot, args.json_out, args.markdown_out) print( json.dumps( { "ok": validation["ok"], "mode": snapshot.get("mode") if snapshot else None, "json": str(args.json_out), "markdown": str(args.markdown_out), "submission_open": validation.get("submission_open"), "age_days": validation.get("age_days"), "findings": validation["findings"], }, indent=2, ) ) raise SystemExit(0 if validation["ok"] else 2) if __name__ == "__main__": main()