mirror of
https://github.com/migatu/vtt_work.git
synced 2026-07-14 13:34:42 +00:00
357 lines
13 KiB
Python
357 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
manage_releases.py
|
||
|
||
Automates synchronization of Git tags with GitHub releases. On each run it:
|
||
* Verifies the repository is clean and the current branch is fully pushed.
|
||
* Ensures HEAD is tagged by comparing the root version.nfo with existing
|
||
release tags and bumping/committing when required.
|
||
* Iterates over every tag in the repository, creating GitHub releases for any
|
||
missing entries and attaching per-directory ZIP archives (excluding
|
||
version.nfo files) that reflect the repository state for that tag.
|
||
|
||
Dependencies:
|
||
- gh CLI (https://cli.github.com/)
|
||
- git available in PATH
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import io
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import tarfile
|
||
import tempfile
|
||
import textwrap
|
||
from pathlib import Path
|
||
from typing import Dict, Iterable, List, Optional, Sequence, Tuple
|
||
from zipfile import ZIP_DEFLATED, ZipFile
|
||
|
||
|
||
def sh(args: Sequence[str], cwd: Path, check: bool = True) -> subprocess.CompletedProcess[str]:
|
||
"""Run a subprocess and return the completed process."""
|
||
result = subprocess.run(
|
||
list(args),
|
||
cwd=str(cwd),
|
||
text=True,
|
||
capture_output=True,
|
||
)
|
||
if check and result.returncode != 0:
|
||
raise RuntimeError(
|
||
f"Command failed ({' '.join(args)}):\n"
|
||
f"{result.stdout}{result.stderr}"
|
||
)
|
||
return result
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description="Synchronize Git tags with GitHub releases.")
|
||
parser.add_argument(
|
||
"--repo",
|
||
default=".",
|
||
help="Ścieżka do repozytorium git (domyślnie: bieżący katalog).",
|
||
)
|
||
parser.add_argument(
|
||
"--remote",
|
||
default="origin",
|
||
help="Nazwa zdalnego repozytorium do pushowania commitów oraz tagów (domyślnie: origin).",
|
||
)
|
||
parser.add_argument(
|
||
"--output-dir",
|
||
default="Release packs",
|
||
help="Katalog wyjściowy, w którym powstaną ZIP-y uploadowane do release (domyślnie: 'Release packs').",
|
||
)
|
||
parser.add_argument(
|
||
"--notes-template",
|
||
default="Automated release for {tag}",
|
||
help="Szablon notatek release (dostępne pola: {tag}).",
|
||
)
|
||
parser.add_argument(
|
||
"--dry-run",
|
||
action="store_true",
|
||
help="Tylko plan – bez commitów, tagowania i publikacji release.",
|
||
)
|
||
return parser.parse_args()
|
||
|
||
|
||
def read_text_if_exists(path: Path) -> Optional[str]:
|
||
if not path.is_file():
|
||
return None
|
||
return path.read_text(encoding="utf-8").strip()
|
||
|
||
|
||
def parse_version(version: str) -> Optional[Tuple[int, ...]]:
|
||
if not version:
|
||
return None
|
||
if not re.fullmatch(r"\d+(?:\.\d+)*", version):
|
||
return None
|
||
return tuple(int(part) for part in version.split("."))
|
||
|
||
|
||
def bump_version(version: str) -> str:
|
||
parts = parse_version(version)
|
||
if parts is None:
|
||
# fallback: append .1 to non-numeric strings
|
||
return f"{version}.1"
|
||
bumped = list(parts)
|
||
bumped[-1] += 1
|
||
return ".".join(str(piece) for piece in bumped)
|
||
|
||
|
||
def sanitize_component(value: str) -> str:
|
||
value = value.strip()
|
||
if not value:
|
||
return "unknown"
|
||
return re.sub(r"[\\s/\\\\]+", "-", value)
|
||
|
||
|
||
class ReleaseManager:
|
||
def __init__(self, args: argparse.Namespace):
|
||
self.args = args
|
||
self.repo = Path(args.repo).resolve()
|
||
self.output_dir = (self.repo / args.output_dir).resolve()
|
||
|
||
# ------------------------------------------------------------------ helpers
|
||
def git(self, *args: str, check: bool = True) -> subprocess.CompletedProcess[str]:
|
||
return sh(args, cwd=self.repo, check=check)
|
||
|
||
def ensure_tooling(self) -> None:
|
||
sh(["gh", "--version"], cwd=self.repo, check=True)
|
||
|
||
def ensure_repo(self) -> None:
|
||
if not (self.repo / ".git").exists():
|
||
raise RuntimeError(f"{self.repo} nie jest repozytorium git.")
|
||
|
||
def ensure_clean_state(self) -> None:
|
||
status = self.git("git", "status", "--porcelain", check=True).stdout.strip()
|
||
if status:
|
||
raise RuntimeError(
|
||
"Wykryto niezakomitowane zmiany. Ukończ commit lub stash przed publikacją release.\n"
|
||
"Release podpisany z nieczystego drzewa może wymagać resetu HEAD do taga."
|
||
)
|
||
|
||
upstream = self.git("git", "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}", check=False)
|
||
if upstream.returncode != 0:
|
||
raise RuntimeError(
|
||
"Brak skonfigurowanego upstreamu dla bieżącej gałęzi – nie można zweryfikować, "
|
||
"czy wszystkie commity zostały wypchnięte."
|
||
)
|
||
ahead = self.git("git", "rev-list", "--count", "@{u}..HEAD", check=True).stdout.strip()
|
||
if ahead and int(ahead) > 0:
|
||
raise RuntimeError(
|
||
"Bieżąca gałąź zawiera niewypchnięte commity. Wypchnij zmiany zanim utworzysz release."
|
||
)
|
||
|
||
def get_head_tag(self) -> Optional[str]:
|
||
result = self.git("git", "describe", "--tags", "--exact-match", "HEAD", check=False)
|
||
tag = result.stdout.strip()
|
||
if result.returncode == 0 and tag:
|
||
return tag
|
||
return None
|
||
|
||
def list_tags(self) -> List[str]:
|
||
tags = self.git("git", "tag", "--sort=creatordate", check=True).stdout.splitlines()
|
||
return [tag.strip() for tag in tags if tag.strip()]
|
||
|
||
def list_version_tags(self) -> List[str]:
|
||
tags = []
|
||
for tag in self.list_tags():
|
||
if parse_version(tag):
|
||
tags.append(tag)
|
||
return tags
|
||
|
||
def read_root_version(self) -> Optional[str]:
|
||
return read_text_if_exists(self.repo / "version.nfo")
|
||
|
||
def write_root_version(self, version: str) -> None:
|
||
target = self.repo / "version.nfo"
|
||
target.write_text(f"{version}\n", encoding="utf-8")
|
||
|
||
def commit_root_version(self, version: str) -> None:
|
||
if self.args.dry_run:
|
||
print(f"[DRY-RUN] git add version.nfo && git commit -m 'release: bump root to v{version}'")
|
||
return
|
||
self.write_root_version(version)
|
||
self.git("git", "add", "version.nfo", check=True)
|
||
self.git("git", "commit", "-m", f"release: bump root to v{version}", check=True)
|
||
self.git("git", "push", self.args.remote, check=True)
|
||
|
||
def tag_exists(self, tag: str) -> bool:
|
||
result = self.git("git", "rev-parse", "-q", "--verify", f"refs/tags/{tag}", check=False)
|
||
return result.returncode == 0
|
||
|
||
def create_tag(self, tag: str, message: str) -> None:
|
||
if self.tag_exists(tag):
|
||
print(f"[INFO] Tag {tag} już istnieje – pomijam tworzenie.")
|
||
return
|
||
if self.args.dry_run:
|
||
print(f"[DRY-RUN] git tag -a {tag} -m \"{message}\"")
|
||
return
|
||
self.git("git", "tag", "-a", tag, "-m", message, check=True)
|
||
self.git("git", "push", self.args.remote, tag, check=True)
|
||
|
||
def ensure_head_tagged(self) -> str:
|
||
head_tag = self.get_head_tag()
|
||
if head_tag:
|
||
return head_tag
|
||
|
||
root_version = self.read_root_version()
|
||
root_tuple = parse_version(root_version) if root_version else None
|
||
|
||
version_tags = self.list_version_tags()
|
||
latest_version = version_tags[-1] if version_tags else None
|
||
latest_tuple = parse_version(latest_version) if latest_version else None
|
||
|
||
if root_version and (root_tuple and (latest_tuple is None or root_tuple > latest_tuple)):
|
||
new_version = root_version
|
||
else:
|
||
base_version = latest_version or root_version or "0.0"
|
||
new_version = bump_version(base_version)
|
||
if not self.args.dry_run:
|
||
print(f"[INFO] Bump root version.nfo -> {new_version}")
|
||
self.commit_root_version(new_version)
|
||
|
||
tag_message = f"release {new_version}"
|
||
self.create_tag(new_version, tag_message)
|
||
return new_version
|
||
|
||
def gather_directories_for_tag(self, tag: str) -> List[str]:
|
||
result = self.git("git", "ls-tree", "-d", "--name-only", tag, check=True)
|
||
candidates = []
|
||
for line in result.stdout.splitlines():
|
||
name = line.strip()
|
||
if not name:
|
||
continue
|
||
candidates.append(name)
|
||
return candidates
|
||
|
||
def version_for_directory(self, tag: str, directory: str) -> str:
|
||
show = self.git("git", "show", f"{tag}:{directory}/version.nfo", check=False)
|
||
if show.returncode == 0:
|
||
version = show.stdout.strip()
|
||
if version:
|
||
return version
|
||
return tag
|
||
|
||
def build_assets_for_tag(self, tag: str) -> List[Path]:
|
||
assets: List[Path] = []
|
||
directories = self.gather_directories_for_tag(tag)
|
||
if not directories:
|
||
print(f"[WARN] Tag {tag} nie zawiera katalogów do zzipowania.")
|
||
return assets
|
||
|
||
tag_output_dir = self.output_dir / sanitize_component(tag)
|
||
if not self.args.dry_run:
|
||
if tag_output_dir.exists():
|
||
shutil.rmtree(tag_output_dir)
|
||
tag_output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
for directory in directories:
|
||
version = self.version_for_directory(tag, directory)
|
||
safe_version = sanitize_component(version)
|
||
safe_dir = sanitize_component(directory)
|
||
zip_name = f"{safe_dir}-v{safe_version}.zip"
|
||
zip_path = tag_output_dir / zip_name
|
||
|
||
if self.args.dry_run:
|
||
print(f"[DRY-RUN] Przygotuj ZIP {zip_path} z taga {tag}:{directory} (pomijając version.nfo)")
|
||
assets.append(zip_path)
|
||
continue
|
||
|
||
# Re-run capturing binary properly.
|
||
archive_proc = subprocess.run(
|
||
["git", "archive", "--format=tar", tag, directory],
|
||
cwd=str(self.repo),
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
)
|
||
if archive_proc.returncode != 0:
|
||
raise RuntimeError(
|
||
f"git archive nie powiódł się dla {tag}:{directory}:\n"
|
||
f"{archive_proc.stderr.decode('utf-8', errors='ignore')}"
|
||
)
|
||
|
||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||
tmp_path = Path(tmp_dir)
|
||
with tarfile.open(fileobj=io.BytesIO(archive_proc.stdout)) as tar:
|
||
tar.extractall(tmp_path)
|
||
|
||
extracted_root = tmp_path / directory
|
||
for extra in extracted_root.rglob("version.nfo"):
|
||
if extra.is_file():
|
||
extra.unlink()
|
||
|
||
with ZipFile(zip_path, "w", ZIP_DEFLATED) as zipf:
|
||
for root, _, files in os.walk(extracted_root):
|
||
for filename in files:
|
||
file_path = Path(root) / filename
|
||
rel_path = file_path.relative_to(tmp_path)
|
||
zipf.write(file_path, rel_path.as_posix())
|
||
|
||
assets.append(zip_path)
|
||
|
||
return assets
|
||
|
||
def release_exists(self, tag: str) -> bool:
|
||
result = self.git("gh", "release", "view", tag, check=False)
|
||
return result.returncode == 0
|
||
|
||
def create_release(self, tag: str, assets: List[Path]) -> None:
|
||
notes = self.args.notes_template.format(tag=tag)
|
||
wrapped = "\n".join(textwrap.wrap(notes, width=100))
|
||
|
||
if self.args.dry_run:
|
||
print(f"[DRY-RUN] gh release create {tag} {' '.join(str(a) for a in assets)}")
|
||
print(f"[DRY-RUN] Notatki:\n{wrapped}")
|
||
return
|
||
|
||
cmd = ["gh", "release", "create", tag, "--title", tag, "--notes", wrapped]
|
||
cmd.extend(str(asset) for asset in assets)
|
||
result = self.git(*cmd, check=False)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(
|
||
f"Nie udało się utworzyć release {tag}:\n{result.stdout}{result.stderr}"
|
||
)
|
||
|
||
def sync_releases(self) -> None:
|
||
tags = self.list_tags()
|
||
if not tags:
|
||
print("[INFO] Brak tagów do synchronizacji.")
|
||
return
|
||
|
||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
for tag in tags:
|
||
if self.release_exists(tag):
|
||
print(f"[INFO] Release {tag} już istnieje – pomijam.")
|
||
continue
|
||
print(f"[INFO] Przygotowuję release dla taga {tag}...")
|
||
assets = self.build_assets_for_tag(tag)
|
||
self.create_release(tag, assets)
|
||
|
||
def run(self) -> None:
|
||
self.ensure_repo()
|
||
self.ensure_tooling()
|
||
self.ensure_clean_state()
|
||
self.ensure_head_tagged()
|
||
self.sync_releases()
|
||
|
||
|
||
def main() -> int:
|
||
args = parse_args()
|
||
manager = ReleaseManager(args)
|
||
try:
|
||
manager.run()
|
||
except RuntimeError as exc:
|
||
print(f"[BŁĄD] {exc}")
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|