github_tools.py

"""
وحدة 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()