amber-backend/scripts/seed-providers.py

"""Load (or refresh) the `providers` rows the setup flow renders.

    python scripts/seed-providers.py                      # local docker compose
    PB_BASE=https://amber.petruzalekr.cz \
      PB_ADMIN_EMAIL=… PB_ADMIN_PASS=… python scripts/seed-providers.py

Reads `scripts/providers.seed.json` and **upserts by `slug`**, so running it
twice is the same as running it once. That matters more than it sounds: prices
and click-paths are edited in the admin UI, and a seeder that inserted blindly
would either duplicate every row or quietly overwrite a correction somebody made
last week. `--force` is the switch for "yes, put the file back over the top".

Without `--force` an existing row is left alone and reported as `kept`, which
makes this safe to run after any deploy — the collection gets populated on a
fresh instance and untouched on one that is already looked after.

Never reads secrets/amber.env: the superuser credentials come from the
environment or from the local defaults, the same way scripts/verify.py takes
them, so nothing here needs access to the family's credential file.
"""
import json, os, pathlib, sys, urllib.error, urllib.request

BASE = os.environ.get("PB_BASE", "http://localhost:8090").rstrip("/")
ADMIN_EMAIL = os.environ.get("PB_ADMIN_EMAIL", "admin@myanime.local")
ADMIN_PASS = os.environ.get("PB_ADMIN_PASS", "Sup3rSecret!123")
FORCE = "--force" in sys.argv
SEED = pathlib.Path(__file__).with_name("providers.seed.json")


def req(method, path, token=None, body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(BASE + path, data=data, method=method)
    r.add_header("Content-Type", "application/json")
    if token:
        r.add_header("Authorization", token)
    try:
        with urllib.request.urlopen(r) as resp:
            return resp.status, json.loads(resp.read() or "{}")
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read() or "{}")


def main():
    rows = json.loads(SEED.read_text(encoding="utf-8"))
    slugs = [r["slug"] for r in rows]
    if len(set(slugs)) != len(slugs):
        sys.exit(f"duplicate slug in {SEED.name}: {slugs}")
    # The page keys its per-service wiring off these four strings; a typo here
    # would render a card that leads nowhere, so it is checked rather than
    # trusted.
    known = {"torbox", "prehrajto", "webshare", "tmdb"}
    unknown = set(slugs) - known
    if unknown:
        sys.exit(f"unknown slug(s) {sorted(unknown)} — the page has no wiring for these")

    st, auth = req("POST", "/api/collections/_superusers/auth-with-password",
                   body={"identity": ADMIN_EMAIL, "password": ADMIN_PASS})
    if st != 200:
        sys.exit(f"superuser auth failed against {BASE}: HTTP {st} {auth}")
    token = auth["token"]

    st, existing = req("GET", "/api/collections/providers/records?perPage=200", token)
    if st != 200:
        sys.exit(f"cannot list providers (is the migration applied?): HTTP {st} {existing}")
    by_slug = {i["slug"]: i for i in existing.get("items", [])}

    for row in rows:
        slug = row["slug"]
        cur = by_slug.get(slug)
        if cur and not FORCE:
            print(f"[kept]    {slug} — already present, --force to overwrite")
            continue
        if cur:
            st, res = req("PATCH", f"/api/collections/providers/records/{cur['id']}",
                          token, row)
            verb = "updated"
        else:
            st, res = req("POST", "/api/collections/providers/records", token, row)
            verb = "created"
        if st not in (200, 201):
            sys.exit(f"[FAIL] {slug}: HTTP {st} {res}")
        print(f"[{verb}] {slug}")

    print(f"\n{len(rows)} provider(s) in {BASE}. "
          "Prices are editable in the admin UI — this file is only the starting point.")


if __name__ == "__main__":
    main()