#!/usr/bin/env python3 # -*- coding: utf-8 -*- import argparse import os import sys import zipfile import tempfile import shutil import subprocess from datetime import datetime, timezone def eprint(*a, **kw): print(*a, file=sys.stderr, **kw) def is_git_repo(path: str) -> bool: return os.path.isdir(os.path.join(path, ".git")) def run(cmd, cwd=None, env=None, check=True, capture=True): result = subprocess.run( cmd, cwd=cwd, env=env, text=True, capture_output=capture, ) if check and result.returncode != 0: eprint(f"[BŁĄD] Komenda nie powiodła się: {' '.join(cmd)}") if capture: eprint(result.stdout) eprint(result.stderr) sys.exit(result.returncode) return result def list_zip_files(zip_dir: str): zips = [] for name in os.listdir(zip_dir): if not name.lower().endswith(".zip"): continue path = os.path.join(zip_dir, name) try: st = os.stat(path) except FileNotFoundError: continue zips.append((st.st_mtime, path)) zips.sort(key=lambda x: x[0]) # najstarsze -> najnowsze return [p for _, p in zips] def find_module_json(root: str): """Zwraca listę absolutnych ścieżek do plików module.json (rekurencyjnie), pomija .git.""" matches = [] for dirpath, dirnames, filenames in os.walk(root): if ".git" in dirnames: dirnames.remove(".git") if "module.json" in filenames: matches.append(os.path.join(dirpath, "module.json")) return matches def copy_contents(src_dir: str, dst_dir: str): """Kopiuje zawartość katalogu src_dir do dst_dir (łącznie z podkatalogami), nadpisując istniejące pliki.""" for item in os.listdir(src_dir): s = os.path.join(src_dir, item) d = os.path.join(dst_dir, item) if os.path.isdir(s): shutil.copytree(s, d, dirs_exist_ok=True) else: os.makedirs(os.path.dirname(d), exist_ok=True) shutil.copy2(s, d) def git_has_staged_changes(repo: str) -> bool: # Po "git add -A" sprawdzamy, czy coś jest w indexie res = subprocess.run(["git", "diff", "--cached", "--quiet"], cwd=repo) return res.returncode == 1 # 1 = są zmiany; 0 = brak; inne = błąd (ignorujemy tutaj) def git_is_clean(repo: str) -> bool: res = run(["git", "status", "--porcelain"], cwd=repo, capture=True, check=False) return res.stdout.strip() == "" def main(): parser = argparse.ArgumentParser( description="Import archiwów ZIP do repozytorium git (od najstarszego), z commitami po każdym ZIPie." ) parser.add_argument("--zip-dir", required=True, help="Katalog z plikami .zip") parser.add_argument("--repo", required=True, help="Katalog repozytorium git (cel)") parser.add_argument( "--allow-dirty", action="store_true", help="Pozwól działać, nawet jeśli repo nie jest czyste (domyślnie: wymagana czystość).", ) parser.add_argument( "--dry-run", action="store_true", help="Tylko pokaż plan (bez rozpakowywania i commitów).", ) args = parser.parse_args() zip_dir = os.path.abspath(args.zip_dir) repo = os.path.abspath(args.repo) if not os.path.isdir(zip_dir): eprint(f"[BŁĄD] Nie ma katalogu: {zip_dir}") sys.exit(1) if not os.path.isdir(repo): eprint(f"[BŁĄD] Nie ma katalogu repozytorium: {repo}") sys.exit(1) if not is_git_repo(repo): eprint(f"[BŁĄD] {repo} nie wygląda na repozytorium git (brak .git).") sys.exit(1) if not args.allow_dirty and not git_is_clean(repo): eprint("[BŁĄD] Repo nie jest czyste. Użyj --allow-dirty, jeśli chcesz mimo to kontynuować.") sys.exit(1) zips = list_zip_files(zip_dir) if not zips: print("[INFO] Brak plików .zip do przetworzenia.") return print(f"[INFO] Znaleziono {len(zips)} archiwów .zip do przetworzenia (od najstarszego do najnowszego).") for i, zip_path in enumerate(zips, start=1): zip_name = os.path.basename(zip_path) mtime = datetime.fromtimestamp(os.path.getmtime(zip_path), tz=timezone.utc) mtime_iso = mtime.strftime("%Y-%m-%dT%H:%M:%SZ") print(f"\n=== [{i}/{len(zips)}] {zip_name} (mtime: {mtime_iso}) ===") if args.dry_run: print(f"[DRY-RUN] Rozpakowałbym {zip_name} i zrobił commit.") continue # Rozpakuj najpierw do katalogu tymczasowego stage_dir = tempfile.mkdtemp(prefix="zipstage_") try: with zipfile.ZipFile(zip_path, "r") as zf: zf.extractall(stage_dir) except zipfile.BadZipFile: eprint(f"[BŁĄD] Uszkodzony zip: {zip_name}. Pomijam.") shutil.rmtree(stage_dir, ignore_errors=True) continue # BONUS: wykrywanie rekursywne struktury katalogów z module.json stage_root_mjson = os.path.join(stage_dir, "module.json") if os.path.isfile(stage_root_mjson): # idealnie: module.json już w korzeniu paczki effective_root = stage_dir print("[OK] Wykryto module.json w korzeniu archiwum.") else: matches = find_module_json(stage_dir) if len(matches) == 1: # jednoznaczny kandydat -> uznaj katalog z module.json za korzeń i skopiuj stamtąd effective_root = os.path.dirname(matches[0]) rel = os.path.relpath(effective_root, stage_dir) print(f"[BONUS] Wykryto zagnieżdżenie: module.json w '{rel}'. " f"Kopiuję zawartość tego katalogu do repozytorium.") else: # 0 lub >1 -> pauza i instrukcja ręczna print("[UWAGA] Nie wykryto jednoznacznie położenia module.json " "(0 lub wiele wystąpień).") if matches: print("Kandydaci:") for p in matches: print(f" - {os.path.relpath(p, stage_dir)}") print("\nProszę **ręcznie** przenieść/rozpakować pliki z poniższego katalogu tymczasowego") print(f"do właściwego miejsca w repo: {repo}") print(f"Katalog tymczasowy: {stage_dir}") print("Cel: po ułożeniu plików, w repo ma być poprawny układ projektu (typowo: module.json w korzeniu).") while True: try: choice = input("Po wykonaniu zmian naciśnij [Enter], aby kontynuować, " "albo wpisz 's' aby pominąć to archiwum, 'q' aby zakończyć: ").strip().lower() except KeyboardInterrupt: print("\nPrzerwano przez użytkownika.") sys.exit(130) if choice == "q": print("Kończę działanie na życzenie użytkownika.") sys.exit(0) if choice == "s": print("Pomijam to archiwum.") shutil.rmtree(stage_dir, ignore_errors=True) effective_root = None break # Enter -> spróbujmy zrobić commit tego, co jest w repo effective_root = None # nic nie kopiujemy automatycznie break # jeśli pominięto if effective_root is None and choice == "s": continue # Jeśli mamy effective_root (auto), kopiujemy zawartość do repo if effective_root: copy_contents(effective_root, repo) # Git add + commit (jeśli są zmiany) run(["git", "add", "-A"], cwd=repo) if git_has_staged_changes(repo): msg = f"Import {zip_name} ({mtime_iso})" env = os.environ.copy() # Ustaw datę commita na mtime pliku ZIP (ładny, historyczny timeline) env["GIT_AUTHOR_DATE"] = mtime_iso env["GIT_COMMITTER_DATE"] = mtime_iso run(["git", "commit", "-m", msg], cwd=repo, env=env) print(f"[OK] Commit: {msg}") else: print("[INFO] Brak zmian po rozpakowaniu — pomijam commit.") # Sprzątanie shutil.rmtree(stage_dir, ignore_errors=True) print("\n[ZAKOŃCZONE] Przetwarzanie archiwów .zip zakończone.") if __name__ == "__main__": main()