github_client.py

from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Any
from urllib.parse import urlparse

import requests


API_BASE = "https://api.github.com"


@dataclass(frozen=True)
class GitHubTarget:
    owner: str
    repository: str | None = None


class GitHubClientError(RuntimeError):
    pass


class GitHubClient:
    def __init__(self, token: str | None = None, timeout: int = 30) -> None:
        self.token = token.strip() if token else None
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update(
            {
                "Accept": "application/vnd.github+json",
                "X-GitHub-Api-Version": "2022-11-28",
                "User-Agent": "github-security-scanner-local",
            }
        )
        if self.token:
            self.session.headers["Authorization"] = f"Bearer {self.token}"

    @staticmethod
    def parse_target(raw_target: str) -> GitHubTarget:
        value = raw_target.strip().rstrip("/")
        if not value:
            raise GitHubClientError("Informe um usuário, repositório ou URL do GitHub.")

        if value.startswith(("http://", "https://")):
            parsed = urlparse(value)
            if parsed.netloc.lower() not in {"github.com", "www.github.com"}:
                raise GitHubClientError("A URL precisa pertencer ao github.com.")
            parts = [part for part in parsed.path.split("/") if part]
        else:
            value = re.sub(r"^github\.com/", "", value, flags=re.I)
            parts = [part for part in value.split("/") if part]

        if not parts or len(parts) > 2:
            raise GitHubClientError("Use o formato usuário ou usuário/repositório.")

        owner = parts[0]
        repository = parts[1].removesuffix(".git") if len(parts) == 2 else None
        valid = re.compile(r"^[A-Za-z0-9_.-]+$")
        if not valid.fullmatch(owner) or (repository and not valid.fullmatch(repository)):
            raise GitHubClientError("O usuário ou repositório contém caracteres inválidos.")

        return GitHubTarget(owner=owner, repository=repository)

    def _request(self, url: str, params: dict[str, Any] | None = None) -> requests.Response:
        try:
            response = self.session.get(url, params=params, timeout=self.timeout)
        except requests.RequestException as exc:
            raise GitHubClientError(
                "Não foi possível conectar à API do GitHub. Verifique sua internet, proxy ou firewall."
            ) from exc
        if response.status_code == 404:
            raise GitHubClientError("Perfil ou repositório não encontrado, ou sem permissão de acesso.")
        if response.status_code == 403:
            rate_remaining = response.headers.get("X-RateLimit-Remaining")
            if rate_remaining == "0":
                raise GitHubClientError("Limite da API do GitHub atingido. Configure GITHUB_TOKEN e tente novamente.")
            raise GitHubClientError("O GitHub recusou o acesso. Verifique o token e suas permissões.")
        if response.status_code >= 400:
            raise GitHubClientError(f"Erro da API do GitHub: HTTP {response.status_code}.")
        return response

    def authenticated_login(self) -> str | None:
        if not self.token:
            return None
        try:
            return str(self._request(f"{API_BASE}/user").json().get("login") or "") or None
        except GitHubClientError:
            return None

    def get_repository(self, owner: str, repository: str) -> dict[str, Any]:
        return self._request(f"{API_BASE}/repos/{owner}/{repository}").json()

    def list_repositories(self, owner: str, include_forks: bool, include_archived: bool) -> list[dict[str, Any]]:
        auth_login = self.authenticated_login()
        if auth_login and auth_login.lower() == owner.lower():
            url = f"{API_BASE}/user/repos"
            params: dict[str, Any] = {
                "per_page": 100,
                "page": 1,
                "affiliation": "owner",
                "visibility": "all",
                "sort": "updated",
            }
        else:
            url = f"{API_BASE}/users/{owner}/repos"
            params = {"per_page": 100, "page": 1, "type": "owner", "sort": "updated"}

        repositories: list[dict[str, Any]] = []
        while True:
            response = self._request(url, params=params)
            page_items = response.json()
            if not isinstance(page_items, list):
                raise GitHubClientError("Resposta inesperada da API do GitHub.")
            repositories.extend(page_items)
            if len(page_items) < 100:
                break
            params["page"] += 1

        filtered = []
        for repo in repositories:
            if not include_forks and repo.get("fork"):
                continue
            if not include_archived and repo.get("archived"):
                continue
            filtered.append(repo)
        return filtered

    def resolve(self, target: GitHubTarget, include_forks: bool, include_archived: bool) -> list[dict[str, Any]]:
        if target.repository:
            repo = self.get_repository(target.owner, target.repository)
            if (repo.get("fork") and not include_forks) or (repo.get("archived") and not include_archived):
                return []
            return [repo]
        return self.list_repositories(target.owner, include_forks, include_archived)