mirror of
https://github.com/migatu/vtt_work.git
synced 2026-07-14 21:38:39 +00:00
sync: root v0.4
This commit is contained in:
@@ -0,0 +1,356 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user