N22/python_scripts/import_valdemarsro_favorites.py

#!/usr/bin/env python3
"""Import Valdemarsro favorites into Mealie (deduplicated by source URL).

This script imports a fixed list of favorite recipes from Valdemarsro by URL,
skipping recipes already present in Mealie.
"""

from __future__ import annotations

import json
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

BASE_URL = "http://10.0.0.142:9925"
SECRETS = Path("/Volumes/homeassistant/secrets.yaml")

FAVORITE_SLUGS = [
    "hjemmelavet-pizza",
    "lasagne",
    "musli-opskrift",
    "lakselasagne",
    "mexicansk-burger-med-hjemmelavet-guacamole",
    "grontsagsfad",
    "humus",
    "indisk_curry_med_kylling",
    "pariserbof",
    "skipperlabskovs",
    "pizzasnegle",
    "luksus-stjerneskud",
    "jordbaer-og-fetasalat-med-glaserede-pecannoedder",
    "spaghetti-bolognese",
    "kylling-med-cornflakes",
    "tarteletter-hoens-asparges",
    "one-pot-pasta",
    "cacio-e-pepe",
    "citronpasta",
    "blomkaalssalat",
    "koedsovs-onepotpasta",
    "kikaertegryde",
    "moerbradboeffer-med-bloede-loeg",
    "stegt-spidskaal",
    "bagt-kylling",
    "ristede-kartoffelskiver-fad",
    "vikingegryde",
    "spidskaalssalat-opskrift",
    "kylling-med-parmesan",
    "bagt-broccoli",
    "pastasalat-med-pesto",
    "nachos-bowl",
    "barbecuesauce",
    "congee-rissuppe-kylling",
    "macaroni-and-cheese",
    "halloween-dessert",
    "feta-pasta-med-tomat",
    "tortellini-i-fad",
    "flyvende-jacob",
]


def read_token() -> str:
    for line in SECRETS.read_text().splitlines():
        if line.strip().startswith("mealie_bearer_token:"):
            return line.split(":", 1)[1].strip().strip('"')
    raise RuntimeError("mealie_bearer_token not found in secrets.yaml")


def api_request(path: str, token: str, method: str = "GET", payload: dict | None = None) -> dict:
    headers = {"Authorization": token}
    data = None
    if payload is not None:
        headers["Content-Type"] = "application/json"
        data = json.dumps(payload).encode("utf-8")

    req = urllib.request.Request(f"{BASE_URL}{path}", headers=headers, data=data, method=method)
    with urllib.request.urlopen(req, timeout=90) as resp:
        raw = resp.read()
    return json.loads(raw) if raw else {}


def fetch_existing_org_urls(token: str) -> set[str]:
    urls: set[str] = set()
    page = 1
    per_page = 100

    while True:
        path = f"/api/recipes?page={page}&perPage={per_page}"
        payload = api_request(path, token)
        items = payload.get("items", []) or []
        if not items:
            break

        for recipe in items:
            org_url = (recipe.get("orgURL") or "").strip()
            if org_url:
                urls.add(org_url.rstrip("/"))

        if len(items) < per_page:
            break
        page += 1

    return urls


def import_recipe_by_url(token: str, url: str) -> tuple[bool, str]:
    try:
        api_request(
            "/api/recipes/create/url",
            token,
            method="POST",
            payload={"url": url, "includeTags": True},
        )
        return True, "imported"
    except urllib.error.HTTPError as exc:
        body = ""
        try:
            body = exc.read().decode("utf-8", errors="ignore")
        except Exception:
            pass
        if exc.code == 422:
            return False, f"422 {body[:200]}"
        return False, f"HTTP {exc.code} {body[:200]}"
    except Exception as exc:  # noqa: BLE001
        return False, str(exc)


def main() -> int:
    token = read_token()
    urls = [f"https://www.valdemarsro.dk/{slug}/" for slug in FAVORITE_SLUGS]

    existing_urls = fetch_existing_org_urls(token)

    to_import = [u for u in urls if u.rstrip("/") not in existing_urls]
    skipped = [u for u in urls if u.rstrip("/") in existing_urls]

    imported: list[str] = []
    failed: list[tuple[str, str]] = []

    for url in to_import:
        ok, msg = import_recipe_by_url(token, url)
        if ok:
            imported.append(url)
        else:
            failed.append((url, msg))

    print(f"TOTAL_LISTED={len(urls)}")
    print(f"SKIPPED_EXISTING={len(skipped)}")
    print(f"IMPORTED={len(imported)}")
    print(f"FAILED={len(failed)}")

    if skipped:
        print("-- skipped existing --")
        for url in skipped:
            print(url)

    if imported:
        print("-- imported --")
        for url in imported:
            print(url)

    if failed:
        print("-- failed --")
        for url, reason in failed:
            print(url)
            print(f"  reason: {reason}")

    # Exit non-zero only if everything failed.
    if failed and not imported:
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())