#!/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 def git_is_clean(repo: str) -> bool: res = run(["git", "status", "--porcelain"], cwd=repo, capture=True, check=False) return res.stdout.strip() == "" def ensure_subdir(repo_root: str, subdir: str) -> str: """Zwraca absolutną ścieżkę docelowego podkatalogu w repo, tworząc go jeśli trzeba. Zabezpiecza przed wychodzeniem poza repo (..).""" target = repo_root if subdir: target = os.path.abspath(os.path.join(repo_root, subdir)) repo_root_abs = os.path.abspath(repo_root) # Bezpieczeństwo: podkatalog musi być wewnątrz repo if os.path.commonpath([target, repo_root_abs]) != repo_root_abs: eprint(f"[BŁĄD] Niedozwolona ścieżka podkatalogu: {subdir}") sys.exit(1) os.makedirs(target, exist_ok=True) return target 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( "--subdir", default="", help="Opcjonalny podkatalog w repo, do którego mają trafić pliki (utworzony jeśli nie istnieje).", ) 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) subdir = args.subdir.strip() 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) target_dir = ensure_subdir(repo, subdir) # gdzie właściwie kopiujemy zawartość ZIPów 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 rel_target = os.path.relpath(target_dir, repo) print(f"[INFO] Cel importu: {target_dir} (rel: {rel_target})") 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} do '{rel_target}' i zrobił commit.") continue # Rozpakuj do katalogu tymczasowego (stage) 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 module.json stage_root_mjson = os.path.join(stage_dir, "module.json") effective_root = None if os.path.isfile(stage_root_mjson): effective_root = stage_dir print("[OK] Wykryto module.json w korzeniu archiwum.") else: matches = find_module_json(stage_dir) if len(matches) == 1: 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 '{rel_target}'.") else: 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: {target_dir}") print(f"Katalog tymczasowy: {stage_dir}") print("Cel: po ułożeniu plików, w docelowym podkatalogu powinna być poprawna struktura (typowo: module.json w korzeniu).") choice = "" 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 in ("", "s"): break if choice == "s": print("Pomijam to archiwum.") shutil.rmtree(stage_dir, ignore_errors=True) continue # Enter -> nic nie kopiujemy automatycznie, zakładamy że użytkownik przeniósł co trzeba # Jeśli mamy effective_root (auto), kopiujemy zawartość do docelowego podkatalogu if effective_root: copy_contents(effective_root, target_dir) # Git add + commit (jeśli są zmiany) – commit z korzenia repo run(["git", "add", "-A"], cwd=repo) if git_has_staged_changes(repo): msg = f"Import {zip_name} do {rel_target} ({mtime_iso})" env = os.environ.copy() 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.") print(f"[INFO] Pliki zostały umieszczone w: {target_dir}") if __name__ == "__main__": main()