502 lines
18 KiB
Python
502 lines
18 KiB
Python
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", ".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)
|