from typing import Dict, List, Optional import os from github import Github from datetime import datetime, UTC class GitHubService: def __init__(self, token: str, username: str, email: str): self.token = token self.username = username self.email = email self.gh = Github(token) self.user = self.gh.get_user() def _new_repo_name(self, task: str) -> str: base = "".join(c if c.isalnum() or c in ("-", "_") else "-" for c in task.lower()) # suffix = nonce return f"{base}"[:90] def find_repo_by_task(self, task: str) -> Optional[str]: """Try to find an existing repo URL derived from the task name. Returns HTML URL if found, else None. """ name = self._new_repo_name(task) try: repo = self.gh.get_repo(f"{self.username}/{name}") return repo.html_url except Exception: try: repo = self.user.get_repo(name) return repo.html_url except Exception: return None def _safe_create_file(self, repo, path: str, message: str, content: str, branch: str = "main") -> None: """Safely create or update a file in the repository, handling the SHA requirement.""" try: # Try to get existing file to get its SHA contents = repo.get_contents(path, ref=branch) if isinstance(contents, list): # If it's a directory, create new file repo.create_file(path, message, content, branch=branch) else: # Update existing file with its SHA repo.update_file(path, message, content, contents.sha, branch=branch) except Exception: # File doesn't exist, create new repo.create_file(path, message, content, branch=branch) def create_repo_with_files(self, task: str, files: Dict[str, str], nonce: str) -> Dict[str, str]: repo_name = self._new_repo_name(task) # Create repo repo = self.user.create_repo( repo_name, private=False, auto_init=True, has_issues=True, license_template="mit" ) # Wait a bit for the repository to be ready import time time.sleep(2) # Single-branch mode: we'll serve Pages directly from main # Set repo homepage and ensure default branch is main repo.edit( homepage=f"https://{self.username}.github.io/{repo_name}/", default_branch="main" ) # Update README with task-specific content readme_content = f"# {task}\n\nGenerated by builder.\n" self._safe_create_file(repo, "README.md", f"chore: init for {task}", readme_content) # Ensure LICENSE exists (should be there from template, but verify) try: repo.get_contents("LICENSE") except Exception: license_content = f"""MIT License Copyright (c) 2025 Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.""" self._safe_create_file(repo, "LICENSE", "chore: add MIT license", license_content) # Add generated files to main for path, content in files.items(): self._safe_create_file(repo, path, f"feat: add {path}", content) # Configure GitHub Pages to serve from main branch root via REST (with retries) try: pages_url = f"https://api.github.com/repos/{self.username}/{repo_name}/pages" headers = { "Accept": "application/vnd.github.v3+json", "Authorization": f"Bearer {self.token}" } import time import httpx max_attempts = 5 backoff = 1.0 last_err: Exception | None = None with httpx.Client(timeout=30.0) as client: for attempt in range(1, max_attempts + 1): try: response = client.post( pages_url, json={"source": {"branch": "main", "path": "/"}}, headers=headers ) # Success or already configured if response.status_code in (201, 204, 409): if response.status_code == 409: print("GitHub Pages already configured (409). Proceeding.") else: print("Configured GitHub Pages to serve from main branch root") last_err = None break # Retry on 5xx errors if 500 <= response.status_code < 600 and attempt < max_attempts: print(f"Warning: Pages config returned {response.status_code}. Retrying in {backoff:.0f}s...") time.sleep(backoff) backoff = min(backoff * 2, 16) continue # Non-retryable error print(f"Warning: Pages configuration (main) returned status {response.status_code}: {response.text}") break except httpx.RequestError as e: last_err = e if attempt < max_attempts: print(f"Warning: Pages setup attempt {attempt} failed: {e}. Retrying in {backoff:.0f}s…") time.sleep(backoff) backoff = min(backoff * 2, 16) continue else: print(f"Warning: Pages setup failed after {max_attempts} attempts: {e}") break if last_err: raise last_err except Exception as e: print(f"Warning: Pages setup encountered an error: {e}") return { "repo_url": repo.html_url, "default_branch": repo.default_branch } def _repo_full_name_from_url(self, repo_html_url: str) -> str: # https://github.com// parts = repo_html_url.rstrip("/").split("/") owner = parts[-2] name = parts[-1] return f"{owner}/{name}" def get_text_files(self, repo_full_name: str, paths: List[str]) -> Dict[str, str]: """Fetch text files from the default branch and return mapping path->text content.""" repo = self.gh.get_repo(repo_full_name) results: Dict[str, str] = {} for path in paths: try: cf = repo.get_contents(path, ref=repo.default_branch) # decoded_content is bytes results[path] = cf.decoded_content.decode("utf-8", errors="replace") except Exception as e: # Skip missing files silently; caller decides what to do print(f"get_text_files: unable to fetch {path}: {e}") return results def update_files_in_repo(self, repo_full_name: str, files: Dict[str, str], commit_message_prefix: str | None = None) -> str: """Create or update provided files on the default branch; return latest commit SHA.""" repo = self.gh.get_repo(repo_full_name) prefix = (commit_message_prefix + ": ") if commit_message_prefix else "" for path, content in files.items(): msg = f"{prefix}update {path}".strip() try: existing = repo.get_contents(path, ref=repo.default_branch) if isinstance(existing, list): repo.create_file(path, msg, content, branch=repo.default_branch) else: repo.update_file(path, msg, content, existing.sha, branch=repo.default_branch) except Exception: repo.create_file(path, msg, content, branch=repo.default_branch) # Return latest commit SHA on default branch return repo.get_branch(repo.default_branch).commit.sha def get_pages_url(self, repo_html_url: str, repo_name: str | None = None) -> str: # https://username.github.io/repo/ user = self.username if not repo_name: repo_name = repo_html_url.rstrip("/").split("/")[-1] return f"https://{user}.github.io/{repo_name}/" def get_latest_commit_sha(self, repo_full_name: str) -> str: repo = self.gh.get_repo(repo_full_name) return repo.get_branch(repo.default_branch).commit.sha