maxp/repoCat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

refactor(cli): simplify repository cleaning logic

Browse Source 
- Replaced direct filesystem and Git operations with catalog-based API
- Streamlined handling of repository properties and deletion logic
- Improved readability and maintainability by reducing redundancy
This commit is contained in:
0xMax42
2025-05-11 14:38:12 +02:00
parent d3ee7ee63d
commit 458d965062
1 changed files with 32 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

126
src/repocat/cli/clean.py
View File

@@ -1,65 +1,29 @@
import subprocess
from typer import Typer, Option, Argument
from pathlib import Path
import shutil
from datetime import datetime, timedelta
from typer import Typer, Option
from rich.console import Console
from rich.table import Table
import os
import shutil
from repocat.config import load_config
from repocat.models.catalog import RepoCatalogState
app = Typer()
console = Console()
def is_git_dirty(path: Path) -> bool:
try:
result = subprocess.run(
["git", "status", "--porcelain"],
cwd=path,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
)
return bool(result.stdout.strip())
except Exception:
return False
def is_git_unpushed(path: Path) -> bool:
try:
# Check if upstream exists
subprocess.run(["git", "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"],
cwd=path, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True)
result = subprocess.run(
["git", "log", "@{u}..HEAD", "--oneline"],
cwd=path,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
)
return bool(result.stdout.strip())
except Exception:
return False
def get_dir_size(path: Path) -> int:
"""Returns size in bytes."""
return sum(f.stat().st_size for f in path.rglob("*") if f.is_file())
def complete_categories(incomplete: str):
config = load_config()
return [c.name for c in config.categories if c.name.startswith(incomplete)]
@app.command("volatile")
def clean_volatile(
dry_run: bool = Option(False, "--dry-run", help="Nur anzeigen, was gelöscht würde"),
category: list[str] = Option(None, "--category", "-c", help="Nur bestimmte Kategorien reinigen", autocompletion=complete_categories),
):
"""
Löscht **bedingungslos** alle Repositories aus `volatile`-Kategorien (z. B. `scratches`).
Löscht **bedingungslos** alle Repositories aus `volatile`-Kategorien (z.\u200bB. `scratches`).
"""
config = load_config()
base = config.base_dir
catalog = RepoCatalogState.from_config(load_config())
table = Table(title="Clean – Volatile (Zerstörung)")
table.add_column("Kategorie")
@@ -70,37 +34,31 @@ def clean_volatile(
deleted = 0
skipped = 0
for cat in config.categories:
if not cat.volatile:
for cat in catalog.categories:
if not cat.config.volatile:
continue
if category and cat.name not in category:
if category and cat.config.name not in category:
continue
path = base / cat.subdir
if not path.exists():
continue
for repo in sorted(path.iterdir()):
if not repo.is_dir():
continue
size_mb = get_dir_size(repo) / 1024 / 1024
for repo in cat.repos:
size_mb = repo.size_mb
if dry_run:
action = "[yellow]Würde löschen[/]"
else:
try:
shutil.rmtree(repo)
shutil.rmtree(repo.path)
action = "[red]Gelöscht[/]"
deleted += 1
except Exception as e:
action = f"[red]Fehler: {e}[/]"
table.add_row(cat.name, repo.name, f"{size_mb:.1f}", action)
table.add_row(cat.config.name, repo.name, f"{size_mb:.1f}", action)
console.print(table)
console.print(f"\n[green]✓ Volatile-Clean abgeschlossen[/] – {deleted} gelöscht, {skipped} übersprungen")
@app.command()
def run(
dry_run: bool = Option(False, "--dry-run", help="Nur anzeigen, was gelöscht würde"),
@@ -109,11 +67,7 @@ def run(
"""
Löscht alte oder volatile Repositories aus den konfigurierten Kategorien.
"""
config = load_config()
now = datetime.now()
base = config.base_dir
console.print(f"[bold yellow]Base-Verzeichnis:[/] {base}")
catalog = RepoCatalogState.from_config(load_config())
table = Table(title="Clean-Ergebnis")
table.add_column("Kategorie")
@@ -125,43 +79,27 @@ def run(
deleted = 0
skipped = 0
for cat in config.categories:
if category and cat.name not in category:
for cat in catalog.categories:
if category and cat.config.name not in category:
continue
if cat.config.is_archive:
continue
if cat.is_archive: continue
path = base / cat.subdir
if not path.exists():
continue
cutoff = now - timedelta(days=cat.days or 0)
for repo in sorted(path.iterdir()):
if not repo.is_dir():
continue
mtime = datetime.fromtimestamp(repo.stat().st_mtime)
age_days = (now - mtime).days
size_mb = get_dir_size(repo) / 1024 / 1024
limit_days = cat.config.days or 0
for repo in cat.repos:
should_delete = (
(cat.volatile)
or (cat.days is not None and mtime < cutoff)
cat.config.volatile
or (cat.config.days is not None and repo.is_expired(limit_days))
)
# Git-Schutz prüfen
is_git = (repo / ".git").exists()
dirty = is_git and is_git_dirty(repo)
unpushed = is_git and is_git_unpushed(repo)
if is_git and (dirty or unpushed):
reason = []
if dirty:
reason.append("uncommitted")
if unpushed:
reason.append("unpushed")
status = "/".join(reason)
if repo.git and repo.git.is_protected:
reasons = []
if repo.git.dirty:
reasons.append("uncommitted")
if repo.git.unpushed:
reasons.append("unpushed")
status = "/".join(reasons)
action = f"[blue]Geschützt ({status})[/]"
skipped += 1
else:
@@ -170,7 +108,7 @@ def run(
action = "[yellow]Würde löschen[/]"
else:
try:
shutil.rmtree(repo)
shutil.rmtree(repo.path)
action = "[red]Gelöscht[/]"
deleted += 1
except Exception as e:
@@ -179,7 +117,7 @@ def run(
action = "[cyan]Keine Aktion[/]"
skipped += 1
table.add_row(cat.name, repo.name, str(age_days), f"{size_mb:.1f}", action)
table.add_row(cat.config.name, repo.name, str(repo.age_days), f"{repo.size_mb:.1f}", action)
console.print(table)
console.print(f"\n[green]✓ Clean abgeschlossen[/] – {deleted} gelöscht, {skipped} übersprungen")