mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-14 21:38:38 +00:00
Syncinc script drop in
This commit is contained in:
@@ -0,0 +1,296 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""
|
||||
Buduje nową gałąź z "próbkowanych" commitów na podstawie tagów z bieżącej gałęzi,
|
||||
a na końcu dodaje stan bieżącego HEAD (jeśli nie jest otagowany).
|
||||
Autor: (drop-in)
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import List, Tuple, Optional
|
||||
|
||||
# --- Logging setup ---
|
||||
logger = logging.getLogger("tag_branch_builder")
|
||||
handler = logging.StreamHandler(sys.stdout)
|
||||
formatter = logging.Formatter("%(levelname)s: %(message)s")
|
||||
handler.setFormatter(formatter)
|
||||
logger.addHandler(handler)
|
||||
logger.setLevel(logging.INFO)
|
||||
|
||||
|
||||
class GitError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def run_git(args: List[str], cwd: Optional[str] = None, check: bool = True) -> str:
|
||||
"""Run a git command and return stdout (stripped)."""
|
||||
cmd = ["git"] + args
|
||||
logger.debug("Running: %s", " ".join(cmd))
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
cmd,
|
||||
cwd=cwd,
|
||||
check=False,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
except Exception as e:
|
||||
raise Exception("Nie udało się uruchomić gita.") from e
|
||||
raise
|
||||
|
||||
if check and proc.returncode != 0:
|
||||
logger.error("Git error (%s): %s", " ".join(cmd), proc.stderr.strip())
|
||||
raise GitError(proc.stderr.strip())
|
||||
return proc.stdout.strip()
|
||||
|
||||
|
||||
def ensure_git_repo() -> None:
|
||||
try:
|
||||
run_git(["rev-parse", "--is-inside-work-tree"])
|
||||
except Exception:
|
||||
logger.exception("To nie wygląda na repozytorium git.")
|
||||
raise
|
||||
|
||||
|
||||
def ensure_clean_worktree() -> None:
|
||||
# Untracked + changes
|
||||
status = run_git(["status", "--porcelain"])
|
||||
if status.strip():
|
||||
raise GitError(
|
||||
"Drzewo robocze nie jest czyste. Zacommituj/stashuj zmiany i spróbuj ponownie."
|
||||
)
|
||||
|
||||
|
||||
def current_branch() -> str:
|
||||
# Returns branch or 'HEAD' when detached
|
||||
ref = run_git(["rev-parse", "--abbrev-ref", "HEAD"])
|
||||
return ref
|
||||
|
||||
|
||||
def head_sha() -> str:
|
||||
return run_git(["rev-parse", "HEAD"])
|
||||
|
||||
|
||||
def sha_has_tag(sha: str) -> List[str]:
|
||||
# tags pointing at sha
|
||||
tags = run_git(["tag", "--points-at", sha])
|
||||
return [t for t in tags.splitlines() if t.strip()]
|
||||
|
||||
|
||||
def tags_merged_into(branch: str) -> List[str]:
|
||||
out = run_git(["tag", "--merged", branch])
|
||||
return [t for t in out.splitlines() if t.strip()]
|
||||
|
||||
|
||||
def tag_commit_sha(tag: str) -> str:
|
||||
return run_git(["rev-list", "-n", "1", tag])
|
||||
|
||||
|
||||
def commit_unix_time(sha: str) -> int:
|
||||
return int(run_git(["show", "-s", "--format=%ct", sha]))
|
||||
|
||||
|
||||
def sort_tags_by_commit_time(tags: List[str]) -> List[Tuple[str, str, int]]:
|
||||
triples = []
|
||||
for t in tags:
|
||||
sha = tag_commit_sha(t)
|
||||
ts = commit_unix_time(sha)
|
||||
triples.append((t, sha, ts))
|
||||
triples.sort(key=lambda x: x[2]) # oldest first
|
||||
return triples
|
||||
|
||||
|
||||
def list_intermediate_messages(old_sha: Optional[str], new_sha: str) -> List[str]:
|
||||
"""
|
||||
Zwróć listę komunikatów commitów POŚREDNICH (wyłącznie) od old_sha do new_sha.
|
||||
Kolejność: od najstarszego do najnowszego.
|
||||
"""
|
||||
if old_sha is None:
|
||||
# Nie ma commitów pośrednich przed pierwszym tagiem
|
||||
return []
|
||||
# ancestry-path: tylko ścieżka od old_sha do new_sha (jeśli wiele rodziców)
|
||||
# Zakres old_sha..new_sha zawiera new_sha, dlatego pominiemy go w output.
|
||||
rng = f"{old_sha}..{new_sha}"
|
||||
try:
|
||||
out = run_git(
|
||||
["log", "--format=%s", "--reverse", "--ancestry-path", rng], check=True
|
||||
)
|
||||
except GitError:
|
||||
# Brak ścieżki (np. tag nie jest potomkiem old_sha) – wtedy nie ma pośrednich
|
||||
return []
|
||||
msgs = [ln for ln in out.splitlines() if ln.strip()]
|
||||
if msgs:
|
||||
# Ostatnia pozycja może być new_sha (w praktyce git log na %s nie odróżnia, ale
|
||||
# jeśli zakres zwróci message z new_sha na końcu, usuniemy go porównując SHA).
|
||||
# Prostsze: usuń ostatni wpis, bo to najnowszy (new_sha).
|
||||
msgs = msgs[:-1]
|
||||
return msgs
|
||||
|
||||
|
||||
def checkout_orphan_branch(new_branch: str) -> None:
|
||||
run_git(["checkout", "--orphan", new_branch])
|
||||
# Usuń wszystko z indeksu i roboczego (z wyjątkiem .git)
|
||||
# Najpierw usuń śledzone:
|
||||
run_git(["rm", "-r", "--quiet", "--cached", "--force", "."], check=False)
|
||||
# Potem pliki robocze:
|
||||
for root, dirs, files in os.walk(".", topdown=False):
|
||||
# pomiń .git
|
||||
if root.startswith("./.git") or root == ".git":
|
||||
continue
|
||||
for name in files:
|
||||
try:
|
||||
os.remove(os.path.join(root, name))
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
for name in dirs:
|
||||
p = os.path.join(root, name)
|
||||
if p == "./.git":
|
||||
continue
|
||||
try:
|
||||
os.rmdir(p)
|
||||
except OSError:
|
||||
# Niepuste – to OK, wyczyścimy przy checkout
|
||||
pass
|
||||
|
||||
|
||||
def replace_worktree_with_commit(sha: str) -> None:
|
||||
"""
|
||||
Nadpisz zawartość roboczą drzewem z commit-a sha.
|
||||
"""
|
||||
# Najpierw usuń aktualne pliki (również nieśledzone), potem wczytaj tree wybranego commita:
|
||||
# 1) git checkout <sha> -- . (zapisze pliki do working tree + index)
|
||||
# 2) git add -A
|
||||
# Aby dopilnować usunięć: zrób czyszczenie przez git rm -r ., potem checkout.
|
||||
run_git(["rm", "-r", "--quiet", "--ignore-unmatch", "."], check=False)
|
||||
# Przywróć pliki ze wskazanego commita:
|
||||
# Uwaga: jeśli repo ma submoduły/large files – to wykracza poza zakres, ale zadziała dla standardowych plików.
|
||||
run_git(["checkout", sha, "--", "."])
|
||||
run_git(["add", "-A"])
|
||||
|
||||
|
||||
def build_commit_message_for_tag(tag: str, intermediates: List[str]) -> str:
|
||||
if not intermediates:
|
||||
return f"Tag: {tag}"
|
||||
lines = ["Tag: " + tag, "", "Intermediate commits (oldest → newest):"]
|
||||
lines += [f"- {m}" for m in intermediates]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def build_commit_message_for_head(since_tag: Optional[str], intermediates: List[str]) -> str:
|
||||
title = "HEAD (unreleased)"
|
||||
hdr = title if since_tag is None else f"{title} since tag {since_tag}"
|
||||
if not intermediates:
|
||||
return hdr
|
||||
lines = [hdr, "", "Intermediate commits (oldest → newest):"]
|
||||
lines += [f"- {m}" for m in intermediates]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def commit_all(message: str) -> None:
|
||||
# Zacommituj wszystko co w indeksie (po replace_worktree_with_commit daliśmy add -A)
|
||||
run_git(["commit", "-m", message])
|
||||
|
||||
|
||||
def push_new_branch(remote_url: str, branch: str) -> None:
|
||||
# Dodaj zdalny 'newrepo' jeśli nie istnieje, ustaw URL i wypchnij
|
||||
remotes = run_git(["remote"]).splitlines()
|
||||
if "newrepo" not in remotes:
|
||||
run_git(["remote", "add", "newrepo", remote_url])
|
||||
else:
|
||||
# podmień URL na wszelki wypadek
|
||||
run_git(["remote", "set-url", "newrepo", remote_url])
|
||||
run_git(["push", "-u", "newrepo", f"{branch}:{branch}"])
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Zbuduj nową gałąź na podstawie tagów z bieżącej gałęzi."
|
||||
)
|
||||
parser.add_argument("new_branch", help="Nazwa nowej gałęzi do utworzenia")
|
||||
parser.add_argument(
|
||||
"--new-repo-url",
|
||||
dest="new_repo_url",
|
||||
default=None,
|
||||
help="(Opcjonalnie) adres URL nowego zdalnego repo – zostanie dodany jako 'newrepo' i wykonany push nowej gałęzi.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-v", "--verbose", action="store_true", help="Bardziej gadatliwe logi"
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.verbose:
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
try:
|
||||
ensure_git_repo()
|
||||
ensure_clean_worktree()
|
||||
base_branch = current_branch()
|
||||
base_head = head_sha()
|
||||
logger.info("Bieżąca gałąź: %s", base_branch)
|
||||
logger.info("HEAD: %s", base_head[:12])
|
||||
|
||||
# Zbierz tagi osiągalne z bieżącej gałęzi
|
||||
merged_tags = tags_merged_into(base_branch)
|
||||
if not merged_tags:
|
||||
logger.warning(
|
||||
"Nie znaleziono tagów osiągalnych z bieżącej gałęzi. Gałąź zostanie zbudowana tylko z HEAD."
|
||||
)
|
||||
|
||||
# Posortuj tagi po czasie commita
|
||||
sorted_tags = sort_tags_by_commit_time(merged_tags)
|
||||
|
||||
# Sprawdź czy HEAD jest otagowany
|
||||
head_tags = sha_has_tag(base_head)
|
||||
head_is_tagged = bool(head_tags)
|
||||
|
||||
# Utwórz orphan branch
|
||||
logger.info("Tworzę sierocą gałąź: %s", args.new_branch)
|
||||
checkout_orphan_branch(args.new_branch)
|
||||
|
||||
prev_sha: Optional[str] = None
|
||||
last_tag_name: Optional[str] = None
|
||||
|
||||
# Dla każdego tagu – commit z jego zawartości
|
||||
for tag_name, tag_sha, _ts in sorted_tags:
|
||||
logger.info("Przetwarzam tag: %s (%s)", tag_name, tag_sha[:12])
|
||||
replace_worktree_with_commit(tag_sha)
|
||||
intermediates = list_intermediate_messages(prev_sha, tag_sha)
|
||||
msg = build_commit_message_for_tag(tag_name, intermediates)
|
||||
commit_all(msg)
|
||||
prev_sha = tag_sha
|
||||
last_tag_name = tag_name
|
||||
|
||||
# Jeśli HEAD nie jest dokładnie ostatnim tagiem – dorzuć "unreleased"
|
||||
if not head_is_tagged:
|
||||
logger.info("Dodaję końcowy commit z bieżącego HEAD (unreleased).")
|
||||
replace_worktree_with_commit(base_head)
|
||||
inter = list_intermediate_messages(prev_sha, base_head)
|
||||
msg = build_commit_message_for_head(last_tag_name, inter)
|
||||
commit_all(msg)
|
||||
else:
|
||||
logger.info("HEAD jest oznaczony tagiem – kończę na ostatnim tagu.")
|
||||
|
||||
# Push do nowego repo jeśli podano
|
||||
if args.new_repo_url:
|
||||
logger.info("Wypycham nową gałąź do: %s", args.new_repo_url)
|
||||
push_new_branch(args.new_repo_url, args.new_branch)
|
||||
|
||||
logger.info("Zakończono pomyślnie.")
|
||||
return 0
|
||||
|
||||
except GitError as ge:
|
||||
logger.error("Błąd gita: %s", ge)
|
||||
return 2
|
||||
except Exception as e:
|
||||
raise Exception("Nieoczekiwany błąd.") from e
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user