""" وحدة GitHub الكاملة باستخدام PyGithub تدعم: - إنشاء/حذف/سرد repos - رفع ملفات ومحتوى - فتح issues و PRs - قراءة محتوى repos - سرد branches و commits """ import logging from typing import List, Dict, Optional, Tuple from github import Github, GithubException, Auth from github.Repository import Repository from github.NamedUser import NamedUser from config import config logger = logging.getLogger(__name__) class GitHubTools: """أدوات GitHub الكاملة - للمالك والمشرفين فقط""" def __init__(self, token: str = ""): self.token = token or config.GITHUB_TOKEN self._client: Optional[Github] = None @property def enabled(self) -> bool: return bool(self.token) def _get_client(self) -> Github: if not self.enabled: raise RuntimeError( "GitHub غير مفعّل. أضف GITHUB_TOKEN كـ Secret في Hugging Face Space." ) if self._client is None: auth = Auth.Token(self.token) self._client = Github(auth=auth, timeout=30) return self._client def get_user_info(self) -> Dict: """معلومات المستخدم المرتبط بالتوكن""" g = self._get_client() user = g.get_user() return { "login": user.login, "name": user.name, "email": user.email, "public_repos": user.public_repos, "followers": user.followers, "bio": user.bio, } def list_repos(self, limit: int = 20) -> List[Dict]: """سرد repos المستخدم""" g = self._get_client() user = g.get_user() repos = [] all_repos = list(user.get_repos(sort="updated", direction="desc")) for repo in all_repos[:limit]: repos.append({ "name": repo.full_name, "description": repo.description or "—", "stars": repo.stargazers_count, "language": repo.language or "—", "private": repo.private, "url": repo.html_url, "updated": repo.updated_at.isoformat() if repo.updated_at else "", }) return repos def create_repo( self, name: str, description: str = "", private: bool = True, auto_init: bool = True, ) -> Dict: """إنشاء repo جديد""" g = self._get_client() user = g.get_user() repo = user.create_repo( name=name, description=description, private=private, auto_init=auto_init, ) return { "name": repo.full_name, "url": repo.html_url, "private": repo.private, "created": True, } def delete_repo(self, repo_full_name: str) -> Dict: """حذف repo""" g = self._get_client() try: repo = g.get_repo(repo_full_name) repo.delete() return {"deleted": True, "repo": repo_full_name} except GithubException as e: return {"deleted": False, "error": str(e.data)} def list_files(self, repo_full_name: str, path: str = "") -> List[Dict]: """سرد محتويات مسار في repo""" g = self._get_client() repo = g.get_repo(repo_full_name) contents = repo.get_contents(path) if not isinstance(contents, list): contents = [contents] result = [] for c in contents: result.append({ "name": c.name, "path": c.path, "type": c.type, # "file" or "dir" "size": c.size, "url": c.html_url, }) return result def read_file(self, repo_full_name: str, path: str) -> Dict: """قراءة محتوى ملف""" g = self._get_client() repo = g.get_repo(repo_full_name) content = repo.get_contents(path) if isinstance(content, list): return {"error": f"المسار {path} هو مجلد وليس ملفاً"} import base64 try: decoded = base64.b64decode(content.content).decode("utf-8") except Exception: decoded = content.content return { "path": content.path, "content": decoded, "size": content.size, "sha": content.sha, } def create_file( self, repo_full_name: str, path: str, content: str, commit_message: str = "", branch: str = "main", ) -> Dict: """إنشاء ملف جديد في repo""" g = self._get_client() repo = g.get_repo(repo_full_name) if not commit_message: commit_message = f"Create {path}" try: result = repo.create_file( path=path, message=commit_message, content=content, branch=branch, ) return { "created": True, "path": path, "commit": result.get("commit", {}).get("sha", ""), "url": result.get("content", {}).get("html_url", ""), } except GithubException as e: return {"created": False, "error": str(e.data)} def update_file( self, repo_full_name: str, path: str, content: str, commit_message: str = "", branch: str = "main", ) -> Dict: """تحديث ملف موجود""" g = self._get_client() repo = g.get_repo(repo_full_name) if not commit_message: commit_message = f"Update {path}" try: existing = repo.get_contents(path, ref=branch) sha = existing.sha if not isinstance(existing, list) else existing[0].sha result = repo.update_file( path=path, message=commit_message, content=content, sha=sha, branch=branch, ) return { "updated": True, "path": path, "commit": result.get("commit", {}).get("sha", ""), } except GithubException as e: return {"updated": False, "error": str(e.data)} def delete_file( self, repo_full_name: str, path: str, commit_message: str = "", branch: str = "main", ) -> Dict: """حذف ملف""" g = self._get_client() repo = g.get_repo(repo_full_name) if not commit_message: commit_message = f"Delete {path}" try: existing = repo.get_contents(path, ref=branch) sha = existing.sha if not isinstance(existing, list) else existing[0].sha repo.delete_file( path=path, message=commit_message, sha=sha, branch=branch, ) return {"deleted": True, "path": path} except GithubException as e: return {"deleted": False, "error": str(e.data)} def list_branches(self, repo_full_name: str) -> List[Dict]: g = self._get_client() repo = g.get_repo(repo_full_name) return [{"name": b.name, "protected": b.protected} for b in repo.get_branches()] def list_commits(self, repo_full_name: str, limit: int = 10) -> List[Dict]: g = self._get_client() repo = g.get_repo(repo_full_name) commits = [] for c in repo.get_commits()[:limit]: commits.append({ "sha": c.sha[:8], "message": c.commit.message.split("\n")[0][:80], "author": c.commit.author.name if c.commit.author else "—", "date": c.commit.author.date.isoformat() if c.commit.author else "", }) return commits def create_issue( self, repo_full_name: str, title: str, body: str = "", labels: Optional[List[str]] = None, ) -> Dict: g = self._get_client() repo = g.get_repo(repo_full_name) try: issue = repo.create_issue( title=title, body=body, labels=labels or [], ) return { "created": True, "number": issue.number, "url": issue.html_url, } except GithubException as e: return {"created": False, "error": str(e.data)} def list_issues(self, repo_full_name: str, state: str = "open", limit: int = 10) -> List[Dict]: g = self._get_client() repo = g.get_repo(repo_full_name) issues = [] for i in repo.get_issues(state=state)[:limit]: issues.append({ "number": i.number, "title": i.title, "state": i.state, "url": i.html_url, "labels": [l.name for l in i.labels], }) return issues def create_pull_request( self, repo_full_name: str, title: str, body: str, head: str, base: str = "main", ) -> Dict: g = self._get_client() repo = g.get_repo(repo_full_name) try: pr = repo.create_pull( title=title, body=body, head=head, base=base, ) return {"created": True, "number": pr.number, "url": pr.html_url} except GithubException as e: return {"created": False, "error": str(e.data)} def fork_repo(self, repo_full_name: str) -> Dict: g = self._get_client() repo = g.get_repo(repo_full_name) try: forked = repo.create_fork() return {"forked": True, "name": forked.full_name, "url": forked.html_url} except GithubException as e: return {"forked": False, "error": str(e.data)} # Singleton github_tools = GitHubTools()