drift1ng/openclaw

1import os2import time3from typing import Any, Dict, List, Optional4 5import gradio as gr6import requests7from requests import exceptions as req_exc8 9 10GITHUB_API = "https://api.github.com"11DEFAULT_WAIT_SECONDS = 6012POLL_INTERVAL_SECONDS = 313 14 15class CodespaceWaker:16    def __init__(self, token: str) -> None:17        self.session = requests.Session()18        self.session.headers.update(19            {20                "Accept": "application/vnd.github+json",21                "Authorization": f"Bearer {token}",22                "X-GitHub-Api-Version": "2022-11-28",23                "User-Agent": "codespace-waker",24            }25        )26 27    def _request(28        self, method: str, path: str, expected: List[int], json: Optional[Dict[str, Any]] = None29    ) -> Dict[str, Any]:30        # Use short connect/read timeouts so UI does not appear stuck.31        response = self.session.request(32            method,33            f"{GITHUB_API}{path}",34            json=json,35            timeout=(6, 12),36        )37        if response.status_code not in expected:38            message = response.text39            raise RuntimeError(f"GitHub API error {response.status_code}: {message}")40        if response.content:41            return response.json()42        return {}43 44    def list_codespaces(self) -> List[Dict[str, Any]]:45        data = self._request("GET", "/user/codespaces?per_page=100", [200])46        return data.get("codespaces", [])47 48    def get_codespace(self, name: str) -> Dict[str, Any]:49        return self._request("GET", f"/user/codespaces/{name}", [200])50 51    def start_codespace(self, name: str) -> None:52        self._request("POST", f"/user/codespaces/{name}/start", [202, 200])53 54 55def pick_codespace(56    all_codespaces: List[Dict[str, Any]], owner: str, repo: str57) -> Dict[str, Any]:58    full_name = f"{owner}/{repo}"59    repo_matches = [c for c in all_codespaces if c.get("repository", {}).get("full_name") == full_name]60    if not repo_matches:61        visible = ", ".join(62            f"{c.get('name')} ({c.get('repository', {}).get('full_name', 'unknown')})"63            for c in all_codespaces[:5]64        )65        if not visible:66            raise RuntimeError(f"No codespaces are visible for this token. Expected repository: {full_name}.")67        raise RuntimeError(68            f"No codespaces found for repository {full_name}. Visible examples: {visible}"69        )70 71    repo_matches.sort(key=lambda x: x.get("last_used_at", ""), reverse=True)72    return repo_matches[0]73 74 75def wake_codespace(owner: str, repo: str, codespace_name: str, wait_seconds: int) -> str:76    token = os.getenv("GITHUB_TOKEN", "").strip()77    if not token:78        return "Error: missing GITHUB_TOKEN environment variable."79 80    owner = owner.strip()81    repo = repo.strip()82    codespace_name = codespace_name.strip()83 84    if not codespace_name and (not owner or not repo):85        return "Error: owner and repo are required."86 87    if wait_seconds <= 0:88        wait_seconds = DEFAULT_WAIT_SECONDS89 90    client = CodespaceWaker(token)91    try:92        # Quick auth sanity check for clearer failure messages.93        user = client._request("GET", "/user", [200])94        login = user.get("login", "unknown")95        if codespace_name:96            try:97                target = client.get_codespace(codespace_name)98            except RuntimeError as exc:99                if " 404" in str(exc) or "404:" in str(exc):100                    return (101                        f"Error: Codespace '{codespace_name}' not found for token user '{login}'. "102                        "Check CODESPACE_NAME and ensure GITHUB_TOKEN belongs to the same GitHub account."103                    )104                raise105        else:106            all_codespaces = client.list_codespaces()107            target = pick_codespace(all_codespaces, owner, repo)108        name = target["name"]109 110        state = target.get("state", "Unknown")111        if state != "Available":112            client.start_codespace(name)113 114        deadline = time.time() + wait_seconds115        while time.time() < deadline:116            details = client.get_codespace(name)117            if details.get("state") == "Available":118                web_url = details.get("web_url", "")119                last_used = details.get("last_used_at", "unknown")120                return (121                    f"Codespace is ready.\n\n"122                    f"Name: {name}\n"123                    f"State: Available\n"124                    f"Last used: {last_used}\n"125                    f"Open: {web_url}"126                )127            time.sleep(POLL_INTERVAL_SECONDS)128 129        return (130            f"Start request sent for '{name}', but it did not become Available "131            f"within {wait_seconds} seconds. Try again."132        )133    except req_exc.Timeout:134        return "Error: request to GitHub timed out. Check network and try again."135    except req_exc.RequestException as exc:136        return f"Error: network request failed: {exc}"137    except Exception as exc:138        return f"Error: {exc}"139 140 141def default_value(env_name: str) -> str:142    return os.getenv(env_name, "").strip()143 144 145demo = gr.Interface(146    fn=wake_codespace,147    inputs=[148        gr.Textbox(149            label="GitHub Owner",150            value=default_value("GITHUB_OWNER"),151            placeholder="e.g. octocat",152        ),153        gr.Textbox(154            label="Repository Name",155            value=default_value("GITHUB_REPO"),156            placeholder="e.g. my-repo",157        ),158        gr.Textbox(159            label="Codespace Name (optional)",160            value=default_value("CODESPACE_NAME"),161            placeholder="leave empty to auto-pick most recently used",162        ),163        gr.Number(164            label="Wait Timeout (seconds)",165            value=int(os.getenv("WAIT_SECONDS", str(DEFAULT_WAIT_SECONDS))),166            precision=0,167        ),168    ],169    outputs=gr.Textbox(label="Result", lines=10),170    title="GitHub Codespace Waker",171    description=(172        "Start your Codespace on demand using GitHub API, then return the web URL. "173        "This does not bypass idle timeout; it automates wake-up."174    ),175)176 177 178if __name__ == "__main__":179    demo.launch(server_name="0.0.0.0", server_port=7860)180