| import json |
| import base64 |
| import urllib.parse |
| import httpx |
| from database import get_token_from_supabase |
|
|
| async def github_get_headers(user_id: str): |
| token_data = await get_token_from_supabase(user_id, "github") |
| if not token_data: |
| return None |
| access_token = token_data.get("access_token") or token_data.get("token") |
| if not access_token: |
| return None |
| return { |
| "Authorization": f"Bearer {access_token}", |
| "Accept": "application/vnd.github+json", |
| "X-GitHub-Api-Version": "2022-11-28", |
| "User-Agent": "Clade-AI-Assistant" |
| } |
|
|
| async def github_list_repos(user_id: str, sort: str = "updated", per_page: int = 10): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected. Please connect GitHub account." |
| try: |
| url = f"https://api.github.com/user/repos?sort={sort}&per_page={int(per_page)}" |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.get(url, headers=headers) |
| if res.status_code != 200: |
| return f"Error fetching GitHub repositories: {res.text}" |
| repos = res.json() |
| if not repos: |
| return "No repositories found for this account." |
| out = [] |
| for r in repos: |
| out.append(f"Repo: {r.get('full_name')} | Private: {r.get('private')} | Default Branch: {r.get('default_branch')} | Stars: {r.get('stargazers_count')} | URL: {r.get('html_url')}") |
| return "\n".join(out) |
| except Exception as e: |
| return f"Error fetching GitHub repos: {str(e)}" |
|
|
| async def github_get_repo(user_id: str, owner: str, repo: str): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}" |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.get(url, headers=headers) |
| if res.status_code != 200: |
| return f"Error fetching repository details: {res.text}" |
| r = res.json() |
| return f"Repo: {r.get('full_name')}\nDescription: {r.get('description')}\nDefault Branch: {r.get('default_branch')}\nStars: {r.get('stargazers_count')}\nForks: {r.get('forks_count')}\nOpen Issues: {r.get('open_issues_count')}\nURL: {r.get('html_url')}" |
| except Exception as e: |
| return f"Error: {str(e)}" |
|
|
| async def github_create_repo(user_id: str, name: str, description: str = "", private: bool = False, auto_init: bool = True): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = "https://api.github.com/user/repos" |
| payload = { |
| "name": name.strip(), |
| "description": description, |
| "private": bool(private), |
| "auto_init": bool(auto_init) |
| } |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.post(url, headers=headers, json=payload) |
| if res.status_code not in [200, 201]: |
| return f"Error creating repository: {res.text}" |
| r = res.json() |
| return f"Successfully created repository '{r.get('full_name')}'. URL: {r.get('html_url')}" |
| except Exception as e: |
| return f"Error creating repository: {str(e)}" |
|
|
| async def github_delete_repo(user_id: str, owner: str, repo: str): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}" |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.delete(url, headers=headers) |
| if res.status_code == 204: |
| return f"Successfully deleted repository {owner}/{repo}." |
| return f"Error deleting repository: {res.text}" |
| except Exception as e: |
| return f"Error deleting repository: {str(e)}" |
|
|
| async def github_get_file_contents(user_id: str, owner: str, repo: str, path: str, ref: str = None): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| clean_path = path.lstrip("/") |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/contents/{clean_path}" |
| if ref: |
| url += f"?ref={ref.strip()}" |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.get(url, headers=headers) |
| if res.status_code != 200: |
| return f"Error reading content: {res.text}" |
| data = res.json() |
| if isinstance(data, list): |
| items = [f"[{item.get('type')}] {item.get('name')} (path: {item.get('path')}, sha: {item.get('sha')})" for item in data] |
| return "\n".join(items) |
| elif isinstance(data, dict): |
| content_b64 = data.get("content", "") |
| encoding = data.get("encoding", "") |
| sha = data.get("sha", "") |
| if encoding == "base64" and content_b64: |
| raw_text = base64.b64decode(content_b64).decode("utf-8", errors="ignore") |
| return f"SHA: {sha}\n\n{raw_text}" |
| return f"SHA: {sha} | Name: {data.get('name')} | Download URL: {data.get('download_url')}" |
| except Exception as e: |
| return f"Error getting file contents: {str(e)}" |
|
|
| async def github_create_or_update_file(user_id: str, owner: str, repo: str, path: str, message: str, content: str, branch: str = None, sha: str = None): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| clean_path = path.lstrip("/") |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/contents/{clean_path}" |
| content_b64 = base64.b64encode(content.encode("utf-8")).decode("utf-8") |
| payload = { |
| "message": message, |
| "content": content_b64 |
| } |
| if branch: |
| payload["branch"] = branch.strip() |
| if sha: |
| payload["sha"] = sha.strip() |
| else: |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| chk = await client.get(url, headers=headers) |
| if chk.status_code == 200: |
| payload["sha"] = chk.json().get("sha") |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.put(url, headers=headers, json=payload) |
| if res.status_code not in [200, 201]: |
| return f"Error committing file to GitHub: {res.text}" |
| data = res.json() |
| return f"Successfully saved '{clean_path}' in {owner}/{repo}. Commit SHA: {data.get('commit', {}).get('sha')}" |
| except Exception as e: |
| return f"Error writing file: {str(e)}" |
|
|
| async def github_delete_file(user_id: str, owner: str, repo: str, path: str, message: str, sha: str, branch: str = None): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| clean_path = path.lstrip("/") |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/contents/{clean_path}" |
| payload = { |
| "message": message, |
| "sha": sha.strip() |
| } |
| if branch: |
| payload["branch"] = branch.strip() |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.request("DELETE", url, headers=headers, json=payload) |
| if res.status_code != 200: |
| return f"Error deleting file: {res.text}" |
| return f"Successfully deleted '{clean_path}' from {owner}/{repo}." |
| except Exception as e: |
| return f"Error deleting file: {str(e)}" |
|
|
| async def github_list_branches(user_id: str, owner: str, repo: str): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/branches" |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.get(url, headers=headers) |
| if res.status_code != 200: |
| return f"Error listing branches: {res.text}" |
| branches = res.json() |
| return "\n".join([f"Branch: {b.get('name')} | Commit: {b.get('commit', {}).get('sha')}" for b in branches]) |
| except Exception as e: |
| return f"Error listing branches: {str(e)}" |
|
|
| async def github_create_branch(user_id: str, owner: str, repo: str, branch: str, from_branch: str = None): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| target_source = from_branch.strip() if from_branch else None |
| if not target_source: |
| r_info = await client.get(f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}", headers=headers) |
| target_source = r_info.json().get("default_branch", "main") |
| ref_res = await client.get(f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/git/ref/heads/{target_source}", headers=headers) |
| if ref_res.status_code != 200: |
| return f"Error resolving source branch '{target_source}': {ref_res.text}" |
| sha = ref_res.json().get("object", {}).get("sha") |
| payload = { |
| "ref": f"refs/heads/{branch.strip()}", |
| "sha": sha |
| } |
| create_res = await client.post(f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/git/refs", headers=headers, json=payload) |
| if create_res.status_code not in [200, 201]: |
| return f"Error creating branch: {create_res.text}" |
| return f"Successfully created branch '{branch}' from '{target_source}' (SHA: {sha})." |
| except Exception as e: |
| return f"Error creating branch: {str(e)}" |
|
|
| async def github_list_commits(user_id: str, owner: str, repo: str, branch: str = None, per_page: int = 5): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/commits?per_page={int(per_page)}" |
| if branch: |
| url += f"&sha={branch.strip()}" |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.get(url, headers=headers) |
| if res.status_code != 200: |
| return f"Error listing commits: {res.text}" |
| commits = res.json() |
| out = [] |
| for c in commits: |
| commit_info = c.get("commit", {}) |
| author = commit_info.get("author", {}).get("name") |
| msg = commit_info.get("message", "").split("\n")[0] |
| out.append(f"Commit: {c.get('sha')[:7]} | Author: {author} | Message: {msg}") |
| return "\n".join(out) |
| except Exception as e: |
| return f"Error listing commits: {str(e)}" |
|
|
| async def github_list_pull_requests(user_id: str, owner: str, repo: str, state: str = "open"): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/pulls?state={state.strip()}" |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.get(url, headers=headers) |
| if res.status_code != 200: |
| return f"Error listing pull requests: {res.text}" |
| prs = res.json() |
| if not prs: |
| return f"No {state} pull requests found in {owner}/{repo}." |
| out = [] |
| for pr in prs: |
| out.append(f"PR #{pr.get('number')}: {pr.get('title')} | State: {pr.get('state')} | Head: {pr.get('head', {}).get('ref')} -> Base: {pr.get('base', {}).get('ref')} | URL: {pr.get('html_url')}") |
| return "\n".join(out) |
| except Exception as e: |
| return f"Error listing PRs: {str(e)}" |
|
|
| async def github_create_pull_request(user_id: str, owner: str, repo: str, title: str, head: str, base: str = "main", body: str = ""): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/pulls" |
| payload = { |
| "title": title.strip(), |
| "head": head.strip(), |
| "base": base.strip(), |
| "body": body |
| } |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.post(url, headers=headers, json=payload) |
| if res.status_code not in [200, 201]: |
| return f"Error creating PR: {res.text}" |
| pr = res.json() |
| return f"Successfully created PR #{pr.get('number')}: '{pr.get('title')}'. URL: {pr.get('html_url')}" |
| except Exception as e: |
| return f"Error creating PR: {str(e)}" |
|
|
| async def github_merge_pull_request(user_id: str, owner: str, repo: str, pull_number: int, commit_title: str = None, merge_method: str = "merge"): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/pulls/{int(pull_number)}/merge" |
| payload = {"merge_method": merge_method.strip()} |
| if commit_title: |
| payload["commit_title"] = commit_title.strip() |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.put(url, headers=headers, json=payload) |
| if res.status_code != 200: |
| return f"Error merging PR: {res.text}" |
| data = res.json() |
| return f"Successfully merged PR #{pull_number}. Message: {data.get('message')}" |
| except Exception as e: |
| return f"Error merging PR: {str(e)}" |
|
|
| async def github_list_issues(user_id: str, owner: str, repo: str, state: str = "open"): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/issues?state={state.strip()}" |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.get(url, headers=headers) |
| if res.status_code != 200: |
| return f"Error listing issues: {res.text}" |
| issues = res.json() |
| if not issues: |
| return f"No {state} issues found in {owner}/{repo}." |
| out = [] |
| for issue in issues: |
| if "pull_request" not in issue: |
| out.append(f"Issue #{issue.get('number')}: {issue.get('title')} | State: {issue.get('state')} | Comments: {issue.get('comments')} | URL: {issue.get('html_url')}") |
| return "\n".join(out) if out else "No regular issues found." |
| except Exception as e: |
| return f"Error listing issues: {str(e)}" |
|
|
| async def github_create_issue(user_id: str, owner: str, repo: str, title: str, body: str = ""): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/issues" |
| payload = {"title": title.strip(), "body": body} |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.post(url, headers=headers, json=payload) |
| if res.status_code not in [200, 201]: |
| return f"Error creating issue: {res.text}" |
| issue = res.json() |
| return f"Successfully created Issue #{issue.get('number')}: '{issue.get('title')}'. URL: {issue.get('html_url')}" |
| except Exception as e: |
| return f"Error creating issue: {str(e)}" |
|
|
| async def github_add_issue_comment(user_id: str, owner: str, repo: str, issue_number: int, body: str): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| url = f"https://api.github.com/repos/{owner.strip()}/{repo.strip()}/issues/{int(issue_number)}/comments" |
| payload = {"body": body} |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.post(url, headers=headers, json=payload) |
| if res.status_code not in [200, 201]: |
| return f"Error adding comment: {res.text}" |
| return f"Successfully posted comment on Issue/PR #{issue_number}." |
| except Exception as e: |
| return f"Error adding comment: {str(e)}" |
|
|
| async def github_search(user_id: str, type: str, query: str): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected." |
| try: |
| valid_types = ["repositories", "code", "issues", "commits", "users"] |
| t = type.lower().strip() |
| if t not in valid_types: |
| t = "repositories" |
| url = f"https://api.github.com/search/{t}?q={urllib.parse.quote(query.strip())}" |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.get(url, headers=headers) |
| if res.status_code != 200: |
| return f"Error searching GitHub: {res.text}" |
| data = res.json() |
| items = data.get("items", []) |
| if not items: |
| return f"No results found for query '{query}' in {t}." |
| out = [] |
| for item in items[:10]: |
| if t == "repositories": |
| out.append(f"Repo: {item.get('full_name')} | Description: {item.get('description')} | Stars: {item.get('stargazers_count')} | URL: {item.get('html_url')}") |
| elif t == "code": |
| out.append(f"File: {item.get('name')} | Path: {item.get('path')} | Repo: {item.get('repository', {}).get('full_name')} | URL: {item.get('html_url')}") |
| elif t == "issues": |
| out.append(f"#{item.get('number')}: {item.get('title')} | State: {item.get('state')} | Repo URL: {item.get('html_url')}") |
| elif t == "users": |
| out.append(f"User: {item.get('login')} | Type: {item.get('type')} | URL: {item.get('html_url')}") |
| else: |
| out.append(f"Item: {item.get('sha', '') or item.get('id', '')} | URL: {item.get('html_url', '')}") |
| return "\n".join(out) |
| except Exception as e: |
| return f"Error searching GitHub: {str(e)}" |
|
|
| async def github_api_request(user_id: str, endpoint: str, method: str = "GET", body_json: str = None): |
| headers = await github_get_headers(user_id) |
| if not headers: |
| return "GitHub is not connected. Please connect GitHub account." |
| try: |
| clean_endpoint = "/" + endpoint.lstrip("/") |
| url = f"https://api.github.com{clean_endpoint}" |
| req_method = method.upper().strip() |
| payload = None |
| if body_json and req_method in ["POST", "PATCH", "PUT"]: |
| try: |
| payload = json.loads(body_json) if isinstance(body_json, str) else body_json |
| except Exception: |
| payload = body_json |
| async with httpx.AsyncClient(timeout=30.0) as client: |
| res = await client.request(req_method, url, headers=headers, json=payload if isinstance(payload, (dict, list)) else None, content=payload if isinstance(payload, str) else None) |
| if res.status_code == 204: |
| return f"GitHub API Request {req_method} {clean_endpoint} completed successfully (204 No Content)." |
| try: |
| data = res.json() |
| return json.dumps(data, indent=2) |
| except Exception: |
| return res.text |
| except Exception as e: |
| return f"Error executing GitHub API Request: {str(e)}" |
|
|
