From 4bc948d77d23a3c8817127bc7a163f0f6989a879 Mon Sep 17 00:00:00 2001 From: Abdussamet Kocak Date: Mon, 28 Sep 2026 04:15:45 +0200 Subject: [PATCH] refactor(snot): Rewrite to postgrest API --- snot/snot.py | 385 ++++++++++++++++++++++++++------------------------- 1 file changed, 200 insertions(+), 185 deletions(-) diff --git a/snot/snot.py b/snot/snot.py index 0ac2c85..ca935f6 100755 --- a/snot/snot.py +++ b/snot/snot.py @@ -1,9 +1,8 @@ #!/usr/bin/env -S uv run --script # /// script -# dependencies = ["psycopg[binary]", "click", "httpx"] +# dependencies = ["click", "httpx"] # /// -import argparse import contextlib import dataclasses import datetime @@ -19,22 +18,22 @@ import socket import subprocess import threading import time -import typing +from collections.abc import Callable +from decimal import ROUND_CEILING, ROUND_FLOOR, Decimal from functools import partial, wraps from os import getenv from pathlib import Path -from typing import Any, Callable +from typing import Any from urllib.parse import parse_qs from wsgiref.simple_server import WSGIRequestHandler, make_server import click import httpx -import psycopg -from psycopg import sql as pgsql logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s") HandlerFunc = Callable[["Request"], "Response"] +client = httpx.Client(base_url="http://klein:18543", timeout=30) @dataclasses.dataclass @@ -70,8 +69,7 @@ def ffprobe(video_path: Path) -> FfprobeResult: str(video_path), ], text=True, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, + capture_output=True, ) try: proc.check_returncode() @@ -79,9 +77,7 @@ def ffprobe(video_path: Path) -> FfprobeResult: logging.error(f"failed to run ffprobe. stderr={e.stderr}") raise output: dict = json.loads(proc.stdout) - video_stream: dict = [ - s for s in output.get("streams", []) if s.get("codec_type") == "video" - ][0] + video_stream: dict = [s for s in output.get("streams", []) if s.get("codec_type") == "video"][0] video_format: dict = output.get("format", {}) width, height = video_stream.get("width"), video_stream.get("height") codec = video_stream.get("codec_name") @@ -94,9 +90,7 @@ def ffprobe(video_path: Path) -> FfprobeResult: tags = video_format.get("tags", {}) try: - sample_w, sample_h = list( - map(int, video_stream["sample_aspect_ratio"].split(":")) - ) + sample_w, sample_h = list(map(int, video_stream["sample_aspect_ratio"].split(":"))) sample_ar = (sample_w / sample_h) or 1.0 except: sample_ar = 1.0 @@ -151,9 +145,7 @@ def retry(max_attempts: int = 3, delay: float = 1.0): except Exception as e: last_exception = e if attempt < max_attempts: - logging.warning( - f"Attempt {attempt} failed: {e}. Retrying in {delay}s..." - ) + logging.warning(f"Attempt {attempt} failed: {e}. Retrying in {delay}s...") time.sleep(delay) else: logging.error(f"All {max_attempts} attempts failed.") @@ -164,49 +156,55 @@ def retry(max_attempts: int = 3, delay: float = 1.0): return decorator -def video_exists(con: psycopg.Connection, file_path: str) -> bool: - stmt = con.execute( - "UPDATE videos SET last_seen_at = NOW() WHERE file_path = %s RETURNING true AS exists", - [file_path], +def url_from_path(input_path: str) -> str: + return input_path.replace("/mnt/box/files/", "https://u201686.your-storagebox.de/files/") + + +def video_exists(file_path: str) -> bool: + response = client.patch( + "/videos", + params={"file_path": f"eq.{file_path}", "select": "file_path"}, + json={"last_seen_at": "now"}, + headers={"Prefer": "return=representation"}, ) - row = stmt.fetchone() - if not row: - return False - return row["exists"] + response.raise_for_status() + return bool(response.json()) def save_video( - con: psycopg.Connection, category: str, file_path: str, ffprobe_data: dict, video_hash: str, ): - sql = """ - INSERT INTO videos(category, file_path, ffprobe, hash_partial) - VALUES (%s, %s, %s, %s) - ON CONFLICT(file_path) DO UPDATE SET - last_seen_at = NOW();""" - con.execute( - sql, - [ - category, - file_path, - json.dumps(ffprobe_data), - video_hash, - ], + response = client.post( + "/videos", + params={"on_conflict": "file_path", "select": "file_path"}, + json={ + "category": category, + "file_path": file_path, + "ffprobe": ffprobe_data, + "hash_partial": video_hash, + }, + headers={"Prefer": "resolution=ignore-duplicates,return=representation"}, ) + response.raise_for_status() + if not response.json(): + response = client.patch( + "/videos", + params={"file_path": f"eq.{file_path}"}, + json={"last_seen_at": "now"}, + ) + response.raise_for_status() -def save_error(con: psycopg.Connection, file_path: str, stderr: str): - sql = """INSERT INTO scan_state(file_path, stderr) VALUES (%s, %s) ON conflict do nothing""" - con.execute( - sql, - [ - file_path, - stderr, - ], +def save_error(file_path: str, stderr: str) -> None: + response = client.post( + "/scan_state", + json={"file_path": file_path, "stderr": stderr}, + headers={"Prefer": "resolution=ignore-duplicates"}, ) + response.raise_for_status() def category_for_file(file_path: Path) -> str: @@ -218,7 +216,7 @@ def category_for_file(file_path: Path) -> str: return "local" -def scan_videos(con: psycopg.Connection, video_paths: list[Path]) -> None: +def scan_videos(video_paths: list[Path]) -> None: new_videos = [] for i, f in enumerate(video_paths): if "/_picks/" in str(f): @@ -226,7 +224,7 @@ def scan_videos(con: psycopg.Connection, video_paths: list[Path]) -> None: progress = f"[{i + 1}/{len(video_paths)}]" if f.is_dir(): # Recursively find files in directory - scan_videos(con, list(f.glob("**/*"))) + scan_videos(list(f.glob("**/*"))) continue if f.suffix.lower() not in [".mp4", ".mkv", ".avi", ".mov", ".wmv", ".webm"]: @@ -234,7 +232,7 @@ def scan_videos(con: psycopg.Connection, video_paths: list[Path]) -> None: file_path = str(f.absolute()) - if video_exists(con, file_path=file_path): + if video_exists(file_path=file_path): logging.debug(f"{progress} video already exists {f=}") continue @@ -243,27 +241,25 @@ def scan_videos(con: psycopg.Connection, video_paths: list[Path]) -> None: probed = ffprobe(f) except Exception as e: logging.error(f"Failed to probe {f}: {e}") - save_error(con, file_path=file_path, stderr=str(e)) + save_error(file_path=file_path, stderr=str(e)) continue file_hash = hash_partial(f) save_video( category=category_for_file(f), - con=con, file_path=file_path, ffprobe_data=probed.__dict__, video_hash=file_hash, ) logging.info(f"{progress} saved: {f.name}") new_videos.append(file_path) - con.commit() if new_videos: - parse_releases(con, new_videos) + parse_releases(new_videos) -def parse_releases(con: psycopg.Connection, file_paths: list[str]) -> None: +def parse_releases(file_paths: list[str]) -> None: api_key = os.getenv("OPENROUTER_API_KEY") if not api_key: logging.warning("OPENROUTER_API_KEY not set, skipping filename parsing") @@ -272,9 +268,7 @@ def parse_releases(con: psycopg.Connection, file_paths: list[str]) -> None: # Chunk file_paths by 20 items to avoid prompt token limits for i, chunk in enumerate(chunked(file_paths, 20), 1): filenames = [Path(p).name for p in chunk] - logging.info( - f"Parsing filenames for {len(filenames)} videos (chunk {i}) via OpenRouter..." - ) + logging.info(f"Parsing filenames for {len(filenames)} videos (chunk {i}) via OpenRouter...") try: parsed_data = parse_filenames_with_ai(filenames, api_key) @@ -294,16 +288,15 @@ def parse_releases(con: psycopg.Connection, file_paths: list[str]) -> None: # Update the release column release_info = {k: v for k, v in item.items() if k != "filename"} - con.execute( - "UPDATE videos SET release = %s WHERE file_path = %s", - [json.dumps(release_info), full_path], + response = client.patch( + "/videos", + params={"file_path": f"eq.{full_path}"}, + json={"release": release_info}, ) - con.commit() + response.raise_for_status() logging.info(f"Successfully updated release info for chunk {i}") except Exception as e: - logging.error( - f"Failed to parse filenames or update database for chunk {i}: {e}" - ) + logging.error(f"Failed to parse filenames or update database for chunk {i}: {e}") @retry(max_attempts=3, delay=0.1) @@ -357,11 +350,7 @@ HookupHotshot.26.02.06.Episode.453.Shrooms.Q.XXX.720p.HEVC.x265.PRT.mp4 -> studi # Clean output: omit empty strings and empty arrays cleaned_data = [] for entry in parsed_data: - cleaned_entry = { - k: v - for k, v in entry.items() - if v != "" and not (isinstance(v, list) and not v) - } + cleaned_entry = {k: v for k, v in entry.items() if v != "" and not (isinstance(v, list) and not v)} actors = cleaned_entry.get("actors", []) if isinstance(actors, list) and len(actors) > 0: actors = [a for a in actors if a and "@" not in a] @@ -523,51 +512,58 @@ class TinyAPI: thread.join(timeout=5) -@contextlib.contextmanager -def connect_db() -> typing.Generator[psycopg.Connection, typing.Any, typing.Any]: - with psycopg.connect( - "postgres://abdus:abdus@db.abdus.dev:5444/snot?sslmode=disable" - ) as conn: - conn.row_factory = psycopg.rows.dict_row - yield conn - - -def handle_search(req: Request, conn: psycopg.Connection) -> Response: +def handle_search(req: Request) -> Response: query: str = req.payload.get("query", "") if not query: return Response.error("Query is empty") - where_sql, where_params = filter_to_where(query) - - query_stmt = pgsql.SQL( - """ - with q as (select - id, - filename_from_path(file_path) as file_name, - file_path, - actors, - size_bytes / 1048576 as size_mb, - (ffprobe->>'width') || 'x' || (ffprobe->>'height') as resolution, - duration_human, - created_at - from videos) - select * from q - where {filter} - order by created_at desc - """ - ).format(filter=pgsql.SQL(where_sql)) - with conn.cursor() as cursor: - rows = cursor.execute(query_stmt, where_params).fetchall() - return Response.from_json( - [ - { - **row, - "download_url": "https://u201686:T6672ICVoWAedECH@u201686.your-storagebox.de/files" - + row["file_path"].replace("/mnt/box/files", ""), - } - for row in rows - ] + params = filter_to_postgrest(query) + params.update( + { + "select": "id,file_path,file_name,actors,size_bytes,duration_human,created_at,ffprobe", + "order": "created_at.desc,id.desc", + } ) + rows = fetch_videos(params=params) + results = [] + for row in rows: + probe = row.get("ffprobe") or {} + width = probe.get("width") + height = probe.get("height") + resolution = f"{width}x{height}" if width is not None and height is not None else None + results.append( + { + "id": row["id"], + "file_name": row["file_name"], + "file_path": row["file_path"], + "actors": row.get("actors"), + "size_mb": row["size_bytes"] // 1048576, + "resolution": resolution, + "duration_human": row.get("duration_human"), + "created_at": row.get("created_at"), + "download_url": url_from_path(input_path=row["file_path"]), + } + ) + return Response.from_json(results) + + +def fetch_videos(params: dict[str, str]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + total: int | None = None + while total is None or len(rows) < total: + response = client.get( + "/videos", + params={**params, "limit": "1000", "offset": str(len(rows))}, + headers={"Prefer": "count=exact"} if total is None else None, + ) + response.raise_for_status() + page = response.json() + if total is None: + total = int(response.headers["Content-Range"].rsplit("/", 1)[1]) + if not page: + break + rows.extend(page) + return rows def tokenize_filter_groups(filter_query: str) -> list[list[str]]: @@ -587,58 +583,72 @@ def tokenize_filter_groups(filter_query: str) -> list[list[str]]: return [group for group in groups if group] -def token_to_condition(token: str) -> tuple[str | None, list[typing.Any]]: +def token_to_conditions(token: str) -> list[str]: if ":" in token: key, value = token.split(":", 1) key = key.strip().lower() value = value.strip() if not value: - return None, [] + return [] if key == "file_name": - return "file_name ILIKE %s", [f"%{value}%"] + return [f"file_name.ilike.{postgrest_quote(f'%{value}%')}"] if key == "size_mb": - match = re.match( - r"^(>=|<=|>|<|=)?\s*(\d+(?:\.\d+)?)$", value, re.IGNORECASE - ) + match = re.match(r"^(>=|<=|>|<|=)?\s*(\d+(?:\.\d+)?)$", value, re.IGNORECASE) if not match: - return None, [] + return [] op = match.group(1) or "=" - num = float(match.group(2)) - return f"size_mb {op} %s", [num] + size_mb = Decimal(match.group(2)) + floor_mb = int(size_mb.to_integral_value(rounding=ROUND_FLOOR)) + ceil_mb = int(size_mb.to_integral_value(rounding=ROUND_CEILING)) + if op == ">": + return [f"size_bytes.gte.{(floor_mb + 1) * 1048576}"] + if op == ">=": + return [f"size_bytes.gte.{ceil_mb * 1048576}"] + if op == "<": + return [f"size_bytes.lt.{ceil_mb * 1048576}"] + if op == "<=": + return [f"size_bytes.lt.{(floor_mb + 1) * 1048576}"] + if size_mb != floor_mb: + return ["size_bytes.gt.0", "size_bytes.lt.0"] + return [ + f"size_bytes.gte.{floor_mb * 1048576}", + f"size_bytes.lt.{(floor_mb + 1) * 1048576}", + ] if key == "actor": - return "actors @> %s::text[]", [[value]] + array_value = value.replace("\\", "\\\\").replace('"', '\\"') + return [f'actors.cs.{{"{array_value}"}}'] - return None, [] + return [] - return "file_name ILIKE %s", [f"%{token}%"] + return [f"file_name.ilike.{postgrest_quote(f'%{token}%')}"] -def filter_to_where(filter_query: str) -> tuple[str, list[typing.Any]]: +def postgrest_quote(value: str) -> str: + if re.search(r"[,(){}\\\"]", value): + return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' + return value + + +def filter_to_postgrest(filter_query: str) -> dict[str, str]: groups = tokenize_filter_groups(filter_query=filter_query) if not groups: - return "TRUE", [] + return {} or_clauses: list[str] = [] - params: list[typing.Any] = [] - for group in groups: and_clauses: list[str] = [] for token in group: - clause, clause_params = token_to_condition(token=token) - if not clause: - continue - and_clauses.append(clause) - params.extend(clause_params) + and_clauses.extend(token_to_conditions(token=token)) if and_clauses: - or_clauses.append("(" + " AND ".join(and_clauses) + ")") + or_clauses.append("and(" + ",".join(and_clauses) + ")") if not or_clauses: - return "TRUE", [] + return {} - return " OR ".join(or_clauses), params + return {"or": "(" + ",".join(or_clauses) + ")"} def handle_home(req: Request, query: str = "") -> Response: @@ -680,12 +690,8 @@ def handle_assets(req: Request, base_path: Path) -> Response: def run_htmlpopup(address: str): exe_path = shutil.which("htmlpopup") if not exe_path: - raise RuntimeError( - "htmlpopup executable not found in PATH. Please install it to use the HTML popup feature." - ) - proc = subprocess.Popen( - [exe_path, "--title", "s·m·u·t", address], text=True, stdout=subprocess.PIPE - ) + raise RuntimeError("htmlpopup executable not found in PATH. Please install it to use the HTML popup feature.") + proc = subprocess.Popen([exe_path, "--title", "s·m·u·t", address], text=True, stdout=subprocess.PIPE) try: yield proc.wait() @@ -711,21 +717,20 @@ def cli(ctx: click.Context): @click.option("--query", type=str, default="", help="Initial query string") def serve(port: int, query: str): """Start the web UI""" - with connect_db() as conn: - handlers = { - "POST /search": partial(handle_search, conn=conn), - "GET /": partial(handle_home, query=query), - "GET /.*": partial(handle_assets, base_path=Path(__file__).parent), - } - api = TinyAPI(handlers=handlers) + handlers = { + "POST /search": handle_search, + "GET /": partial(handle_home, query=query), + "GET /.*": partial(handle_assets, base_path=Path(__file__).parent), + } + api = TinyAPI(handlers=handlers) - with api.serve(port=port) as server_url: - logging.info(f"Server running at {server_url}") - logging.info("Available endpoints:") - for path in handlers.keys(): - logging.info(f" {path}") - with run_htmlpopup(server_url): - logging.info("HTML popup started. Press Ctrl+C to stop.") + with api.serve(port=port) as server_url: + logging.info(f"Server running at {server_url}") + logging.info("Available endpoints:") + for path in handlers: + logging.info(f" {path}") + with run_htmlpopup(server_url): + logging.info("HTML popup started. Press Ctrl+C to stop.") @cli.command() @@ -736,8 +741,7 @@ def scan(paths: list[Path]): click.echo("No paths provided to scan.") return - with connect_db() as con: - scan_videos(con=con, video_paths=list(paths)) + scan_videos(video_paths=list(paths)) @cli.command("mark-deletion") @@ -748,37 +752,48 @@ def mark_deletion(paths: list[Path]): click.echo("No paths provided.") return - with connect_db() as con: - for path in paths: - file_name = path.name - rows = con.execute( - "UPDATE videos SET marked_for_deletion_at = NOW() WHERE file_name = %s RETURNING file_path", - [file_name], - ).fetchall() - if rows: - for row in rows: - click.echo(f"Marked for deletion: {row['file_path']}") - else: - click.echo(f"Not found in database: {file_name}", err=True) - continue + for path in paths: + file_name = path.name + response = client.patch( + "/videos", + params={ + "file_name": f"eq.{postgrest_quote(file_name)}", + "select": "file_path", + }, + json={"marked_for_deletion_at": "now"}, + headers={"Prefer": "return=representation"}, + ) + response.raise_for_status() + rows = response.json() + if rows: + for row in rows: + click.echo(f"Marked for deletion: {row['file_path']}") + else: + click.echo(f"Not found in database: {file_name}", err=True) + continue - if not click.get_text_stream("stdin").isatty(): - continue + if not click.get_text_stream("stdin").isatty(): + continue - stem = path.stem - siblings = con.execute( - "SELECT file_path, file_name FROM videos WHERE file_name != %s AND file_name LIKE %s AND marked_for_deletion_at IS NULL", - [file_name, f"{stem}.%"], - ).fetchall() - for sibling in siblings: - if click.confirm(f" Also mark {sibling['file_name']}?", default=False): - con.execute( - "UPDATE videos SET marked_for_deletion_at = NOW() WHERE file_path = %s", - [sibling["file_path"]], - ) - click.echo(f" Marked for deletion: {sibling['file_path']}") - - con.commit() + stem = path.stem + siblings = fetch_videos( + params={ + "select": "file_path,file_name", + "file_name": f"like.{postgrest_quote(f'{stem}.%')}", + "marked_for_deletion_at": "is.null", + "order": "file_path.asc", + } + ) + siblings = [row for row in siblings if row["file_name"] != file_name] + for sibling in siblings: + if click.confirm(f" Also mark {sibling['file_name']}?", default=False): + response = client.patch( + "/videos", + params={"file_path": f"eq.{sibling['file_path']}"}, + json={"marked_for_deletion_at": "now"}, + ) + response.raise_for_status() + click.echo(f" Marked for deletion: {sibling['file_path']}") if __name__ == "__main__":