refractioning

src/download/admin_links.py

@@ -0,0 +1,192 @@
+#!/usr/bin/env python3
+"""Admin CLI: link operations."""
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+import db
+from functions import load_config_variables
+
+
+def prompt_yes_no(message: str) -> bool:
+    while True:
+        raw = input(f"{message} [y/n]: ").strip().lower()
+        if raw in ("y", "yes"):
+            return True
+        if raw in ("n", "no"):
+            return False
+
+
+def parse_list_file(path: Path) -> dict:
+    enabled: set[str] = set()
+    disabled: set[str] = set()
+    if not path.is_file():
+        return {"enabled": enabled, "disabled": disabled}
+    with open(path, "r", encoding="utf-8") as r_file:
+        for raw in r_file:
+            line = raw.strip()
+            if not line:
+                continue
+            if line.startswith("#"):
+                url = line.lstrip("#").strip()
+                if url:
+                    disabled.add(db.normalize_url(url))
+                continue
+            enabled.add(db.normalize_url(line))
+    return {"enabled": enabled, "disabled": disabled}
+
+
+def cmd_add(args: argparse.Namespace) -> None:
+    configs = load_config_variables()
+    with db.connect(configs) as conn:
+        result = db.add_link(
+            conn, args.user, args.url, assume_yes=args.assume_yes, source="manual"
+        )
+        if result["status"] == "removed" and not args.assume_yes:
+            removed_at = result.get("removed_at", "unknown")
+            if prompt_yes_no(f"Link was deleted on {removed_at}. Re-add?"):
+                result = db.add_link(
+                    conn, args.user, args.url, assume_yes=True, source="manual"
+                )
+        row = result.get("row")
+        if row and row["banned_at"]:
+            print(f"Warning: link is banned ({row['banned_reason'] or 'no reason'})")
+        if row and not row["enabled"]:
+            print("Warning: link is disabled")
+        conn.commit()
+    print(result["status"])
+
+
+def cmd_disable(args: argparse.Namespace) -> None:
+    with db.connect() as conn:
+        ok = db.set_enabled(conn, args.user, args.url, enabled=False)
+        if ok:
+            conn.commit()
+    print("ok" if ok else "not found")
+
+
+def cmd_enable(args: argparse.Namespace) -> None:
+    with db.connect() as conn:
+        ok = db.set_enabled(conn, args.user, args.url, enabled=True)
+        if ok:
+            conn.commit()
+    print("ok" if ok else "not found")
+
+
+def cmd_ban(args: argparse.Namespace) -> None:
+    with db.connect() as conn:
+        ok = db.set_banned(conn, args.user, args.url, banned=True, reason=args.reason)
+        if ok:
+            conn.commit()
+    print("ok" if ok else "not found")
+
+
+def cmd_unban(args: argparse.Namespace) -> None:
+    with db.connect() as conn:
+        ok = db.set_banned(conn, args.user, args.url, banned=False)
+        if ok:
+            conn.commit()
+    print("ok" if ok else "not found")
+
+
+def cmd_remove(args: argparse.Namespace) -> None:
+    with db.connect() as conn:
+        ok = db.remove_link(conn, args.user, args.url)
+        if ok:
+            conn.commit()
+    print("ok" if ok else "not found")
+
+
+def cmd_rename(args: argparse.Namespace) -> None:
+    with db.connect() as conn:
+        result = db.rename_link(conn, args.user, args.old_url, args.new_url)
+        if result["status"] == "renamed":
+            conn.commit()
+    print(result["status"])
+
+
+def cmd_list(args: argparse.Namespace) -> None:
+    users = args.user or None
+    with db.connect() as conn:
+        rows = db.get_links(
+            conn,
+            users=users,
+            include_disabled=args.disabled,
+            include_banned=args.banned,
+        )
+    for row in rows:
+        status = "enabled" if row["enabled"] else "disabled"
+        if row["banned_at"]:
+            status = "banned"
+        print(f"{row['user_name']} [{status}] {row['url_original']}")
+
+
+def cmd_import(_: argparse.Namespace) -> None:
+    configs = load_config_variables()
+    with db.connect(configs) as conn:
+        imported_paths = []
+        for entry in configs["users"]:
+            user = entry["name"]
+            lists_dir = Path(configs["global"]["lists-dir"]) / user
+            master = lists_dir / "watch.txt"
+            result = db.import_master_list(conn, user, master)
+            if result["status"] == "ok":
+                imported_paths.append(str(master))
+            print(f"{user}: {result}")
+            if result.get("duplicates"):
+                print(f"{user} duplicates:")
+                for dup in result["duplicates"]:
+                    print(f"  {dup}")
+        if imported_paths:
+            print("Imported lists:")
+            for path in imported_paths:
+                print(f"  {path}")
+        conn.commit()
+
+
+def cmd_validate_import(_: argparse.Namespace) -> None:
+    configs = load_config_variables()
+    with db.connect(configs) as conn:
+        for entry in configs["users"]:
+            user = entry["name"]
+            lists_dir = Path(configs["global"]["lists-dir"]) / user
+            master = lists_dir / "watch.txt"
+            list_sets = parse_list_file(master)
+
+            rows = db.get_links_by_user(conn, user)
+            db_enabled = set()
+            db_disabled = set()
+            for row in rows:
+                norm = db.normalize_url(row["url_original"])
+                if row["enabled"] and not row["banned_at"]:
+                    db_enabled.add(norm)
+                else:
+                    db_disabled.add(norm)
+
+            missing_enabled = list_sets["enabled"] - db_enabled
+            missing_disabled = list_sets["disabled"] - db_disabled
+            extra_enabled = db_enabled - list_sets["enabled"]
+            extra_disabled = db_disabled - list_sets["disabled"]
+
+            print(f"{user}:")
+            if missing_enabled:
+                print("  Missing enabled in DB:")
+                for url in sorted(missing_enabled):
+                    print(f"    {url}")
+            if missing_disabled:
+                print("  Missing disabled in DB:")
+                for url in sorted(missing_disabled):
+                    print(f"    {url}")
+            if extra_enabled:
+                print("  Extra enabled in DB:")
+                for url in sorted(extra_enabled):
+                    print(f"    {url}")
+            if extra_disabled:
+                print("  Extra disabled in DB:")
+                for url in sorted(extra_disabled):
+                    print(f"    {url}")
+            if not any(
+                [missing_enabled, missing_disabled, extra_enabled, extra_disabled]
+            ):
+                print("  OK")