Final Universal Git Analyzer v11
This commit is contained in:
444
git_analyzer.py
Normal file
444
git_analyzer.py
Normal file
@@ -0,0 +1,444 @@
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
LANG_EXTENSIONS = {
|
||||
"Python": [".py"],
|
||||
"C": [".c", ".h"],
|
||||
"C++": [".cpp", ".cxx", ".cc", ".hpp", ".hh", ".hxx"],
|
||||
"C#": [".cs"],
|
||||
"Java": [".java"],
|
||||
"JavaScript": [".js", ".jsx", ".mjs", ".cjs"],
|
||||
"TypeScript": [".ts", ".tsx"],
|
||||
"Bash / Shell": [".sh", ".bash", ".zsh", ".ksh"],
|
||||
"Batchfile": [".bat", ".cmd"],
|
||||
"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 run_git(args, cwd=None):
|
||||
result = subprocess.run(
|
||||
["git"] + args,
|
||||
cwd=cwd,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
timeout=60,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "Ошибка git")
|
||||
return result.stdout.strip()
|
||||
|
||||
|
||||
def run_git_allow_fail(args, cwd=None):
|
||||
return subprocess.run(
|
||||
["git"] + args,
|
||||
cwd=cwd,
|
||||
text=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
timeout=60,
|
||||
)
|
||||
|
||||
|
||||
def normalize_link(link):
|
||||
return link.strip()
|
||||
|
||||
|
||||
def repo_name_from_url(url):
|
||||
name = url.rstrip("/").split("/")[-1]
|
||||
if name.endswith(".git"):
|
||||
name = name[:-4]
|
||||
return name or "repository"
|
||||
|
||||
|
||||
def count_code_lines(text):
|
||||
return sum(1 for line in text.splitlines() if line.strip())
|
||||
|
||||
|
||||
def format_russian_datetime(value):
|
||||
if not value or value == "—":
|
||||
return "—"
|
||||
value = value.strip()
|
||||
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 = line.strip()
|
||||
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": parts[1],
|
||||
"author": parts[2],
|
||||
"date": format_russian_datetime(parts[3]),
|
||||
"timestamp": int(parts[4]) if parts[4].isdigit() else 0,
|
||||
"message": 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(x.strip() for x in out.splitlines() if x.strip()))
|
||||
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": parts[0], "author": parts[1], "date": format_russian_datetime(parts[2]), "timestamp": int(parts[3]) if parts[3].isdigit() else 0, "message": 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": parts[0], "author": parts[1], "date": format_russian_datetime(parts[2]), "timestamp": int(parts[3]) if parts[3].isdigit() else 0, "message": 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(x.strip() for x in out.splitlines() if x.strip())
|
||||
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 [x.strip() for x in out.splitlines() if x.strip()]
|
||||
|
||||
|
||||
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": parts[1],
|
||||
"author": parts[2],
|
||||
"date": format_russian_datetime(parts[3]),
|
||||
"timestamp": int(parts[4]) if parts[4].isdigit() else 0,
|
||||
"message": 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 = line.strip()
|
||||
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": 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)
|
||||
Reference in New Issue
Block a user