feat(imports): report bulk-import progress to the background-tasks registry

Register sales / purchase / VAT bulk imports with the background-tasks
registry: emit per-item progress, honour cancellation requests between
items, and finish/cancel the job so the global widget can track them.

invoice_az/api.py

@@ -1,7 +1,7 @@
 import frappe
 import requests
 import json
-from frappe.utils import nowdate, now, cint, now_datetime, get_datetime
+from frappe.utils import nowdate, now, cint, now_datetime, get_datetime, getdate
 from frappe.model.document import Document
 import re
 from difflib import SequenceMatcher
@@ -11,6 +11,7 @@ import time
 
 # Импорты из модуля аутентификации
 from invoice_az.auth import record_etaxes_activity, get_default_asan_login
+from invoice_az import background_tasks
 from .utils import resolve_customer_group
 from .sales_api import get_sales_invoices
 
@@ -2374,9 +2375,19 @@ def process_single_invoice_for_units(token, invoice_id, source_type='purchase'):
         }
     
 @frappe.whitelist()
-def process_invoice_parties_from_list(invoice_list_data, token):
+def process_invoice_parties_from_list(invoice_list_data, token=None, load_customers=1, load_suppliers=1):
-    """Обрабатывает контрагентов из списка инвойсов с разделением на customers и suppliers"""
+    """Обрабатывает контрагентов прямо из СПИСКА инвойсов (без захода внутрь документа).
+
+    Контрагенты (customers/suppliers) полностью содержатся в списке инвойсов
+    (sender/receiver -> name + tin), поэтому для них не нужен запрос деталей
+    каждого инвойса. Это на порядок быстрее, чем построчная обработка через
+    get_invoice_details(). load_customers/load_suppliers позволяют ограничить
+    обработку только нужным типом.
+    """
     try:
+        load_customers = int(load_customers)
+        load_suppliers = int(load_suppliers)
+
         if isinstance(invoice_list_data, str):
             invoices = json.loads(invoice_list_data)
         else:
@@ -2414,7 +2425,7 @@ def process_invoice_parties_from_list(invoice_list_data, token):
                         
                         if source == 'inbox':
                             # Inbox: Sender = Supplier
-                            if sender_name not in batch_unique_suppliers:
+                            if load_suppliers and sender_name not in batch_unique_suppliers:
                                 existing = frappe.db.exists('E-Taxes Suppliers', sender_name)
 
                                 if not existing:
@@ -2428,7 +2439,7 @@ def process_invoice_parties_from_list(invoice_list_data, token):
                                     }
                         else:
                             # Outbox: Sender = Customer
-                            if sender_name not in batch_unique_customers:
+                            if load_customers and sender_name not in batch_unique_customers:
                                 existing = frappe.db.exists('E-Taxes Customers', sender_name)
                                 
                                 if not existing:
@@ -2450,7 +2461,7 @@ def process_invoice_parties_from_list(invoice_list_data, token):
                     
                     receiver_name = ' '.join(receiver_name.split())
                     
-                    if receiver_name and receiver_tin:
+                    if receiver_name and receiver_tin and load_customers:
                         # Логика: Receiver всегда Customer (и для inbox, и для outbox)
                         if receiver_name not in batch_unique_customers:
                             existing = frappe.db.exists('E-Taxes Customers', receiver_name)
@@ -2513,6 +2524,10 @@ def process_invoice_parties_from_list(invoice_list_data, token):
             except Exception:
                 total_failed += 1
 
+        # Bulk loop -> commit once after the loop (CLAUDE.md §12)
+        if total_customers_created or total_suppliers_created:
+            frappe.db.commit()
+
         return {
             'success': True,
             'customers_created': total_customers_created,
@@ -3772,6 +3787,45 @@ def get_reference_data_lists(data_type, limit=100, offset=0):
         frappe.log_error(f"Error getting reference data list for {data_type}: {str(e)}\n{frappe.get_traceback()}", "Reference Data List Error")
         return {'success': False, 'message': str(e)}
 
+
+@frappe.whitelist()
+def get_settings_workflow_state():
+    """Lightweight reference-data counts that drive the setup wizard on the
+    E-Taxes Settings form. Used to figure out the user's next step.
+
+    Returns per-entity total and 'new' (still unmapped) counts. A mapping
+    step is considered complete once its 'new' count is zero (nothing left
+    to map), so an entity with no cached records reads as already done.
+    """
+    try:
+        def counts(doctype):
+            if not frappe.db.exists('DocType', doctype):
+                return 0, 0
+            total = frappe.db.count(doctype)
+            new = frappe.db.count(doctype, {'status': 'New'})
+            return total, new
+
+        i_total, i_new = counts('E-Taxes Item')
+        c_total, c_new = counts('E-Taxes Customers')
+        s_total, s_new = counts('E-Taxes Suppliers')
+        u_total, u_new = counts('E-Taxes Unit')
+
+        return {
+            'success': True,
+            'reference_total': i_total + c_total + s_total + u_total,
+            'items_total': i_total, 'items_new': i_new,
+            'customers_total': c_total, 'customers_new': c_new,
+            'suppliers_total': s_total, 'suppliers_new': s_new,
+            'units_total': u_total, 'units_new': u_new,
+        }
+    except Exception as e:
+        frappe.log_error(
+            f"Error getting settings workflow state: {str(e)}\n{frappe.get_traceback()}",
+            "Settings Workflow State Error",
+        )
+        return {'success': False, 'message': str(e)}
+
+
 @frappe.whitelist()
 def process_single_invoice_for_reference_data(token, invoice_id, source_type='purchase',
                                                load_items=1, load_units=1,
@@ -4325,6 +4379,7 @@ def import_bulk_purchase_invoices(invoice_ids, token, warehouse=None):
         invoice_ids = json.loads(invoice_ids)
 
     user = frappe.session.user
+    job_id = background_tasks.new_job_id()
 
     frappe.enqueue(
         _process_bulk_purchase_import,
@@ -4332,15 +4387,17 @@ def import_bulk_purchase_invoices(invoice_ids, token, warehouse=None):
         token=token,
         warehouse=warehouse,
         user=user,
+        bg_job_id=job_id,
         queue="default",
         timeout=1200,
     )
 
-    return {"success": True, "enqueued": True, "total": len(invoice_ids)}
+    return {"success": True, "enqueued": True, "total": len(invoice_ids), "job_id": job_id}
 
 
-def _process_bulk_purchase_import(invoice_ids, token, warehouse, user):
+def _process_bulk_purchase_import(invoice_ids, token, warehouse, user, bg_job_id=None):
     """Background job: fetch details and import each purchase invoice with realtime progress."""
+    job_id = bg_job_id
     frappe.set_user(user)
     record_etaxes_activity()
 
@@ -4348,7 +4405,19 @@ def _process_bulk_purchase_import(invoice_ids, token, warehouse, user):
     imported_count = 0
     errors = []
 
+    background_tasks.register_task(job_id, user, "purchase_import", "Importing purchase invoices", total=total)
+
     for idx, invoice_id in enumerate(invoice_ids):
+        if job_id and background_tasks.is_cancel_requested(job_id):
+            background_tasks.finish_task(job_id, user, "cancelled")
+            frappe.publish_realtime(
+                "purchase_import_complete",
+                {"total": total, "imported": imported_count, "errors": errors, "cancelled": True},
+                user=user,
+            )
+            return
+        background_tasks.update_task(job_id, user, current=idx, total=total,
+                                     message=f"Importing invoice {idx + 1}/{total}")
         try:
             details = get_invoice_details(token, invoice_id)
         except Exception:
@@ -4417,8 +4486,379 @@ def _process_bulk_purchase_import(invoice_ids, token, warehouse, user):
             user=user,
         )
 
+    background_tasks.finish_task(job_id, user, "done")
     frappe.publish_realtime(
         "purchase_import_complete",
         {"total": total, "imported": imported_count, "errors": errors},
         user=user,
     )
+
+
+# ======= REFERENCE DATA LOADING (BACKGROUND) =======
+#
+# Previously the whole "Data Loading" flow (fetch invoices -> create parties ->
+# create items/units) ran as a recursive loop in the browser, so a page reload
+# killed it and a modal progress bar froze the page. It now runs entirely in a
+# background worker that mirrors progress into the global Background Tasks widget
+# (see invoice_az/background_tasks.py) and can be cancelled from there.
+
+REFERENCE_LOADING_COMPLETE_EVENT = "reference_data_loading_complete"
+_REF_FETCH_PAGE_CAP = 500  # safety net against a never-ending hasMore (≤100k invoices)
+
+
+@frappe.whitelist()
+def start_reference_data_loading_bg(date_from, date_to, load_items=0,
+                                    load_customers=0, load_suppliers=0, load_units=0):
+    """Enqueue the reference-data loading worker and return its job_id.
+
+    `date_from` / `date_to` arrive as plain dates (YYYY-MM-DD) from the dialog;
+    we format them the way the E-Taxes filter expects (DD-MM-YYYY HH:MM)."""
+    record_etaxes_activity()
+
+    load_items = 1 if cint(load_items) else 0
+    load_customers = 1 if cint(load_customers) else 0
+    load_suppliers = 1 if cint(load_suppliers) else 0
+    load_units = 1 if cint(load_units) else 0
+
+    if not (load_items or load_customers or load_suppliers or load_units):
+        return {"success": False, "message": "Please select at least one data type to load."}
+
+    try:
+        df = getdate(date_from)
+        dt = getdate(date_to)
+    except Exception:
+        return {"success": False, "message": "Invalid date range."}
+    if df > dt:
+        return {"success": False, "message": "Date From cannot be later than Date To."}
+
+    date_from_str = df.strftime("%d-%m-%Y 00:00")
+    date_to_str = dt.strftime("%d-%m-%Y 23:59")
+
+    job_id = background_tasks.new_job_id()
+    frappe.enqueue(
+        _process_reference_data_loading,
+        date_from=date_from_str,
+        date_to=date_to_str,
+        load_items=load_items,
+        load_customers=load_customers,
+        load_suppliers=load_suppliers,
+        load_units=load_units,
+        user=frappe.session.user,
+        bg_job_id=job_id,  # NOT `job_id`: that name is reserved by frappe.enqueue
+        queue="default",
+        timeout=7200,
+    )
+    return {"success": True, "enqueued": True, "job_id": job_id}
+
+
+def _ref_fetch_source(date_from, date_to, source, job_id, user, all_invoices, fatal):
+    """Paginate one source (inbox/outbox) into `all_invoices`.
+
+    Returns (ok, token). On a failure, inbox is fatal (ok=False), outbox is not
+    (ok=True, we just stop with whatever we have) — mirroring the old JS logic."""
+    loader = load_single_invoice_batch if source == "inbox" else load_single_sales_invoice_batch
+    label = "purchase" if source == "inbox" else "sales"
+    token = None
+    offset = 0
+    for _page in range(_REF_FETCH_PAGE_CAP):
+        if background_tasks.is_cancel_requested(job_id):
+            return None, token  # signal cancel to caller via token=None + sentinel
+        result = loader(date_from, date_to, offset)
+        if not result.get("success"):
+            return (not fatal), token
+        token = token or result.get("token")
+        batch = result.get("invoices") or []
+        for inv in batch:
+            inv["_source"] = source
+        all_invoices.extend(batch)
+        background_tasks.update_task(
+            job_id, user,
+            message=f"Finding {label} invoices: {len(all_invoices)} found...",
+        )
+        if not (result.get("hasMore") and batch):
+            break
+        offset += 200
+    return True, token
+
+
+def _process_reference_data_loading(date_from, date_to, load_items, load_customers,
+                                    load_suppliers, load_units, user, bg_job_id=None):
+    """Background worker — fetch invoices, then create parties + items/units,
+    publishing progress to the Background Tasks registry the whole way."""
+    # `bg_job_id` (not `job_id`) because frappe.enqueue reserves the `job_id` kwarg
+    # for the RQ job identifier and would not forward it to us.
+    job_id = bg_job_id
+    frappe.set_user(user)
+    record_etaxes_activity()
+
+    accumulated = {
+        "items_created": 0, "units_created": 0,
+        "customers_created": 0, "suppliers_created": 0,
+    }
+
+    def _cancel_and_finish():
+        background_tasks.finish_task(job_id, user, "cancelled")
+        frappe.publish_realtime(
+            REFERENCE_LOADING_COMPLETE_EVENT, {"cancelled": True}, user=user
+        )
+
+    def _is_cancelled():
+        return background_tasks.is_cancel_requested(job_id)
+
+    background_tasks.register_task(
+        job_id, user, "reference_data_loading", "Loading reference data", total=0
+    )
+
+    try:
+        # ---- Phase 1: fetch invoice lists ----
+        background_tasks.update_task(job_id, user, message="Finding invoices...")
+        all_invoices = []
+        token = None
+
+        ok, tok = _ref_fetch_source(date_from, date_to, "inbox", job_id, user, all_invoices, fatal=True)
+        token = token or tok
+        if ok is None:  # cancelled mid-fetch
+            return _cancel_and_finish()
+        if not ok:
+            background_tasks.finish_task(job_id, user, "error", "Failed to fetch purchase invoices.")
+            frappe.publish_realtime(
+                REFERENCE_LOADING_COMPLETE_EVENT,
+                {"error": "fetch_failed", "message": "Failed to fetch purchase invoices."},
+                user=user,
+            )
+            return
+
+        # outbox (sales) only matters for customers / items / units
+        outbox_needed = bool(load_customers or load_items or load_units)
+        if outbox_needed and not _is_cancelled():
+            _ok2, tok2 = _ref_fetch_source(date_from, date_to, "outbox", job_id, user, all_invoices, fatal=False)
+            token = token or tok2
+            if _ok2 is None:
+                return _cancel_and_finish()
+
+        if _is_cancelled():
+            return _cancel_and_finish()
+
+        if not all_invoices:
+            background_tasks.finish_task(job_id, user, "done", "No invoices found")
+            frappe.publish_realtime(
+                REFERENCE_LOADING_COMPLETE_EVENT,
+                {"summary": accumulated, "empty": True},
+                user=user,
+            )
+            return
+
+        total_invoices = len(all_invoices)
+        needs_detail = bool(load_items or load_units)
+        needs_parties = bool(load_customers or load_suppliers)
+
+        # The bar tracks the VISIBLE per-invoice work (0..total_invoices). When both
+        # phases run, the parties pass is a fast pre-step (one bulk call per 500
+        # invoices) and must NOT consume half the bar — otherwise the slow detail
+        # loop appears to "start at 50%". So parties drive the bar only when there
+        # is no detail phase; otherwise they just tick the message while the bar
+        # stays at 0 until the per-invoice loop begins.
+        grand_total = total_invoices
+        background_tasks.update_task(
+            job_id, user, current=0, total=grand_total,
+            message=f"Found {total_invoices} invoices. Processing...",
+        )
+
+        # ---- Phase 2: parties straight from the invoice list (chunked) ----
+        if needs_parties:
+            parties_only = not needs_detail
+            done_parties = 0
+            CHUNK = 500
+            for start in range(0, total_invoices, CHUNK):
+                if _is_cancelled():
+                    return _cancel_and_finish()
+                chunk = all_invoices[start:start + CHUNK]
+                res = process_invoice_parties_from_list(chunk, token, load_customers, load_suppliers)
+                if isinstance(res, dict) and res.get("success"):
+                    accumulated["customers_created"] += res.get("customers_created", 0) or 0
+                    accumulated["suppliers_created"] += res.get("suppliers_created", 0) or 0
+                done_parties += len(chunk)
+                made = accumulated["customers_created"] + accumulated["suppliers_created"]
+                background_tasks.update_task(
+                    job_id, user,
+                    current=(done_parties if parties_only else 0),
+                    total=grand_total,
+                    message=(f"Counterparties: {made} created" if parties_only
+                             else f"Reading counterparties... {made} created"),
+                )
+
+        # ---- Phase 3: items/units from invoice details (per invoice) ----
+        if needs_detail:
+            # Parties were already handled from the list above; disable them here
+            # so we don't do the work twice.
+            for idx, inv in enumerate(all_invoices):
+                if _is_cancelled():
+                    return _cancel_and_finish()
+                source = inv.get("_source", "inbox")
+                source_type = "sales" if source == "outbox" else "purchase"
+                res = process_single_invoice_for_reference_data(
+                    token, inv.get("id"), source_type,
+                    load_items=load_items, load_units=load_units,
+                    load_customers=0, load_suppliers=0,
+                )
+                if isinstance(res, dict):
+                    if res.get("new_token"):
+                        token = res["new_token"]
+                    accumulated["items_created"] += res.get("items_created", 0) or 0
+                    accumulated["units_created"] += res.get("units_created", 0) or 0
+                if idx % 5 == 0 or idx == total_invoices - 1:
+                    background_tasks.update_task(
+                        job_id, user, current=idx + 1, total=grand_total,
+                        message=f"Processing invoice {idx + 1}/{total_invoices}",
+                    )
+
+        background_tasks.finish_task(job_id, user, "done")
+        frappe.publish_realtime(
+            REFERENCE_LOADING_COMPLETE_EVENT, {"summary": accumulated}, user=user
+        )
+
+    except Exception as e:
+        frappe.log_error(
+            f"Reference data loading failed: {e}\n{frappe.get_traceback()}",
+            "Reference Data Loading Error",
+        )
+        background_tasks.finish_task(job_id, user, "error", str(e))
+        frappe.publish_realtime(
+            REFERENCE_LOADING_COMPLETE_EVENT,
+            {"error": "exception", "message": str(e)},
+            user=user,
+        )
+        raise
+
+
+# ======= ITEMS / UNITS SYNC FROM INVOICES (BACKGROUND) =======
+#
+# The E-Taxes Item / E-Taxes Unit list views had a "Load from E-Taxes" button
+# that ran a client-side loop (fetch invoice pages, then extract items/units per
+# invoice) with a blocking modal that died on reload. These now run server-side
+# through the same Background Tasks widget, reusing the exact backend functions
+# the client loop used so behaviour is unchanged.
+
+ITEMS_LOADING_COMPLETE_EVENT = "items_loading_complete"
+UNITS_LOADING_COMPLETE_EVENT = "units_loading_complete"
+
+
+@frappe.whitelist()
+def start_items_loading_bg(date_from, date_to):
+    """Background version of the E-Taxes Item list 'Load from E-Taxes'.
+    `date_from`/`date_to` arrive already formatted (DD-MM-YYYY HH:MM) by the dialog."""
+    record_etaxes_activity()
+    if not date_from or not date_to:
+        return {"success": False, "message": "Date range required."}
+    job_id = background_tasks.new_job_id()
+    frappe.enqueue(
+        _process_entity_loading,
+        date_from=date_from, date_to=date_to, entity="items",
+        user=frappe.session.user, bg_job_id=job_id,
+        queue="default", timeout=7200,
+    )
+    return {"success": True, "enqueued": True, "job_id": job_id}
+
+
+@frappe.whitelist()
+def start_units_loading_bg(date_from, date_to):
+    """Background version of the E-Taxes Unit list 'Load from E-Taxes'."""
+    record_etaxes_activity()
+    if not date_from or not date_to:
+        return {"success": False, "message": "Date range required."}
+    job_id = background_tasks.new_job_id()
+    frappe.enqueue(
+        _process_entity_loading,
+        date_from=date_from, date_to=date_to, entity="units",
+        user=frappe.session.user, bg_job_id=job_id,
+        queue="default", timeout=7200,
+    )
+    return {"success": True, "enqueued": True, "job_id": job_id}
+
+
+def _process_entity_loading(date_from, date_to, entity, user, bg_job_id=None):
+    """Shared worker for the items-only / units-only list-view syncs: paginate
+    the invoice list, then extract items/units from each invoice's details —
+    the same functions the old client loop used, now with widget progress/cancel."""
+    job_id = bg_job_id
+    frappe.set_user(user)
+    record_etaxes_activity()
+
+    if entity == "units":
+        fetch_fn, process_fn = load_units_from_invoices, process_single_invoice_for_units
+        feature, label, complete_event = "units_loading", "Loading units from invoices", UNITS_LOADING_COMPLETE_EVENT
+    else:
+        fetch_fn, process_fn = load_items_from_invoices, process_single_invoice_for_items
+        feature, label, complete_event = "items_loading", "Loading items from invoices", ITEMS_LOADING_COMPLETE_EVENT
+
+    background_tasks.register_task(job_id, user, feature, label, total=0)
+    created = 0
+
+    def cancelled():
+        return background_tasks.is_cancel_requested(job_id)
+
+    def finish_cancel():
+        background_tasks.finish_task(job_id, user, "cancelled")
+        frappe.publish_realtime(complete_event, {"cancelled": True}, user=user)
+
+    try:
+        # ---- fetch (paginate inbox+outbox, mirroring the old client loop) ----
+        background_tasks.update_task(job_id, user, message="Finding invoices...")
+        all_invoices, token, offset = [], None, 0
+        for _page in range(_REF_FETCH_PAGE_CAP):
+            if cancelled():
+                return finish_cancel()
+            res = fetch_fn(date_from, date_to, 200, offset)
+            if not res.get("success") and res.get("error") == "unauthorized":
+                res = fetch_fn(date_from, date_to, 200, offset)  # token re-read from DB; retry once
+            if not res.get("success"):
+                background_tasks.finish_task(job_id, user, "error", res.get("message"))
+                frappe.publish_realtime(
+                    complete_event,
+                    {"error": res.get("error", "fetch_failed"), "message": res.get("message")},
+                    user=user,
+                )
+                return
+            token = token or res.get("token")
+            batch = res.get("invoices") or []
+            all_invoices.extend(batch)
+            background_tasks.update_task(job_id, user, message=f"Finding invoices: {len(all_invoices)} found...")
+            if not (res.get("hasMore") and batch):
+                break
+            offset += len(batch)
+
+        if cancelled():
+            return finish_cancel()
+        if not all_invoices:
+            background_tasks.finish_task(job_id, user, "done", "No invoices found")
+            frappe.publish_realtime(complete_event, {"created": 0, "empty": True}, user=user)
+            return
+
+        # ---- process each invoice ----
+        total = len(all_invoices)
+        background_tasks.update_task(job_id, user, current=0, total=total,
+                                     message=f"Found {total} invoices. Processing...")
+        for idx, inv in enumerate(all_invoices):
+            if cancelled():
+                return finish_cancel()
+            source_type = "sales" if inv.get("_source") == "outbox" else "purchase"
+            res = process_fn(token, inv.get("id"), source_type)
+            if isinstance(res, dict) and res.get("error") == "unauthorized":
+                fresh = get_default_asan_login()
+                if fresh.get("found") and fresh.get("main_token"):
+                    token = fresh["main_token"]
+                    res = process_fn(token, inv.get("id"), source_type)
+            if isinstance(res, dict):
+                created += res.get("created_count", 0) or 0
+            if idx % 5 == 0 or idx == total - 1:
+                background_tasks.update_task(job_id, user, current=idx + 1, total=total,
+                                             message=f"Processing invoice {idx + 1}/{total}")
+
+        background_tasks.finish_task(job_id, user, "done")
+        frappe.publish_realtime(complete_event, {"created": created}, user=user)
+
+    except Exception as e:
+        frappe.log_error(f"{label} failed: {e}\n{frappe.get_traceback()}", "Entity Loading Error")
+        background_tasks.finish_task(job_id, user, "error", str(e))
+        frappe.publish_realtime(complete_event, {"error": "exception", "message": str(e)}, user=user)
+        raise

invoice_az/sales_api.py

@@ -1,6 +1,8 @@
 import frappe
 import json
 import requests
+
+from invoice_az import background_tasks
 from datetime import datetime
 import re
 
@@ -886,6 +888,7 @@ def import_bulk_sales_invoices(invoice_ids, token, warehouse=None):
         invoice_ids = json.loads(invoice_ids)
 
     user = frappe.session.user
+    job_id = background_tasks.new_job_id()
 
     frappe.enqueue(
         _process_bulk_sales_import,
@@ -893,15 +896,17 @@ def import_bulk_sales_invoices(invoice_ids, token, warehouse=None):
         token=token,
         warehouse=warehouse,
         user=user,
+        bg_job_id=job_id,
         queue="default",
         timeout=1200,
     )
 
-    return {"success": True, "enqueued": True, "total": len(invoice_ids)}
+    return {"success": True, "enqueued": True, "total": len(invoice_ids), "job_id": job_id}
 
 
-def _process_bulk_sales_import(invoice_ids, token, warehouse, user):
+def _process_bulk_sales_import(invoice_ids, token, warehouse, user, bg_job_id=None):
     """Background job: fetch details and import each sales invoice with realtime progress."""
+    job_id = bg_job_id
     frappe.set_user(user)
     record_etaxes_activity()
 
@@ -909,7 +914,19 @@ def _process_bulk_sales_import(invoice_ids, token, warehouse, user):
     imported_count = 0
     errors = []
 
+    background_tasks.register_task(job_id, user, "sales_import", "Importing sales invoices", total=total)
+
     for idx, invoice_id in enumerate(invoice_ids):
+        if job_id and background_tasks.is_cancel_requested(job_id):
+            background_tasks.finish_task(job_id, user, "cancelled")
+            frappe.publish_realtime(
+                "sales_import_complete",
+                {"total": total, "imported": imported_count, "errors": errors, "cancelled": True},
+                user=user,
+            )
+            return
+        background_tasks.update_task(job_id, user, current=idx, total=total,
+                                     message=f"Importing invoice {idx + 1}/{total}")
         try:
             details = get_sales_invoice_details(token, invoice_id)
         except Exception:
@@ -972,6 +989,7 @@ def _process_bulk_sales_import(invoice_ids, token, warehouse, user):
             user=user,
         )
 
+    background_tasks.finish_task(job_id, user, "done")
     frappe.publish_realtime(
         "sales_import_complete",
         {"total": total, "imported": imported_count, "errors": errors},

invoice_az/vat_api.py

@@ -9,6 +9,7 @@ from datetime import datetime
 from frappe.utils import now_datetime, getdate
 from invoice_az.auth import record_etaxes_activity, get_default_asan_login
 from invoice_az.utils import resolve_customer_group
+from invoice_az import background_tasks
 
 # API Endpoint
 VAT_OPERATIONS_URL = "https://new.e-taxes.gov.az/api/po/vatacc/public/v1/operation/find.outbox"
@@ -1135,6 +1136,7 @@ def import_bulk_vat_operations(operations_data, company=None, create_as_draft=0,
         auto_create_customers = 0
 
     user = frappe.session.user
+    job_id = background_tasks.new_job_id()
 
     frappe.enqueue(
         _process_bulk_vat_import,
@@ -1143,15 +1145,17 @@ def import_bulk_vat_operations(operations_data, company=None, create_as_draft=0,
         create_as_draft=create_as_draft,
         auto_create_customers=auto_create_customers,
         user=user,
+        bg_job_id=job_id,
         queue="default",
         timeout=1200,
     )
 
-    return {"success": True, "enqueued": True, "total": len(operations_data)}
+    return {"success": True, "enqueued": True, "total": len(operations_data), "job_id": job_id}
 
 
-def _process_bulk_vat_import(operations, company, create_as_draft, auto_create_customers, user):
+def _process_bulk_vat_import(operations, company, create_as_draft, auto_create_customers, user, bg_job_id=None):
     """Background job: import each VAT operation with realtime progress."""
+    job_id = bg_job_id
     frappe.set_user(user)
     record_etaxes_activity()
 
@@ -1159,7 +1163,20 @@ def _process_bulk_vat_import(operations, company, create_as_draft, auto_create_c
     imported_count = 0
     errors = []
 
+    background_tasks.register_task(job_id, user, "vat_import", "Importing VAT operations", total=total)
+
     for idx, operation in enumerate(operations):
+        if job_id and background_tasks.is_cancel_requested(job_id):
+            background_tasks.finish_task(job_id, user, "cancelled")
+            frappe.publish_realtime(
+                "vat_import_complete",
+                {"total": total, "imported": imported_count, "errors": errors,
+                 "create_as_draft": create_as_draft, "cancelled": True},
+                user=user,
+            )
+            return
+        background_tasks.update_task(job_id, user, current=idx, total=total,
+                                     message=f"Importing operation {idx + 1}/{total}")
         try:
             result = import_vat_operations(
                 operations_data=json.dumps([operation]),
@@ -1195,6 +1212,7 @@ def _process_bulk_vat_import(operations, company, create_as_draft, auto_create_c
             user=user,
         )
 
+    background_tasks.finish_task(job_id, user, "done")
     frappe.publish_realtime(
         "vat_import_complete",
         {