refactor(snot): Rewrite to postgrest API

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__":