my-homelab-configs/infra/heimdall/seed-heimdall.py

#!/usr/bin/env python3
import json
import os
import sqlite3
import time
from datetime import datetime, timezone


DB_CANDIDATES = (
    "/config/www/app.sqlite",
    "/config/app.sqlite",
)
LINKS_PATH = os.environ.get("HEIMDALL_LINKS_PATH", "/seed/links.json")
RESYNC_SECONDS = int(os.environ.get("HEIMDALL_LINK_RESYNC_SECONDS", "3600"))


def find_database():
    while True:
        for path in DB_CANDIDATES:
            if os.path.exists(path):
                return path
        time.sleep(5)


def table_exists(conn, name):
    row = conn.execute(
        "select name from sqlite_master where type = 'table' and name = ?",
        (name,),
    ).fetchone()
    return row is not None


def columns(conn, table):
    return {row[1] for row in conn.execute(f"pragma table_info({table})")}


def quote_identifier(name):
    return '"' + name.replace('"', '""') + '"'


def ensure_public_dashboard(conn, user_id):
    if not table_exists(conn, "users"):
        return
    user_columns = columns(conn, "users")
    if "public_front" not in user_columns:
        return
    conn.execute(
        'update "users" set "public_front" = 1 where "id" = ?',
        (user_id,),
    )


def ensure_dashboard_tag(conn, item_id, tag_id, now):
    if not table_exists(conn, "item_tag"):
        return
    item_tag_columns = columns(conn, "item_tag")
    if not {"item_id", "tag_id"}.issubset(item_tag_columns):
        return
    existing = conn.execute(
        'select 1 from "item_tag" where "item_id" = ? and "tag_id" = ?',
        (item_id, tag_id),
    ).fetchone()
    if existing:
        return
    values = {
        "item_id": item_id,
        "tag_id": tag_id,
        "created_at": now,
        "updated_at": now,
    }
    insert_columns = [name for name in values if name in item_tag_columns]
    placeholders = ", ".join("?" for _ in insert_columns)
    quoted_insert_columns = ", ".join(quote_identifier(name) for name in insert_columns)
    conn.execute(
        f'insert into "item_tag" ({quoted_insert_columns}) values ({placeholders})',
        [values[name] for name in insert_columns],
    )


def home_dashboard_tag_id(conn):
    item_columns = columns(conn, "items")
    if not {"id", "title"}.issubset(item_columns):
        return 0
    row = conn.execute(
        'select "id" from "items" where "title" = ? order by "id" limit 1',
        ("app.dashboard",),
    ).fetchone()
    if row:
        return row[0]
    return 0


def wait_for_items_table(db_path):
    while True:
        try:
            conn = sqlite3.connect(db_path)
            if table_exists(conn, "items"):
                return conn
            conn.close()
        except sqlite3.Error:
            pass
        time.sleep(5)


def upsert_links(conn, links):
    item_columns = columns(conn, "items")
    seed_user_id = 1
    home_dashboard_tag = home_dashboard_tag_id(conn)
    now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")

    ensure_public_dashboard(conn, seed_user_id)

    for order, link in enumerate(links):
        existing = conn.execute(
            'select "id" from "items" where "title" = ?',
            (link["title"],),
        ).fetchone()
        values = {
            "title": link["title"],
            "url": link["url"],
            "description": None,
            "colour": link.get("colour"),
            "pinned": 1,
            "order": order,
            "type": 0,
            "user_id": seed_user_id,
            "class": None,
            "appid": None,
            "appdescription": link.get("description"),
            "created_at": now,
            "updated_at": now,
            "deleted_at": None,
        }

        if existing:
            update_columns = [
                name
                for name in values
                if name in item_columns and name not in ("id", "title", "created_at")
            ]
            assignments = ", ".join(f"{quote_identifier(name)} = ?" for name in update_columns)
            params = [values[name] for name in update_columns]
            params.append(existing[0])
            conn.execute(f'update "items" set {assignments} where "id" = ?', params)
            ensure_dashboard_tag(conn, existing[0], home_dashboard_tag, now)
            continue

        insert_columns = [name for name in values if name in item_columns and name != "deleted_at"]
        placeholders = ", ".join("?" for _ in insert_columns)
        quoted_insert_columns = ", ".join(quote_identifier(name) for name in insert_columns)
        conn.execute(
            f'insert into "items" ({quoted_insert_columns}) values ({placeholders})',
            [values[name] for name in insert_columns],
        )
        new_item_id = conn.execute("select last_insert_rowid()").fetchone()[0]
        ensure_dashboard_tag(conn, new_item_id, home_dashboard_tag, now)

    conn.commit()


def load_links():
    with open(LINKS_PATH, encoding="utf-8") as links_file:
        return json.load(links_file)


while True:
    db_path = find_database()
    connection = wait_for_items_table(db_path)
    try:
        upsert_links(connection, load_links())
        time.sleep(RESYNC_SECONDS)
    except (OSError, sqlite3.Error, json.JSONDecodeError) as error:
        print(f"Failed to seed Heimdall links: {error}", flush=True)
        time.sleep(10)
    finally:
        connection.close()