Lib

gistfile1.txt

import subprocess
import json
import os
import sys
from typing import List, Set
import unittest

import requests


class CompareChangesError(Exception):
    pass


class CompareChangesLib:
    
    def __init__(
        self,
        owner: str,
        repo: str,
        access_token: str,
        local_repo_path: str,
        branch_a: str,
        branch_b: str
    ):
        self.owner = owner
        self.repo = repo
        self.access_token = access_token
        self.local_repo_path = os.path.abspath(local_repo_path)
        self.branch_a = branch_a
        self.branch_b = branch_b
        
        if not os.path.isdir(self.local_repo_path):
            raise CompareChangesError(f"Указанная папка '{self.local_repo_path}' не существует или не является директорией.")
        
    def _run_git_command(self, args: List[str]) -> str:
        try:
            result = subprocess.run(
                ["git"] + args,
                cwd=self.local_repo_path,
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                check=True
            )
        except subprocess.CalledProcessError as e:
            raise CompareChangesError(
                f"Ошибка при вызове git {' '.join(args)}: {e.stderr}"
            )
        return result.stdout.strip()
    
    def _get_merge_base_sha(self) -> str:
        self._run_git_command(["rev-parse", "--verify", self.branch_a])
        self._run_git_command(["rev-parse", "--verify", self.branch_b])
        
        merge_base_sha = self._run_git_command(["merge-base", self.branch_a, self.branch_b])
        if not merge_base_sha:
            raise CompareChangesError(
                f"Не удалось найти merge base между {self.branch_a} и {self.branch_b}"
            )
        return merge_base_sha
    
    def _get_changed_files_local(self, merge_base_sha: str) -> Set[str]:
        raw_diff = self._run_git_command(["diff", "--name-only", f"{merge_base_sha}..{self.branch_b}"])
        if not raw_diff.strip():
            return set()  # изменений нет
        
        changed_paths = set(raw_diff.splitlines())
        
        actually_changed = set()
        for path in changed_paths:
            file_diff = self._run_git_command([
                "diff", f"{merge_base_sha}..{self.branch_b}", "--", path
            ])
            if file_diff.strip():
                actually_changed.add(path)
        
        return actually_changed
    
    def _get_changed_files_remote(self, merge_base_sha: str) -> Set[str]:
        compare_url = f"https://api.github.com/repos/{self.owner}/{self.repo}/compare/{merge_base_sha}...{self.branch_a}"
        
        headers = {
            "Accept": "application/vnd.github.v3+json",
            "Authorization": f"token {self.access_token}",
        }
        
        response = requests.get(compare_url, headers=headers, timeout=30)
        if response.status_code != 200:
            raise CompareChangesError(
                f"Не удалось получить данные compare из GitHub. "
                f"Status code: {response.status_code}, ответ: {response.text}"
            )
        
        compare_data = response.json()
        if "files" not in compare_data:
            raise CompareChangesError(
                "Некорректный формат ответа GitHub Compare: поле 'files' отсутствует."
            )
        
        changed_files = set()
        for file_info in compare_data["files"]:
            filename = file_info.get("filename", "")
            if not filename:
                continue

            status = file_info.get("status", "")
            changes_count = file_info.get("changes", 0)
            patch = file_info.get("patch", None)

            if status in ("added", "removed", "modified"):
                if status in ("added", "removed"):
                    changed_files.add(filename)
                elif status == "modified":
                    if changes_count > 0:
                        changed_files.add(filename)
                    else:
                        if patch and patch.strip():
                            changed_files.add(filename)
        
        return changed_files
    
    def get_files_changed_in_both_branches(self) -> List[str]:
        merge_base_sha = self._get_merge_base_sha()
        
        local_changed = self._get_changed_files_local(merge_base_sha)
        remote_changed = self._get_changed_files_remote(merge_base_sha)
        
        both_changed = local_changed.intersection(remote_changed)
        
        return sorted(both_changed)

class TestCompareChangesLib(unittest.TestCase):
    def test_init_with_invalid_repo_path(self):
        """Проверка, что при несуществующей папке вылетит ошибка."""
        with self.assertRaises(CompareChangesError):
            CompareChangesLib(
                owner="owner",
                repo="repo",
                access_token="token",
                local_repo_path="/path/that/does/not/exist",
                branch_a="main",
                branch_b="feature"
            )


if __name__ == "__main__":
    # Запуск тестов
    unittest.main(argv=sys.argv, exit=True)