Spaces:
Sleeping
Sleeping
| import os | |
| import base64 | |
| import json | |
| import tempfile | |
| import subprocess | |
| import time | |
| import uuid | |
| from http import HTTPStatus | |
| from pathlib import Path | |
| from fastapi import FastAPI, Request, Response, BackgroundTasks | |
| import requests | |
| GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN") | |
| API_SECRET = os.environ.get("API_SECRET") | |
| GH_API = "https://api.github.com" | |
| app = FastAPI() | |
| def run(cmd, cwd=None): | |
| p = subprocess.run(cmd, shell=True, cwd=cwd, capture_output=True, text=True) | |
| if p.returncode != 0: | |
| raise RuntimeError(p.stderr) | |
| return p.stdout.strip() | |
| def create_minimal_site(tmpdir, task, attachments): | |
| p = Path(tmpdir) | |
| p.mkdir(parents=True, exist_ok=True) | |
| sample = None | |
| for a in attachments: | |
| name = a.get("name") | |
| url = a.get("url") | |
| if not name or not url: | |
| continue | |
| if url.startswith("data:"): | |
| header, b64 = url.split(",", 1) | |
| data = base64.b64decode(b64) | |
| (p / name).write_bytes(data) | |
| sample = name | |
| index = p / "index.html" | |
| content = f"""<!doctype html> | |
| <html><head><meta charset="utf-8"><title>{task}</title></head><body> | |
| <h1 id="title">{task}</h1> | |
| <div id="demo"></div> | |
| <script> | |
| const params = new URLSearchParams(location.search); | |
| const url = params.get('url') || '{sample or ""}'; | |
| if (url) {{ | |
| document.getElementById('demo').innerHTML = `<img id="img" src="${{url}}" alt="captcha image">`; | |
| }} | |
| setTimeout(() => {{ | |
| const s = document.createElement('div'); | |
| s.id = 'solved'; | |
| s.textContent = 'SAMPLE_SOLUTION'; | |
| document.body.appendChild(s); | |
| }}, 1000); | |
| </script> | |
| </body></html>""" | |
| index.write_text(content) | |
| (p / "README.md").write_text(f"# {task}\n\nAuto-generated site.\n") | |
| (p / "LICENSE").write_text("MIT License\n") | |
| (p / ".nojekyll").write_text("") | |
| return tmpdir | |
| def create_github_repo(repo_name, tmpdir): | |
| headers = {"Authorization": f"token {GITHUB_TOKEN}", "Accept": "application/vnd.github+json"} | |
| data = {"name": repo_name, "private": False, "auto_init": False} | |
| r = requests.post(f"{GH_API}/user/repos", json=data, headers=headers) | |
| if r.status_code not in (200, 201): | |
| raise RuntimeError(f"create repo failed: {r.status_code} {r.text}") | |
| repo = r.json() | |
| clone_url = repo["clone_url"] | |
| run("git init", cwd=tmpdir) | |
| run("git add .", cwd=tmpdir) | |
| run('git -c user.name="auto" -c user.email="auto@example.com" commit -m "initial"', cwd=tmpdir) | |
| run(f"git remote add origin {clone_url}", cwd=tmpdir) | |
| run("git branch -M main", cwd=tmpdir) | |
| run("git push -u origin main", cwd=tmpdir) | |
| return repo["html_url"] | |
| def enable_pages(owner, repo): | |
| headers = {"Authorization": f"token {GITHUB_TOKEN}", "Accept": "application/vnd.github+json"} | |
| data = {"source": {"branch": "main", "path": "/"}} | |
| requests.post(f"{GH_API}/repos/{owner}/{repo}/pages", json=data, headers=headers) | |
| time.sleep(2) | |
| pages = requests.get(f"{GH_API}/repos/{owner}/{repo}", headers=headers).json().get("html_url") | |
| if pages: | |
| return f"https://{owner}.github.io/{repo}/" | |
| return None | |
| def post_evaluation(evaluation_url, payload): | |
| delay = 1 | |
| for _ in range(8): | |
| r = requests.post(evaluation_url, json=payload, headers={"Content-Type": "application/json"}) | |
| if r.status_code == 200: | |
| return True | |
| time.sleep(delay) | |
| delay *= 2 | |
| return False | |
| def process_task(body): | |
| try: | |
| email = body.get("email") | |
| task = body.get("task") or f"task-{uuid.uuid4().hex[:6]}" | |
| nonce = body.get("nonce") | |
| round_index = body.get("round", 1) | |
| attachments = body.get("attachments", []) | |
| tmpdir = tempfile.mkdtemp(prefix="genrepo_") | |
| create_minimal_site(tmpdir, task, attachments) | |
| repo_name = f"{task}-{uuid.uuid4().hex[:5]}" | |
| repo_url = create_github_repo(repo_name, tmpdir) | |
| owner = requests.get(f"{GH_API}/user", headers={"Authorization": f"token {GITHUB_TOKEN}"}).json()["login"] | |
| pages_url = enable_pages(owner, repo_name) | |
| commit_sha = run("git rev-parse HEAD", cwd=tmpdir) | |
| payload = { | |
| "email": email, | |
| "task": task, | |
| "round": round_index, | |
| "nonce": nonce, | |
| "repo_url": repo_url, | |
| "commit_sha": commit_sha, | |
| "pages_url": pages_url | |
| } | |
| eval_url = body.get("evaluation_url") | |
| if eval_url: | |
| post_evaluation(eval_url, payload) | |
| except Exception as e: | |
| print("Error in process_task:", e) | |
| async def task_endpoint(req: Request, background_tasks: BackgroundTasks): | |
| body = await req.json() | |
| secret = body.get("secret") | |
| if secret != API_SECRET: | |
| return Response(status_code=HTTPStatus.UNAUTHORIZED.value, content=json.dumps({"error": "invalid secret"})) | |
| ack = {"status": "ok"} | |
| background_tasks.add_task(process_task, body) | |
| return ack | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 7860))) | |