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"""
{task}
{task}
"""
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)
@app.post("/api/task")
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