Add standalone "Tipy dne" script with systemd service and timer
This commit is contained in:
@@ -0,0 +1,130 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Standalone "Tipy dne" generator — deployable next to a Filmotéka (e.g. cron).
|
||||
|
||||
On each run it reads the pool index (``<pool>/.Curator.!index``), picks N random
|
||||
pool movies, clears the tips folder in the output and recreates it as hardlinks.
|
||||
It does not import Curator — only the Python standard library — so it can live on
|
||||
the server with the Filmotéka, independent of the app.
|
||||
|
||||
Paths come from CLI args, or fall back to the global config (``.Curator.!gtag``,
|
||||
keys ``pool_dir`` / ``filmoteka_dir``) when ``--config`` is given or one is found.
|
||||
|
||||
Examples:
|
||||
python scripts/tipy_dne.py --pool /mnt/srv/Filmoteka --output /mnt/srv/Filmy/Filmoteka_v2
|
||||
python scripts/tipy_dne.py --count 15 # uses the repo's global config
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import random
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
INDEX_FILENAME = ".Curator.!index"
|
||||
GLOBAL_CONFIG_FILENAME = ".Curator.!gtag"
|
||||
DEFAULT_SUBFOLDER = "Tipy dne"
|
||||
DEFAULT_COUNT = 15
|
||||
|
||||
|
||||
def _load_config_paths(config_path: Path) -> tuple[Path | None, Path | None]:
|
||||
"""Read (pool_dir, filmoteka_dir) from a global config file, if present."""
|
||||
try:
|
||||
with open(config_path, "r", encoding="utf-8") as f:
|
||||
cfg = json.load(f)
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return None, None
|
||||
pool = cfg.get("pool_dir")
|
||||
out = cfg.get("filmoteka_dir")
|
||||
return (Path(pool) if pool else None, Path(out) if out else None)
|
||||
|
||||
|
||||
def _pool_movies(pool_dir: Path) -> list[Path]:
|
||||
"""Existing pool movie files listed in the index (copy-as-is folders aren't
|
||||
indexed, so this is the pool/Filmy set only)."""
|
||||
index_path = pool_dir / INDEX_FILENAME
|
||||
with open(index_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
movies: list[Path] = []
|
||||
for rel in data.get("movies", {}):
|
||||
path = pool_dir / rel
|
||||
if path.is_file():
|
||||
movies.append(path)
|
||||
return movies
|
||||
|
||||
|
||||
def _clear_folder(folder: Path) -> int:
|
||||
"""Remove the files in ``folder`` (the old tips). Returns how many removed."""
|
||||
removed = 0
|
||||
if folder.exists():
|
||||
for child in folder.iterdir():
|
||||
if child.is_file() or child.is_symlink():
|
||||
try:
|
||||
child.unlink()
|
||||
removed += 1
|
||||
except OSError as exc:
|
||||
print(f" ! nelze smazat {child}: {exc}", file=sys.stderr)
|
||||
return removed
|
||||
|
||||
|
||||
def generate_tips(pool_dir: Path, output_dir: Path, count: int, subfolder: str) -> int:
|
||||
"""Refresh the tips folder with ``count`` random pool movies as hardlinks."""
|
||||
movies = _pool_movies(pool_dir)
|
||||
if not movies:
|
||||
print("Index neobsahuje žádné filmy.", file=sys.stderr)
|
||||
return 0
|
||||
|
||||
tips_dir = output_dir / subfolder
|
||||
cleared = _clear_folder(tips_dir)
|
||||
tips_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
chosen = random.sample(movies, min(count, len(movies)))
|
||||
created = 0
|
||||
for src in chosen:
|
||||
target = tips_dir / src.name
|
||||
try:
|
||||
if target.exists():
|
||||
target.unlink()
|
||||
os.link(src, target)
|
||||
created += 1
|
||||
print(f" + {src.name}")
|
||||
except OSError as exc:
|
||||
print(f" ! {src.name}: {exc}", file=sys.stderr)
|
||||
|
||||
print(f"\nTipy dne: smazáno {cleared}, vytvořeno {created} (z {len(movies)} filmů)")
|
||||
return created
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Vygeneruje složku Tipy dne.")
|
||||
parser.add_argument("--pool", type=Path, help="Pool root (obsahuje .Curator.!index)")
|
||||
parser.add_argument("--output", type=Path, help="Výstupní složka Filmotéky")
|
||||
parser.add_argument("--count", type=int, default=DEFAULT_COUNT,
|
||||
help=f"Počet náhodných titulů (výchozí {DEFAULT_COUNT})")
|
||||
parser.add_argument("--subfolder", default=DEFAULT_SUBFOLDER,
|
||||
help=f'Název složky tipů (výchozí "{DEFAULT_SUBFOLDER}")')
|
||||
parser.add_argument("--config", type=Path,
|
||||
help="Cesta ke global configu (.Curator.!gtag) pro doplnění cest")
|
||||
args = parser.parse_args()
|
||||
|
||||
pool, output = args.pool, args.output
|
||||
if pool is None or output is None:
|
||||
config_path = args.config or (
|
||||
Path(__file__).resolve().parent.parent / GLOBAL_CONFIG_FILENAME)
|
||||
cfg_pool, cfg_out = _load_config_paths(config_path)
|
||||
pool = pool or cfg_pool
|
||||
output = output or cfg_out
|
||||
|
||||
if pool is None or output is None:
|
||||
parser.error("Chybí --pool a/nebo --output (a nenalezeny v global configu).")
|
||||
if not (pool / INDEX_FILENAME).is_file():
|
||||
parser.error(f"Index nenalezen: {pool / INDEX_FILENAME}")
|
||||
|
||||
created = generate_tips(pool, output, args.count, args.subfolder)
|
||||
sys.exit(0 if created else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user