From ddb8292186f4e74b8558ab261f5fc84a91b22689 Mon Sep 17 00:00:00 2001 From: Michal Tuszowski Date: Wed, 17 Sep 2025 23:58:30 +0200 Subject: [PATCH] a --- import.py | 175 +++++++++++++++++++++++++++++++----------------------- 1 file changed, 102 insertions(+), 73 deletions(-) diff --git a/import.py b/import.py index 4cb8324..2999365 100644 --- a/import.py +++ b/import.py @@ -9,6 +9,7 @@ import tempfile import shutil import subprocess from datetime import datetime, timezone +from collections import defaultdict def eprint(*a, **kw): print(*a, file=sys.stderr, **kw) @@ -56,27 +57,6 @@ def find_module_json(root: str): 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 (..).""" @@ -84,13 +64,93 @@ def ensure_subdir(repo_root: str, subdir: str) -> str: 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 build_repo_index(repo_root: str): + """Mapa: basename -> [abspathy] (bez .git).""" + index = defaultdict(list) + for dirpath, dirnames, filenames in os.walk(repo_root): + if ".git" in dirnames: + dirnames.remove(".git") + for f in filenames: + index[f].append(os.path.join(dirpath, f)) + return index + +def git_mv_or_rename(repo_root: str, src_abs: str, dest_abs: str): + """Przenosi plik używając git mv, a w razie potrzeby zwykłego move. Tworzy katalog docelowy.""" + if os.path.abspath(src_abs) == os.path.abspath(dest_abs): + return + os.makedirs(os.path.dirname(dest_abs), exist_ok=True) + rel_src = os.path.relpath(src_abs, repo_root) + rel_dest = os.path.relpath(dest_abs, repo_root) + # Spróbuj git mv (utrzymanie historii) + res = subprocess.run(["git", "mv", "-f", rel_src, rel_dest], cwd=repo_root) + if res.returncode != 0: + # Jeśli plik nie był w git lub inny problem — zwykłe przeniesienie + shutil.move(src_abs, dest_abs) + +def mirror_zip_into_target(zip_root: str, target_dir: str, repo_root: str, dry_run: bool): + """Odwzorowuje strukturę zip_root wewnątrz target_dir. + - Jeśli plik o tej samej nazwie istnieje w repo w innej ścieżce i dest nie istnieje -> przenosimy (git mv). + - W przeciwnym razie kopiujemy (nadpisując). + """ + repo_index = build_repo_index(repo_root) + + for dirpath, dirnames, filenames in os.walk(zip_root): + rel_dir = os.path.relpath(dirpath, zip_root) + dest_dir = target_dir if rel_dir in (".", "") else os.path.join(target_dir, rel_dir) + if dry_run: + print(f"[PLAN] Utworzę katalog: {os.path.relpath(dest_dir, repo_root)}") + else: + os.makedirs(dest_dir, exist_ok=True) + + for fn in filenames: + src = os.path.join(dirpath, fn) + dest = os.path.join(dest_dir, fn) + + # 1) Unikaj duplikatów: jeśli dest nie istnieje, a w repo jest dokładnie jeden kandydat o tej samej nazwie -> przenieś + if not os.path.exists(dest): + candidates = [p for p in repo_index.get(fn, []) if os.path.abspath(p) != os.path.abspath(dest)] + # Odfiltruj ewentualne kandydaty wewnątrz .git (i tak ich nie było dodanych) + candidates = [p for p in candidates if "/.git/" not in p.replace("\\", "/")] + if len(candidates) == 1: + prev = candidates[0] + rel_prev = os.path.relpath(prev, repo_root) + rel_new = os.path.relpath(dest, repo_root) + if dry_run: + print(f"[MOVE] {rel_prev} -> {rel_new}") + else: + git_mv_or_rename(repo_root, prev, dest) + # aktualizuj indeks + repo_index[fn] = [dest if p == prev else p for p in repo_index.get(fn, [])] + elif len(candidates) > 1: + # Nie zgadujemy, żeby nie zrobić złej migracji — kopiujemy i ostrzegamy + print(f"[INFO] Wiele kandydatów o nazwie '{fn}' w repo ({len(candidates)}). " + f"Kopiuję do {os.path.relpath(dest, repo_root)} (bez przenoszenia).") + + # 2) Nadpisz/utwórz plik zawartością z ZIPa (repo ma odzwierciedlać ZIP) + if dry_run: + print(f"[COPY] {os.path.relpath(src, zip_root)} -> {os.path.relpath(dest, repo_root)}") + else: + os.makedirs(os.path.dirname(dest), exist_ok=True) + shutil.copy2(src, dest) + + # 3) Upewnij się, że indeks widzi plik w nowym miejscu (dla kolejnych kolizji) + if dest not in repo_index.get(fn, []): + repo_index[fn].append(dest) + +def git_has_staged_changes(repo: str) -> bool: + 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 main(): parser = argparse.ArgumentParser( description="Import archiwów ZIP do repozytorium git (od najstarszego), z commitami po każdym ZIPie." @@ -128,7 +188,7 @@ def main(): 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 + target_dir = ensure_subdir(repo, subdir) 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ć.") @@ -147,14 +207,13 @@ def main(): 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.") + print(f"[DRY-RUN] Odtworzyłbym strukturę {zip_name} w '{rel_target}', " + f"z przenosinami plików o tej samej nazwie.") continue - # Rozpakuj do katalogu tymczasowego (stage) stage_dir = tempfile.mkdtemp(prefix="zipstage_") try: with zipfile.ZipFile(zip_path, "r") as zf: @@ -164,53 +223,24 @@ def main(): 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.") + # Ustal punkt wejścia importu + matches = find_module_json(stage_dir) + if len(matches) == 1: + zip_root = os.path.dirname(matches[0]) + rel = os.path.relpath(zip_root, stage_dir) + print(f"[OK] Wykryto module.json w '{rel}'. Odtwarzam strukturę tego katalogu.") + elif len(matches) == 0: + zip_root = stage_dir + print("[INFO] Brak module.json. Odtwarzam pełną strukturę ZIPa w repo (merge z istniejącymi katalogami).") 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) + # wiele module.json — to ryzykowne; pozostawiamy do ręcznej interwencji + print("[UWAGA] Wykryto wiele plików module.json w archiwum. " + "Ta sytuacja jest niejednoznaczna — pomijam ten ZIP.") + shutil.rmtree(stage_dir, ignore_errors=True) + continue - 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) + # Mirror + przenosiny + mirror_zip_into_target(zip_root, target_dir, repo, dry_run=False) # Git add + commit (jeśli są zmiany) – commit z korzenia repo run(["git", "add", "-A"], cwd=repo) @@ -224,11 +254,10 @@ def main(): 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}") + print(f"[INFO] Pliki zostały odzwierciedlone w: {target_dir}") if __name__ == "__main__": main()