conjurer/sync.py

#!/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())