import os import shutil import subprocess import tempfile from pathlib import Path from datetime import datetime from dataclasses import dataclass, field @dataclass class CommitInfo: hash: str short: str author: str date: str timestamp: int message: str @dataclass class BranchInfo: name: str ref: str total_files: int = 0 total_lines: int = 0 @dataclass class RepositoryInfo: link: str repo_name: str default_branch: str branches: list = field(default_factory=list) commits: list = field(default_factory=list) total_files: int = 0 total_lines: int = 0 LANG_EXTENSIONS = { "Python": [".py"], "C": [".c", ".h"], "C++": [".cpp", ".cxx", ".cc", ".hpp", ".hh", ".hxx"], "C#": [".cs", ".csx"], "Java": [".java"], "JavaScript": [".js", ".jsx", ".mjs", ".cjs"], "TypeScript": [".ts", ".tsx"], "Bash / Shell": [".sh", ".bash", ".zsh", ".ksh"], "Batchfile": [".bat", ".cmd", ".btm"], "PowerShell": [".ps1", ".psm1", ".psd1"], "Go": [".go"], "PHP": [".php", ".phtml", ".php3", ".php4", ".php5", ".phps"], } SPECIAL_FILENAMES = { "dockerfile": "Dockerfile", } IGNORE_DIRS = {".git", "__pycache__", "node_modules", ".venv", "venv", "dist", "build"} def safe_text(value, default=""): if value is None: return default return str(value) def safe_strip(value, default=""): return safe_text(value, default).strip() def _subprocess_no_window_kwargs(): if os.name != "nt": return {} kwargs = {} try: kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW except AttributeError: pass try: startupinfo = subprocess.STARTUPINFO() startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW startupinfo.wShowWindow = 0 kwargs["startupinfo"] = startupinfo except Exception: pass return kwargs def run_git(args, cwd=None): result = subprocess.run( ["git", "-c", "core.quotepath=false"] + args, cwd=cwd, text=True, encoding="utf-8", errors="replace", stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL, timeout=60, **_subprocess_no_window_kwargs(), ) if result.returncode != 0: raise RuntimeError(safe_strip(result.stderr) or safe_strip(result.stdout) or "Ошибка git") return safe_strip(result.stdout) def run_git_allow_fail(args, cwd=None): return subprocess.run( ["git", "-c", "core.quotepath=false"] + args, cwd=cwd, text=True, encoding="utf-8", errors="replace", stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL, timeout=60, **_subprocess_no_window_kwargs(), ) def normalize_link(link): link = safe_strip(link) if not link: raise RuntimeError("Пустая ссылка на репозиторий") if "github.com/" in link: before = link proto = "" rest = link if "://" in rest: proto, rest = rest.split("://", 1) proto += "://" parts = rest.split("/") if len(parts) >= 3 and parts[0].lower() == "github.com": repo = parts[2] if repo.endswith(".git"): repo = repo[:-4] link = proto + "/".join([parts[0], parts[1], repo]) return link def repo_name_from_url(url): url = normalize_link(url) name = url.rstrip("/").split("/")[-1] if name.endswith(".git"): name = name[:-4] return name or "repository" def count_code_lines(text): text = safe_text(text) return sum(1 for line in text.splitlines() if safe_strip(line)) def format_russian_datetime(value): value = safe_strip(value) if not value or value == "—": return "—" for fmt in ("%Y-%m-%d %H:%M:%S %z", "%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%d %H:%M:%S"): try: dt = datetime.strptime(value, fmt) return dt.strftime("%d.%m.%Y %H:%M") except ValueError: continue parts = value.split() if len(parts) >= 2: date_part = parts[0] time_part = parts[1][:5] date_bits = date_part.split("-") if len(date_bits) == 3: return f"{date_bits[2]}.{date_bits[1]}.{date_bits[0]} {time_part}" return value def detect_language(path): lower = path.lower() filename = Path(path).name.lower() if filename in SPECIAL_FILENAMES: return SPECIAL_FILENAMES[filename] for language, exts in LANG_EXTENSIONS.items(): if any(lower.endswith(ext) for ext in exts): return language return None def safe_branch_name(remote_branch): if "/" in remote_branch: return remote_branch.split("/", 1)[1] return remote_branch def clone_repo(link): temp_dir = tempfile.mkdtemp(prefix="uga_repo_") repo_dir = os.path.join(temp_dir, "repo") try: run_git(["clone", "--quiet", normalize_link(link), repo_dir]) run_git(["fetch", "--all", "--prune", "--quiet"], cwd=repo_dir) return temp_dir, repo_dir except Exception: shutil.rmtree(temp_dir, ignore_errors=True) raise def get_default_branch(repo_dir): try: ref = run_git(["symbolic-ref", "refs/remotes/origin/HEAD"], cwd=repo_dir) return ref.replace("refs/remotes/origin/", "") except Exception: try: return run_git(["rev-parse", "--abbrev-ref", "HEAD"], cwd=repo_dir) except Exception: return "main" def get_all_remote_branches(repo_dir): output = run_git(["branch", "-r"], cwd=repo_dir) branches = [] for line in output.splitlines(): b = safe_strip(line) if not b or "->" in b: continue if b.startswith("origin/"): branches.append(b) branches = list(dict.fromkeys(branches)) def key(b): name = safe_branch_name(b) if name in ("main", "master"): return (0, name) return (1, name) return sorted(branches, key=key) def get_commits(repo_dir, limit=10): fmt = "%H%x1f%h%x1f%an%x1f%ad%x1f%at%x1f%s" out = run_git(["log", "--all", f"--max-count={limit}", "--date=iso", f"--pretty=format:{fmt}"], cwd=repo_dir) raw_lines = out.splitlines() commits = [] for idx, line in enumerate(raw_lines, start=1): parts = line.split("\x1f") if len(parts) >= 6: commits.append({ "number": len(raw_lines) - idx + 1, "hash": parts[0], "short": safe_strip(parts[1], "—"), "author": safe_strip(parts[2], "—"), "date": format_russian_datetime(parts[3]), "timestamp": int(safe_strip(parts[4])) if safe_strip(parts[4]).isdigit() else 0, "message": safe_strip(parts[5], "—"), }) commits.reverse() return commits def get_all_authors(repo_dir): try: out = run_git(["log", "--all", "--format=%an"], cwd=repo_dir) return sorted(set(safe_strip(x) for x in out.splitlines() if safe_strip(x))) except Exception: return [] def first_commit_info(repo_dir): fmt = "%H%x1f%an%x1f%ad%x1f%at%x1f%s" out = run_git(["log", "--all", "--reverse", "--date=iso", f"--pretty=format:{fmt}"], cwd=repo_dir) line = out.splitlines()[0] if out else "" parts = line.split("\x1f") if len(parts) >= 5: return {"hash": safe_strip(parts[0]), "author": safe_strip(parts[1], "—"), "date": format_russian_datetime(parts[2]), "timestamp": int(safe_strip(parts[3])) if safe_strip(parts[3]).isdigit() else 0, "message": safe_strip(parts[4], "—")} return {"hash": "", "author": "—", "date": "—", "timestamp": 0, "message": "—"} def last_commit_info(repo_dir): fmt = "%H%x1f%an%x1f%ad%x1f%at%x1f%s" out = run_git(["log", "--all", "--max-count=1", "--date=iso", f"--pretty=format:{fmt}"], cwd=repo_dir) parts = out.split("\x1f") if len(parts) >= 5: return {"hash": safe_strip(parts[0]), "author": safe_strip(parts[1], "—"), "date": format_russian_datetime(parts[2]), "timestamp": int(safe_strip(parts[3])) if safe_strip(parts[3]).isdigit() else 0, "message": safe_strip(parts[4], "—")} return {"hash": "", "author": "—", "date": "—", "timestamp": 0, "message": "—"} def files_changed_in_commit(repo_dir, commit_hash): if not commit_hash: return set() for args in (["diff-tree", "--root", "--no-commit-id", "--name-only", "-r", commit_hash], ["diff-tree", "--no-commit-id", "--name-only", "-r", commit_hash]): try: out = run_git(args, cwd=repo_dir) return set(safe_strip(x) for x in out.splitlines() if safe_strip(x)) except Exception: continue return set() def list_files_at_ref(repo_dir, ref): try: out = run_git(["ls-tree", "-r", "--name-only", ref], cwd=repo_dir) except Exception: return [] return [safe_strip(x) for x in out.splitlines() if safe_strip(x)] def read_file_at_ref(repo_dir, ref, path): try: return run_git(["show", f"{ref}:{path}"], cwd=repo_dir) except Exception: return "" def get_branch_commits(repo_dir, branch_ref): """Коммиты конкретной ветки от старых к новым, с timestamp. Это нужно, чтобы на старой версии не показывать ветку, которой тогда ещё не было. """ fmt = "%H%x1f%h%x1f%an%x1f%ad%x1f%at%x1f%s" try: out = run_git(["log", branch_ref, "--reverse", "--date=iso", f"--pretty=format:{fmt}"], cwd=repo_dir) except Exception: return [] result = [] for line in out.splitlines(): parts = line.split("\x1f") if len(parts) >= 6: result.append({ "hash": parts[0], "short": safe_strip(parts[1], "—"), "author": safe_strip(parts[2], "—"), "date": format_russian_datetime(parts[3]), "timestamp": int(safe_strip(parts[4])) if safe_strip(parts[4]).isdigit() else 0, "message": safe_strip(parts[5], "—"), }) return result def branch_first_unique_timestamp(repo_dir, branch_ref, default_ref): """Когда ветка реально стала отдельной. Git не хранит дату создания ветки, поэтому берём первый коммит ветки, которого нет в основной ветке. Если таких нет — ветка считается старой. """ if branch_ref == default_ref: return None result = run_git_allow_fail(["log", "--reverse", "--format=%at", f"{default_ref}..{branch_ref}"], cwd=repo_dir) if result.returncode == 0: for line in result.stdout.splitlines(): line = safe_strip(line) if line.isdigit(): return int(line) return None def latest_branch_commit_at_time(repo_dir, branch_ref, selected_timestamp): commits = get_branch_commits(repo_dir, branch_ref) if not commits: return None current = None for commit in commits: if commit["timestamp"] <= selected_timestamp: current = commit else: break return current def get_visible_branches_at_commit(repo_dir, selected_commit): all_branches = get_all_remote_branches(repo_dir) default_branch = get_default_branch(repo_dir) default_ref = f"origin/{default_branch}" if default_ref not in all_branches and all_branches: default_ref = all_branches[0] selected_ts = selected_commit.get("timestamp", 0) visible = [] for branch_ref in all_branches: if branch_ref == default_ref: visible.append(branch_ref) continue first_unique_ts = branch_first_unique_timestamp(repo_dir, branch_ref, default_ref) if first_unique_ts is not None and first_unique_ts <= selected_ts: visible.append(branch_ref) elif first_unique_ts is None and latest_branch_commit_at_time(repo_dir, branch_ref, selected_ts): visible.append(branch_ref) return visible def analyze_files_for_branches(repo_dir, selected_commit): selected_hash = selected_commit.get("hash") selected_ts = selected_commit.get("timestamp", 0) branches = get_visible_branches_at_commit(repo_dir, selected_commit) changed = files_changed_in_commit(repo_dir, selected_hash) if selected_hash else set() result = [] for branch_ref in branches: branch_name = safe_branch_name(branch_ref) branch_commit = latest_branch_commit_at_time(repo_dir, branch_ref, selected_ts) if not branch_commit: continue ref_for_files = branch_commit["hash"] ref_note = "состояние на выбранной версии" files = list_files_at_ref(repo_dir, ref_for_files) languages = {} total_files = 0 total_lines = 0 for path in files: if any(part in IGNORE_DIRS for part in Path(path).parts): continue language = detect_language(path) if not language: continue content = read_file_at_ref(repo_dir, ref_for_files, path) lines = count_code_lines(content) total_files += 1 total_lines += lines languages.setdefault(language, []).append({ "path": path, "lines": lines, "changed": path in changed, }) result.append({ "name": branch_name, "ref": branch_ref, "ref_note": ref_note, "branch_commit": branch_commit, "languages": languages, "total_files": total_files, "total_lines": total_lines, }) return result def analyze_repository(link, selected_index=None): temp_dir, repo_dir = clone_repo(link) try: commits = get_commits(repo_dir, limit=10) if not commits: raise RuntimeError("В репозитории нет коммитов") if selected_index is None: selected_index = len(commits) - 1 selected_index = max(0, min(selected_index, len(commits) - 1)) selected_commit = commits[selected_index] default_branch = get_default_branch(repo_dir) visible_branches = get_visible_branches_at_commit(repo_dir, selected_commit) first = first_commit_info(repo_dir) last = last_commit_info(repo_dir) authors = get_all_authors(repo_dir) branch_files = analyze_files_for_branches(repo_dir, selected_commit) total_files = sum(b["total_files"] for b in branch_files) total_lines = sum(b["total_lines"] for b in branch_files) return { "link": normalize_link(link), "repo_name": repo_name_from_url(link), "default_branch": default_branch, "branches_count": len(visible_branches), "branches": [safe_branch_name(b) for b in visible_branches], "first": first, "last": last, "authors": authors, "commits": commits, "selected_index": selected_index, "selected_commit": selected_commit, "branch_files": branch_files, "total_files": total_files, "total_lines": total_lines, } finally: shutil.rmtree(temp_dir, ignore_errors=True) def format_analysis(data): lines = [] lines.append(f"Ссылка: {data['link']}") lines.append(f"Название репозитория: {data['repo_name']}") lines.append(f"Основная ветка: {data['default_branch']}") lines.append(f"Количество веток в выбранной версии: {data['branches_count']}") lines.append("Ветки в выбранной версии: " + (", ".join(data['branches']) if data['branches'] else "—")) lines.append("") lines.append(f"Автор по первому коммиту: {data['first']['author']}") lines.append("Авторы / соавторы по коммитам:") for author in data["authors"]: lines.append(f"- {author}") lines.append("") lines.append(f"Дата создания по первому коммиту: {data['first']['date']}") lines.append(f"Дата последнего изменения: {data['last']['date']}") lines.append(f"Всего коммитов в списке: {len(data['commits'])}") lines.append("") lines.append("Последние 10 коммитов:") for idx, commit in enumerate(data["commits"], start=1): marker = " <= текущая версия" if idx - 1 == data["selected_index"] else "" lines.append(f"{idx}. {commit['short']} | {commit['author']} | {commit['date']} | {commit['message']}{marker}") lines.append("") selected = data["selected_commit"] lines.append("Текущая выбранная версия:") lines.append(f"Коммит: {data['selected_index'] + 1} ({selected['short']})") lines.append(f"Автор коммита: {selected['author']}") lines.append(f"Дата: {selected['date']}") lines.append(f"Комментарий: {selected['message']}") lines.append("") lines.append("Файлы по языкам и веткам:") lines.append("") for branch in data["branch_files"]: branch_commit = branch.get("branch_commit", {}) commit_note = f", коммит ветки: {branch_commit.get('short', '—')}" if branch_commit else "" lines.append(f"Ветка: {branch['name']} ({branch['ref_note']}{commit_note})") lines.append(f"Файлов кода: {branch['total_files']}, строк кода: {branch['total_lines']}") if not branch["languages"]: lines.append(" Файлов поддерживаемых языков нет") for language, files in branch["languages"].items(): lang_lines = sum(f["lines"] for f in files) lines.append(f" {language}: файлов {len(files)}, строк {lang_lines}") for f in files: changed = " (изменён в этом коммите)" if f["changed"] else "" lines.append(f" - {f['path']}: {f['lines']} строк{changed}") lines.append("") lines.append(f"Итого файлов кода по всем веткам выбранной версии: {data['total_files']}") lines.append(f"Итого строк кода по всем веткам выбранной версии: {data['total_lines']}") return "\n".join(lines) class GitAnalyzer: """Главный класс анализа Git-репозиториев. В старых версиях проекта логика была набором функций. В OOP-версии интерфейс создаёт объект GitAnalyzer и вызывает его методы. Так проще строить диаграмму классов и объяснять систему как набор объектов. """ def __init__(self): self.languages = LANG_EXTENSIONS self.ignore_dirs = IGNORE_DIRS def safe_text(self, value, default=""): return safe_text(value, default) def safe_strip(self, value, default=""): return safe_strip(value, default) def run_git(self, args, cwd=None): return run_git(args, cwd) def run_git_allow_fail(self, args, cwd=None): return run_git_allow_fail(args, cwd) def normalize_link(self, link): return normalize_link(link) def repo_name_from_url(self, url): return repo_name_from_url(url) def detect_language(self, path): return detect_language(path) def clone_repo(self, link): return clone_repo(link) def get_default_branch(self, repo_dir): return get_default_branch(repo_dir) def get_all_remote_branches(self, repo_dir): return get_all_remote_branches(repo_dir) def get_commits(self, repo_dir, limit=10): return get_commits(repo_dir, limit) def get_all_authors(self, repo_dir): return get_all_authors(repo_dir) def first_commit_info(self, repo_dir): return first_commit_info(repo_dir) def last_commit_info(self, repo_dir): return last_commit_info(repo_dir) def files_changed_in_commit(self, repo_dir, commit_hash): return files_changed_in_commit(repo_dir, commit_hash) def get_visible_branches_at_commit(self, repo_dir, selected_commit): return get_visible_branches_at_commit(repo_dir, selected_commit) def analyze_files_for_branches(self, repo_dir, selected_commit): return analyze_files_for_branches(repo_dir, selected_commit) def analyze_repository(self, link, selected_index=None): data = analyze_repository(link, selected_index) self.last_repository_object = RepositoryInfo( link=data.get("link", ""), repo_name=data.get("repo_name", ""), default_branch=data.get("default_branch", ""), branches=data.get("branches", []), commits=data.get("commits", []), total_files=data.get("total_files", 0), total_lines=data.get("total_lines", 0), ) return data def format_analysis(self, data): return format_analysis(data)