Immich × Databricks

# Section 00 — run once to store Immich credentials in Databricks Secrets
import requests

# ── Fill these in, run once, then DELETE this cell ────────────────────
DATABRICKS_PAT  = "dapi..."
WORKSPACE_URL   = "https://adb-xxxxxxxx.azuredatabricks.net"
IMMICH_BASE_URL = "https://your-immich-url.com"
IMMICH_API_KEY  = "your_immich_api_key_here"
# ─────────────────────────────────────────────────────────────────────

headers = {"Authorization": f"Bearer {DATABRICKS_PAT}", "Content-Type": "application/json"}
base = f"{WORKSPACE_URL}/api/2.0/secrets"

# Create the secret scope — harmless if it already exists
def create_scope(name):
    r = requests.post(f"{base}/scopes/create", headers=headers,
                      json={"scope": name, "initial_manage_principal": "users"})
    if r.status_code == 200: print(f"Scope '{name}' created.")
    elif "RESOURCE_ALREADY_EXISTS" in r.text: print(f"Scope '{name}' already exists.")
    else: raise Exception(r.text)

# Store a single key/value pair in the scope
def put_secret(scope, key, value):
    r = requests.post(f"{base}/put", headers=headers,
                      json={"scope": scope, "key": key, "string_value": value})
    if r.status_code == 200: print(f"  secret '{key}' stored.")
    else: raise Exception(r.text)

create_scope("immich")
put_secret("immich", "api_key",  IMMICH_API_KEY)
put_secret("immich", "base_url", IMMICH_BASE_URL)
print("\nDone. Now DELETE this cell.")

# Section 01a — load credentials from Databricks Secrets into session variables
import requests

# ── Your Immich credentials ───────────────────────────────────────────────
# Read from Databricks Secrets — never hardcode these in a notebook cell
API_KEY  = dbutils.secrets.get(scope="immich", key="api_key")
BASE_URL = dbutils.secrets.get(scope="immich", key="base_url").rstrip("/")
HEADERS  = {"x-api-key": API_KEY, "Accept": "application/json"}  # Immich expects the API key in this header on every request

print("Credentials loaded.")

# Section 01b — imports, catalog/schema setup, SQL context, connectivity check
import json
from datetime import datetime, timezone, timedelta
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, from_json, to_timestamp, year, month,
    dayofmonth, lower, current_timestamp, lit, when, count
)
from pyspark.sql.types import (
    StructType, StructField, StringType, LongType,
    BooleanType, DoubleType, IntegerType, TimestampType
)

# Re-use the existing Spark session if one is already running
spark = SparkSession.builder.getOrCreate()

# ── Catalog & schemas ─────────────────────────────────────────────────────
# Discover the catalog name dynamically — avoids hardcoding "workspace"
CATALOG = spark.sql("SELECT current_catalog()").collect()[0][0]
print(f"Catalog : {CATALOG}")

# Create the three medallion schemas if they don't exist yet
for schema in ["immich_bronze", "immich_silver", "immich_gold"]:
    spark.sql(f"CREATE SCHEMA IF NOT EXISTS {CATALOG}.{schema}")
    print(f"  schema ready: {CATALOG}.{schema}")

# ── Set SQL context ───────────────────────────────────────────────────────
# Without this, SQL cells resolve against the 'default' catalog and fail
spark.sql(f"USE CATALOG {CATALOG}")
print(f"  catalog set: {CATALOG}")

# ── Connectivity check ────────────────────────────────────────────────────
# Quick ping to confirm the API key and URL are correct
resp = requests.get(f"{BASE_URL}/api/server/statistics", headers=HEADERS)
resp.raise_for_status()  # Raises an exception immediately if the request fails
stats = resp.json()
print(f"\nImmich connection OK")
print(f"  Photos : {stats.get('photos','?')}")

# Section 02 — destructive reset: drops all bronze/silver/gold tables
# Loop over all three medallion schemas and drop every table in each
for schema in ["immich_bronze", "immich_silver", "immich_gold"]:
    full = f"{CATALOG}.{schema}"
    tables = spark.sql(f"SHOW TABLES IN {full}").collect()
    if tables:
        for row in tables:
            spark.sql(f"DROP TABLE IF EXISTS {full}.{row['tableName']}")
            print(f"  dropped: {full}.{row['tableName']}")
    else:
        print(f"  {full}: already empty")
print("\nAll layers cleared.")

# Section 03 — explore: inspect the raw API response structure for one asset
resp = requests.post(f"{BASE_URL}/api/search/metadata", headers=HEADERS,
                     json={"page": 1, "size": 1, "withExif": True})
resp.raise_for_status()
data = resp.json()

assets_block = data.get("assets", {})
envelope = {k: v for k, v in assets_block.items() if k != "items"}
print("=== Pagination envelope ===")
print(json.dumps(envelope, indent=2))
print("\n=== Full structure of one asset ===")
print(json.dumps(assets_block["items"][0], indent=2))

# Section 04 — explore: check which fields are populated across a sample of 100 assets
resp = requests.post(f"{BASE_URL}/api/search/metadata", headers=HEADERS,
                     json={"page": 1, "size": 100, "withExif": True})
sample = resp.json().get("assets", {}).get("items", [])
n = len(sample)
top_counts, exif_counts = {}, {}  # Separate counters for top-level fields and nested exifInfo fields

# Count non-empty values for each field across the sample
for asset in sample:
    for k, v in asset.items():
        if v is not None and v != "" and v != [] and v != "0:00:00.00000":
            top_counts[k] = top_counts.get(k, 0) + 1
    for k, v in (asset.get("exifInfo") or {}).items():
        if v is not None and v != "":
            exif_counts[k] = exif_counts.get(k, 0) + 1

print(f"Sample: {n} assets\n")
print("=== Top-level fields ===")
# Print a simple ASCII bar chart — width proportional to fill rate
for k, v in sorted(top_counts.items(), key=lambda x: -x[1]):
    bar = "█" * int(v / n * 30)
    print(f"  {k:<28} {bar:<30} {v}/{n}")
print("\n=== exifInfo fields ===")
for k, v in sorted(exif_counts.items(), key=lambda x: -x[1]):
    bar = "█" * int(v / n * 30)
    print(f"  {k:<28} {bar:<30} {v}/{n}")

# Section 05a — bronze full refresh: paginate through all assets from the Immich API
def fetch_all_assets(base_url, headers, page_size=500):
    """
    Fetch all assets with EXIF inline.
    withExif=True is required — without it exifInfo is omitted.
    Stops when a page returns fewer items than page_size.
    """
    all_assets, page = [], 1  # Accumulator list and page counter
    while True:
        resp = requests.post(f"{base_url}/api/search/metadata", headers=headers,
                             json={"page": page, "size": page_size, "withExif": True})
        resp.raise_for_status()
        items = resp.json().get("assets", {}).get("items", [])  # Safely navigate the nested response structure
        all_assets.extend(items)
        print(f"  Page {page}: +{len(items)}  (total: {len(all_assets)})")
        # A short page means we've reached the last page — stop paginating
        if len(items) < page_size:
            break
        page += 1
    print(f"\nFetch complete. Total: {len(all_assets)}")
    return all_assets

# ── Run the fetch ───────────────────────────────────────────────────────
print("Fetching all assets from Immich...")
raw_assets = fetch_all_assets(BASE_URL, HEADERS)

# Section 05c — bronze full refresh: upsert assets into bronze, flag deletions via SCD
import json
from pyspark.sql.functions import from_json, current_timestamp, lit, col
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

raw_schema    = StructType([StructField("raw_json", StringType(), True)])
id_schema     = StructType([StructField("id", StringType())])
BRONZE_ASSETS = f"{CATALOG}.immich_bronze.assets_raw"

# ── Build incoming DataFrame ─────────────────────────────────────────────
# Set of all asset IDs returned by Immich this run — used for deletion detection
incoming_ids = {a["id"] for a in raw_assets}

df_incoming = (
    spark.createDataFrame([(json.dumps(a),) for a in raw_assets], raw_schema)  # Serialise each asset dict to a JSON string — stored raw in bronze
    .withColumn("_p",          from_json(col("raw_json"), id_schema))
    .withColumn("asset_id",    col("_p.id"))
    .withColumn("ingested_at", current_timestamp())
    .withColumn("source",      lit("immich_api"))
    .withColumn("scd_is_deleted",  lit(False))  # All incoming assets are active — deletions are handled in step 2 below
    .withColumn("scd_deleted_at",  lit(None).cast(TimestampType()))
    .drop("_p")
)

# ── First run vs subsequent runs ────────────────────────────────────────
table_exists = spark.catalog.tableExists(BRONZE_ASSETS)

# First run: no table exists yet so just write directly
if not table_exists:
    print("First run — creating bronze table...")
    (df_incoming.write.format("delta").mode("overwrite").saveAsTable(BRONZE_ASSETS))
    print(f"  Wrote {df_incoming.count()} rows")
else:
    df_incoming.createOrReplaceTempView("incoming_assets")

    # Upsert: insert new rows, update changed rows
    spark.sql(f"""
        MERGE INTO {BRONZE_ASSETS} AS target
        USING incoming_assets AS source
        ON target.asset_id = source.asset_id
        WHEN MATCHED THEN UPDATE SET
            target.raw_json    = source.raw_json,
            target.ingested_at = source.ingested_at,
            target.scd_is_deleted  = false,
            target.scd_deleted_at  = NULL
        WHEN NOT MATCHED THEN INSERT *
    """)

    # Flag assets no longer in Immich as deleted
    # Build a comma-separated list of active IDs to use in the UPDATE predicate
    ids_list = ", ".join([f"'{i}'" for i in incoming_ids])
    spark.sql(f"""
        UPDATE {BRONZE_ASSETS}
        SET scd_is_deleted = true, scd_deleted_at = current_timestamp()
        WHERE scd_is_deleted = false
          AND asset_id NOT IN ({ids_list})
    """)

    # Summary stats
    df_b    = spark.table(BRONZE_ASSETS)
    total   = df_b.count()
    active  = df_b.where(col("scd_is_deleted") == False).count()
    deleted = total - active
    print(f"  Total rows : {total}")
    print(f"  Active     : {active}")
    print(f"  Deleted    : {deleted}")

# Section 05d — bronze optional: fetch albums from Immich and write to bronze (run 07b after this)
import json, requests
from pyspark.sql.functions import current_timestamp, lit
from pyspark.sql.types import StructType, StructField, StringType

raw_schema    = StructType([StructField("raw_json", StringType(), True)])
BRONZE_ALBUMS = f"{CATALOG}.immich_bronze.albums_raw"
print("Fetching albums...")
resp = requests.get(f"{BASE_URL}/api/albums", headers=HEADERS)
resp.raise_for_status()
raw_albums = resp.json()
print(f"  Found {len(raw_albums)} albums")

df_albums = (
    spark.createDataFrame([(json.dumps(a),) for a in raw_albums], raw_schema)
    .withColumn("ingested_at", current_timestamp())
    .withColumn("source",      lit("immich_api"))
)
# Albums are small — always full replace rather than merge
(df_albums.write.format("delta").mode("overwrite").saveAsTable(BRONZE_ALBUMS))
print(f"  Wrote {df_albums.count()} rows to bronze.albums_raw")

# Section 05e — bronze optional: fetch named people and their photo counts (run 07d after this)
import json, requests
from pyspark.sql.functions import current_timestamp, lit
from pyspark.sql.types import StructType, StructField, StringType

raw_schema     = StructType([StructField("raw_json", StringType(), True)])
BRONZE_PEOPLE  = f"{CATALOG}.immich_bronze.people_raw"

# Fetch all named people — Immich returns unnamed/hidden people too so we filter
print("Fetching people...")
resp = requests.get(
    f"{BASE_URL}/api/people",
    headers=HEADERS,
    params={"withHidden": False}
)
resp.raise_for_status()
all_people = resp.json().get("people", [])

# Keep only people who have been given a name in Immich
named_people = [p for p in all_people if p.get("name")]
print(f"  Total people detected : {len(all_people)}")
print(f"  Named people          : {len(named_people)}")

# Fetch photo count per person from the statistics endpoint
print("  Fetching per-person statistics...")
for person in named_people:
    stats_resp = requests.get(
        f"{BASE_URL}/api/people/{person['id']}/statistics",
        headers=HEADERS
    )
    if stats_resp.status_code == 200:
        person["statistics"] = stats_resp.json()  # adds assets count
    else:
        person["statistics"] = {"assets": 0}

# Write to bronze — always full replace (small dataset)
df_people = (
    spark.createDataFrame([(json.dumps(p),) for p in named_people], raw_schema)
    .withColumn("ingested_at", current_timestamp())
    .withColumn("source",      lit("immich_api"))
)
(df_people.write.format("delta").mode("overwrite").saveAsTable(BRONZE_PEOPLE))
print(f"  Wrote {df_people.count()} rows to bronze.people_raw")

# Section 05f — bronze incremental: fetch only assets updated since last run and merge into bronze
import json
from datetime import timedelta
from pyspark.sql.functions import from_json, current_timestamp, lit, col
from pyspark.sql.types import StructType, StructField, StringType, TimestampType

raw_schema    = StructType([StructField("raw_json", StringType(), True)])
BRONZE_ASSETS = f"{CATALOG}.immich_bronze.assets_raw"

# ── Find last ingestion timestamp ───────────────────────────────────────
last_run = spark.sql(f"""
    SELECT MAX(ingested_at) AS ts FROM {BRONZE_ASSETS} WHERE scd_is_deleted = false
""").collect()[0]["ts"]

if last_run is None:
    print("No bronze data found — run the full refresh (cell 5) first.")
else:
    # 1hr buffer for clock skew
    since = (last_run - timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%S.000Z")
    print(f"Fetching assets updated since: {since}")

    # ── Fetch only changed/new assets ───────────────────────────────────
    updated, page, page_size = [], 1, 500
    while True:
        resp = requests.post(f"{BASE_URL}/api/search/metadata", headers=HEADERS,
                             json={"page": page, "size": page_size,
                                   "withExif": True, "updatedAfter": since})
        resp.raise_for_status()
        items = resp.json().get("assets", {}).get("items", [])  # Safely navigate the nested response structure
        updated.extend(items)
        print(f"  Page {page}: +{len(items)}  (total: {len(updated)})")
        # A short page means we've reached the last page — stop paginating
        if len(items) < page_size:
            break
        page += 1

    print(f"\n{len(updated)} new/updated assets found")

    # ── Merge into bronze ───────────────────────────────────────────────
    # Only proceed with the merge if there is actually something new
    if updated:
        df_new = (
            spark.createDataFrame([(json.dumps(a),) for a in updated], raw_schema)
            .withColumn("_p", from_json(col("raw_json"), StructType([StructField("id", StringType())])))
            .withColumn("asset_id",    col("_p.id"))
            .withColumn("ingested_at", current_timestamp())
            .withColumn("source",      lit("immich_api"))
            .withColumn("scd_is_deleted",  lit(False))  # All incoming assets are active — deletions are handled in step 2 below
            .withColumn("scd_deleted_at",  lit(None).cast(TimestampType()))
            .drop("_p")
        )
        df_new.createOrReplaceTempView("incoming_assets")
        spark.sql(f"""
            MERGE INTO {BRONZE_ASSETS} AS target
            USING incoming_assets AS source
            ON target.asset_id = source.asset_id
            WHEN MATCHED THEN UPDATE SET
                target.raw_json    = source.raw_json,
                target.ingested_at = source.ingested_at,
                target.scd_is_deleted  = false,
                target.scd_deleted_at  = NULL
            WHEN NOT MATCHED THEN INSERT *
        """)
        print("  Merged into bronze.")
    else:
        print("  Nothing to update.")

# Section 06 — verify: inspect bronze table schema, row counts, and SCD deletion state
df = spark.table(f"{CATALOG}.immich_bronze.assets_raw")
print("=== bronze.assets_raw schema ===")
df.printSchema()

# Count active vs SCD-deleted rows
total   = df.count()
active  = df.where(col("scd_is_deleted") == False).count()
deleted = df.where(col("scd_is_deleted") == True).count()
print(f"\n  Total rows : {total}")
print(f"  Active     : {active}")
print(f"  Deleted    : {deleted}")

# ── Show recently SCD-deleted assets if any ─────────────────────────────
if deleted > 0:
    print("\n=== Recently deleted assets ===")
    (df.where(col("scd_is_deleted") == True)
       .select("asset_id", "scd_deleted_at", "ingested_at")
       .orderBy(col("scd_deleted_at").desc())
       .limit(5).show(truncate=False))

# Preview one raw JSON row to confirm the structure looks right
sample = df.where(col("scd_is_deleted") == False).limit(1).collect()[0]["raw_json"]
print("\n=== Sample raw_json (first 400 chars) ===")
print(sample[:400] + "...")

# Section 07a — silver: parse raw JSON into typed columns, filter SCD-deleted rows
from pyspark.sql.functions import (
    col, from_json, to_timestamp, year, month,
    dayofmonth, lower, current_timestamp, count, when
)
from pyspark.sql.types import (
    StructType, StructField, StringType, LongType,
    BooleanType, DoubleType, IntegerType
)

# Define the schema for the nested exifInfo object
# Field names must match exactly what Immich returns (case-sensitive)
exif_schema = StructType([
    StructField("make",             StringType()),
    StructField("model",            StringType()),
    StructField("lensModel",        StringType()),
    StructField("fNumber",          DoubleType()),
    StructField("focalLength",      DoubleType()),
    StructField("iso",              IntegerType()),
    StructField("exposureTime",     StringType()),
    StructField("latitude",         DoubleType()),
    StructField("longitude",        DoubleType()),
    StructField("city",             StringType()),
    StructField("state",            StringType()),
    StructField("country",          StringType()),
    StructField("description",      StringType()),
    StructField("timeZone",         StringType()),
    StructField("dateTimeOriginal", StringType()),
    StructField("rating",           IntegerType()),
    StructField("exifImageWidth",   IntegerType()),
    StructField("exifImageHeight",  IntegerType()),
    StructField("fileSizeInByte",   LongType()),
])
# Top-level asset schema — exifInfo is nested inside as a struct
asset_schema = StructType([
    StructField("id",               StringType()),
    StructField("originalFileName", StringType()),
    StructField("type",             StringType()),
    StructField("fileCreatedAt",    StringType()),
    StructField("localDateTime",    StringType()),
    StructField("isFavorite",       BooleanType()),
    StructField("isArchived",       BooleanType()),
    StructField("visibility",       StringType()),
    StructField("duration",         StringType()),
    StructField("exifInfo",         exif_schema),
])

# ── Parse and flatten ───────────────────────────────────────────────────
print("Parsing bronze assets to silver...")
df_silver = (
    spark.table(f"{CATALOG}.immich_bronze.assets_raw")
    .where(col("scd_is_deleted") == False)  # Only process active assets — SCD-flagged deletions stay in bronze only
    .withColumn("data", from_json(col("raw_json"), asset_schema))  # Parse the raw JSON string into a typed struct column
    .select(
        col("data.id").alias("asset_id"),
        col("data.originalFileName").alias("file_name"),
        lower(col("data.type")).alias("media_type"),
        to_timestamp(col("data.fileCreatedAt")).alias("created_at"),
        to_timestamp(col("data.localDateTime")).alias("local_datetime"),
        year(to_timestamp(col("data.localDateTime"))).alias("year"),
        month(to_timestamp(col("data.localDateTime"))).alias("month"),
        dayofmonth(to_timestamp(col("data.localDateTime"))).alias("day"),
        col("data.isFavorite").alias("is_favourite"),
        col("data.isArchived").alias("is_archived"),
        col("data.visibility").alias("visibility"),
        col("data.duration").alias("video_duration"),
        col("data.exifInfo.make").alias("camera_make"),
        col("data.exifInfo.model").alias("camera_model"),
        col("data.exifInfo.lensModel").alias("lens_model"),
        col("data.exifInfo.fNumber").alias("f_number"),
        col("data.exifInfo.iso").alias("iso"),
        col("data.exifInfo.latitude").alias("latitude"),
        col("data.exifInfo.longitude").alias("longitude"),
        col("data.exifInfo.city").alias("city"),
        col("data.exifInfo.country").alias("country"),
        col("data.exifInfo.fileSizeInByte").alias("file_size_bytes"),
        col("data.exifInfo.exifImageWidth").alias("width_px"),
        col("data.exifInfo.exifImageHeight").alias("height_px"),
        col("data.exifInfo.timeZone").alias("timezone"),
        col("data.exifInfo.rating").alias("rating"),
        col("ingested_at")
    )
    .dropDuplicates(["asset_id"])  # Defensive dedup in case of any duplicate IDs
    .where(col("asset_id").isNotNull())  # Drop any rows where JSON parsing failed
)
# ── Write and summarise ─────────────────────────────────────────────────
(df_silver.write.format("delta").mode("overwrite").saveAsTable(f"{CATALOG}.immich_silver.assets"))

# Summary counts
total    = df_silver.count()
with_gps = df_silver.where(col("latitude").isNotNull()).count()
rated    = df_silver.where(col("rating").isNotNull()).count()
print(f"  Total  : {total}")
print(f"  GPS    : {with_gps} ({round(with_gps/total*100)}%)")
print(f"  Rated  : {rated} ({round(rated/total*100)}%)")
print(f"\nWritten to: {CATALOG}.immich_silver.assets")

# Section 07b — silver optional: parse bronze albums into typed silver table
# Requires cell 05d (album fetch) to have been run first.
# Skips gracefully if bronze.albums_raw doesn't exist yet.
if not spark.catalog.tableExists(f"{CATALOG}.immich_bronze.albums_raw"):
    print("Skipping — bronze.albums_raw not found. Run cell 05d first.")
else:
    album_schema = StructType([
        StructField("id",            StringType()),
        StructField("albumName",     StringType()),
        StructField("description",   StringType()),
        StructField("assetCount",    IntegerType()),
        StructField("startDate",     StringType()),
        StructField("endDate",       StringType()),
        StructField("shared",        BooleanType()),
        StructField("hasSharedLink", BooleanType()),
    ])
    # Parse raw JSON into typed columns and write to silver
    df_albums = (
        spark.table(f"{CATALOG}.immich_bronze.albums_raw")
        .withColumn("data", from_json(col("raw_json"), album_schema))
        .select(
            col("data.id").alias("album_id"),
            col("data.albumName").alias("album_name"),
            col("data.description").alias("description"),
            col("data.assetCount").alias("asset_count"),
            to_timestamp(col("data.startDate")).alias("start_date"),
            to_timestamp(col("data.endDate")).alias("end_date"),
            col("data.shared").alias("is_shared"),
            col("data.hasSharedLink").alias("has_shared_link"),
            col("ingested_at")
        )
        .dropDuplicates(["album_id"])
    )
    (df_albums.write.format("delta").mode("overwrite").saveAsTable(f"{CATALOG}.immich_silver.albums"))
    print(f"  Albums : {df_albums.count()}")
    print(f"  Shared : {df_albums.where(col('is_shared') == True).count()}")

# Section 07c — silver: strip CBD / city centre suffixes from city names in place
from pyspark.sql.functions import regexp_replace, col

# Strip common CBD / city centre suffixes from city names:
# "Melbourne CBD" → "Melbourne"
# "Sydney city centre" → "Sydney"
# "Brisbane Central Business District" → "Brisbane"
# Read silver, apply the regex substitution, write back in place
df_cleaned = spark.table(f"{CATALOG}.immich_silver.assets").withColumn(
    "city",
    regexp_replace(col("city"), " (CBD|city centre|Central Business District)$", "")
)

(df_cleaned.write
    .format("delta")
    .mode("overwrite")
    .saveAsTable(f"{CATALOG}.immich_silver.assets"))

# Show distinct cities so you can spot-check the result
print("City clean-up applied. Distinct cities now in silver:")
spark.table(f"{CATALOG}.immich_silver.assets").select("city").distinct().orderBy("city").limit(30).show(truncate=False)

# Section 07d — silver optional: parse bronze people into typed silver table
from pyspark.sql.functions import col, from_json, current_timestamp
from pyspark.sql.types import StructType, StructField, StringType, BooleanType, IntegerType

# Only runs if cell 05e has been run first
if not spark.catalog.tableExists(f"{CATALOG}.immich_bronze.people_raw"):
    print("Skipping — bronze.people_raw not found. Run cell 05e first.")
else:
    stats_schema = StructType([StructField("assets", IntegerType())])
    person_schema = StructType([
        StructField("id",          StringType()),
        StructField("name",        StringType()),
        StructField("birthDate",   StringType()),
        StructField("isFavorite",  BooleanType()),
        StructField("isHidden",    BooleanType()),
        StructField("statistics",  stats_schema),
    ])

    df_people = (
        spark.table(f"{CATALOG}.immich_bronze.people_raw")
        .withColumn("data", from_json(col("raw_json"), person_schema))
        .select(
            col("data.id").alias("person_id"),
            col("data.name").alias("name"),
            col("data.birthDate").alias("birth_date"),
            col("data.isFavorite").alias("is_favourite"),
            col("data.isHidden").alias("is_hidden"),
            col("data.statistics.assets").alias("photo_count"),
            col("ingested_at")
        )
        .dropDuplicates(["person_id"])
        .where(col("person_id").isNotNull())
    )
    (df_people.write.format("delta").mode("overwrite").saveAsTable(f"{CATALOG}.immich_silver.people"))
    print(f"  People : {df_people.count()}")
    df_people.show(truncate=False)

# Section 08 — verify: check silver schema, null rates, and sample rows
df = spark.table(f"{CATALOG}.immich_silver.assets")
print("=== silver.assets schema ===")
df.printSchema()

print("=== 5 sample rows ===")
(df.select("asset_id", "file_name", "media_type", "year", "month",
           "camera_make", "camera_model", "city", "country",
           "file_size_bytes", "rating")
   .limit(5).show(truncate=35))

print("=== Null counts in key columns ===")
# Check null rates on columns that are commonly sparse
cols_to_check = ["camera_make", "camera_model", "city", "country",
                 "latitude", "longitude", "year", "file_size_bytes",
                 "rating", "timezone"]
# Aggregate null count for each column in one pass
null_counts = df.select([
    count(when(col(c).isNull(), c)).alias(c) for c in cols_to_check
]).collect()[0].asDict()
total = df.count()
print(f"  (out of {total} rows)")
# Print each column's null rate as a percentage
for c, n in null_counts.items():
    pct = round(n / total * 100)
    print(f"  {c:<20} {n:>6} nulls  ({pct}%)")

# Section 08b — photos to delete in Immich
# All photos should be rated 1 (first place) before upload
# Anything unrated or rated 2+ should not exist — find them here then go delete in Immich
df = spark.table(f"{CATALOG}.immich_silver.assets")
total = df.count()

# Photos that are unrated or not first place
to_delete = df.where(
    col("rating").isNull() | (col("rating") != 1)
)
n = to_delete.count()
print(f"Photos to delete : {n} of {total} ({round(n/total*100)}%)")

if n == 0:
    print("  All good — every photo is rated 1.")
else:
    # Break down by rating value so you know what you're dealing with
    print("\n=== Breakdown by rating ===")
    (to_delete
        .groupBy("rating")
        .count()
        .orderBy("rating")
        .show())

    # Break down by year to see when the problem photos are from
    print("=== Breakdown by year ===")
    (to_delete
        .groupBy("year")
        .count()
        .orderBy("year")
        .show())

    # Full list — take this to Immich and delete these
    print("=== Full list (most recent first) ===")
    (to_delete
        .select("file_name", "rating", "local_datetime", "city", "country")
        .orderBy(col("local_datetime").desc())
        .show(50, truncate=False))

# Section 09 — debug: run this if gold tables return no rows after 09a–09f
# Run this before 9a to confirm silver has data and the gold query will work
df = spark.table(f"{CATALOG}.immich_silver.assets")

print("=== Row count ===")
print(f"  Total rows : {df.count()}")

print("=== year distinct values ===")
df.groupBy("year").count().orderBy("year").show()

print("=== Direct test of 9a query logic ===")
spark.sql(f"""
    SELECT year, month, COUNT(*) AS asset_count
    FROM {CATALOG}.immich_silver.assets
        GROUP BY year, month
    ORDER BY year, month
    LIMIT 5
""").show()

print("=== Current schema context ===")
print(spark.sql("SELECT current_catalog(), current_schema()").collect()[0])

-- Section 09a — gold: photo counts aggregated by year and month
-- Count photos per year/month with storage and favourite stats
-- Library is images only so no media_type filter needed
CREATE OR REPLACE TABLE immich_gold.photos_by_month AS
SELECT
    year,
    month,
    COUNT(*)                                    AS photo_count,
    COUNT(*) FILTER (WHERE is_favourite = true) AS favourite_count,
    ROUND(SUM(file_size_bytes) / 1e9, 2)        AS total_size_gb,
    COUNT(DISTINCT camera_make)                 AS distinct_cameras
FROM immich_silver.assets
GROUP BY year, month
ORDER BY year, month;

-- Section 09b — gold: photo counts by country and city with avg GPS coordinates
-- Aggregate photo counts by country and city
-- avg_lat/avg_lon: NULL for rows with no GPS data (used by Leaflet map in section 13)
CREATE OR REPLACE TABLE immich_gold.geography AS
SELECT
    COALESCE(country, 'No GPS data') AS country,  -- group nulls together
    city,
    COUNT(*)                         AS photo_count,
    COUNT(DISTINCT camera_model)     AS cameras_used,
    MIN(local_datetime)              AS earliest_photo,
    MAX(local_datetime)              AS latest_photo,
    AVG(latitude)                    AS avg_lat,   -- NULL when no GPS data
    AVG(longitude)                   AS avg_lon    -- NULL when no GPS data
FROM immich_silver.assets
GROUP BY country, city
ORDER BY photo_count DESC;

-- Section 09c — gold: shooting stats per camera body
-- Summarise shooting stats per camera body
-- Useful for tracking when you switched gear and how each body was used
CREATE OR REPLACE TABLE immich_gold.camera_summary AS
SELECT
    COALESCE(camera_make,  'Unknown') AS make,
    COALESCE(camera_model, 'Unknown') AS model,
    COUNT(*)                          AS photo_count,
    ROUND(AVG(f_number), 1)           AS avg_f_number,
    ROUND(AVG(iso), 0)                AS avg_iso,
    MIN(year)                         AS first_used_year,
    MAX(year)                         AS last_used_year
FROM immich_silver.assets
GROUP BY camera_make, camera_model
ORDER BY photo_count DESC;

-- Section 09d — gold: rating distribution with action flag for non-first-place photos
-- Rating convention: 1 = first place (best), 5 = last
-- This library should only contain rating=1 photos
-- The action column flags anything else for review/deletion in Immich
CREATE OR REPLACE TABLE immich_gold.ratings_summary AS
SELECT
    rating,
    COUNT(*)                     AS photo_count,
    COUNT(DISTINCT country)      AS countries,
    COUNT(DISTINCT camera_model) AS cameras,
    MIN(year)                    AS earliest_year,
    MAX(year)                    AS latest_year,
    CASE WHEN rating = 1 THEN 'OK' ELSE 'Review / delete' END AS action
FROM immich_silver.assets
WHERE rating IS NOT NULL
GROUP BY rating
ORDER BY rating ASC;

-- Section 09e — gold optional: album stats (requires cells 05d and 07b)
-- Only run if albums were fetched (cell 05d) and parsed (cell 07b)
CREATE OR REPLACE TABLE immich_gold.album_stats AS
SELECT
    album_name,
    asset_count,
    is_shared,
    has_shared_link,
    start_date,
    end_date,
    DATEDIFF(end_date, start_date) AS date_span_days  -- days from first to last photo in the album
FROM immich_silver.albums
ORDER BY asset_count DESC;

-- Section 09f — gold optional: people ranked by photo count (requires cells 05e and 07d)
-- Ranked list of named people by photo count
-- Only run if bronze.people_raw was populated by cell 05e
CREATE OR REPLACE TABLE immich_gold.people_summary AS
SELECT
    name,
    photo_count,
    birth_date,
    is_favourite,
    -- rank by photo count so the most-photographed person is first
    RANK() OVER (ORDER BY photo_count DESC) AS rank
FROM immich_silver.people
ORDER BY photo_count DESC;

-- Section 09g — gold: lens usage stats
-- Separate table from camera_summary since one body is used with multiple lenses
-- Joined back to camera_summary to show which body each lens was used with
CREATE OR REPLACE TABLE immich_gold.lens_summary AS
SELECT
    COALESCE(lens_model, 'Unknown')  AS lens,
    COALESCE(camera_make, 'Unknown') AS camera_make,
    COALESCE(camera_model, 'Unknown') AS camera_model,
    COUNT(*)                           AS photo_count,
    ROUND(AVG(f_number), 1)            AS avg_f_number,
    ROUND(AVG(iso), 0)                 AS avg_iso,
    MIN(year)                          AS first_used_year,
    MAX(year)                          AS last_used_year
FROM immich_silver.assets
GROUP BY lens_model, camera_make, camera_model
ORDER BY photo_count DESC;

-- Section 10a — explore: total photos per year
-- Total photos per year — quick overview of library growth over time
SELECT year, SUM(photo_count) AS total_photos
FROM immich_gold.photos_by_month
GROUP BY year
ORDER BY year;

-- Section 10b — explore: countries ranked by photo count
-- Countries ranked by photo count — excludes assets with no GPS data
SELECT country, SUM(photo_count) AS photos
FROM immich_gold.geography
WHERE country != 'No GPS data'
GROUP BY country
ORDER BY photos DESC
LIMIT 20;

-- Section 10c — explore: cities ranked by photo count
-- Cities ranked by photo count — only rows where city was resolved from GPS
SELECT city, country, SUM(photo_count) AS photos
FROM immich_gold.geography
WHERE city IS NOT NULL
  AND country != 'No GPS data'
GROUP BY city, country
ORDER BY photos DESC
LIMIT 30;

-- Section 10d — explore: first-place photos by location, highlights non-1-star for review
-- Where are the first-place photos from?
-- needs_review highlights any non-1-star rated photos by location
SELECT
    country,
    city,
    COUNT(*) FILTER (WHERE rating = 1)                         AS first_place,
    COUNT(*) FILTER (WHERE rating IS NOT NULL AND rating <> 1) AS needs_review,
    COUNT(*) FILTER (WHERE rating IS NOT NULL)                AS total_rated
FROM immich_silver.assets
WHERE country IS NOT NULL
GROUP BY country, city
ORDER BY first_place DESC
LIMIT 20;

# Section 10e — explore: matplotlib bar chart of photos per year
import matplotlib.pyplot as plt
df = spark.sql(f"""
    SELECT year, SUM(asset_count) AS photos
    FROM {CATALOG}.immich_gold.photos_by_month
    WHERE year IS NOT NULL
    GROUP BY year ORDER BY year
""").toPandas()
fig, ax = plt.subplots(figsize=(12, 4))
ax.bar(df["year"], df["photos"], color="#1D9E75", width=0.7)
ax.set_xlabel("Year"); ax.set_ylabel("Photos"); ax.set_title("Photos per year")
for _, row in df.iterrows():
    ax.text(row["year"], row["photos"] + 20, str(row["photos"]), ha="center", fontsize=9)
plt.tight_layout(); display(fig)

# Section 10f — explore: matplotlib horizontal bar chart of top countries
df = spark.sql(f"""
    SELECT country, SUM(photo_count) AS photos
    FROM {CATALOG}.immich_gold.geography
    WHERE country != 'No GPS data'
    GROUP BY country ORDER BY photos DESC LIMIT 15
""").toPandas()
fig, ax = plt.subplots(figsize=(10, 5))
ax.barh(df["country"][::-1], df["photos"][::-1], color="#378ADD")
ax.set_xlabel("Photos"); ax.set_title("Top 15 countries")
plt.tight_layout(); display(fig)

-- Section 10g — explore: join lens_summary with camera_summary to see
-- which lenses were used on which bodies, and how each combination performed
SELECT
    l.lens,
    l.camera_make || ' ' || l.camera_model  AS camera,
    l.photo_count                            AS lens_photos,
    c.photo_count                            AS body_total_photos,
    -- what percentage of this body's shots used this lens
    ROUND(l.photo_count * 100.0 / c.photo_count, 1) AS pct_of_body,
    l.avg_f_number,
    l.avg_iso,
    l.first_used_year,
    l.last_used_year
FROM immich_gold.lens_summary l
JOIN immich_gold.camera_summary c
  ON  l.camera_make  = c.make
  AND l.camera_model = c.model
WHERE l.lens != 'Unknown'
  AND c.make       != 'Unknown'
ORDER BY l.photo_count DESC;

0 3 * * *       # every day at 3am       ← use for daily incremental
0 3 * * 0       # every sunday at 3am    ← use for weekly full refresh
0 */6 * * *     # every 6 hours
0 3 1 * *       # first of every month at 3am
0 3 * * 1-5     # weekdays only at 3am

SELECT year, SUM(photo_count) AS total_photos
FROM workspace.immich_gold.photos_by_month
WHERE year IS NOT NULL
GROUP BY year ORDER BY year;

SELECT year, month, SUM(photo_count) AS photos
FROM workspace.immich_gold.photos_by_month
WHERE year IS NOT NULL
GROUP BY year, month ORDER BY year, month;

SELECT country, SUM(photo_count) AS photos
FROM workspace.immich_gold.geography
WHERE country != 'No GPS data'
GROUP BY country ORDER BY photos DESC LIMIT 20;

SELECT city || ', ' || country AS location,
       SUM(photo_count) AS photos
FROM workspace.immich_gold.geography
WHERE city IS NOT NULL AND country != 'No GPS data'
GROUP BY city, country ORDER BY photos DESC LIMIT 30;

SELECT make || ' ' || model AS camera, photo_count, first_used_year, last_used_year
FROM workspace.immich_gold.camera_summary
WHERE make != 'Unknown'
ORDER BY photo_count DESC LIMIT 15;

SELECT rating, photo_count, action
FROM workspace.immich_gold.ratings_summary
ORDER BY rating ASC;

SELECT lens, camera_make, camera_model, photo_count,
       avg_f_number, avg_iso, first_used_year, last_used_year
FROM workspace.immich_gold.lens_summary
WHERE lens != 'Unknown'
ORDER BY photo_count DESC;

SELECT
    SUM(photo_count)                          AS total_photos,
    ROUND(SUM(total_size_gb), 1)                AS total_gb,
    (SELECT COUNT(DISTINCT country) FROM workspace.immich_gold.geography
     WHERE country != 'No GPS data')         AS countries_visited,
    (SELECT COUNT(*) FROM workspace.immich_silver.assets
     WHERE rating = 1)                            AS first_place_photos
FROM workspace.immich_gold.photos_by_month;

# Section 13a — map: export geography gold table to JSON for the Leaflet map
import json

# Pull geography rows that have GPS coordinates
geo_df = spark.sql(f"""
    SELECT country, city,
           ROUND(avg_lat, 4) AS lat,
           ROUND(avg_lon, 4) AS lon,
           photo_count, earliest_photo, latest_photo
    FROM {CATALOG}.immich_gold.geography
    WHERE avg_lat IS NOT NULL AND avg_lon IS NOT NULL
      AND country != 'No GPS data'
    ORDER BY photo_count DESC
""").toPandas()

total_photos    = int(geo_df["photo_count"].sum())
total_locations = len(geo_df)
print(f"Locations with GPS : {total_locations}")
print(f"Photos mapped      : {total_photos}")

# Serialise to JSON — embedded directly in the app
locations_json = json.dumps(geo_df.to_dict(orient="records"))

# Save to FileStore so the app can read it
dbutils.fs.put("/FileStore/immich_map_data.json", locations_json, overwrite=True)
print("Saved to /FileStore/immich_map_data.json")

# Section 13b — map: build self-contained Leaflet HTML and save to FileStore
# Build a self-contained dark-themed Leaflet map with circle markers
# Marker size is proportional to photo count at that location
html_content = f"""
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Photo Map</title>
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/leaflet/1.9.4/leaflet.min.css">
<style>
  *{{box-sizing:border-box;margin:0;padding:0}}
  body{{background:#0f1117;color:#e8eaf0;font-family:system-ui,sans-serif}}
  #header{{padding:1rem 1.5rem;background:#161b27;border-bottom:1px solid rgba(255,255,255,.08)}}
  #header h1{{font-size:18px;font-weight:400}}
  #header p{{font-size:13px;color:#8b90a0;margin-top:.2rem}}
  #map{{height:calc(100vh - 62px)}}
  .leaflet-popup-content-wrapper{{background:#1e2535;border:1px solid rgba(255,255,255,.12);color:#e8eaf0;border-radius:8px}}
  .leaflet-popup-tip{{background:#1e2535}}
  .popup-city{{font-size:15px;font-weight:500;margin-bottom:.3rem}}
  .popup-count{{font-size:22px;font-weight:300;color:#4ade9e}}
  .popup-label{{font-size:11px;color:#8b90a0;text-transform:uppercase;letter-spacing:.06em}}
  .popup-dates{{font-size:12px;color:#8b90a0;margin-top:.4rem}}
</style>
</head>
<body>
<div id="header">
  <h1>Photo Map</h1>
  <p>{total_photos:,} photos &middot; {total_locations} locations</p>
</div>
<div id="map"></div>
<script src="https://cdnjs.cloudflare.com/ajax/libs/leaflet/1.9.4/leaflet.min.js"></script>
<script>
const locations = {locations_json};
const map = L.map('map').setView([20,10],2);
L.tileLayer('https://{{s}}.basemaps.cartocdn.com/dark_all/{{z}}/{{x}}/{{y}}{{r}}.png',{{
  attribution:'&copy; OpenStreetMap &copy; CARTO',subdomains:'abcd',maxZoom:19
}}).addTo(map);
const maxCount = Math.max(...locations.map(l=>l.photo_count));
locations.forEach(loc=>{{
  if(!loc.lat||!loc.lon)return;
  const r = Math.max(6, Math.sqrt(loc.photo_count/maxCount)*40);
  const circle = L.circleMarker([loc.lat,loc.lon],{{
    radius:r,fillColor:'#4ade9e',color:'#4ade9e',
    weight:1,opacity:0.8,fillOpacity:0.25
  }}).addTo(map);
  const city  = loc.city||loc.country;
  const from  = loc.earliest_photo?loc.earliest_photo.substring(0,10):'';
  const to    = loc.latest_photo?loc.latest_photo.substring(0,10):'';
  circle.bindPopup(`
    <div class="popup-city">${{city}}</div>
    <div class="popup-label">${{loc.country}}</div>
    <div class="popup-count">${{loc.photo_count.toLocaleString()}}</div>
    <div class="popup-label">photos</div>
    <div class="popup-dates">${{from}} &rarr; ${{to}}</div>
  `);
}});
</script>
</body>
</html>"""

# Save to FileStore
dbutils.fs.put("/FileStore/photo_map.html", html_content, overwrite=True)
print("Saved to /FileStore/photo_map.html")
print("Download via: Catalog → Files → FileStore → photo_map.html")