invoice_az/invoice_az/master_data/sync.py

"""Diff-based sync for master data shipped as JSON.

The legacy approach (fixtures dir) re-imports every record on every `bench
migrate` via DELETE+INSERT. For static reference data with thousands of rows
that's a multi-minute hit. This module instead reconciles the DB against the
JSON on every migrate: it reads the current rows in one query and applies only
the delta — insert new rows, update changed rows, delete rows missing from the
file (or disable / skip when they are referenced by existing documents).

The diff is intentionally driven by the *DB state*, not a stored file hash:
if rows go missing from the table (manual delete, restore from an old dump),
the next migrate restores them. A full diff costs one SELECT plus an in-memory
comparison — sub-second even for ~13k rows.
"""

from __future__ import annotations

import json
import os
from collections.abc import Iterable

import frappe

LOG_PREFIX = "invoice_az"


def _load_records(path: str, key_field: str) -> dict[str, dict]:
    with open(path, encoding="utf-8") as f:
        raw = json.load(f)
    if not isinstance(raw, list):
        frappe.throw(f"{path}: expected a JSON array")

    out: dict[str, dict] = {}
    for rec in raw:
        key = rec.get(key_field)
        if not key:
            continue
        out[str(key)] = rec
    return out


def _fetch_existing(doctype: str, fields: Iterable[str], key_field: str) -> dict[str, dict]:
    rows = frappe.db.sql(
        f"SELECT {', '.join('`' + f + '`' for f in fields)} FROM `tab{doctype}`",
        as_dict=True,
    )
    return {str(row[key_field]): row for row in rows}


def _try_remove(doctype: str, name: str, has_disabled: bool) -> str:
    """Smart-delete: try hard delete; on link conflicts soft-disable or skip.

    Returns one of: "deleted", "disabled", "skipped".
    """
    try:
        frappe.delete_doc(
            doctype,
            name,
            ignore_permissions=True,
            ignore_missing=True,
            delete_permanently=True,
        )
        return "deleted"
    except frappe.LinkExistsError:
        if has_disabled:
            frappe.db.set_value(doctype, name, "disabled", 1, update_modified=False)
            return "disabled"
        frappe.logger().warning(
            f"[{LOG_PREFIX}] {doctype} '{name}' is referenced; "
            f"cannot delete and no 'disabled' field — kept as-is"
        )
        return "skipped"


def sync_master_data(
    doctype: str,
    json_path: str,
    key_field: str = "name",
    *,
    is_tree: bool = False,
    manage_deletes: bool = True,
) -> dict:
    """Reconcile a doctype against a JSON file, applying only the delta.

    Runs a full diff against the current DB state every call — no file-hash
    fast-path, so rows that go missing from the table are always restored.

    is_tree: doctype is a nested set — insert via db_insert() and rebuild
        lft/rgt once at the end (avoids per-row tree recomputation and the
        need to insert parents before children).
    manage_deletes: when False, rows present in the DB but not in the JSON
        are left untouched. Use False when the JSON is only a subset of the
        table.
    """
    if not os.path.exists(json_path):
        frappe.logger().warning(f"[{LOG_PREFIX}] master data file not found: {json_path}")
        return {"skipped": True}

    target = _load_records(json_path, key_field)
    if not target:
        return {"empty_file": True}

    meta = frappe.get_meta(doctype)
    sample = next(iter(target.values()))
    value_fields = [
        f for f in sample.keys()
        if f != key_field and (meta.has_field(f) or f == "name")
    ]
    select_fields = [key_field, *value_fields] if key_field not in value_fields else value_fields
    existing = _fetch_existing(doctype, select_fields, key_field)

    has_disabled = meta.has_field("disabled")

    inserted = updated = deleted = disabled = skipped = 0

    to_insert = [(key, rec) for key, rec in target.items() if key not in existing]
    to_check = [(key, rec) for key, rec in target.items() if key in existing]

    # Bulk-insert path: avoids thousands of ORM .insert() round-trips, and for
    # tree doctypes also sidesteps per-row nested-set recomputation (lft/rgt are
    # rebuilt once at the end). db_insert() skips before_insert/after_insert
    # hooks — appropriate for static master data.
    BULK_THRESHOLD = 100
    use_bulk = is_tree or len(to_insert) >= BULK_THRESHOLD
    if use_bulk:
        for key, rec in to_insert:
            doc = frappe.new_doc(doctype)
            if key_field == "name":
                doc.name = key
                doc.flags.name_set = True
            else:
                doc.set(key_field, key)
            for field in value_fields:
                if field in rec:
                    doc.set(field, rec[field])
            if is_tree:
                # placeholder bounds; rebuild_tree() fixes them after all inserts
                doc.lft = 0
                doc.rgt = 0
            doc.db_insert()
            inserted += 1
            if inserted % 500 == 0:
                frappe.db.commit()
        frappe.db.commit()
    else:
        for key, rec in to_insert:
            doc = frappe.new_doc(doctype)
            if key_field == "name":
                doc.name = key
            else:
                doc.set(key_field, key)
            for field in value_fields:
                if field in rec:
                    doc.set(field, rec[field])
            doc.flags.ignore_permissions = True
            doc.flags.ignore_mandatory = True
            doc.flags.ignore_links = True
            doc.insert(ignore_if_duplicate=True)
            inserted += 1

    for key, rec in to_check:
        db_row = existing[key]
        diff = {f: rec[f] for f in value_fields if f in rec and (db_row.get(f) or "") != (rec[f] or "")}
        if diff:
            frappe.db.set_value(doctype, key, diff, update_modified=False)
            updated += 1

    # Delete rows not present in the JSON. Skipped when the JSON is only a
    # subset of the table.
    if manage_deletes:
        for key in set(existing.keys()) - set(target.keys()):
            outcome = _try_remove(doctype, key, has_disabled)
            if outcome == "deleted":
                deleted += 1
            elif outcome == "disabled":
                disabled += 1
            else:
                skipped += 1

    # Tree doctypes: recompute lft/rgt once for the whole tree.
    if is_tree and (inserted or updated or deleted):
        from frappe.utils.nestedset import rebuild_tree

        rebuild_tree(doctype)

    frappe.db.commit()

    summary = {
        "doctype": doctype,
        "inserted": inserted,
        "updated": updated,
        "deleted": deleted,
        "disabled": disabled,
        "skipped": skipped,
    }
    if inserted or updated or deleted or disabled or skipped:
        print(
            f"[{LOG_PREFIX}] {doctype}: "
            f"+{inserted} ~{updated} -{deleted} (disabled {disabled}, skipped {skipped})"
        )
    return summary


def sync_e_taxes_item_group() -> dict:
    path = frappe.get_app_path("invoice_az", "master_data", "e_taxes_item_group.json")
    return sync_master_data("E-Taxes Item Group", path, key_field="name")


def sync_classification_code() -> dict:
    path = frappe.get_app_path("invoice_az", "master_data", "classification_code.json")
    return sync_master_data("Classification code", path, key_field="name")


def sync_all() -> None:
    """Entry point for after_migrate — runs all master_data syncs sequentially."""
    sync_e_taxes_item_group()
    sync_classification_code()