diff --git a/invoice_az/api.py b/invoice_az/api.py index 37c83c8..a6669d8 100644 --- a/invoice_az/api.py +++ b/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,14 +2375,24 @@ def process_single_invoice_for_units(token, invoice_id, source_type='purchase'): } @frappe.whitelist() -def process_invoice_parties_from_list(invoice_list_data, token): - """Обрабатывает контрагентов из списка инвойсов с разделением на customers и suppliers""" +def process_invoice_parties_from_list(invoice_list_data, token=None, load_customers=1, load_suppliers=1): + """Обрабатывает контрагентов прямо из СПИСКА инвойсов (без захода внутрь документа). + + Контрагенты (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: invoices = invoice_list_data - + total_customers_created = 0 total_suppliers_created = 0 total_skipped = 0 @@ -2414,9 +2425,9 @@ 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: batch_unique_suppliers[sender_name] = { 'etaxes_party_name': sender_name, @@ -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) @@ -2512,7 +2523,11 @@ def process_invoice_parties_from_list(invoice_list_data, token): total_skipped += 1 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, @@ -3771,7 +3786,46 @@ def get_reference_data_lists(data_type, limit=100, offset=0): except Exception as e: 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 diff --git a/invoice_az/sales_api.py b/invoice_az/sales_api.py index fec448a..1f0d4f4 100644 --- a/invoice_az/sales_api.py +++ b/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}, diff --git a/invoice_az/vat_api.py b/invoice_az/vat_api.py index 255b62e..f5ff8d3 100644 --- a/invoice_az/vat_api.py +++ b/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", {