From ee99fc957dfb057e5b1911309e0d076c80e6bf5f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jan=20Doubravsk=C3=BD?= Date: Mon, 29 Jun 2026 15:33:44 +0200 Subject: [PATCH] Add "Tipy dne" folder and fix stale genre folder cleanup --- CHANGELOG.md | 15 +++ pyproject.toml | 2 +- scripts/filter_magnets_gui.py | 87 --------------- scripts/rargb_magnets.py | 196 --------------------------------- src/core/hardlink_manager.py | 95 ++++++++++++++-- src/ui/qt_app.py | 15 ++- tests/test_hardlink_manager.py | 57 +++++++++- 7 files changed, 165 insertions(+), 302 deletions(-) delete mode 100644 scripts/filter_magnets_gui.py delete mode 100644 scripts/rargb_magnets.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a6925de..d86864d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,21 @@ Each version entry uses these sections (include only those that apply): ## Unreleased +## 1.6.0 — 2026-06-16 + +### Added +- Filmotéka generation now also creates a **"Tipy dne"** folder with **10 random + pool movies** (hardlinks; copy-as-is folders like Seriály are excluded). The + folder is refreshed each run (`HardlinkManager.generate_random_tips`). + +### Fixed +- Stale **root-level genre folders** were left in the Filmotéka output when the + last movie carrying a genre dropped that tag: obsolete-link cleanup only + scanned genre folders still present on current movies, so an emptied one was + never visited. Cleanup now also sweeps other root-level folders (excluding the + grouping roots and `reserved_subfolders` — copy-as-is mirrors, "Tipy dne") and + removes the now-empty folder. Grouping folders (Dle roku/…) were already fine. + ## 1.5.0 — 2026-06-16 ### Added diff --git a/pyproject.toml b/pyproject.toml index 59d50ee..4dfbbc6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "curator" -version = "1.5.0" +version = "1.6.0" description = "" authors = [ {name = "jan.doubravsky@gmail.com"} diff --git a/scripts/filter_magnets_gui.py b/scripts/filter_magnets_gui.py deleted file mode 100644 index 9604757..0000000 --- a/scripts/filter_magnets_gui.py +++ /dev/null @@ -1,87 +0,0 @@ -#!/usr/bin/env python3 -"""Minimal PySide6 GUI for filtering magnet lists from ``rargb_magnets.py``. - -Just a text box on top and a list below — type to filter live (same syntax as -the CLI: space-separated AND terms, ``-term`` to exclude). Double-click or press -Enter on a row to copy its magnet link to the clipboard. - - python tools/filter_magnets_gui.py [files/glob/dir ...] - -With no arguments it loads ``magnets_*.txt`` from the current directory. The -loading/filtering logic is reused from ``filter_magnets.py`` in this folder. -""" - -from __future__ import annotations - -import sys -from pathlib import Path - -# Reuse the CLI tool's parsing/filtering (same folder). -sys.path.insert(0, str(Path(__file__).resolve().parent)) -from filter_magnets import Entry, load_entries, apply_filter, resolve_inputs # noqa: E402 - -from PySide6.QtCore import Qt # noqa: E402 -from PySide6.QtWidgets import ( # noqa: E402 - QApplication, QWidget, QVBoxLayout, QLineEdit, QListWidget, QListWidgetItem, -) - - -class MagnetFilter(QWidget): - def __init__(self, entries: list[Entry]) -> None: - super().__init__() - self.entries = entries - - layout = QVBoxLayout(self) - layout.setContentsMargins(6, 6, 6, 6) - - self.search = QLineEdit() - self.search.setPlaceholderText("filtr… (např. 1080p 2022 -hindi) — ↵/dvojklik = kopírovat magnet") - self.search.setClearButtonEnabled(True) - self.search.textChanged.connect(self._refilter) - layout.addWidget(self.search) - - self.list = QListWidget() - self.list.itemActivated.connect(self._copy) # Enter / double-click - layout.addWidget(self.list) - - self.resize(820, 600) - self._refilter("") - self.search.setFocus() - - def _refilter(self, text: str) -> None: - self.list.clear() - for entry in apply_filter(self.entries, text): - short = entry.magnet.split("&", 1)[0] # only the part before the first & - item = QListWidgetItem(f"{entry.name}\n{short}") - item.setData(Qt.UserRole, short) - item.setToolTip(short) - self.list.addItem(item) - self._update_title() - - def _copy(self, item: QListWidgetItem) -> None: - QApplication.clipboard().setText(item.data(Qt.UserRole)) - self._update_title(copied=item.text()) - - def _update_title(self, copied: str | None = None) -> None: - base = f"Magnet filtr — {self.list.count()} / {len(self.entries)}" - self.setWindowTitle(f"{base} ✓ zkopírováno" if copied else base) - - -def main() -> None: - paths = [p for p in resolve_inputs(sys.argv[1:]) if p.exists()] - if not paths: - print("Žádné vstupní soubory (magnets_*.txt) nenalezeny.", file=sys.stderr) - sys.exit(1) - entries = load_entries(paths) - if not entries: - print("Vstupní soubory neobsahují žádné magnet odkazy.", file=sys.stderr) - sys.exit(1) - - app = QApplication(sys.argv) - window = MagnetFilter(entries) - window.show() - sys.exit(app.exec()) - - -if __name__ == "__main__": - main() diff --git a/scripts/rargb_magnets.py b/scripts/rargb_magnets.py deleted file mode 100644 index f96e853..0000000 --- a/scripts/rargb_magnets.py +++ /dev/null @@ -1,196 +0,0 @@ -#!/usr/bin/env python3 -"""Standalone scraper: collect magnet links from a rargb.to search. - -Given a search query it walks every results page -(``https://rargb.to/search/?search=`` and ``/search//?search=``), -opens each torrent's detail page and saves its magnet link. - -This is a self-contained tool — it only needs ``requests`` and -``beautifulsoup4`` and does not import anything from the Curator project. - -Examples: - python scripts/rargb_magnets.py "ubuntu 24.04" - python scripts/rargb_magnets.py test --output test_magnets.txt --max-pages 3 - python scripts/rargb_magnets.py test --tsv # also write namemagnet - -Be considerate: a polite delay is inserted between requests by default. Use the -results responsibly and respect the target site's terms and your local law. -""" - -from __future__ import annotations - -import re -import sys -import time -import argparse -from pathlib import Path -from urllib.parse import quote, urljoin - -import requests -from bs4 import BeautifulSoup - -BASE_URL = "https://rargb.to" -HEADERS = { - "User-Agent": ( - "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " - "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" - ), - "Accept-Language": "en-US,en;q=0.9", -} -MAGNET_RE = re.compile(r"magnet:\?[^\"'\s<>]+") - - -def search_page_url(query: str, page: int) -> str: - """URL of the N-th results page for a query (page 1 has no number).""" - q = quote(query) - if page <= 1: - return f"{BASE_URL}/search/?search={q}" - return f"{BASE_URL}/search/{page}/?search={q}" - - -def fetch(session: requests.Session, url: str, timeout: float, retries: int) -> str | None: - """GET ``url`` and return the HTML, or None after exhausting retries.""" - for attempt in range(1, retries + 1): - try: - resp = session.get(url, headers=HEADERS, timeout=timeout) - resp.raise_for_status() - return resp.text - except requests.RequestException as exc: - wait = attempt * 2 - print(f" ! chyba ({attempt}/{retries}) u {url}: {exc} — čekám {wait}s", - file=sys.stderr) - time.sleep(wait) - return None - - -def parse_result_links(html: str) -> list[tuple[str, str]]: - """Return (name, detail_url) for each result row on a search page.""" - soup = BeautifulSoup(html, "html.parser") - results: list[tuple[str, str]] = [] - seen: set[str] = set() - for row in soup.select("tr.lista2"): - link = row.find("a", href=re.compile(r"^/torrent/")) - if not link: - continue - href = link.get("href") - if not href or href in seen: - continue - seen.add(href) - name = link.get("title") or link.get_text(strip=True) or href - results.append((name.strip(), urljoin(BASE_URL, href))) - return results - - -def parse_last_page(html: str) -> int: - """Best-effort highest page number from the pager (1 if none found).""" - pages = [int(n) for n in re.findall(r"/search/(\d+)/\?search=", html)] - return max(pages) if pages else 1 - - -def extract_magnet(html: str) -> str | None: - """First magnet link found on a torrent detail page, or None.""" - match = MAGNET_RE.search(html) - return match.group(0) if match else None - - -def scrape(query: str, max_pages: int | None, delay: float, - timeout: float, retries: int) -> list[tuple[str, str]]: - """Walk all result pages and return a de-duplicated [(name, magnet)] list.""" - session = requests.Session() - collected: list[tuple[str, str]] = [] - seen_magnets: set[str] = set() - seen_details: set[str] = set() - - first_html = fetch(session, search_page_url(query, 1), timeout, retries) - if first_html is None: - print("Nepodařilo se načíst první stránku výsledků.", file=sys.stderr) - return collected - - last_page = parse_last_page(first_html) - if max_pages is not None: - last_page = min(last_page, max_pages) - print(f"Dotaz: {query!r} — stránek k projití: ~{last_page}") - - page = 1 - while True: - html = first_html if page == 1 else fetch( - session, search_page_url(query, page), timeout, retries) - if html is None: - break - - rows = parse_result_links(html) - new_rows = [(n, u) for n, u in rows if u not in seen_details] - if not new_rows: - # No fresh results → past the last real page; stop. - break - - print(f"[strana {page}] nalezeno položek: {len(new_rows)}") - for name, detail_url in new_rows: - seen_details.add(detail_url) - time.sleep(delay) - detail_html = fetch(session, detail_url, timeout, retries) - if detail_html is None: - print(f" - {name}: detail se nenačetl", file=sys.stderr) - continue - magnet = extract_magnet(detail_html) - if not magnet: - print(f" - {name}: magnet nenalezen", file=sys.stderr) - continue - if magnet in seen_magnets: - continue - seen_magnets.add(magnet) - collected.append((name, magnet)) - print(f" + {name}") - - if max_pages is not None and page >= max_pages: - break - page += 1 - if page > last_page: - # Probe one page past the detected last page in case the pager was - # windowed; the empty-results check above will stop us if it's truly - # the end. - last_page = page - time.sleep(delay) - - return collected - - -def main() -> None: - parser = argparse.ArgumentParser( - description="Vyparsuje magnet odkazy z vyhledávání na rargb.to.") - parser.add_argument("query", help="Vyhledávací dotaz (např. \"ubuntu 24.04\")") - parser.add_argument("-o", "--output", type=Path, - help="Výstupní soubor (výchozí: magnets_.txt)") - parser.add_argument("--max-pages", type=int, default=None, - help="Maximální počet stránek (výchozí: všechny)") - parser.add_argument("--delay", type=float, default=1.0, - help="Prodleva mezi requesty v sekundách (výchozí: 1.0)") - parser.add_argument("--timeout", type=float, default=20.0, - help="Timeout requestu v sekundách (výchozí: 20)") - parser.add_argument("--retries", type=int, default=3, - help="Počet pokusů při chybě (výchozí: 3)") - parser.add_argument("--tsv", action="store_true", - help="Uložit i \\t vedle čistých magnetů") - args = parser.parse_args() - - output = args.output or Path( - f"magnets_{re.sub(r'[^A-Za-z0-9._-]+', '_', args.query).strip('_')}.txt") - - results = scrape(args.query, args.max_pages, args.delay, args.timeout, args.retries) - - if not results: - print("Nenalezeny žádné magnet odkazy.") - sys.exit(1) - - output.write_text("".join(f"{magnet}\n" for _, magnet in results), encoding="utf-8") - print(f"\nUloženo {len(results)} magnet odkazů do: {output}") - - if args.tsv: - tsv_path = output.with_suffix(".tsv") - tsv_path.write_text( - "".join(f"{name}\t{magnet}\n" for name, magnet in results), encoding="utf-8") - print(f"Uloženo také název+magnet do: {tsv_path}") - - -if __name__ == "__main__": - main() diff --git a/src/core/hardlink_manager.py b/src/core/hardlink_manager.py index c8a8d89..8c00d40 100644 --- a/src/core/hardlink_manager.py +++ b/src/core/hardlink_manager.py @@ -18,6 +18,7 @@ Example: └── film.mkv (hardlink) """ import os +import random from pathlib import Path from typing import List, Tuple, Optional, Dict, Set from .file import File @@ -115,25 +116,40 @@ class HardlinkManager: def _managed_top_dirs( self, files: List[File], roots: Optional[Dict[str, str]], transforms: Optional[Dict[str, str]] = None, + reserved_subfolders: Optional[Set[str]] = None, ) -> Optional[Set[str]]: """Top-level output folders owned by the tag tree (None = all of them). - For a category with a non-empty root the root folder is owned; for a - category placed at the output root (empty root, e.g. genres) each of its - (transformed) tag values is its own top-level folder. This lets cleanup - skip unrelated root entries such as the copy-as-is mirror (Seriály). + A category with a non-empty root owns that root folder. A category placed + at the output root (empty root, e.g. genres) owns its own folders at the + root — and, so a genre whose last movie dropped the tag still gets its + now-stale folder cleaned, *every* root-level folder is treated as owned + except the grouping roots and any ``reserved_subfolders`` (copy-as-is + mirrors, "Tipy dne", …). The obsolete check still only removes links to + current pool files, so reserved/foreign folders are doubly protected. """ if roots is None: return None - tops: Set[str] = set() + grouping_roots = {folder for folder in roots.values() if folder} + has_root_level = any(not folder for folder in roots.values()) + tops: Set[str] = set(grouping_roots) + + # genre folders currently present on the movies for cat, folder in roots.items(): - if folder: - tops.add(folder) - else: + if not folder: for file_obj in files: for tag in file_obj.tags: if tag.category == cat: tops.add(self._folder_value(tag, transforms)) + + # plus any other root-level folder (catches genres that lost their last + # movie), minus grouping roots and reserved (mirrors / Tipy dne / …) + if has_root_level and self.output_dir.exists(): + reserved = set(reserved_subfolders or set()) + for entry in self.output_dir.iterdir(): + if (entry.is_dir() and entry.name not in grouping_roots + and entry.name not in reserved): + tops.add(entry.name) return tops def create_structure_for_files( @@ -258,6 +274,53 @@ class HardlinkManager: return success_count, fail_count + def generate_random_tips( + self, + files: List[File], + count: int = 10, + subfolder: str = "Tipy dne", + dry_run: bool = False, + ) -> Tuple[int, int]: + """(Re)fill ``subfolder`` with ``count`` random files as hardlinks. + + The folder is emptied first so each run picks a fresh random selection. + ``files`` should be the pool movies only (copy-as-is mirrors excluded). + + Returns: + Tuple of (created_links, failed_links) + """ + base = self.output_dir / subfolder + + if not dry_run and base.exists(): + for child in base.iterdir(): + if child.is_file(): + try: + child.unlink() + except OSError: + pass + + chosen = random.sample(files, min(count, len(files))) + created = 0 + fail = 0 + for file_obj in chosen: + target = base / file_obj.filename + try: + if not dry_run: + base.mkdir(parents=True, exist_ok=True) + if target.exists(): + if self._is_same_file(file_obj.file_path, target): + created += 1 + continue + target = self._get_unique_name(target) + os.link(file_obj.file_path, target) + self.created_links.append(target) + created += 1 + except OSError as e: + self.errors.append((file_obj.file_path, str(e))) + fail += 1 + + return created, fail + def _is_same_file(self, path1: Path, path2: Path) -> bool: """Check if two paths point to the same file (same inode).""" try: @@ -359,6 +422,7 @@ class HardlinkManager: category_roots: Optional[Dict[str, str]] = None, category_transforms: Optional[Dict[str, str]] = None, category_filename_templates: Optional[Dict[str, str]] = None, + reserved_subfolders: Optional[Set[str]] = None, ) -> List[Tuple[Path, Path]]: """ Find hardlinks in the output directory that no longer match file tags. @@ -411,7 +475,8 @@ class HardlinkManager: continue # Scan only the tag-tree's own top-level folders (skip copy-as-is mirrors) - top_dirs = self._managed_top_dirs(files, roots, category_transforms) + top_dirs = self._managed_top_dirs( + files, roots, category_transforms, reserved_subfolders) for top in self.output_dir.iterdir(): if not top.is_dir(): continue @@ -441,6 +506,7 @@ class HardlinkManager: category_roots: Optional[Dict[str, str]] = None, category_transforms: Optional[Dict[str, str]] = None, category_filename_templates: Optional[Dict[str, str]] = None, + reserved_subfolders: Optional[Set[str]] = None, ) -> Tuple[int, List[Path]]: """ Remove hardlinks that no longer match file tags. @@ -453,13 +519,15 @@ class HardlinkManager: ``categories`` when given). category_transforms: Optional category → transform-name map for folders. category_filename_templates: Optional category → hardlink-name template. + reserved_subfolders: Output top-level folders to never touch (mirrors, + "Tipy dne", …). Returns: Tuple of (removed_count, list_of_removed_paths) """ obsolete = self.find_obsolete_links( files, categories, category_roots, category_transforms, - category_filename_templates) + category_filename_templates, reserved_subfolders) removed_paths = [] if dry_run: @@ -485,6 +553,7 @@ class HardlinkManager: category_roots: Optional[Dict[str, str]] = None, category_transforms: Optional[Dict[str, str]] = None, category_filename_templates: Optional[Dict[str, str]] = None, + reserved_subfolders: Optional[Set[str]] = None, ) -> Tuple[int, int, int, int]: """ Synchronize hardlink structure with current file tags. @@ -501,6 +570,8 @@ class HardlinkManager: ``categories`` when given). category_transforms: Optional category → transform-name map for folders. category_filename_templates: Optional category → hardlink-name template. + reserved_subfolders: Output top-level folders to never touch (mirrors, + "Tipy dne", …). Returns: Tuple of (created, create_failed, removed, remove_failed) @@ -508,12 +579,12 @@ class HardlinkManager: # First find how many obsolete links there are obsolete_count = len(self.find_obsolete_links( files, categories, category_roots, category_transforms, - category_filename_templates)) + category_filename_templates, reserved_subfolders)) # Remove obsolete links removed, removed_paths = self.remove_obsolete_links( files, categories, dry_run, category_roots, category_transforms, - category_filename_templates + category_filename_templates, reserved_subfolders ) remove_failed = obsolete_count - removed if not dry_run else 0 diff --git a/src/ui/qt_app.py b/src/ui/qt_app.py index cc6aa6a..bc99a25 100644 --- a/src/ui/qt_app.py +++ b/src/ui/qt_app.py @@ -957,12 +957,14 @@ class QtApp(QMainWindow): QMessageBox.information(self, "Filmotéka", "Pool je prázdný.") return manager = HardlinkManager(out) + reserved = set(self.filehandler.copyasis_folders) | {"Tipy dne"} created, create_fail, removed, remove_fail = manager.sync_structure( files, category_roots=self.filehandler.filmoteka_category_roots(), category_transforms=self.filehandler.filmoteka_category_transforms(), category_filename_templates=( self.filehandler.filmoteka_category_filename_templates()), + reserved_subfolders=reserved, ) # Copy-as-is folders (e.g. Seriály): mirror each 1:1 (hardlinked) @@ -976,17 +978,22 @@ class QtApp(QMainWindow): mirrored += m_created mirror_fail += m_failed + # "Tipy dne": 10 random pool movies (copy-as-is folders excluded) + tips, tips_fail = manager.generate_random_tips(files, count=10) + msg = ( f"Filmy — vytvořeno: {created}, odebráno zastaralých: {removed}\n" - f"Copy-as-is — zrcadleno: {mirrored}" + f"Copy-as-is — zrcadleno: {mirrored}\n" + f"Tipy dne — náhodně: {tips}" ) - if create_fail or remove_fail or mirror_fail: - msg += f"\nSelhalo: {create_fail + remove_fail + mirror_fail}" + if create_fail or remove_fail or mirror_fail or tips_fail: + msg += f"\nSelhalo: {create_fail + remove_fail + mirror_fail + tips_fail}" QMessageBox.warning(self, "Filmotéka dokončena s chybami", msg) else: QMessageBox.information(self, "Filmotéka vygenerována", msg) self.status.showMessage( - f"Filmotéka: filmy +{created}/-{removed}, copy-as-is +{mirrored}", 5000 + f"Filmotéka: filmy +{created}/-{removed}, copy-as-is +{mirrored}, " + f"tipy +{tips}", 5000 ) def edit_copyasis_folders(self) -> None: diff --git a/tests/test_hardlink_manager.py b/tests/test_hardlink_manager.py index 030a9a2..5546f69 100644 --- a/tests/test_hardlink_manager.py +++ b/tests/test_hardlink_manager.py @@ -136,11 +136,30 @@ class TestHardlinkManager: mirror_link = mirror / "file1.txt" os.link(temp_source_dir / "file1.txt", mirror_link) - manager.sync_structure(files_with_tags, category_roots=roots) + manager.sync_structure( + files_with_tags, category_roots=roots, + reserved_subfolders={"Seriály"}) - # The mirror (not a managed tag folder) is left alone + # The mirror (reserved folder) is left alone assert mirror_link.exists() + def test_sync_removes_root_genre_folder_when_last_movie_drops_tag( + self, temp_source_dir, temp_output_dir, tag_manager + ): + """A genre folder at the output root is cleaned when no movie has it.""" + f = File(temp_source_dir / "file1.txt", tag_manager) + f.tags.clear() + f.add_tag(Tag("žánr", "Akční")) + roots = {"žánr": ""} + manager = HardlinkManager(temp_output_dir) + manager.sync_structure([f], category_roots=roots) + assert (temp_output_dir / "Akční").is_dir() + + # last movie with the tag loses it → folder must go + f.tags.clear() + manager.sync_structure([f], category_roots=roots) + assert not (temp_output_dir / "Akční").exists() + def test_category_transform_groups_folder_by_band( self, temp_source_dir, temp_output_dir, tag_manager ): @@ -202,6 +221,40 @@ class TestHardlinkManager: assert [p.name for p in folder.iterdir()] == ["1962 - Dr. No.txt"] assert removed == 0 # nothing treated as obsolete on the second run + def test_generate_random_tips_creates_hardlinks( + self, files_with_tags, temp_output_dir + ): + """Tips folder gets `count` random pool files as hardlinks.""" + manager = HardlinkManager(temp_output_dir) + created, fail = manager.generate_random_tips(files_with_tags, count=2) + + tips = temp_output_dir / "Tipy dne" + assert created == 2 + assert fail == 0 + links = sorted(p.name for p in tips.iterdir()) + assert len(links) == 2 + # each is a hardlink to one of the source files + srcs = {f.filename for f in files_with_tags} + assert set(links) <= srcs + + def test_generate_random_tips_caps_at_available( + self, files_with_tags, temp_output_dir + ): + """Asking for more than available yields all of them, no error.""" + manager = HardlinkManager(temp_output_dir) + created, _ = manager.generate_random_tips(files_with_tags, count=99) + assert created == len(files_with_tags) + + def test_generate_random_tips_refreshes_each_run( + self, files_with_tags, temp_output_dir + ): + """A second run replaces the previous selection (folder is emptied).""" + manager = HardlinkManager(temp_output_dir) + manager.generate_random_tips(files_with_tags, count=1) + manager.generate_random_tips(files_with_tags, count=1) + tips = temp_output_dir / "Tipy dne" + assert len(list(tips.iterdir())) == 1 # not accumulated to 2 + def test_dry_run(self, files_with_tags, temp_output_dir): """Test dry run (bez skutečného vytváření)""" manager = HardlinkManager(temp_output_dir)