"""Выполнение очистки: удаление выбранных находок с журналом и защитой.""" from __future__ import annotations import fnmatch import json import os import time from dataclasses import dataclass, field from typing import Callable, Iterable from .model import Finding from .util import Cancel, clear_contents, delete_path, human, is_safe_target @dataclass class CleanOptions: to_recycle_bin: bool = True # False = удалять безвозвратно dry_run: bool = False stop_on_error: bool = False journal_dir: str = "" @dataclass class ItemResult: finding: Finding ok: bool freed: int = 0 removed: int = 0 skipped: int = 0 message: str = "" @dataclass class CleanResult: items: list[ItemResult] = field(default_factory=list) started: float = 0.0 finished: float = 0.0 cancelled: bool = False journal_path: str = "" @property def freed(self) -> int: return sum(i.freed for i in self.items) @property def ok_count(self) -> int: return sum(1 for i in self.items if i.ok) @property def fail_count(self) -> int: return sum(1 for i in self.items if not i.ok) def summary(self) -> str: dur = max(0.0, (self.finished or time.time()) - self.started) head = (f"Освобождено {human(self.freed)} · успешно {self.ok_count} · " f"с ошибками {self.fail_count} · {dur:.0f} с") if self.cancelled: head = "ОТМЕНЕНО. " + head return head def _free_space_bytes(paths: Iterable[str]) -> dict[str, int]: import shutil out: dict[str, int] = {} for p in paths: drive = os.path.splitdrive(os.path.abspath(p))[0] or "C:" root = drive + "\\" if root in out: continue try: out[root] = shutil.disk_usage(root).free except OSError: continue return out def _delete_masked_files(path: str, masks: list[str], recycle: bool) -> tuple[int, int, int, list[str]]: """Удалить в каталоге только файлы, подходящие под маски.""" freed = removed = skipped = 0 errs: list[str] = [] try: entries = list(os.scandir(path)) except OSError as e: return 0, 0, 0, [str(e)] for e in entries: try: if not e.is_file(follow_symlinks=False): continue if not any(fnmatch.fnmatch(e.name.lower(), m.lower()) for m in masks): continue size = e.stat(follow_symlinks=False).st_size except OSError: continue ok, err = delete_path(e.path, recycle) if ok: removed += 1 freed += size else: skipped += 1 if err and len(errs) < 10: errs.append(f"{e.name}: {err}") return freed, removed, skipped, errs def clean( findings: Iterable[Finding], opts: CleanOptions, cancel: Cancel | None = None, progress: Callable[[str, float], None] | None = None, ) -> CleanResult: """Удалить выбранные находки. Возвращает подробный результат по каждой.""" items = list(findings) res = CleanResult(started=time.time()) total = max(1, len(items)) before = _free_space_bytes(f.path for f in items) journal: list[dict] = [] for i, f in enumerate(items): if cancel is not None and cancel.cancelled: res.cancelled = True break if progress is not None: progress(f"{f.title} — {human(f.size)}", i / total) safe, why = is_safe_target(f.path) if not f.delete_contents_only and not safe: res.items.append(ItemResult(f, False, message=f"пропущено: {why}")) continue if not os.path.exists(f.path): res.items.append(ItemResult(f, True, message="уже отсутствует")) continue if opts.dry_run: res.items.append(ItemResult(f, True, freed=f.size, removed=f.files, message="пробный запуск — ничего не удалено")) continue masks = f.extra.get("file_masks") or [] try: if f.members: # группа дубликатов freed = removed = skipped = 0 errs: list[str] = [] unit = int(f.extra.get("unit") or 0) for member in f.members: ok, err = delete_path(member, opts.to_recycle_bin) if ok: removed += 1 freed += unit else: skipped += 1 if err: errs.append(f"{os.path.basename(member)}: {err}") item = ItemResult(f, skipped == 0, freed, removed, skipped, "; ".join(errs[:3])) elif masks: freed, removed, skipped, errs = _delete_masked_files( f.path, masks, opts.to_recycle_bin) item = ItemResult(f, skipped == 0, freed, removed, skipped, "; ".join(errs[:3])) elif f.delete_contents_only: done, skipped, errs = clear_contents(f.path, opts.to_recycle_bin) # оцениваем освобождённое по доле удалённых объектов share = done / max(1, done + skipped) item = ItemResult(f, skipped == 0, int(f.size * share), done, skipped, "; ".join(errs[:3])) else: ok, err = delete_path(f.path, opts.to_recycle_bin) item = ItemResult(f, ok, f.size if ok else 0, 1 if ok else 0, 0 if ok else 1, err) except Exception as e: # noqa: BLE001 item = ItemResult(f, False, message=str(e)) res.items.append(item) journal.append({ "time": time.strftime("%Y-%m-%d %H:%M:%S"), "path": f.path, "title": f.title, "category": f.category.label, "risk": f.risk.label, "mode": ("содержимое" if f.delete_contents_only else "копии" if f.members else "целиком"), "recycle_bin": opts.to_recycle_bin, "ok": item.ok, "freed": item.freed, "removed": item.removed, "skipped": item.skipped, "message": item.message, }) if opts.stop_on_error and not item.ok: break res.finished = time.time() # уточняем реально освобождённое место по свободному объёму дисков after = _free_space_bytes(f.path for f in items) res_real = sum(max(0, after.get(k, 0) - v) for k, v in before.items()) if opts.dry_run: res_real = 0 if journal and not opts.dry_run: res.journal_path = _write_journal(journal, opts, res, res_real) if progress is not None: progress("Очистка завершена", 1.0) return res def _write_journal(entries: list[dict], opts: CleanOptions, res: CleanResult, real_freed: int) -> str: base = opts.journal_dir or os.path.join( os.environ.get("LOCALAPPDATA", os.path.expanduser("~")), "SysDevCleaner", "journal") try: os.makedirs(base, exist_ok=True) name = time.strftime("clean-%Y%m%d-%H%M%S.json") path = os.path.join(base, name) with open(path, "w", encoding="utf-8") as fh: json.dump({ "started": res.started, "finished": res.finished, "to_recycle_bin": opts.to_recycle_bin, "estimated_freed": res.freed, "disk_delta": real_freed, "items": entries, }, fh, ensure_ascii=False, indent=2) return path except OSError: return "" def list_journals(limit: int = 30) -> list[tuple[str, str, int]]: """Последние журналы очистки: (путь, дата, освобождено).""" base = os.path.join(os.environ.get("LOCALAPPDATA", os.path.expanduser("~")), "SysDevCleaner", "journal") if not os.path.isdir(base): return [] out: list[tuple[str, str, int]] = [] try: names = sorted(os.listdir(base), reverse=True)[:limit] except OSError: return [] for n in names: p = os.path.join(base, n) try: with open(p, encoding="utf-8") as fh: data = json.load(fh) out.append((p, time.strftime("%d.%m.%Y %H:%M", time.localtime(data.get("started", 0))), int(data.get("estimated_freed", 0)))) except Exception: continue return out