"""
assets_store.py — SQLite asset inventory store (DVI-1185 P1, board-approved
Option B: Togen is the source of truth for the Asset File data).

Holds the asset inventory previously maintained in the SharePoint "Asset File"
workbook (4 team sheets + Master Data controlled vocabulary). The importer
reads that workbook once (and on demand for re-import during the transition
window), flagging data-quality exceptions instead of silently fixing or
dropping rows. Search/filtering for the Equipment tool Assets tab and future
consumers (WO pickers, Oversight, export) read from here.

Flask-independent (mirrors scada_store.py) so app.py, scripts, and tests can
all import it. assets.db lives alongside the app's other state files.

Schema
------
assets(asset_tag UNIQUE, tag_num, asset_type, description, area, form_id,
       length, width, height, manufacturer, model_no, serial_no, notes,
       material, source_sheet, extra, sp_hash) — one row per asset. tag_num
    is the numeric form of the tag when it is all digits (sort key). extra
    keeps unmapped source columns as JSON so imports are lossless. sp_hash
    fingerprints the source row as of the last import/sync (P3 merge guard).
vocab(kind, value, sort) — controlled vocabulary from the Master Data sheet
    (kinds: asset_type, form_id, area), in sheet order.
meta(key, value) — import bookkeeping (last_import_at/source/report).
"""

import hashlib
import json
import os
import re
import sqlite3
import threading
from datetime import datetime
from pathlib import Path

ASSETS_DB_FILE = Path(
    os.environ.get("ASSETS_DB_FILE", Path(__file__).resolve().parent / "assets.db"))

# Extracted/attached asset photos live here as <dir>/<asset_tag>/<filename>
# (DVI-1185 P2); the photos table is the index over this tree.
ASSETS_PHOTO_DIR = Path(
    os.environ.get("ASSETS_PHOTO_DIR",
                   Path(__file__).resolve().parent / "asset_photos"))

# Serialize writes within one process; SQLite handles cross-process locking.
_write_lock = threading.Lock()

# Detail rows in the exceptions report are capped so a pathological workbook
# can't bloat the stored report; per-issue counts always stay complete.
EXCEPTION_DETAIL_CAP = 500

VOCAB_KINDS = ("asset_type", "form_id", "area")

# Master Data is a single column with section headers introducing each vocab.
_MASTER_SECTIONS = {
    "asset types": "asset_type",
    "form id's": "form_id",
    "form ids": "form_id",
    "area": "area",
    "areas": "area",
}

# Normalized source header -> assets column. Photo columns are deliberately
# unmapped: the workbook's photo cells hold stale #VALUE! errors (linkage is
# already lost); real photo attachments arrive in P2.
_HEADER_MAP = {
    "assettag": "asset_tag",
    "assettagno": "asset_tag",
    "assettagnumber": "asset_tag",
    "assettype": "asset_type",
    "description": "description",
    "area": "area",
    "formid": "form_id",
    "formids": "form_id",
    "length": "length",
    "lengthin": "length",
    "width": "width",
    "widthin": "width",
    "height": "height",
    "heightin": "height",
    "manufacturer": "manufacturer",
    "model": "model_no",
    "modelno": "model_no",
    "modelnumber": "model_no",
    "serial": "serial_no",
    "serialno": "serial_no",
    "serialnumber": "serial_no",
    "notes": "notes",
    "material": "material",
}

_PHOTO_HEADER_RE = re.compile(r"^photo\d*$")

ASSET_FIELDS = (
    "asset_tag", "asset_type", "description", "area", "form_id",
    "length", "width", "height", "manufacturer", "model_no", "serial_no",
    "notes", "material", "source_sheet",
)

_SEARCH_COLUMNS = (
    "asset_tag", "description", "manufacturer", "model_no", "serial_no",
    "form_id", "notes",
)


def _connect():
    conn = sqlite3.connect(str(ASSETS_DB_FILE), timeout=30)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=10000")
    return conn


def _add_column_if_missing(conn, cols, column, ddl):
    """Additive ALTER TABLE assets ADD COLUMN, concurrency-safe.

    `_write_lock` is a threading.Lock — it only serializes threads within one
    process, not the several gunicorn worker processes that boot concurrently on
    a deploy restart. Each worker reads `cols` (a PRAGMA table_info snapshot),
    sees the column missing, and races on the ALTER; N-1 lose and would crash
    with `OperationalError: duplicate column name` (DVI-1488, seen DVI-1486).
    The ADD COLUMN is inherently idempotent, so swallowing exactly that error is
    the KISS fix: whichever worker wins adds the column, the losers no-op. Any
    other OperationalError still propagates."""
    if column in cols:
        return
    try:
        conn.execute("ALTER TABLE assets ADD COLUMN %s %s" % (column, ddl))
    except sqlite3.OperationalError as exc:
        if "duplicate column name" not in str(exc).lower():
            raise


def init_assets_db():
    """Create tables/indexes if missing. Idempotent; call at import/startup."""
    with _write_lock, _connect() as conn:
        conn.execute(
            """CREATE TABLE IF NOT EXISTS assets (
                   id INTEGER PRIMARY KEY AUTOINCREMENT,
                   asset_tag TEXT NOT NULL UNIQUE,
                   tag_num INTEGER,
                   asset_type TEXT NOT NULL DEFAULT '',
                   description TEXT NOT NULL DEFAULT '',
                   area TEXT NOT NULL DEFAULT '',
                   form_id TEXT NOT NULL DEFAULT '',
                   length TEXT NOT NULL DEFAULT '',
                   width TEXT NOT NULL DEFAULT '',
                   height TEXT NOT NULL DEFAULT '',
                   manufacturer TEXT NOT NULL DEFAULT '',
                   model_no TEXT NOT NULL DEFAULT '',
                   serial_no TEXT NOT NULL DEFAULT '',
                   notes TEXT NOT NULL DEFAULT '',
                   material TEXT NOT NULL DEFAULT '',
                   source_sheet TEXT NOT NULL DEFAULT '',
                   extra TEXT NOT NULL DEFAULT '',
                   created_at TEXT NOT NULL DEFAULT (datetime('now')),
                   updated_at TEXT NOT NULL DEFAULT (datetime('now'))
               )""")
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_assets_type ON assets(asset_type)")
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_assets_area ON assets(area)")
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_assets_form ON assets(form_id)")
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_assets_tagnum ON assets(tag_num)")
        conn.execute(
            """CREATE TABLE IF NOT EXISTS vocab (
                   kind TEXT NOT NULL,
                   value TEXT NOT NULL,
                   sort INTEGER NOT NULL DEFAULT 0,
                   UNIQUE(kind, value)
               )""")
        conn.execute(
            "CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT)")
        conn.execute(
            """CREATE TABLE IF NOT EXISTS photos (
                   asset_tag TEXT NOT NULL,
                   filename TEXT NOT NULL,
                   sort INTEGER NOT NULL DEFAULT 0,
                   UNIQUE(asset_tag, filename)
               )""")
        # sp_hash (DVI-1185 P3) fingerprints the source-workbook row as of the
        # last sync, so sync_workbook can tell "source changed" apart from
        # "Togen edited". Added via migration for pre-P3 databases.
        cols = {r[1] for r in conn.execute("PRAGMA table_info(assets)")}
        _add_column_if_missing(
            conn, cols, "sp_hash", "TEXT NOT NULL DEFAULT ''")
        # Equipment-ops fields (DVI-1472 P2): the Equipment Registry
        # (wor_equipment.json) merges into assets.db, so an asset carries the
        # ops state the registry record used to hold. All additive with safe
        # defaults so existing assets are unaffected. rts_departments is a JSON
        # dict {dept: bool} (app owns the dept vocabulary); the booleans are
        # stored as 0/1 INTEGER.
        for column, ddl in (
                ("rts_required", "INTEGER NOT NULL DEFAULT 0"),
                ("rts_departments", "TEXT NOT NULL DEFAULT ''"),
                # include_in_oversight is TRI-STATE text: '' (off),
                # 'stationary', or 'mobile' (the Oversight lorawan_mobile layer
                # + mm-eas-options key on == 'mobile'); a legacy bool True is
                # kept truthy. Stored as TEXT, never a 0/1 int.
                ("include_in_oversight", "TEXT NOT NULL DEFAULT ''"),
                ("oversight_dev_eui", "TEXT NOT NULL DEFAULT ''"),
                ("out_of_service", "INTEGER NOT NULL DEFAULT 0"),
                ("offsite", "INTEGER NOT NULL DEFAULT 0"),
                # equipment_seeded_at is set when an asset is created by the WO
                # auto-seed / registry migration (source_sheet=EQUIPMENT_SHEET),
                # so the shim can report a stable created_at.
                ("equipment_created_at", "TEXT NOT NULL DEFAULT ''"),
                # model_id (DVI-1472 P3, D6): optional FK into asset_models. The
                # existing asset_type category vocab is untouched — a machine
                # MODEL is a distinct concept. NULL = no model assigned. Kept out
                # of ASSET_FIELDS/_sp_hash (write path handles it separately) so
                # the import/sync fingerprint is unchanged.
                ("model_id", "INTEGER")):
            _add_column_if_missing(conn, cols, column, ddl)
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_assets_model ON assets(model_id)")
        # equipment_aliases maps a normalized equipment name (the old registry
        # key) to an asset_tag, so every WO/manual/Oversight consumer that keys
        # by free-text equipment name resolves to the merged asset. One asset
        # can carry several aliases (name variants of one machine).
        conn.execute(
            """CREATE TABLE IF NOT EXISTS equipment_aliases (
                   norm_name TEXT NOT NULL UNIQUE,
                   name TEXT NOT NULL DEFAULT '',
                   asset_tag TEXT NOT NULL,
                   created_at TEXT NOT NULL DEFAULT (datetime('now'))
               )""")
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_equipment_aliases_tag"
            " ON equipment_aliases(asset_tag)")
        # Asset models (DVI-1472 P3, D6): a named machine model
        # (name/manufacturer/model_no/description). Assets reference one via
        # assets.model_id; sub-assemblies attached to a model are inherited by
        # every asset of that model.
        conn.execute(
            """CREATE TABLE IF NOT EXISTS asset_models (
                   id INTEGER PRIMARY KEY AUTOINCREMENT,
                   name TEXT NOT NULL,
                   manufacturer TEXT NOT NULL DEFAULT '',
                   model_no TEXT NOT NULL DEFAULT '',
                   description TEXT NOT NULL DEFAULT '',
                   created_at TEXT NOT NULL DEFAULT (datetime('now')),
                   updated_at TEXT NOT NULL DEFAULT (datetime('now'))
               )""")
        # Sub-assemblies (DVI-1472 P3, D7): a named part-of-a-machine that
        # attaches EITHER to one asset (parent_kind='asset', parent_ref=asset_tag)
        # OR to a model (parent_kind='model', parent_ref=str(model id), inherited
        # by every asset of that model). sort orders them within their parent.
        conn.execute(
            """CREATE TABLE IF NOT EXISTS sub_assemblies (
                   id INTEGER PRIMARY KEY AUTOINCREMENT,
                   name TEXT NOT NULL,
                   parent_kind TEXT NOT NULL,
                   parent_ref TEXT NOT NULL,
                   description TEXT NOT NULL DEFAULT '',
                   sort INTEGER NOT NULL DEFAULT 0,
                   created_at TEXT NOT NULL DEFAULT (datetime('now')),
                   updated_at TEXT NOT NULL DEFAULT (datetime('now'))
               )""")
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_sub_assemblies_parent"
            " ON sub_assemblies(parent_kind, parent_ref, sort)")
        # Parts inventory (DVI-1472 P4, D8-A). `parts` is the catalog + current
        # stock level (part_no is the catalog key, seeded from
        # rpm_known_parts.json). `part_links` associates a part with a model /
        # asset / sub-assembly. `part_movements` is an append-only ledger of
        # every stock change (WO usage + manual adjustments) — on_hand is the
        # running total the ledger explains, so a movement never rewrites a WO
        # sidecar. A usage movement is idempotent per wo_part_id (the stable RPM
        # part id) so a WO's parts (re-read/re-saved often) decrement exactly
        # once. Quantities are REAL to tolerate fractional stock. below-min /
        # negative are FLAGS only (house rule) — usage never blocks.
        conn.execute(
            """CREATE TABLE IF NOT EXISTS parts (
                   id INTEGER PRIMARY KEY AUTOINCREMENT,
                   part_no TEXT NOT NULL UNIQUE,
                   name TEXT NOT NULL DEFAULT '',
                   description TEXT NOT NULL DEFAULT '',
                   on_hand REAL NOT NULL DEFAULT 0,
                   min_qty REAL NOT NULL DEFAULT 0,
                   location TEXT NOT NULL DEFAULT '',
                   created_at TEXT NOT NULL DEFAULT (datetime('now')),
                   updated_at TEXT NOT NULL DEFAULT (datetime('now'))
               )""")
        conn.execute(
            """CREATE TABLE IF NOT EXISTS part_links (
                   id INTEGER PRIMARY KEY AUTOINCREMENT,
                   part_id INTEGER NOT NULL,
                   target_kind TEXT NOT NULL,
                   target_ref TEXT NOT NULL,
                   created_at TEXT NOT NULL DEFAULT (datetime('now')),
                   UNIQUE(part_id, target_kind, target_ref)
               )""")
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_part_links_target"
            " ON part_links(target_kind, target_ref)")
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_part_links_part"
            " ON part_links(part_id)")
        conn.execute(
            """CREATE TABLE IF NOT EXISTS part_movements (
                   id INTEGER PRIMARY KEY AUTOINCREMENT,
                   part_id INTEGER NOT NULL,
                   delta REAL NOT NULL,
                   kind TEXT NOT NULL,
                   reason TEXT NOT NULL DEFAULT '',
                   wo_ref TEXT NOT NULL DEFAULT '',
                   wo_part_id TEXT NOT NULL DEFAULT '',
                   asset_tag TEXT NOT NULL DEFAULT '',
                   on_hand_after REAL NOT NULL DEFAULT 0,
                   actor TEXT NOT NULL DEFAULT '',
                   created_at TEXT NOT NULL DEFAULT (datetime('now'))
               )""")
        conn.execute(
            "CREATE INDEX IF NOT EXISTS idx_part_movements_part"
            " ON part_movements(part_id, id)")
        # One usage movement per WO part id — the idempotency guard so re-reading
        # a WO's parts never double-decrements. Partial index (wo_part_id != '')
        # leaves manual adjustments (no wo_part_id) unconstrained.
        conn.execute(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_part_movements_wopart"
            " ON part_movements(wo_part_id) WHERE wo_part_id != ''")
        # Low-stock reorder-alert cooldown (DVI-1472 P5). One row per part that
        # has been alerted, holding the last alert time — so a part sitting below
        # its reorder point doesn't re-alert on every movement. The stamp is
        # cleared (row deleted) when a part rises back above min, so a future dip
        # alerts immediately. This is only cooldown state; the actual send lives
        # in app.py (Graph email / Teams webhook), keeping this module
        # Flask-independent.
        conn.execute(
            """CREATE TABLE IF NOT EXISTS part_alerts (
                   part_id INTEGER PRIMARY KEY,
                   last_alert_at TEXT NOT NULL DEFAULT '',
                   on_hand_at_alert REAL NOT NULL DEFAULT 0,
                   min_qty_at_alert REAL NOT NULL DEFAULT 0
               )""")


def set_meta(key, value):
    with _write_lock, _connect() as conn:
        conn.execute(
            "INSERT INTO meta (key, value) VALUES (?, ?)"
            " ON CONFLICT(key) DO UPDATE SET value = excluded.value",
            (key, "" if value is None else str(value)))


def get_meta(key):
    with _connect() as conn:
        row = conn.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone()
    return row[0] if row else None


# ---------------------------------------------------------------------------
# Import
# ---------------------------------------------------------------------------

def _clean(value):
    """Cell -> trimmed text. Integers lose the trailing .0 float form; error
    values like #VALUE! are treated per-caller (photo cells vs data cells)."""
    if value is None:
        return ""
    if isinstance(value, float) and value.is_integer():
        value = int(value)
    return str(value).replace("\xa0", " ").strip()


def _norm_header(value):
    return re.sub(r"[^a-z0-9]", "", _clean(value).lower())


def _is_error_value(text):
    return text.startswith("#") and text.endswith("!")


def _parse_master_vocab(ws):
    """Master Data sheet -> {kind: [values]} preserving sheet order.

    The sheet is one column: a section header ("Asset Types", "Form ID's",
    "Area") followed by that section's values until the next header."""
    vocab = {k: [] for k in VOCAB_KINDS}
    current = None
    for row in ws.iter_rows(values_only=True):
        text = _clean(row[0] if row else None)
        if not text:
            continue
        section = _MASTER_SECTIONS.get(text.lower())
        if section:
            current = section
            continue
        if current and text not in vocab[current]:
            vocab[current].append(text)
    return vocab


def _sp_hash(record):
    """Fingerprint of a parsed source row (data fields + extra JSON). Stored
    per asset at import/sync time so the next sync can tell whether the
    source row changed (source wins) or only Togen did (Togen edit kept)."""
    basis = "\x1f".join(
        [record[f] for f in ASSET_FIELDS] + [record.get("_extra", "")])
    return hashlib.sha256(basis.encode("utf-8")).hexdigest()


def _sheet_team(ws):
    """Row 1 is a team banner ("Team #1" | names). Best-effort label."""
    first = next(ws.iter_rows(min_row=1, max_row=1, values_only=True), None)
    if not first:
        return ""
    parts = [_clean(c) for c in first[:2]]
    return " — ".join(p for p in parts if p)


def _find_header_row(ws):
    """Locate the header row (contains "Asset Tag") within the first rows.

    Returns (row_index, {col_index: field_or_None}) or (None, {}) when the
    sheet has no recognizable asset table (Master Data / Totals)."""
    for idx, row in enumerate(ws.iter_rows(min_row=1, max_row=5, values_only=True), start=1):
        normed = [_norm_header(c) for c in row]
        if any(h.startswith("assettag") for h in normed):
            mapping = {}
            for col, h in enumerate(normed):
                if not h or _PHOTO_HEADER_RE.match(h):
                    mapping[col] = "_photo" if h else None
                    continue
                mapping[col] = _HEADER_MAP.get(h, "_unknown:" + _clean(row[col]))
            return idx, mapping
    return None, {}


def _parse_workbook(path, source_name=None):
    """Stream the workbook into ``(vocab, report, prepared)`` without touching
    the database — shared by the full-replace importer and the merge sync.

    Exceptions are FLAGGED, never silently fixed or dropped (plan §2B):
    - rows missing an Asset Tag # or duplicating one are skipped + listed;
    - unknown Asset Type / Area values (no case-insensitive vocab match) are
      imported verbatim + listed;
    - case-drift vocab matches (e.g. "mh2") are canonicalized to the vocab
      casing and counted as normalized;
    - photo cells holding a real (non-#VALUE!) value are listed — photos are
      not imported until P2;
    - blank Asset Type / Area are tallied in counts only (226 Tools rows have
      no Area; listing each would drown real anomalies).
    """
    from openpyxl import load_workbook

    wb = load_workbook(path, read_only=True, data_only=True)
    try:
        vocab = {k: [] for k in VOCAB_KINDS}
        if "Master Data" in wb.sheetnames:
            vocab = _parse_master_vocab(wb["Master Data"])
        vocab_ci = {
            kind: {v.lower(): v for v in values}
            for kind, values in vocab.items()
        }

        report = {
            "imported_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "source_name": source_name or getattr(path, "name", None) or str(path),
            "sheets": [],
            "vocab": {k: len(v) for k, v in vocab.items()},
            "total_rows": 0,
            "imported": 0,
            "skipped_blank": 0,
            "normalized": 0,
            "exception_counts": {},
            "exceptions": [],
        }
        seen_tags = {}
        prepared = []

        def flag(issue, sheet, row, tag, value=""):
            report["exception_counts"][issue] = (
                report["exception_counts"].get(issue, 0) + 1)
            if len(report["exceptions"]) < EXCEPTION_DETAIL_CAP:
                report["exceptions"].append({
                    "issue": issue, "sheet": sheet, "row": row,
                    "tag": tag, "value": value,
                })

        def tally(issue):
            report["exception_counts"][issue] = (
                report["exception_counts"].get(issue, 0) + 1)

        for sheet_name in wb.sheetnames:
            # EXPORT_ALL_SHEET duplicates the team sheets in our own exports;
            # skipping it lets an exported workbook re-import cleanly.
            if sheet_name in ("Master Data", "Totals", "All Assets"):
                continue
            ws = wb[sheet_name]
            header_row, mapping = _find_header_row(ws)
            if header_row is None:
                report["sheets"].append({
                    "name": sheet_name, "team": _sheet_team(ws),
                    "data_rows": 0, "imported": 0, "skipped_blank": 0,
                    "note": "no asset table found",
                })
                continue
            stats = {"name": sheet_name, "team": _sheet_team(ws),
                     "data_rows": 0, "imported": 0, "skipped_blank": 0}
            for row_idx, row in enumerate(
                    ws.iter_rows(min_row=header_row + 1, values_only=True),
                    start=header_row + 1):
                record = {f: "" for f in ASSET_FIELDS}
                record["source_sheet"] = sheet_name
                extra = {}
                any_value = False
                for col, field in mapping.items():
                    if field is None or col >= len(row):
                        continue
                    text = _clean(row[col])
                    if not text:
                        continue
                    if field == "_photo":
                        if not _is_error_value(text):
                            flag("photo value present (not imported until P2)",
                                 sheet_name, row_idx,
                                 record.get("asset_tag", ""), text[:80])
                        continue
                    if _is_error_value(text):
                        flag("formula error value", sheet_name, row_idx,
                             record.get("asset_tag", ""), text)
                        continue
                    any_value = True
                    if field.startswith("_unknown:"):
                        extra[field.split(":", 1)[1]] = text
                    else:
                        record[field] = text
                if not any_value:
                    stats["skipped_blank"] += 1
                    continue
                stats["data_rows"] += 1
                report["total_rows"] += 1
                tag = record["asset_tag"]
                if not tag:
                    flag("missing asset tag (row skipped)", sheet_name,
                         row_idx, "", record["description"][:60])
                    continue
                if tag in seen_tags:
                    flag("duplicate asset tag (row skipped)", sheet_name,
                         row_idx, tag,
                         "first seen on sheet %s row %s" % seen_tags[tag])
                    continue
                seen_tags[tag] = (sheet_name, row_idx)
                for field, kind in (("asset_type", "asset_type"),
                                    ("area", "area"),
                                    ("form_id", "form_id")):
                    val = record[field]
                    if not val:
                        if field in ("asset_type", "area"):
                            tally("blank %s" % field.replace("_", " "))
                        continue
                    if val in vocab_ci[kind].values():
                        continue
                    canonical = vocab_ci[kind].get(val.lower())
                    if canonical:
                        record[field] = canonical
                        report["normalized"] += 1
                    elif field != "form_id":
                        # Form ID free text is common (descriptive values);
                        # only type/area are treated as controlled.
                        flag("unknown %s (imported verbatim)"
                             % field.replace("_", " "),
                             sheet_name, row_idx, tag, val[:60])
                record["_extra"] = json.dumps(extra) if extra else ""
                record["_tag_num"] = int(tag) if tag.isdigit() else None
                prepared.append(record)
                stats["imported"] += 1
                report["imported"] += 1
            report["skipped_blank"] += stats["skipped_blank"]
            report["sheets"].append(stats)
    finally:
        wb.close()
    return vocab, report, prepared


def import_workbook(path, source_name=None):
    """Full-replace import of the Asset File workbook. Returns the report dict
    (also persisted to meta.last_import_report). Parse-level exceptions are
    flagged, never silently fixed or dropped — see _parse_workbook."""
    vocab, report, prepared = _parse_workbook(path, source_name)
    init_assets_db()
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with _write_lock, _connect() as conn:
        conn.execute("DELETE FROM assets")
        conn.execute("DELETE FROM vocab")
        conn.executemany(
            """INSERT INTO assets (asset_tag, tag_num, asset_type, description,
                   area, form_id, length, width, height, manufacturer,
                   model_no, serial_no, notes, material, source_sheet, extra,
                   sp_hash, created_at, updated_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            [(r["asset_tag"], r["_tag_num"], r["asset_type"], r["description"],
              r["area"], r["form_id"], r["length"], r["width"], r["height"],
              r["manufacturer"], r["model_no"], r["serial_no"], r["notes"],
              r["material"], r["source_sheet"], r["_extra"], _sp_hash(r),
              now, now)
             for r in prepared])
        conn.executemany(
            "INSERT OR IGNORE INTO vocab (kind, value, sort) VALUES (?, ?, ?)",
            [(kind, value, i)
             for kind, values in vocab.items()
             for i, value in enumerate(values)])
    set_meta("last_import_at", report["imported_at"])
    set_meta("last_import_source", report["source_name"])
    set_meta("last_import_report", json.dumps(report))
    return report


# ---------------------------------------------------------------------------
# Read API (Equipment Assets tab + future consumers)
# ---------------------------------------------------------------------------

def _row_to_asset(row, columns):
    asset = dict(zip(columns, row))
    extra = asset.pop("extra", "")
    try:
        asset["extra"] = json.loads(extra) if extra else {}
    except ValueError:
        asset["extra"] = {}
    asset.pop("tag_num", None)
    asset.pop("sp_hash", None)
    return asset


def _attach_photos(conn, assets):
    for a in assets:
        a["photos"] = []
    by_tag = {a["asset_tag"]: a for a in assets}
    if not by_tag:
        return assets
    placeholders = ",".join("?" * len(by_tag))
    for tag, filename in conn.execute(
            f"SELECT asset_tag, filename FROM photos"
            f" WHERE asset_tag IN ({placeholders}) ORDER BY asset_tag, sort",
            list(by_tag)):
        by_tag[tag]["photos"].append(filename)
    return assets


def search_assets(q=None, asset_type=None, area=None, form_id=None,
                  sheet=None, limit=50, offset=0):
    """Filtered asset page: {"total": n, "assets": [...]} ordered by tag.

    ``q`` is a case-insensitive substring match across tag/description/
    manufacturer/model/serial/form id/notes; the other filters are exact."""
    where, params = [], []
    if q:
        like = "%" + str(q).strip() + "%"
        where.append("(" + " OR ".join(
            f"{c} LIKE ? COLLATE NOCASE" for c in _SEARCH_COLUMNS) + ")")
        params.extend([like] * len(_SEARCH_COLUMNS))
    for column, value in (("asset_type", asset_type), ("area", area),
                          ("form_id", form_id), ("source_sheet", sheet)):
        if value:
            where.append(f"{column} = ?")
            params.append(value)
    clause = (" WHERE " + " AND ".join(where)) if where else ""
    limit = max(1, min(int(limit or 50), 500))
    offset = max(0, int(offset or 0))
    with _connect() as conn:
        total = conn.execute(
            "SELECT COUNT(*) FROM assets" + clause, params).fetchone()[0]
        cur = conn.execute(
            "SELECT * FROM assets" + clause +
            " ORDER BY tag_num IS NULL, tag_num, asset_tag LIMIT ? OFFSET ?",
            params + [limit, offset])
        columns = [d[0] for d in cur.description]
        assets = _attach_photos(
            conn, [_row_to_asset(r, columns) for r in cur.fetchall()])
    return {"total": total, "assets": assets}


def get_asset(tag):
    with _connect() as conn:
        cur = conn.execute(
            "SELECT * FROM assets WHERE asset_tag = ?", (str(tag).strip(),))
        row = cur.fetchone()
        if not row:
            return None
        asset = _row_to_asset(row, [d[0] for d in cur.description])
        return _attach_photos(conn, [asset])[0]


def set_photos(tag, filenames):
    """Replace the photo index rows for one asset (files live on disk)."""
    with _write_lock, _connect() as conn:
        conn.execute("DELETE FROM photos WHERE asset_tag = ?", (str(tag),))
        conn.executemany(
            "INSERT OR IGNORE INTO photos (asset_tag, filename, sort)"
            " VALUES (?, ?, ?)",
            [(str(tag), name, i) for i, name in enumerate(filenames)])


def photos_for(tag):
    """Ordered photo filenames for one asset."""
    with _connect() as conn:
        return [f for (f,) in conn.execute(
            "SELECT filename FROM photos WHERE asset_tag = ? ORDER BY sort",
            (str(tag),))]


def add_photos(tag, filenames):
    """Append photo index rows after the asset's existing photos."""
    if not filenames:
        return
    with _write_lock, _connect() as conn:
        base = conn.execute(
            "SELECT COALESCE(MAX(sort), -1) + 1 FROM photos"
            " WHERE asset_tag = ?", (str(tag),)).fetchone()[0]
        conn.executemany(
            "INSERT OR IGNORE INTO photos (asset_tag, filename, sort)"
            " VALUES (?, ?, ?)",
            [(str(tag), name, base + i) for i, name in enumerate(filenames)])


def remove_photo(tag, filename):
    """Drop one photo index row; returns True when a row was removed.

    The caller owns deleting the file on disk (it may want to keep it)."""
    with _write_lock, _connect() as conn:
        cur = conn.execute(
            "DELETE FROM photos WHERE asset_tag = ? AND filename = ?",
            (str(tag), str(filename)))
        return cur.rowcount > 0


def photo_count():
    with _connect() as conn:
        return conn.execute("SELECT COUNT(*) FROM photos").fetchone()[0]


def vocab_values(kind):
    """Controlled vocabulary for one kind, in Master Data order."""
    with _connect() as conn:
        return [v for (v,) in conn.execute(
            "SELECT value FROM vocab WHERE kind = ? ORDER BY sort", (kind,))]


def filter_options():
    """Distinct filter values for the UI: vocab order first, then any extra
    values present in the data (so imported-verbatim outliers stay findable)."""
    options = {}
    with _connect() as conn:
        for kind, column in (("asset_type", "asset_type"), ("area", "area"),
                             ("form_id", "form_id")):
            ordered = vocab_values(kind)
            present = {v for (v,) in conn.execute(
                f"SELECT DISTINCT {column} FROM assets WHERE {column} != ''")}
            extras = sorted(present - set(ordered))
            options[kind] = [v for v in ordered if v in present] + extras
        options["sheet"] = [v for (v,) in conn.execute(
            "SELECT DISTINCT source_sheet FROM assets"
            " WHERE source_sheet != '' ORDER BY source_sheet")]
    return options


def asset_count():
    with _connect() as conn:
        return conn.execute("SELECT COUNT(*) FROM assets").fetchone()[0]


# ---------------------------------------------------------------------------
# Reporting helpers (DVI-1250 P3): efficient GROUP BY breakdowns + a
# data-completeness scan for Auditor -> Reporting. Read-only; they take the
# same optional filters as search_assets so a report can be scoped like the
# list. Flag-not-fix: completeness only counts gaps, never mutates.
# ---------------------------------------------------------------------------

# Group-by fields the inventory-breakdown report offers -> display label.
GROUP_FIELDS = {
    "asset_type": "Type",
    "area": "Area",
    "source_sheet": "Team",
    "form_id": "Form ID",
}


def _asset_filter_clause(q=None, asset_type=None, area=None, form_id=None,
                         sheet=None):
    """Shared WHERE builder for the reporting helpers — mirrors search_assets'
    case-insensitive substring ``q`` + exact type/area/form/sheet filters.
    Returns ``(clause, params)`` where clause is ``""`` or ``" WHERE ..."``."""
    where, params = [], []
    if q:
        like = "%" + str(q).strip() + "%"
        where.append("(" + " OR ".join(
            f"{c} LIKE ? COLLATE NOCASE" for c in _SEARCH_COLUMNS) + ")")
        params.extend([like] * len(_SEARCH_COLUMNS))
    for column, value in (("asset_type", asset_type), ("area", area),
                          ("form_id", form_id), ("source_sheet", sheet)):
        if value:
            where.append(f"{column} = ?")
            params.append(value)
    clause = (" WHERE " + " AND ".join(where)) if where else ""
    return clause, params


def group_counts(field, q=None, asset_type=None, area=None, form_id=None,
                 sheet=None):
    """Asset counts grouped by one field (Type / Area / Team / Form ID),
    highest count first. Blank values are bucketed under "(blank)" so a data
    gap is visible rather than dropped. Returns
    ``{"total": n, "groups": [{"value", "count"}, ...]}``."""
    if field not in GROUP_FIELDS:
        raise ValueError("unknown group field: %r" % (field,))
    clause, params = _asset_filter_clause(q, asset_type, area, form_id, sheet)
    with _connect() as conn:
        total = conn.execute(
            "SELECT COUNT(*) FROM assets" + clause, params).fetchone()[0]
        rows = conn.execute(
            "SELECT CASE WHEN {0} = '' THEN '(blank)' ELSE {0} END AS v,"
            " COUNT(*) FROM assets{1} GROUP BY v".format(field, clause),
            params).fetchall()
    groups = [{"value": v, "count": c} for v, c in rows]
    groups.sort(key=lambda g: (-g["count"], g["value"].lower()))
    return {"total": total, "groups": groups}


def completeness(q=None, asset_type=None, area=None, form_id=None, sheet=None):
    """Data-completeness scan: how many assets are missing key fields
    (Asset Type / Area / Serial # / photo). Flag, not fix. Returns
    ``{"total": n, "missing": {asset_type, area, serial_no, photo}}``."""
    clause, params = _asset_filter_clause(q, asset_type, area, form_id, sheet)
    prefix = clause + (" AND " if clause else " WHERE ")
    with _connect() as conn:
        total = conn.execute(
            "SELECT COUNT(*) FROM assets" + clause, params).fetchone()[0]

        def miss(cond):
            return conn.execute(
                "SELECT COUNT(*) FROM assets" + prefix + cond,
                params).fetchone()[0]

        missing = {
            "asset_type": miss("asset_type = ''"),
            "area": miss("area = ''"),
            "serial_no": miss("serial_no = ''"),
            "photo": miss("NOT EXISTS (SELECT 1 FROM photos p"
                          " WHERE p.asset_tag = assets.asset_tag)"),
        }
    return {"total": total, "missing": missing}


# ---------------------------------------------------------------------------
# Write API (DVI-1185 P2: edit/add from the Equipment Assets tab)
# ---------------------------------------------------------------------------

EDITABLE_FIELDS = (
    "asset_tag", "asset_type", "description", "area", "form_id",
    "length", "width", "height", "manufacturer", "model_no", "serial_no",
    "notes", "material",
)

# Assets created in Togen (rather than imported from a team inventory sheet)
# carry this marker so the sheet filter still tells the two apart.
CREATED_SHEET = "Added in Togen"


def _canonical_vocab(field, value):
    """Case-drift vocab match -> canonical casing (same rule as the importer);
    unknown values are kept verbatim, never rejected (flag-not-fix)."""
    if field not in ("asset_type", "area", "form_id") or not value:
        return value
    for v in vocab_values(field):
        if v.lower() == value.lower():
            return v
    return value


def save_asset(fields, original_tag=None):
    """Create (``original_tag`` falsy) or update one asset.

    Only the EDITABLE_FIELDS present in ``fields`` are written (partial
    update); tag renames are allowed when the new tag is free. Returns
    ``(asset, changes, error)`` — ``changes`` maps field -> {"from", "to"}
    for the audit trail; on validation failure asset/changes are None and
    ``error`` is a user-facing message.
    """
    init_assets_db()
    clean = {}
    for field in EDITABLE_FIELDS:
        if field in fields and fields[field] is not None:
            clean[field] = _canonical_vocab(field, _clean(fields[field]))
    # model_id (DVI-1472 P3, D6) is handled apart from ASSET_FIELDS so the
    # import/sync fingerprint stays stable. Present-but-blank clears the model.
    model_provided = "model_id" in fields
    model_val = None
    if model_provided:
        raw = fields.get("model_id")
        if raw not in (None, "", 0, "0"):
            try:
                model_val = int(raw)
            except (TypeError, ValueError):
                return None, None, "Invalid asset model."
    original_tag = _clean(original_tag) if original_tag else ""
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    with _write_lock, _connect() as conn:
        if original_tag:
            cur = conn.execute(
                "SELECT * FROM assets WHERE asset_tag = ?", (original_tag,))
            row = cur.fetchone()
            if not row:
                return None, None, (
                    "Asset %s no longer exists — reload the list." % original_tag)
            current = dict(zip([d[0] for d in cur.description], row))
            new_tag = clean.get("asset_tag", current["asset_tag"])
            if not new_tag:
                return None, None, "Asset Tag # is required."
            if new_tag != current["asset_tag"] and conn.execute(
                    "SELECT 1 FROM assets WHERE asset_tag = ?",
                    (new_tag,)).fetchone():
                return None, None, "Asset tag %s is already in use." % new_tag
            changes = {}
            for field, value in clean.items():
                if value != current[field]:
                    changes[field] = {"from": current[field], "to": value}
                    current[field] = value
            if changes:
                conn.execute(
                    "UPDATE assets SET " +
                    ", ".join("%s = ?" % f for f in ASSET_FIELDS) +
                    ", tag_num = ?, updated_at = ? WHERE id = ?",
                    [current[f] for f in ASSET_FIELDS] +
                    [int(new_tag) if new_tag.isdigit() else None, now,
                     current["id"]])
                if new_tag != original_tag:
                    # Photos are keyed by tag (index rows + on-disk dir), so
                    # a rename must carry them over or they would orphan.
                    conn.execute(
                        "UPDATE photos SET asset_tag = ? WHERE asset_tag = ?",
                        (new_tag, original_tag))
                    old_dir = ASSETS_PHOTO_DIR / original_tag
                    new_dir = ASSETS_PHOTO_DIR / new_tag
                    if old_dir.is_dir() and not new_dir.exists():
                        try:
                            old_dir.rename(new_dir)
                        except OSError:
                            pass  # index kept; files stay reachable via re-extract
            if model_provided and model_val != current.get("model_id"):
                if model_val is not None and not conn.execute(
                        "SELECT 1 FROM asset_models WHERE id = ?",
                        (model_val,)).fetchone():
                    return None, None, "Unknown asset model."
                changes["model_id"] = {"from": current.get("model_id"),
                                       "to": model_val}
                conn.execute(
                    "UPDATE assets SET model_id = ?, updated_at = ? WHERE id = ?",
                    (model_val, now, current["id"]))
            final_tag = new_tag
        else:
            tag = clean.get("asset_tag", "")
            if not tag:
                return None, None, "Asset Tag # is required."
            if conn.execute("SELECT 1 FROM assets WHERE asset_tag = ?",
                            (tag,)).fetchone():
                return None, None, "Asset tag %s is already in use." % tag
            record = {f: "" for f in ASSET_FIELDS}
            record.update(clean)
            record["source_sheet"] = CREATED_SHEET
            conn.execute(
                "INSERT INTO assets (" + ", ".join(ASSET_FIELDS) +
                ", tag_num, extra, created_at, updated_at)"
                " VALUES (" + ", ".join("?" * len(ASSET_FIELDS)) + ", ?, '', ?, ?)",
                [record[f] for f in ASSET_FIELDS] +
                [int(tag) if tag.isdigit() else None, now, now])
            changes = {f: {"from": "", "to": v}
                       for f, v in clean.items() if v}
            if model_provided and model_val is not None:
                if not conn.execute(
                        "SELECT 1 FROM asset_models WHERE id = ?",
                        (model_val,)).fetchone():
                    return None, None, "Unknown asset model."
                conn.execute(
                    "UPDATE assets SET model_id = ? WHERE asset_tag = ?",
                    (model_val, tag))
                changes["model_id"] = {"from": None, "to": model_val}
            final_tag = tag
    return get_asset(final_tag), changes, None


# ---------------------------------------------------------------------------
# Asset models & sub-assemblies (DVI-1472 P3, D6/D7)
# ---------------------------------------------------------------------------

_MODEL_FIELDS = ("name", "manufacturer", "model_no", "description")
_SUB_ASSEMBLY_PARENTS = ("asset", "model")

# Parts (DVI-1472 P4). Catalog fields a user edits directly (on_hand is NOT
# here — stock only moves through the movement ledger). part_links target_kind
# values mirror the sub-assembly parent vocabulary plus sub_assembly itself.
_PART_FIELDS = ("part_no", "name", "description", "min_qty", "location")
_PART_LINK_KINDS = ("model", "asset", "sub_assembly")
_PART_SEARCH_COLUMNS = ("part_no", "name", "description", "location")


def list_asset_models():
    """All models ordered by name, each with the count of assets using it (so
    the UI can warn before a delete)."""
    with _connect() as conn:
        counts = {mid: n for mid, n in conn.execute(
            "SELECT model_id, COUNT(*) FROM assets"
            " WHERE model_id IS NOT NULL GROUP BY model_id")}
        cur = conn.execute(
            "SELECT * FROM asset_models ORDER BY name COLLATE NOCASE, id")
        cols = [d[0] for d in cur.description]
        out = []
        for row in cur.fetchall():
            m = dict(zip(cols, row))
            m["asset_count"] = counts.get(m["id"], 0)
            out.append(m)
    return out


def get_asset_model(model_id):
    try:
        model_id = int(model_id)
    except (TypeError, ValueError):
        return None
    with _connect() as conn:
        cur = conn.execute(
            "SELECT * FROM asset_models WHERE id = ?", (model_id,))
        row = cur.fetchone()
        if not row:
            return None
        return dict(zip([d[0] for d in cur.description], row))


def save_asset_model(fields, model_id=None):
    """Create (``model_id`` falsy) or update a model. Returns
    ``(model, changes, error)`` — ``changes`` maps field -> {from,to}."""
    init_assets_db()
    clean = {f: _clean(fields.get(f, "")) for f in _MODEL_FIELDS}
    if not clean["name"]:
        return None, None, "Model name is required."
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with _write_lock, _connect() as conn:
        if model_id:
            cur = conn.execute(
                "SELECT * FROM asset_models WHERE id = ?", (int(model_id),))
            row = cur.fetchone()
            if not row:
                return None, None, "Model no longer exists — reload the list."
            current = dict(zip([d[0] for d in cur.description], row))
            changes = {}
            for f in _MODEL_FIELDS:
                if clean[f] != current[f]:
                    changes[f] = {"from": current[f], "to": clean[f]}
                    current[f] = clean[f]
            if changes:
                conn.execute(
                    "UPDATE asset_models SET name = ?, manufacturer = ?,"
                    " model_no = ?, description = ?, updated_at = ? WHERE id = ?",
                    [current[f] for f in _MODEL_FIELDS] + [now, int(model_id)])
            new_id = int(model_id)
        else:
            cur = conn.execute(
                "INSERT INTO asset_models (name, manufacturer, model_no,"
                " description, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)",
                [clean[f] for f in _MODEL_FIELDS] + [now, now])
            new_id = cur.lastrowid
            changes = {f: {"from": "", "to": v} for f, v in clean.items() if v}
    return get_asset_model(new_id), changes, None


def delete_asset_model(model_id):
    """Delete a model (and its model-scoped sub-assemblies). Refuses while any
    asset still references it (flag-not-fix) — returns ``(ok, error)``."""
    init_assets_db()
    try:
        model_id = int(model_id)
    except (TypeError, ValueError):
        return False, "Model no longer exists."
    with _write_lock, _connect() as conn:
        if not conn.execute(
                "SELECT 1 FROM asset_models WHERE id = ?", (model_id,)).fetchone():
            return False, "Model no longer exists."
        n = conn.execute(
            "SELECT COUNT(*) FROM assets WHERE model_id = ?", (model_id,)
        ).fetchone()[0]
        if n:
            return False, (
                "%d asset%s still use this model — reassign them first."
                % (n, "" if n == 1 else "s"))
        conn.execute(
            "DELETE FROM sub_assemblies WHERE parent_kind = 'model'"
            " AND parent_ref = ?", (str(model_id),))
        conn.execute("DELETE FROM asset_models WHERE id = ?", (model_id,))
    return True, None


def get_sub_assembly(sub_id):
    try:
        sub_id = int(sub_id)
    except (TypeError, ValueError):
        return None
    with _connect() as conn:
        cur = conn.execute(
            "SELECT * FROM sub_assemblies WHERE id = ?", (sub_id,))
        row = cur.fetchone()
        if not row:
            return None
        return dict(zip([d[0] for d in cur.description], row))


def list_sub_assemblies(parent_kind, parent_ref):
    """Sub-assemblies for one parent (asset tag or model id), ordered by sort."""
    if parent_kind not in _SUB_ASSEMBLY_PARENTS:
        return []
    with _connect() as conn:
        cur = conn.execute(
            "SELECT * FROM sub_assemblies WHERE parent_kind = ? AND parent_ref = ?"
            " ORDER BY sort, id", (parent_kind, str(parent_ref)))
        cols = [d[0] for d in cur.description]
        return [dict(zip(cols, r)) for r in cur.fetchall()]


def sub_assemblies_for_asset(tag):
    """{own, inherited, model} for an asset detail (D7): its own sub-assemblies
    plus those inherited from its model (each flagged ``inherited``)."""
    asset = get_asset(tag)
    if not asset:
        return {"own": [], "inherited": [], "model": None}
    own = list_sub_assemblies("asset", asset["asset_tag"])
    for s in own:
        s["inherited"] = False
    inherited, model = [], None
    mid = asset.get("model_id")
    if mid:
        model = get_asset_model(mid)
        if model:
            inherited = list_sub_assemblies("model", str(int(mid)))
            for s in inherited:
                s["inherited"] = True
                s["model_name"] = model["name"]
    return {"own": own, "inherited": inherited, "model": model}


def save_sub_assembly(fields, sub_id=None):
    """Create or update a sub-assembly. On create ``parent_kind``/``parent_ref``
    are required and validated; on update the parent is immutable. Returns
    ``(sub, changes, error)``."""
    init_assets_db()
    name = _clean(fields.get("name", ""))
    desc = _clean(fields.get("description", ""))
    if not name:
        return None, None, "Sub-assembly name is required."
    sort_provided = "sort" in fields and str(fields.get("sort") or "").strip() != ""
    sort_val = 0
    if sort_provided:
        try:
            sort_val = int(fields.get("sort"))
        except (TypeError, ValueError):
            sort_provided = False
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with _write_lock, _connect() as conn:
        if sub_id:
            cur = conn.execute(
                "SELECT * FROM sub_assemblies WHERE id = ?", (int(sub_id),))
            row = cur.fetchone()
            if not row:
                return None, None, "Sub-assembly no longer exists."
            current = dict(zip([d[0] for d in cur.description], row))
            new = {"name": name, "description": desc}
            if sort_provided:
                new["sort"] = sort_val
            changes = {}
            for f, v in new.items():
                if v != current[f]:
                    changes[f] = {"from": current[f], "to": v}
                    current[f] = v
            if changes:
                conn.execute(
                    "UPDATE sub_assemblies SET name = ?, description = ?,"
                    " sort = ?, updated_at = ? WHERE id = ?",
                    (current["name"], current["description"], current["sort"],
                     now, int(sub_id)))
            new_id = int(sub_id)
        else:
            parent_kind = fields.get("parent_kind")
            parent_ref = _clean(fields.get("parent_ref", ""))
            if parent_kind not in _SUB_ASSEMBLY_PARENTS or not parent_ref:
                return None, None, (
                    "A sub-assembly must attach to an asset or a model.")
            if parent_kind == "asset":
                if not conn.execute(
                        "SELECT 1 FROM assets WHERE asset_tag = ?",
                        (parent_ref,)).fetchone():
                    return None, None, "Asset not found."
            else:  # model
                try:
                    parent_ref = str(int(parent_ref))
                except (TypeError, ValueError):
                    return None, None, "Model not found."
                if not conn.execute(
                        "SELECT 1 FROM asset_models WHERE id = ?",
                        (int(parent_ref),)).fetchone():
                    return None, None, "Model not found."
            if not sort_provided:
                sort_val = (conn.execute(
                    "SELECT COALESCE(MAX(sort), -1) + 1 FROM sub_assemblies"
                    " WHERE parent_kind = ? AND parent_ref = ?",
                    (parent_kind, parent_ref)).fetchone()[0])
            cur = conn.execute(
                "INSERT INTO sub_assemblies (name, parent_kind, parent_ref,"
                " description, sort, created_at, updated_at)"
                " VALUES (?, ?, ?, ?, ?, ?, ?)",
                (name, parent_kind, parent_ref, desc, sort_val, now, now))
            new_id = cur.lastrowid
            changes = {"name": {"from": "", "to": name}}
            if desc:
                changes["description"] = {"from": "", "to": desc}
    return get_sub_assembly(new_id), changes, None


def delete_sub_assembly(sub_id):
    """Delete one sub-assembly. Returns ``(record, error)`` (record is the
    deleted row, for the audit summary)."""
    init_assets_db()
    with _write_lock, _connect() as conn:
        cur = conn.execute(
            "SELECT * FROM sub_assemblies WHERE id = ?", (int(sub_id),))
        row = cur.fetchone()
        if not row:
            return None, "Sub-assembly no longer exists."
        record = dict(zip([d[0] for d in cur.description], row))
        conn.execute("DELETE FROM sub_assemblies WHERE id = ?", (int(sub_id),))
    return record, None


# ---------------------------------------------------------------------------
# Parts inventory (DVI-1472 P4, D8-A)
# ---------------------------------------------------------------------------

def _coerce_qty(value, default=0.0):
    """Parse a quantity to a float (signed). Blank/garbage -> default. Tolerant
    of stray text ('2 ea', ' 3.5 ') by taking the leading numeric token, so a
    WO part's free-text qty still decrements stock."""
    if value is None:
        return default
    if isinstance(value, bool):
        return default
    if isinstance(value, (int, float)):
        try:
            return float(value)
        except (TypeError, ValueError):
            return default
    m = re.match(r"\s*(-?\d+(?:\.\d+)?)", str(value))
    if not m:
        return default
    try:
        return float(m.group(1))
    except ValueError:
        return default


def _num(value):
    """REAL -> int when whole (so on_hand renders '5' not '5.0'), else float."""
    try:
        f = float(value)
    except (TypeError, ValueError):
        return value
    return int(f) if f == int(f) else f


def _part_row(row, cols):
    """Shape a parts row: numeric on_hand/min_qty + below_min/negative flags.

    negative  = on_hand < 0 (over-consumed). below_min = a reorder point is set
    (min_qty > 0) and on_hand has reached/fallen below it. `low` = either — the
    digest/badge signal. Flags only; nothing here blocks a movement."""
    p = dict(zip(cols, row))
    on_hand = _num(p.get("on_hand", 0))
    min_qty = _num(p.get("min_qty", 0))
    p["on_hand"] = on_hand
    p["min_qty"] = min_qty
    negative = float(p.get("on_hand") or 0) < 0
    below_min = float(min_qty or 0) > 0 and float(on_hand or 0) <= float(min_qty)
    p["negative"] = negative
    p["below_min"] = below_min
    p["low"] = negative or below_min
    return p


def seed_parts_catalog(catalog):
    """Seed the parts catalog from an external {part_no: name} map (the app's
    rpm_known_parts.json). Idempotent: inserts only part numbers not already
    present — never clobbers an edited name or existing stock. Returns the
    number of catalog rows added."""
    init_assets_db()
    if not isinstance(catalog, dict) or not catalog:
        return 0
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    added = 0
    with _write_lock, _connect() as conn:
        existing = {r[0] for r in conn.execute("SELECT part_no FROM parts")}
        for part_no, name in catalog.items():
            pn = _clean(part_no)
            if not pn or pn in existing:
                continue
            conn.execute(
                "INSERT OR IGNORE INTO parts (part_no, name, created_at,"
                " updated_at) VALUES (?, ?, ?, ?)", (pn, _clean(name), now, now))
            existing.add(pn)
            added += 1
    return added


def part_count():
    with _connect() as conn:
        return conn.execute("SELECT COUNT(*) FROM parts").fetchone()[0]


def get_part(part_id):
    try:
        part_id = int(part_id)
    except (TypeError, ValueError):
        return None
    with _connect() as conn:
        cur = conn.execute("SELECT * FROM parts WHERE id = ?", (part_id,))
        row = cur.fetchone()
        if not row:
            return None
        return _part_row(row, [d[0] for d in cur.description])


def get_part_by_no(part_no):
    pn = _clean(part_no)
    if not pn:
        return None
    with _connect() as conn:
        cur = conn.execute("SELECT * FROM parts WHERE part_no = ?", (pn,))
        row = cur.fetchone()
        if not row:
            return None
        return _part_row(row, [d[0] for d in cur.description])


def list_parts(q=None, low_stock=False, limit=100, offset=0):
    """Filtered parts page: {"total": n, "parts": [...]}. `low_stock` restricts
    to parts flagged below-min or negative. Ordered by part_no."""
    where, params = [], []
    if q:
        like = "%" + str(q).strip() + "%"
        where.append("(" + " OR ".join(
            f"{c} LIKE ? COLLATE NOCASE" for c in _PART_SEARCH_COLUMNS) + ")")
        params.extend([like] * len(_PART_SEARCH_COLUMNS))
    if low_stock:
        where.append("(on_hand < 0 OR (min_qty > 0 AND on_hand <= min_qty))")
    clause = (" WHERE " + " AND ".join(where)) if where else ""
    limit = max(1, min(int(limit or 100), 500))
    offset = max(0, int(offset or 0))
    with _connect() as conn:
        total = conn.execute(
            "SELECT COUNT(*) FROM parts" + clause, params).fetchone()[0]
        cur = conn.execute(
            "SELECT * FROM parts" + clause +
            " ORDER BY part_no COLLATE NOCASE LIMIT ? OFFSET ?",
            params + [limit, offset])
        cols = [d[0] for d in cur.description]
        parts = [_part_row(r, cols) for r in cur.fetchall()]
    return {"total": total, "parts": parts}


def low_stock_parts(limit=200):
    """Parts flagged below-min or negative, most-negative-margin first — the
    Assets low-stock digest + the P5 reorder-notification source."""
    with _connect() as conn:
        cur = conn.execute(
            "SELECT * FROM parts WHERE on_hand < 0"
            " OR (min_qty > 0 AND on_hand <= min_qty)"
            " ORDER BY (on_hand - min_qty), part_no COLLATE NOCASE LIMIT ?",
            (int(limit),))
        cols = [d[0] for d in cur.description]
        return [_part_row(r, cols) for r in cur.fetchall()]


# ---------------------------------------------------------------------------
# Low-stock reorder alerts (DVI-1472 P5)
# ---------------------------------------------------------------------------

_PARTS_NOTIFY_KEY = "parts_notify_config"
_PARTS_NOTIFY_DEFAULTS = {
    "enabled": False,
    "emails": [],
    "teams_webhook": "",
    "cooldown_hours": 24.0,
}


def get_parts_notify_config():
    """The low-stock notification config (recipients, optional Teams webhook,
    per-part cooldown hours, enabled). Stored as a JSON meta row; a missing/bad
    value returns the defaults so the feature is off until an admin configures
    it."""
    cfg = dict(_PARTS_NOTIFY_DEFAULTS)
    cfg["emails"] = []
    raw = get_meta(_PARTS_NOTIFY_KEY)
    if raw:
        try:
            data = json.loads(raw)
        except (ValueError, TypeError):
            data = None
        if isinstance(data, dict):
            cfg["enabled"] = bool(data.get("enabled"))
            cfg["emails"] = [e for e in (data.get("emails") or [])
                             if isinstance(e, str)]
            cfg["teams_webhook"] = _clean(data.get("teams_webhook", ""))
            try:
                cfg["cooldown_hours"] = max(0.0, float(
                    data.get("cooldown_hours", 24)))
            except (TypeError, ValueError):
                cfg["cooldown_hours"] = 24.0
    return cfg


def save_parts_notify_config(cfg):
    """Persist the low-stock notification config, sanitized. Emails are
    lowercased, deduped, and kept only if they look like addresses; the cooldown
    is clamped non-negative. Returns the cleaned config."""
    init_assets_db()
    emails, seen = [], set()
    for e in (cfg.get("emails") or []):
        e2 = _clean(e).lower()
        if e2 and "@" in e2 and e2 not in seen:
            emails.append(e2)
            seen.add(e2)
    try:
        cooldown = max(0.0, float(cfg.get("cooldown_hours", 24)))
    except (TypeError, ValueError):
        cooldown = 24.0
    clean = {
        "enabled": bool(cfg.get("enabled")),
        "emails": emails,
        "teams_webhook": _clean(cfg.get("teams_webhook", "")),
        "cooldown_hours": cooldown,
    }
    set_meta(_PARTS_NOTIFY_KEY, json.dumps(clean))
    return clean


def _low_predicate(prefix=""):
    """SQL fragment: the row is low (negative or below a set reorder point).
    ``prefix`` qualifies the column (e.g. 'p.')."""
    return ("({0}on_hand < 0 OR ({0}min_qty > 0 AND {0}on_hand <= {0}min_qty))"
            .format(prefix))


def part_alert_due(part_id, cooldown_minutes=1440):
    """True if the part is currently low AND due for a reorder alert — never
    alerted, or its last alert was at least ``cooldown_minutes`` ago. The
    per-part cooldown is what stops a below-min part re-alerting on every
    movement."""
    init_assets_db()
    cd_days = max(0.0, float(cooldown_minutes)) / 1440.0
    with _connect() as conn:
        row = conn.execute(
            "SELECT " + _low_predicate("p.") + " AS low, a.last_alert_at,"
            " (julianday('now', 'localtime') - julianday(a.last_alert_at)) AS age"
            " FROM parts p LEFT JOIN part_alerts a ON a.part_id = p.id"
            " WHERE p.id = ?", (int(part_id),)).fetchone()
    if not row or not row[0]:
        return False
    if not row[1]:
        return True
    return row[2] is None or row[2] >= cd_days


def low_stock_alert_candidates(cooldown_minutes=1440, limit=200):
    """Low-stock parts due for a reorder alert (never alerted or cooldown
    elapsed), most-negative-margin first. Feeds a digest / scheduled sweep."""
    init_assets_db()
    cd_days = max(0.0, float(cooldown_minutes)) / 1440.0
    with _connect() as conn:
        cur = conn.execute(
            "SELECT p.* FROM parts p LEFT JOIN part_alerts a ON a.part_id = p.id"
            " WHERE " + _low_predicate("p.") + " AND (a.last_alert_at IS NULL"
            " OR a.last_alert_at = ''"
            " OR (julianday('now', 'localtime') - julianday(a.last_alert_at))"
            " >= ?) ORDER BY (p.on_hand - p.min_qty), p.part_no COLLATE NOCASE"
            " LIMIT ?", (cd_days, int(limit)))
        cols = [d[0] for d in cur.description]
        return [_part_row(r, cols) for r in cur.fetchall()]


def record_low_stock_alert(part_ids):
    """Stamp the reorder-alert time (now) for each part id, starting its
    cooldown."""
    if not part_ids:
        return
    init_assets_db()
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with _write_lock, _connect() as conn:
        for pid in part_ids:
            row = conn.execute(
                "SELECT on_hand, min_qty FROM parts WHERE id = ?",
                (int(pid),)).fetchone()
            oh, mq = (float(row[0] or 0), float(row[1] or 0)) if row \
                else (0.0, 0.0)
            conn.execute(
                "INSERT INTO part_alerts (part_id, last_alert_at,"
                " on_hand_at_alert, min_qty_at_alert) VALUES (?, ?, ?, ?)"
                " ON CONFLICT(part_id) DO UPDATE SET"
                " last_alert_at = excluded.last_alert_at,"
                " on_hand_at_alert = excluded.on_hand_at_alert,"
                " min_qty_at_alert = excluded.min_qty_at_alert",
                (int(pid), now, oh, mq))


def clear_low_stock_alert(part_id):
    """Drop a part's cooldown stamp — called when it rises back above its
    reorder point so a future dip alerts immediately (no stale cooldown)."""
    init_assets_db()
    with _write_lock, _connect() as conn:
        conn.execute("DELETE FROM part_alerts WHERE part_id = ?", (int(part_id),))


def save_part(fields, part_id=None):
    """Create (``part_id`` falsy) or update a part's catalog fields (part_no,
    name, description, min_qty, location). on_hand is NEVER set here — stock only
    moves through adjust_stock / record_wo_part_usage so the ledger stays
    authoritative. Returns ``(part, changes, error)``; part_no must be unique."""
    init_assets_db()
    part_no = _clean(fields.get("part_no", ""))
    if not part_no:
        return None, None, "Part number is required."
    clean = {
        "part_no": part_no,
        "name": _clean(fields.get("name", "")),
        "description": _clean(fields.get("description", "")),
        "min_qty": _coerce_qty(fields.get("min_qty", 0), 0.0),
        "location": _clean(fields.get("location", "")),
    }
    if clean["min_qty"] < 0:
        return None, None, "Minimum quantity cannot be negative."
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with _write_lock, _connect() as conn:
        clash = conn.execute(
            "SELECT id FROM parts WHERE part_no = ?", (part_no,)).fetchone()
        if clash and (not part_id or int(clash[0]) != int(part_id)):
            return None, None, "Part number %r already exists." % part_no
        if part_id:
            cur = conn.execute("SELECT * FROM parts WHERE id = ?", (int(part_id),))
            row = cur.fetchone()
            if not row:
                return None, None, "Part no longer exists — reload the list."
            current = dict(zip([d[0] for d in cur.description], row))
            changes = {}
            for f in _PART_FIELDS:
                cur_val = _num(current[f]) if f == "min_qty" else current[f]
                new_val = clean[f]
                if str(cur_val) != str(new_val):
                    changes[f] = {"from": cur_val, "to": _num(new_val)
                                  if f == "min_qty" else new_val}
            if changes:
                conn.execute(
                    "UPDATE parts SET part_no = ?, name = ?, description = ?,"
                    " min_qty = ?, location = ?, updated_at = ? WHERE id = ?",
                    [clean[f] for f in _PART_FIELDS] + [now, int(part_id)])
            new_id = int(part_id)
        else:
            cur = conn.execute(
                "INSERT INTO parts (part_no, name, description, min_qty,"
                " location, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
                [clean[f] for f in _PART_FIELDS] + [now, now])
            new_id = cur.lastrowid
            changes = {f: {"from": "", "to": clean[f]} for f in _PART_FIELDS
                       if clean[f] not in ("", 0, 0.0)}
    return get_part(new_id), changes, None


def delete_part(part_id):
    """Delete a part. Refuses while a movement ledger exists for it (the ledger
    is the audit trail — flag-not-fix). Cascades its links. Returns
    ``(part, error)`` where part is the deleted row for the audit summary."""
    init_assets_db()
    part = get_part(part_id)
    if not part:
        return None, "Part no longer exists."
    with _write_lock, _connect() as conn:
        n = conn.execute(
            "SELECT COUNT(*) FROM part_movements WHERE part_id = ?",
            (int(part_id),)).fetchone()[0]
        if n:
            return None, (
                "This part has %d stock movement%s — adjust it to zero instead"
                " of deleting." % (n, "" if n == 1 else "s"))
        conn.execute("DELETE FROM part_links WHERE part_id = ?", (int(part_id),))
        conn.execute("DELETE FROM parts WHERE id = ?", (int(part_id),))
    return part, None


def adjust_stock(part_id, delta, reason="", actor="", kind="adjustment"):
    """Append a manual stock movement (signed ``delta``) and recompute on_hand.
    Below-min/negative are flags, never blocked. Returns ``(part, movement,
    error)``."""
    init_assets_db()
    part = get_part(part_id)
    if not part:
        return None, None, "Part no longer exists."
    d = _coerce_qty(delta, None) if not isinstance(delta, (int, float)) \
        else float(delta)
    if d is None or d == 0:
        return None, None, "Enter a non-zero adjustment."
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with _write_lock, _connect() as conn:
        cur = conn.execute(
            "SELECT on_hand FROM parts WHERE id = ?", (int(part_id),)).fetchone()
        new_on_hand = float(cur[0] or 0) + d
        conn.execute("UPDATE parts SET on_hand = ?, updated_at = ? WHERE id = ?",
                     (new_on_hand, now, int(part_id)))
        mid = conn.execute(
            "INSERT INTO part_movements (part_id, delta, kind, reason, actor,"
            " on_hand_after, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
            (int(part_id), d, kind, _clean(reason), _clean(actor), new_on_hand,
             now)).lastrowid
    return get_part(part_id), get_part_movement(mid), None


def record_wo_part_usage(part_no, name, qty, wo_ref, wo_part_id, asset_tag="",
                         actor=""):
    """Record a Work-Order part usage: resolve/create the catalog part by
    part_no, append a ``usage`` movement (delta = -qty), and decrement on_hand.

    IDEMPOTENT per ``wo_part_id`` (the stable RPM part id) — a WO's parts are
    read/re-saved often, but usage is recorded exactly once. below-min/negative
    only FLAG (never block). Returns a dict describing the outcome:
    ``{recorded, part, below_min, negative, reason}``. recorded=False (no stock
    change) when part_no is blank, qty<=0, or this wo_part_id was already logged.
    """
    init_assets_db()
    pn = _clean(part_no)
    wpid = _clean(wo_part_id)
    q = _coerce_qty(qty, 0.0)
    if q <= 0:
        # A missing/zero qty on a listed part still means "one used" if it
        # carries a part number — mirror the paper form's implicit single unit.
        q = 1.0 if pn else 0.0
    if not pn:
        return {"recorded": False, "part": None, "reason": "no_part_no"}
    if q <= 0:
        return {"recorded": False, "part": None, "reason": "no_qty"}
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with _write_lock, _connect() as conn:
        if wpid and conn.execute(
                "SELECT 1 FROM part_movements WHERE wo_part_id = ?",
                (wpid,)).fetchone():
            return {"recorded": False, "part": None, "reason": "already_recorded"}
        row = conn.execute(
            "SELECT id, on_hand FROM parts WHERE part_no = ?", (pn,)).fetchone()
        if row is None:
            pid = conn.execute(
                "INSERT INTO parts (part_no, name, created_at, updated_at)"
                " VALUES (?, ?, ?, ?)", (pn, _clean(name), now, now)).lastrowid
            on_hand = 0.0
        else:
            pid, on_hand = int(row[0]), float(row[1] or 0)
        new_on_hand = on_hand - q
        conn.execute("UPDATE parts SET on_hand = ?, updated_at = ? WHERE id = ?",
                     (new_on_hand, now, pid))
        conn.execute(
            "INSERT INTO part_movements (part_id, delta, kind, reason, wo_ref,"
            " wo_part_id, asset_tag, on_hand_after, actor, created_at)"
            " VALUES (?, ?, 'usage', '', ?, ?, ?, ?, ?, ?)",
            (pid, -q, _clean(wo_ref), wpid, _clean(asset_tag), new_on_hand,
             _clean(actor), now))
    part = get_part(pid)
    return {"recorded": True, "part": part, "below_min": part["below_min"],
            "negative": part["negative"], "reason": ""}


def get_part_movement(movement_id):
    with _connect() as conn:
        cur = conn.execute(
            "SELECT * FROM part_movements WHERE id = ?", (int(movement_id),))
        row = cur.fetchone()
        if not row:
            return None
        m = dict(zip([d[0] for d in cur.description], row))
    m["delta"] = _num(m["delta"])
    m["on_hand_after"] = _num(m["on_hand_after"])
    return m


def list_part_movements(part_id, limit=100):
    """Ledger for one part, newest first."""
    with _connect() as conn:
        cur = conn.execute(
            "SELECT * FROM part_movements WHERE part_id = ?"
            " ORDER BY id DESC LIMIT ?", (int(part_id), int(limit)))
        cols = [d[0] for d in cur.description]
        out = []
        for r in cur.fetchall():
            m = dict(zip(cols, r))
            m["delta"] = _num(m["delta"])
            m["on_hand_after"] = _num(m["on_hand_after"])
            out.append(m)
    return out


def _link_target_exists(conn, target_kind, target_ref):
    if target_kind == "asset":
        return conn.execute("SELECT 1 FROM assets WHERE asset_tag = ?",
                            (target_ref,)).fetchone() is not None
    if target_kind == "model":
        try:
            return conn.execute("SELECT 1 FROM asset_models WHERE id = ?",
                                (int(target_ref),)).fetchone() is not None
        except (TypeError, ValueError):
            return False
    if target_kind == "sub_assembly":
        try:
            return conn.execute("SELECT 1 FROM sub_assemblies WHERE id = ?",
                                (int(target_ref),)).fetchone() is not None
        except (TypeError, ValueError):
            return False
    return False


def link_part(part_id, target_kind, target_ref):
    """Associate a part with a model / asset / sub-assembly. Idempotent.
    Returns ``(link, error)``."""
    init_assets_db()
    if target_kind not in _PART_LINK_KINDS:
        return None, "Unknown link target."
    ref = _clean(target_ref)
    if target_kind in ("model", "sub_assembly"):
        try:
            ref = str(int(ref))
        except (TypeError, ValueError):
            return None, "Invalid link target."
    if not ref:
        return None, "A link needs a target."
    with _write_lock, _connect() as conn:
        if not conn.execute("SELECT 1 FROM parts WHERE id = ?",
                            (int(part_id),)).fetchone():
            return None, "Part no longer exists."
        if not _link_target_exists(conn, target_kind, ref):
            return None, "Link target not found."
        conn.execute(
            "INSERT OR IGNORE INTO part_links (part_id, target_kind, target_ref)"
            " VALUES (?, ?, ?)", (int(part_id), target_kind, ref))
        row = conn.execute(
            "SELECT id, part_id, target_kind, target_ref FROM part_links"
            " WHERE part_id = ? AND target_kind = ? AND target_ref = ?",
            (int(part_id), target_kind, ref)).fetchone()
    return {"id": row[0], "part_id": row[1], "target_kind": row[2],
            "target_ref": row[3]}, None


def unlink_part(link_id):
    """Remove one part association. Returns ``(record, error)``."""
    init_assets_db()
    with _write_lock, _connect() as conn:
        cur = conn.execute(
            "SELECT id, part_id, target_kind, target_ref FROM part_links"
            " WHERE id = ?", (int(link_id),))
        row = cur.fetchone()
        if not row:
            return None, "Link no longer exists."
        rec = dict(zip([d[0] for d in cur.description], row))
        conn.execute("DELETE FROM part_links WHERE id = ?", (int(link_id),))
    return rec, None


def parts_for_target(target_kind, target_ref):
    """Parts linked directly to one target (model / asset / sub_assembly), each
    carrying its stock + the link_id (so the UI can unlink)."""
    if target_kind not in _PART_LINK_KINDS:
        return []
    ref = _clean(target_ref)
    if target_kind in ("model", "sub_assembly"):
        try:
            ref = str(int(ref))
        except (TypeError, ValueError):
            return []
    with _connect() as conn:
        cur = conn.execute(
            "SELECT p.*, l.id AS link_id FROM part_links l"
            " JOIN parts p ON p.id = l.part_id"
            " WHERE l.target_kind = ? AND l.target_ref = ?"
            " ORDER BY p.part_no COLLATE NOCASE", (target_kind, ref))
        cols = [d[0] for d in cur.description]
        out = []
        for r in cur.fetchall():
            p = _part_row(r, cols)
            p["link_id"] = dict(zip(cols, r))["link_id"]
            out.append(p)
    return out


def part_links_detail(part_id):
    """The associations for one part, each resolved to a human label — feeds the
    part detail pane's link list. ``[{link_id, target_kind, target_ref, label}]``."""
    with _connect() as conn:
        rows = conn.execute(
            "SELECT id, target_kind, target_ref FROM part_links"
            " WHERE part_id = ? ORDER BY id", (int(part_id),)).fetchall()
    out = []
    for lid, kind, ref in rows:
        label = ref
        if kind == "model":
            m = get_asset_model(ref)
            label = m["name"] if m else ("model %s" % ref)
        elif kind == "sub_assembly":
            s = get_sub_assembly(ref)
            label = s["name"] if s else ("sub-assembly %s" % ref)
        out.append({"link_id": lid, "target_kind": kind, "target_ref": ref,
                    "label": label})
    return out


def parts_for_asset(tag):
    """Every part associated with an asset for its detail view: linked directly,
    inherited via its model, and via its own + model-inherited sub-assemblies.
    Returns a flat de-duplicated list (first source wins), each annotated with
    ``via`` (asset|model|sub_assembly) + ``via_label``."""
    asset = get_asset(tag)
    if not asset:
        return {"parts": []}
    out, seen = [], set()

    def _add(items, via, via_label):
        for p in items:
            if p["id"] in seen:
                continue
            seen.add(p["id"])
            p["via"] = via
            p["via_label"] = via_label
            out.append(p)

    _add(parts_for_target("asset", asset["asset_tag"]), "asset", "")
    mid = asset.get("model_id")
    model = get_asset_model(mid) if mid else None
    if model:
        _add(parts_for_target("model", str(int(mid))), "model", model["name"])
    subs = sub_assemblies_for_asset(tag)
    for s in (subs.get("own", []) + subs.get("inherited", [])):
        _add(parts_for_target("sub_assembly", str(s["id"])), "sub_assembly",
             s["name"])
    return {"parts": out}


# EXPORT_ALL_SHEET names the normalized "All Assets" table a former Togen
# export wrote alongside the per-team sheets; the importer still SKIPS a sheet
# by this name (see _parse_workbook) so any such legacy workbook round-trips
# through /assets/import without every row flagging as a duplicate. The
# outbound export + the SharePoint sync itself were removed in DVI-1472 P1.
EXPORT_ALL_SHEET = "All Assets"


def import_health():
    """Import bookkeeping for the UI: when/what was last imported."""
    report_raw = get_meta("last_import_report")
    counts = {}
    if report_raw:
        try:
            report = json.loads(report_raw)
            counts = report.get("exception_counts", {})
        except ValueError:
            pass
    return {
        "last_import_at": get_meta("last_import_at"),
        "last_import_source": get_meta("last_import_source"),
        "assets": asset_count(),
        "exception_counts": counts,
    }


def last_import_report():
    raw = get_meta("last_import_report")
    if not raw:
        return None
    try:
        return json.loads(raw)
    except ValueError:
        return None


# ---------------------------------------------------------------------------
# Equipment Registry merge (DVI-1472 P2, D2/D3): assets.db is the single system
# of record for what used to live in wor_equipment.json. The app's
# _load_wor_equipment / _save_wor_equipment become a thin shim over
# load_equipment_registry / save_equipment_registry here, so every consumer
# (RtS/EVH sign-off, WO history, Manuals index, Oversight equipment +
# lorawan_mobile, Makdash counts, mm-eas-options) keeps its {norm_name: record}
# view unchanged. WO sidecars still store the free-text equipment name; the
# equipment_aliases table resolves that name to a merged asset.
# ---------------------------------------------------------------------------

# Assets created by the WO auto-seed / one-time registry migration carry this
# source_sheet so the sheet filter tells them apart from imported inventory.
EQUIPMENT_SHEET = "Equipment Registry"

# Default RtS departments for a freshly-tracked equipment asset — mirrors the
# app's _RTS_DEPTS. The app's PATCH route merges the real per-dept booleans.
_DEFAULT_RTS_DEPTS = ("maintenance", "operations", "safety", "quality")


def _norm_equipment_name(name):
    """Registry key: lowercase + whitespace-collapsed. Mirrors the app's
    _normalize_equipment so aliases match the WO sidecars' free-text names."""
    return " ".join((name or "").strip().lower().split())


def _oversight_from_store(text):
    """Stored include_in_oversight text -> the legacy value shape. '' -> False
    (off, factory default); 'true' -> True (legacy generic opt-in); otherwise
    the string ('stationary' | 'mobile')."""
    text = (text or "").strip()
    if not text or text.lower() in ("false", "0", "off", "none"):
        return False
    if text.lower() == "true":
        return True
    return text


def _oversight_to_store(value):
    """Legacy include_in_oversight value -> stored text. False/None/'' -> '';
    True -> 'true'; a string is kept trimmed."""
    if value is True:
        return "true"
    if isinstance(value, str):
        s = value.strip()
        return "" if s.lower() in ("", "false", "0", "off", "none") else s
    return ""


def _departments_from_store(raw):
    if not raw:
        return {}
    try:
        parsed = json.loads(raw)
    except ValueError:
        return {}
    return {str(k): bool(v) for k, v in parsed.items()} if isinstance(parsed, dict) else {}


def _equip_record_from_row(row):
    """Build the legacy {asset_id, name, rts_*, include_in_oversight, ...}
    record from an (alias JOIN asset) row, so the app-side shim returns exactly
    what wor_equipment.json used to hold."""
    created = row.get("equipment_created_at") or row.get("alias_created_at") or ""
    return {
        "asset_id": row["norm_name"],
        "name": row.get("alias_name") or row.get("description") or row["asset_tag"],
        "created_at": created,
        "rts_required": bool(row.get("rts_required")),
        "rts_departments": _departments_from_store(row.get("rts_departments")),
        "include_in_oversight": _oversight_from_store(row.get("include_in_oversight")),
        "oversight_dev_eui": row.get("oversight_dev_eui") or "",
        "asset_no": row["asset_tag"],
        "out_of_service": bool(row.get("out_of_service")),
        "offsite": bool(row.get("offsite")),
        # Extra breadcrumb (harmless to legacy consumers): the merged asset tag.
        "asset_tag": row["asset_tag"],
    }


_EQUIP_JOIN_SQL = (
    "SELECT al.norm_name AS norm_name, al.name AS alias_name,"
    " al.created_at AS alias_created_at, a.asset_tag AS asset_tag,"
    " a.description AS description, a.equipment_created_at AS equipment_created_at,"
    " a.rts_required AS rts_required, a.rts_departments AS rts_departments,"
    " a.include_in_oversight AS include_in_oversight,"
    " a.oversight_dev_eui AS oversight_dev_eui,"
    " a.out_of_service AS out_of_service, a.offsite AS offsite"
    " FROM equipment_aliases al JOIN assets a ON a.asset_tag = al.asset_tag")


def load_equipment_registry():
    """Return {norm_name: record} exactly like the old _load_wor_equipment().

    One entry per alias; ops fields come from the linked asset. Read half of
    the app-side compatibility shim."""
    init_assets_db()
    out = {}
    with _connect() as conn:
        cur = conn.execute(_EQUIP_JOIN_SQL)
        columns = [d[0] for d in cur.description]
        for row in cur.fetchall():
            rec = dict(zip(columns, row))
            out[rec["norm_name"]] = _equip_record_from_row(rec)
    return out


def _unique_equipment_tag(conn, base):
    base = _clean(base) or "Equipment"
    tag = base
    n = 2
    while conn.execute("SELECT 1 FROM assets WHERE asset_tag = ?",
                       (tag,)).fetchone():
        tag = "%s (%d)" % (base, n)
        n += 1
    return tag


def _insert_equipment_asset(conn, tag, record, now):
    """Create a new asset row for a registry record with no matching asset."""
    name = _clean(record.get("name") or "")
    created = _clean(record.get("created_at") or "") or now
    conn.execute(
        "INSERT INTO assets (asset_tag, tag_num, description, source_sheet,"
        " equipment_created_at, created_at, updated_at)"
        " VALUES (?, ?, ?, ?, ?, ?, ?)",
        (tag, int(tag) if tag.isdigit() else None, name, EQUIPMENT_SHEET,
         created, now, now))


def _link_alias(conn, norm_name, name, asset_tag, created):
    conn.execute(
        "INSERT OR IGNORE INTO equipment_aliases (norm_name, name, asset_tag,"
        " created_at) VALUES (?, ?, ?, ?)",
        (norm_name, _clean(name or ""), asset_tag, created))


def _write_equipment_fields(conn, asset_tag, record, now):
    """Write the ops fields from a registry record onto its linked asset."""
    depts = record.get("rts_departments")
    depts_json = json.dumps(depts) if isinstance(depts, dict) else ""
    conn.execute(
        "UPDATE assets SET rts_required = ?, rts_departments = ?,"
        " include_in_oversight = ?, oversight_dev_eui = ?,"
        " out_of_service = ?, offsite = ?, updated_at = ? WHERE asset_tag = ?",
        (1 if record.get("rts_required") else 0, depts_json,
         _oversight_to_store(record.get("include_in_oversight")),
         _clean(record.get("oversight_dev_eui") or ""),
         1 if record.get("out_of_service") else 0,
         1 if record.get("offsite") else 0, now, asset_tag))


def save_equipment_registry(registry):
    """Write half of the shim: upsert each {norm_name: record} entry into
    assets.db. An entry whose norm_name has no alias yet is the WO auto-seed
    path -> it links to an existing asset named by record['asset_no'] if that
    tag exists, else CREATES a new asset (source_sheet=EQUIPMENT_SHEET) + alias.
    Existing aliases update their linked asset's ops fields. Entries are never
    deleted here (flag-not-fix)."""
    init_assets_db()
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with _write_lock, _connect() as conn:
        alias_map = {r[0]: r[1] for r in conn.execute(
            "SELECT norm_name, asset_tag FROM equipment_aliases")}
        for norm_name, record in registry.items():
            if not norm_name:
                continue
            asset_tag = alias_map.get(norm_name)
            if asset_tag is None:
                candidate = _clean(record.get("asset_no") or "")
                if candidate and conn.execute(
                        "SELECT 1 FROM assets WHERE asset_tag = ?",
                        (candidate,)).fetchone():
                    asset_tag = candidate
                else:
                    asset_tag = _unique_equipment_tag(
                        conn, record.get("name") or norm_name)
                    _insert_equipment_asset(conn, asset_tag, record, now)
                _link_alias(conn, norm_name, record.get("name"), asset_tag,
                            _clean(record.get("created_at") or "") or now)
                alias_map[norm_name] = asset_tag
            _write_equipment_fields(conn, asset_tag, record, now)


def equipment_aliases_for(asset_tag):
    """Ordered [{norm_name, name}] equipment aliases pointing at one asset —
    used by the Assets detail to gather WO history across every name variant."""
    with _connect() as conn:
        return [{"norm_name": n, "name": nm} for (n, nm) in conn.execute(
            "SELECT norm_name, name FROM equipment_aliases"
            " WHERE asset_tag = ? ORDER BY created_at, norm_name",
            (str(asset_tag),))]


def equipment_record_for_asset(asset_tag):
    """The legacy registry record for one asset (first alias), or None when the
    asset carries no equipment alias. Lets the Assets detail render RtS/Service/
    Oversight without a second store."""
    with _connect() as conn:
        cur = conn.execute(
            _EQUIP_JOIN_SQL + " WHERE al.asset_tag = ? ORDER BY al.created_at,"
            " al.norm_name LIMIT 1", (str(asset_tag),))
        columns = [d[0] for d in cur.description]
        row = cur.fetchone()
        if not row:
            return None
        return _equip_record_from_row(dict(zip(columns, row)))


def asset_tag_for_equipment(norm_name):
    """Resolve a normalized equipment name to its merged asset_tag, or None."""
    with _connect() as conn:
        row = conn.execute(
            "SELECT asset_tag FROM equipment_aliases WHERE norm_name = ?",
            (_norm_equipment_name(norm_name),)).fetchone()
    return row[0] if row else None


def ensure_equipment_alias(asset_tag, name=None):
    """Link an EXISTING asset to an equipment alias so it enters the registry
    view (the consolidated Assets detail's "Track as equipment"). Never creates
    a new asset. Returns (norm_name, error): the primary alias norm_name for
    the asset, or an existing one if already tracked. Errors when the asset
    doesn't exist or no usable name can be derived."""
    tag = _clean(asset_tag)
    with _write_lock, _connect() as conn:
        row = conn.execute(
            "SELECT description FROM assets WHERE asset_tag = ?", (tag,)).fetchone()
        if row is None:
            return None, "Asset %s no longer exists." % tag
        existing = conn.execute(
            "SELECT norm_name FROM equipment_aliases WHERE asset_tag = ?"
            " ORDER BY created_at, norm_name LIMIT 1", (tag,)).fetchone()
        if existing:
            return existing[0], None
        display = _clean(name) or _clean(row[0]) or tag
        norm = _norm_equipment_name(display)
        if not norm:
            return None, "Give the equipment a name to track it."
        clash = conn.execute(
            "SELECT asset_tag FROM equipment_aliases WHERE norm_name = ?",
            (norm,)).fetchone()
        if clash and clash[0] != tag:
            return None, ("Another asset already uses the equipment name %r."
                          % display)
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        conn.execute(
            "INSERT OR IGNORE INTO equipment_aliases (norm_name, name,"
            " asset_tag, created_at) VALUES (?, ?, ?, ?)",
            (norm, display, tag, now))
        # Seed the RtS departments default the same way _make_equipment_record
        # would, so a freshly-tracked asset starts with all depts enabled.
        cur = conn.execute(
            "SELECT rts_departments, equipment_created_at FROM assets"
            " WHERE asset_tag = ?", (tag,)).fetchone()
        if cur is not None and not (cur[0] or "").strip():
            conn.execute(
                "UPDATE assets SET rts_departments = ?,"
                " equipment_created_at = COALESCE(NULLIF(equipment_created_at,''),?)"
                " WHERE asset_tag = ?",
                (json.dumps({d: True for d in _DEFAULT_RTS_DEPTS}), now, tag))
        return norm, None


def migrate_equipment_registry(registry):
    """One-time idempotent merge of a legacy wor_equipment.json dict into
    assets.db (DVI-1472 P2 D3). For each record match to an asset by
    asset_no == asset_tag, else exact normalized-name match against
    tag/description; else create a new asset (source_sheet=EQUIPMENT_SHEET).
    Ambiguous (multiple) name matches are reported, NOT linked (flag-not-fix).
    Re-running is safe: a name already aliased is skipped (never re-clobbers
    admin edits). Returns + persists a report dict."""
    init_assets_db()
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    report = {
        "migrated_at": now, "total": 0, "already_linked": 0,
        "linked_by_asset_no": 0, "linked_by_name": 0, "created": 0,
        "ambiguous": [], "exception_counts": {},
    }

    def flag(issue):
        report["exception_counts"][issue] = (
            report["exception_counts"].get(issue, 0) + 1)

    with _write_lock, _connect() as conn:
        existing = {r[0] for r in conn.execute(
            "SELECT norm_name FROM equipment_aliases")}
        name_index = {}
        for tag, desc in conn.execute(
                "SELECT asset_tag, description FROM assets"):
            for key in {_norm_equipment_name(tag), _norm_equipment_name(desc)}:
                if key:
                    name_index.setdefault(key, set()).add(tag)
        for norm_name, record in registry.items():
            if not norm_name:
                continue
            report["total"] += 1
            if norm_name in existing:
                report["already_linked"] += 1
                continue
            asset_tag, kind = None, None
            asset_no = _clean(record.get("asset_no") or "")
            if asset_no and conn.execute(
                    "SELECT 1 FROM assets WHERE asset_tag = ?",
                    (asset_no,)).fetchone():
                asset_tag, kind = asset_no, "asset_no"
            else:
                cands = name_index.get(norm_name) or set()
                if len(cands) == 1:
                    asset_tag, kind = next(iter(cands)), "name"
                elif len(cands) > 1:
                    report["ambiguous"].append({
                        "name": record.get("name") or norm_name,
                        "candidates": sorted(cands),
                    })
                    flag("ambiguous name match (not linked)")
                    continue
            if asset_tag is None:
                asset_tag = _unique_equipment_tag(
                    conn, record.get("name") or norm_name)
                _insert_equipment_asset(conn, asset_tag, record, now)
                kind = "created"
                name_index.setdefault(norm_name, set()).add(asset_tag)
            _link_alias(conn, norm_name, record.get("name"), asset_tag,
                        _clean(record.get("created_at") or "") or now)
            _write_equipment_fields(conn, asset_tag, record, now)
            existing.add(norm_name)
            if kind == "asset_no":
                report["linked_by_asset_no"] += 1
            elif kind == "name":
                report["linked_by_name"] += 1
            else:
                report["created"] += 1
    set_meta("equipment_migration_report", json.dumps(report))
    set_meta("equipment_migration_at", now)
    return report


def equipment_migration_report():
    raw = get_meta("equipment_migration_report")
    if not raw:
        return None
    try:
        return json.loads(raw)
    except ValueError:
        return None