github_pr.py

"""Open a GitHub PR against the website repo for an approved submission.

The site owner still reviews and merges the PR by hand — this only removes
the manual data-entry step of splicing a new object into the right
src/data/*.js array.
"""

import base64
import os

import requests

import formatters
from about import GITHUB_API, GITHUB_REPO


def _headers() -> dict:
    token = os.environ["GITHUB_TOKEN"]
    return {
        "Authorization": f"Bearer {token}",
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }


def _default_branch() -> str:
    resp = requests.get(f"{GITHUB_API}/repos/{GITHUB_REPO}", headers=_headers())
    resp.raise_for_status()
    return resp.json()["default_branch"]


def _get_file(path: str, ref: str) -> tuple[str, str]:
    """Return (content, sha) of a file at ref."""
    resp = requests.get(
        f"{GITHUB_API}/repos/{GITHUB_REPO}/contents/{path}",
        headers=_headers(),
        params={"ref": ref},
    )
    resp.raise_for_status()
    data = resp.json()
    content = base64.b64decode(data["content"]).decode("utf-8")
    return content, data["sha"]


def _create_branch(branch: str, from_sha: str) -> None:
    resp = requests.post(
        f"{GITHUB_API}/repos/{GITHUB_REPO}/git/refs",
        headers=_headers(),
        json={"ref": f"refs/heads/{branch}", "sha": from_sha},
    )
    resp.raise_for_status()


def _branch_sha(branch: str) -> str:
    resp = requests.get(
        f"{GITHUB_API}/repos/{GITHUB_REPO}/git/ref/heads/{branch}", headers=_headers()
    )
    resp.raise_for_status()
    return resp.json()["object"]["sha"]


def _update_file(path: str, branch: str, new_content: str, sha: str, message: str) -> None:
    resp = requests.put(
        f"{GITHUB_API}/repos/{GITHUB_REPO}/contents/{path}",
        headers=_headers(),
        json={
            "message": message,
            "content": base64.b64encode(new_content.encode("utf-8")).decode("ascii"),
            "sha": sha,
            "branch": branch,
        },
    )
    resp.raise_for_status()


def _open_pr(branch: str, base: str, title: str, body: str) -> str:
    resp = requests.post(
        f"{GITHUB_API}/repos/{GITHUB_REPO}/pulls",
        headers=_headers(),
        json={"title": title, "head": branch, "base": base, "body": body},
    )
    resp.raise_for_status()
    return resp.json()["html_url"]


def open_pr_for_submission(submission: dict, fields: dict) -> str:
    """Splice the approved entry into the right data file and open a PR.

    `submission` is the row from submissions.py; `fields` are the
    reviewer-filled values matching formatters.TARGETS[type_]'s schema.
    """
    type_ = submission["type"]
    file_path = formatters.target_file(type_)
    entry_block = formatters.render_entry(type_, fields)
    org_id = fields.get("orgId")

    base = _default_branch()
    base_sha = _branch_sha(base)

    slug = fields.get("id", submission["id"][:8])
    branch = f"add/{type_}-{slug}"
    _create_branch(branch, base_sha)

    content, sha = _get_file(file_path, branch)
    new_content = formatters.insert_entry(content, type_, entry_block, org_id)
    _update_file(
        file_path, branch, new_content, sha,
        message=f"Add {type_}: {fields.get('name') or fields.get('title') or slug}",
    )

    title = f"Add {type_}: {fields.get('name') or fields.get('title') or slug}"
    body = (
        f"Auto-generated from a huggingscience.co submission via the review Space.\n\n"
        f"- **Submission id:** `{submission['id']}`\n"
        f"- **Submitted at:** {submission['submitted_at']}\n"
        f"- **Source:** {submission['source']}\n\n"
        f"**Original description:**\n{submission['description']}\n"
    )
    return _open_pr(branch, base, title, body)