diff --git a/invoice_az/amas_api.py b/invoice_az/amas_api.py index 9bb441a..3873274 100644 --- a/invoice_az/amas_api.py +++ b/invoice_az/amas_api.py @@ -15,6 +15,7 @@ import http.client import json import random import string +from concurrent.futures import ThreadPoolExecutor, as_completed from urllib.parse import unquote import frappe @@ -89,6 +90,102 @@ AMAS_GENERIC_UNAVAILABLE_MESSAGE = ( ) +def _post_amas(session_data, csrf_token, cookies, endpoint, data): + """Stateless POST to ƏMAS. No DB access. + + Used both by stateful `make_amas_request` and by parallel workers that + can't safely touch Frappe ORM. Caller manages session/csrf/cookies state. + + Returns: + On success: {"success": True, "data": , "cookies": } + On 419 / response-level CSRF: {"success": False, "csrf_error": True, ...} + On 401: {"success": False, "session_expired": True, ...} + Otherwise: {"success": False, "message": ..., "status_code": ..., "exception": ...} + """ + headers = AMAS_HEADERS.copy() + headers["X-CSRF-TOKEN"] = csrf_token or "" + headers["MAGUS-REQUEST-NUMBER"] = generate_request_number() + + for k in ("requestNumber", "_token", "currentTabToken"): + data.pop(k, None) + data["requestNumber"] = headers["MAGUS-REQUEST-NUMBER"] + data["_token"] = csrf_token or "" + + token = session_data.get("token") + if token: + data["currentTabToken"] = token + + try: + response = requests.post( + f"{AMAS_BASE_URL}{endpoint}", + data=data, + headers=headers, + cookies=cookies, + timeout=60, + ) + except requests.exceptions.RequestException as e: + return { + "success": False, + "message": AMAS_GENERIC_UNAVAILABLE_MESSAGE, + "exception": str(e), + } + + new_cookies = parse_cookies_from_response(response) + merged_cookies = {**cookies, **new_cookies} if new_cookies else cookies + + if response.status_code == 419: + return { + "success": False, + "csrf_error": True, + "message": "CSRF token rejected (HTTP 419)", + "cookies": merged_cookies, + } + + if response.status_code == 401: + return { + "success": False, + "session_expired": True, + "message": "Session expired (HTTP 401)", + "cookies": merged_cookies, + } + + if response.status_code != 200: + return { + "success": False, + "message": AMAS_GENERIC_UNAVAILABLE_MESSAGE, + "status_code": response.status_code, + "cookies": merged_cookies, + } + + try: + result = response.json() + except json.JSONDecodeError: + return { + "success": False, + "message": "Invalid response from ƏMAS", + "cookies": merged_cookies, + } + + response_info = result.get("response", {}) if isinstance(result, dict) else {} + error_code = response_info.get("code", "") + error_msg = str(response_info.get("message", "")) + + csrf_codes = ("CSRF", "TOKENS_ARE_NOT_SAME", "CSRF_TOKEN_MISMATCH") + if ( + error_code in csrf_codes + or "CSRF" in error_msg + or "TOKEN" in error_msg.upper() + ): + return { + "success": False, + "csrf_error": True, + "message": f"CSRF rejected by API: code={error_code}", + "cookies": merged_cookies, + } + + return {"success": True, "data": result, "cookies": merged_cookies} + + def humanize_amas_error(code, message): """Translate raw ƏMAS API error codes/messages into a user-friendly response dict. @@ -452,15 +549,46 @@ def refresh_csrf_token_internal(asan_login_name): return session_data, doc.amas_csrf_token -def make_amas_request(asan_login_name, endpoint, data, _retried=False): +def make_amas_request(asan_login_name, endpoint, data, _retried=False, _state=None): """Make an authenticated request to ƏMAS API. - Uses the cached CSRF token + session cookies (no upfront dashboard download). - Cookies and tokens are refreshed from each response's Set-Cookie. Only when - ƏMAS actually rejects the CSRF (HTTP 419 or response code in CSRF / - TOKENS_ARE_NOT_SAME / CSRF_TOKEN_MISMATCH) we fall back to fetching a fresh - token from /core.dashboard and retry the request once. + Two modes: + - **Stateful** (default, _state=None): reads session/cookies from the Asan Login + document, calls _post_amas, persists rotated cookies back to DB. On real + CSRF rejection (HTTP 419 or response code CSRF/TOKENS_ARE_NOT_SAME/ + CSRF_TOKEN_MISMATCH) refreshes the token from /core.dashboard and retries + the request once. + - **Stateless** (_state is a mutable dict with keys session_data, csrf_token, + cookies): used by parallel workers. Reads/updates only the in-memory state; + never touches the DB. CSRF errors are bubbled up to the caller — the caller + decides whether to abort the parallel batch. """ + if _state is not None: + session_data = _state["session_data"] + csrf_token = _state.get("csrf_token") or session_data.get("xsrf_token", "") + cookies = _state.get("cookies") or session_data.get("all_cookies", {}) + + result = _post_amas(session_data, csrf_token, cookies, endpoint, data) + + # Persist rotated cookies for the session, but DO NOT overwrite csrf_token + # from the XSRF-TOKEN cookie: that cookie holds Laravel's *encrypted* CSRF, + # while the body `_token` field expects the *plaintext* CSRF (from the + # `` tag). The plaintext CSRF is stable per session, + # so we keep the initial value and only refresh it via /core.dashboard + # (refresh_csrf_token_internal) if the API actually rejects it. + if result.get("cookies"): + _state["cookies"] = result["cookies"] + + if result.get("success"): + return {"success": True, "data": result["data"]} + + out = {"success": False, "message": result.get("message", AMAS_GENERIC_UNAVAILABLE_MESSAGE)} + if result.get("csrf_error"): + out["csrf_error"] = True + if result.get("session_expired"): + out["session_expired"] = True + return out + session_data = get_amas_session_data(asan_login_name) if not session_data: return { @@ -469,119 +597,70 @@ def make_amas_request(asan_login_name, endpoint, data, _retried=False): } doc = frappe.get_doc("Asan Login", asan_login_name) - - headers = AMAS_HEADERS.copy() csrf_token = doc.amas_csrf_token or session_data.get("xsrf_token", "") - headers["X-CSRF-TOKEN"] = csrf_token - headers["MAGUS-REQUEST-NUMBER"] = generate_request_number() - cookies = session_data.get("all_cookies", {}) - # Reset per-request fields (in case of retry, regenerate them) - for k in ("requestNumber", "_token", "currentTabToken"): - data.pop(k, None) - data["requestNumber"] = headers["MAGUS-REQUEST-NUMBER"] - data["_token"] = csrf_token + result = _post_amas(session_data, csrf_token, cookies, endpoint, data) - # The token from config is already in format "{uuid}.{accountOid}" — don't append accountOid again - token = session_data.get("token") - if token: - data["currentTabToken"] = token - - try: - url = f"{AMAS_BASE_URL}{endpoint}" - - response = requests.post( - url, - data=data, - headers=headers, - cookies=cookies, - timeout=60 - ) - - new_cookies = parse_cookies_from_response(response) - if new_cookies: - cookies.update(new_cookies) - session_data["all_cookies"] = cookies - - if "XSRF-TOKEN" in new_cookies: - decoded_token = unquote(new_cookies["XSRF-TOKEN"]) - doc.amas_csrf_token = decoded_token - session_data["xsrf_token"] = decoded_token + new_cookies = result.get("cookies") + if new_cookies and new_cookies is not cookies: + session_data["all_cookies"] = new_cookies + # NOTE: Do NOT overwrite amas_csrf_token / xsrf_token from the rotated + # XSRF-TOKEN cookie. That cookie holds Laravel's *encrypted* CSRF; the + # body `_token` field expects the *plaintext* CSRF (from the meta tag + # on /core.dashboard). The plaintext CSRF is stable per session — we + # only refresh it via refresh_csrf_token_internal() on a real CSRF + # rejection. + doc.amas_session = json.dumps(session_data) + doc.amas_last_activity = frappe.utils.now_datetime() + doc.save(ignore_permissions=True) + frappe.db.commit() + if result.get("success"): + api_data = result["data"] + if isinstance(api_data, dict) and api_data.get("token"): + session_data["token"] = api_data["token"] doc.amas_session = json.dumps(session_data) - doc.amas_last_activity = frappe.utils.now_datetime() doc.save(ignore_permissions=True) frappe.db.commit() + return {"success": True, "data": api_data} - if response.status_code == 419: - if not _retried: - refresh_csrf_token_internal(asan_login_name) - return make_amas_request(asan_login_name, endpoint, data, _retried=True) - return { - "success": False, - "csrf_error": True, - "message": "CSRF token expired. Please reconnect to ƏMAS." - } - - if response.status_code == 401: - doc.amas_auth_status = "Error" - doc.save(ignore_permissions=True) - frappe.db.commit() - return { - "success": False, - "session_expired": True, - "message": "ƏMAS session expired. Please reconnect." - } - - if response.status_code == 200: - try: - result = response.json() - - response_info = result.get("response", {}) if isinstance(result, dict) else {} - error_code = response_info.get("code", "") - error_msg = str(response_info.get("message", "")) - - csrf_codes = ("CSRF", "TOKENS_ARE_NOT_SAME", "CSRF_TOKEN_MISMATCH") - csrf_failed = ( - error_code in csrf_codes - or "CSRF" in error_msg - or "TOKEN" in error_msg.upper() - ) - if csrf_failed: - if not _retried: - refresh_csrf_token_internal(asan_login_name) - return make_amas_request(asan_login_name, endpoint, data, _retried=True) - frappe.log_error( - f"ƏMAS CSRF error after retry: code={error_code} msg={error_msg}", - "ƏMAS Token Error" - ) - return { - "success": False, - "csrf_error": True, - "message": "Token error. Please reconnect to ƏMAS." - } - - if isinstance(result, dict) and result.get("token"): - session_data["token"] = result.get("token") - doc.amas_session = json.dumps(session_data) - doc.save(ignore_permissions=True) - frappe.db.commit() - - return {"success": True, "data": result} - except json.JSONDecodeError: - return {"success": False, "message": "Invalid response from ƏMAS"} - - # Unexpected HTTP status + if result.get("csrf_error"): + if not _retried: + refresh_csrf_token_internal(asan_login_name) + return make_amas_request(asan_login_name, endpoint, data, _retried=True) frappe.log_error( - f"ƏMAS API error: status {response.status_code}\nResponse: {response.text[:500]}", + f"ƏMAS CSRF error after retry: {result.get('message')}", + "ƏMAS Token Error" + ) + return { + "success": False, + "csrf_error": True, + "message": "Token error. Please reconnect to ƏMAS." + } + + if result.get("session_expired"): + doc.amas_auth_status = "Error" + doc.save(ignore_permissions=True) + frappe.db.commit() + return { + "success": False, + "session_expired": True, + "message": "ƏMAS session expired. Please reconnect." + } + + if "status_code" in result: + frappe.log_error( + f"ƏMAS API error: status {result['status_code']}", + "ƏMAS API Error" + ) + elif result.get("exception"): + frappe.log_error( + f"ƏMAS API network error: {result['exception']}", "ƏMAS API Error" ) - return {"success": False, "message": AMAS_GENERIC_UNAVAILABLE_MESSAGE} - except requests.exceptions.RequestException as e: - frappe.log_error(f"ƏMAS API network error: {str(e)}", "ƏMAS API Error") - return {"success": False, "message": AMAS_GENERIC_UNAVAILABLE_MESSAGE} + return {"success": False, "message": result.get("message", AMAS_GENERIC_UNAVAILABLE_MESSAGE)} @frappe.whitelist() @@ -1675,7 +1754,7 @@ def create_single_employee_from_amas(asan_login_name, employee_data, company, cr @frappe.whitelist() -def create_single_employee_from_amas(asan_login_name, emp_data, company, create_designation=0): +def create_single_employee_from_amas(asan_login_name, emp_data, company, create_designation=0, prefetched_detail=None): """ Create/update a SINGLE Employee record from ƏMAS data. This is called from frontend in a loop for progress bar updates. @@ -1685,6 +1764,9 @@ def create_single_employee_from_amas(asan_login_name, emp_data, company, create_ emp_data: Single employee data dict from ƏMAS API company: Company name create_designation: 1/0 to create designation if not found + prefetched_detail: Optional dict from a prior get_employee_detail call. + When supplied, the per-employee 7-call ƏMAS fetch is skipped — used + by the parallel bulk import worker which fetches details upfront. Returns: { @@ -1724,11 +1806,12 @@ def create_single_employee_from_amas(asan_login_name, emp_data, company, create_ doc_no = emp_data.get("doc_no") doc_type = emp_data.get("doc_type", "docType_47") - # Fetch detail data for complete employee information - detail_data = None + # Fetch detail data for complete employee information (or use the + # one already fetched in parallel by the bulk-import worker). + detail_data = prefetched_detail full_name = str(emp_data.get("full_name") or "").strip() - if doc_oid: + if detail_data is None and doc_oid: detail_result = get_employee_detail(asan_login_name, doc_oid, doc_no, doc_type) if detail_result.get('success'): detail_data = detail_result.get('data') @@ -2276,7 +2359,7 @@ def create_employees_from_amas(asan_login_name, employees, company, create_desig # (called when viewing employee details in ƏMAS, not from list report) -def get_edit_form_data(asan_login_name, doc_oid, doc_no, doc_type): +def get_edit_form_data(asan_login_name, doc_oid, doc_no, doc_type, _state=None): """ Call eroom.getEditForm to get contract form data. Returns: EmasContractsEnt, EmasContractDetailsEnt, EmasContractTextDetailsEnt @@ -2287,38 +2370,38 @@ def get_edit_form_data(asan_login_name, doc_oid, doc_no, doc_type): "docNo": doc_no, "docType": doc_type } - return make_amas_request(asan_login_name, "/service/eroom.getEditForm", data) + return make_amas_request(asan_login_name, "/service/eroom.getEditForm", data, _state=_state) -def get_staff_data(asan_login_name, staff_oid): +def get_staff_data(asan_login_name, staff_oid, _state=None): """Call embas.getStaffData for position info""" data = {"staffOid": staff_oid} - return make_amas_request(asan_login_name, "/service/embas.getStaffData", data) + return make_amas_request(asan_login_name, "/service/embas.getStaffData", data, _state=_state) -def get_person_data(asan_login_name, entity_oid): +def get_person_data(asan_login_name, entity_oid, _state=None): """Call refdata.getPersonDataByEntOid for personal data""" data = {"entityOid": entity_oid} # Fixed: was "entOid", should be "entityOid" - return make_amas_request(asan_login_name, "/service/refdata.getPersonDataByEntOid", data) + return make_amas_request(asan_login_name, "/service/refdata.getPersonDataByEntOid", data, _state=_state) -def get_address_data(asan_login_name, address_oid): +def get_address_data(asan_login_name, address_oid, _state=None): """Call refdata.getAddressData for work address""" data = {"addressOid": address_oid} - return make_amas_request(asan_login_name, "/service/refdata.getAddressData", data) + return make_amas_request(asan_login_name, "/service/refdata.getAddressData", data, _state=_state) -def get_doc_main_data(asan_login_name, doc_oid, doc_no, doc_type): +def get_doc_main_data(asan_login_name, doc_oid, doc_no, doc_type, _state=None): """Call eroom.getDocMainData for document metadata""" data = { "docOid": doc_oid, "docNo": doc_no, "docType": doc_type } - return make_amas_request(asan_login_name, "/service/eroom.getDocMainData", data) + return make_amas_request(asan_login_name, "/service/eroom.getDocMainData", data, _state=_state) -def get_contract_attachments(asan_login_name, doc_oid, attach_type="attachType_5"): +def get_contract_attachments(asan_login_name, doc_oid, attach_type="attachType_5", _state=None): """ Call eroom.getAttachListByType to get list of contract attachments. Returns list of attachments with fileId for downloading. @@ -2327,23 +2410,27 @@ def get_contract_attachments(asan_login_name, doc_oid, attach_type="attachType_5 "docOid[]": doc_oid, "attachType": attach_type } - return make_amas_request(asan_login_name, "/service/eroom.getAttachListByType", data) + return make_amas_request(asan_login_name, "/service/eroom.getAttachListByType", data, _state=_state) -def download_contract_file(asan_login_name, file_id): +def download_contract_file(asan_login_name, file_id, _state=None): """ Call request/readFromStorage to download contract HTML file. Returns HTML content of the employment contract. """ data = {"fileId": file_id} - return make_amas_request(asan_login_name, "/request/readFromStorage", data) + return make_amas_request(asan_login_name, "/request/readFromStorage", data, _state=_state) @frappe.whitelist() -def get_employee_detail(asan_login_name, doc_oid, doc_no, doc_type="docType_47"): +def get_employee_detail(asan_login_name, doc_oid, doc_no, doc_type="docType_47", _state=None): """ Fetch complete employee detail from ƏMAS. - Makes 7 API calls to collect all data. + Makes up to 7 sequential API calls to collect all data. + + When `_state` is provided, runs in stateless mode (no DB access) so this + can be called from worker threads. CSRF errors are propagated to the + caller via {"success": False, "csrf_error": True}. Returns combined dict with: - EmasContractsEnt, EmasContractDetailsEnt, EmasContractTextDetailsEnt @@ -2354,7 +2441,9 @@ def get_employee_detail(asan_login_name, doc_oid, doc_no, doc_type="docType_47") try: # 1. Get contract form data - form_result = get_edit_form_data(asan_login_name, doc_oid, doc_no, doc_type) + form_result = get_edit_form_data(asan_login_name, doc_oid, doc_no, doc_type, _state=_state) + if form_result and form_result.get('csrf_error'): + return form_result if form_result and form_result.get('success'): data = form_result.get('data') if data: @@ -2366,22 +2455,24 @@ def get_employee_detail(asan_login_name, doc_oid, doc_no, doc_type="docType_47") if 'EmasContractsEnt' in combined_data and isinstance(combined_data['EmasContractsEnt'], list) and len(combined_data['EmasContractsEnt']) > 0: staff_oid = combined_data['EmasContractsEnt'][0].get('staff') if staff_oid: - staff_result = get_staff_data(asan_login_name, staff_oid) + staff_result = get_staff_data(asan_login_name, staff_oid, _state=_state) + if staff_result and staff_result.get('csrf_error'): + return staff_result if staff_result and staff_result.get('success'): data = staff_result.get('data') if data: - # staffData is on top level, not in resultData combined_data['staffData'] = data.get('staffData') # 3. Get person data (need entity_oid from contract) if 'EmasContractsEnt' in combined_data and isinstance(combined_data['EmasContractsEnt'], list) and len(combined_data['EmasContractsEnt']) > 0: entity_oid = combined_data['EmasContractsEnt'][0].get('entity') if entity_oid: - person_result = get_person_data(asan_login_name, entity_oid) + person_result = get_person_data(asan_login_name, entity_oid, _state=_state) + if person_result and person_result.get('csrf_error'): + return person_result if person_result and person_result.get('success'): data = person_result.get('data') if data: - # iamasBean and addressBean are on top level, not in resultData combined_data['iamasBean'] = data.get('iamasBean') combined_data['addressBean'] = data.get('addressBean') @@ -2389,20 +2480,21 @@ def get_employee_detail(asan_login_name, doc_oid, doc_no, doc_type="docType_47") if 'EmasContractDetailsEnt' in combined_data and isinstance(combined_data['EmasContractDetailsEnt'], list) and len(combined_data['EmasContractDetailsEnt']) > 0: address_oid = combined_data['EmasContractDetailsEnt'][0].get('address') if address_oid: - addr_result = get_address_data(asan_login_name, address_oid) + addr_result = get_address_data(asan_login_name, address_oid, _state=_state) + if addr_result and addr_result.get('csrf_error'): + return addr_result if addr_result and addr_result.get('success'): data = addr_result.get('data') if data: - # addressData is on top level, not in resultData combined_data['workAddress'] = data.get('addressData') # 5. Get document main data - doc_result = get_doc_main_data(asan_login_name, doc_oid, doc_no, doc_type) + doc_result = get_doc_main_data(asan_login_name, doc_oid, doc_no, doc_type, _state=_state) + if doc_result and doc_result.get('csrf_error'): + return doc_result if doc_result and doc_result.get('success'): data = doc_result.get('data') if data: - # Document fields are on top level, not in resultData - # Build docMainData from top-level fields combined_data['docMainData'] = { 'docOid': data.get('docOid'), 'docNo': data.get('docNo'), @@ -2413,16 +2505,19 @@ def get_employee_detail(asan_login_name, doc_oid, doc_no, doc_type="docType_47") } # 6. Get contract document (HTML) - attachments_result = get_contract_attachments(asan_login_name, doc_oid) + attachments_result = get_contract_attachments(asan_login_name, doc_oid, _state=_state) + if attachments_result and attachments_result.get('csrf_error'): + return attachments_result if attachments_result and attachments_result.get('success'): data = attachments_result.get('data') if data and 'attachList' in data: attach_list = data.get('attachList', []) if attach_list and len(attach_list) > 0: - # Get the first attachment (main contract document) file_id = attach_list[0].get('fileId') if file_id: - file_result = download_contract_file(asan_login_name, file_id) + file_result = download_contract_file(asan_login_name, file_id, _state=_state) + if file_result and file_result.get('csrf_error'): + return file_result if file_result and file_result.get('success'): file_data = file_result.get('data') if file_data: @@ -2637,7 +2732,10 @@ def on_delete_employee(doc, method): @frappe.whitelist() def import_bulk_employees(asan_login_name, employees_data, company, create_designation=0): - """Enqueue bulk employee import as a background job with realtime progress via Socket.IO.""" + """Enqueue bulk employee import as a background job with realtime progress via Socket.IO. + + Refuses to enqueue if an import for the same Asan Login is already running. + """ if isinstance(employees_data, str): employees_data = json.loads(employees_data) @@ -2646,6 +2744,22 @@ def import_bulk_employees(asan_login_name, employees_data, company, create_desig except (ValueError, TypeError): create_designation = 0 + if frappe.db.get_value("Asan Login", asan_login_name, "amas_import_running"): + return { + "success": False, + "already_running": True, + "message": ( + "An ƏMAS import is already running for this Asan Login. " + "Wait for it to finish, or click Cancel on the progress bar." + ), + } + + # Mark as running BEFORE enqueue so a fast double-click can't slip past. + # Reset cancel flag from any previous run. + frappe.db.set_value("Asan Login", asan_login_name, "amas_import_running", 1) + frappe.db.set_value("Asan Login", asan_login_name, "amas_import_cancel_requested", 0) + frappe.db.commit() + user = frappe.session.user frappe.enqueue( @@ -2662,8 +2776,98 @@ def import_bulk_employees(asan_login_name, employees_data, company, create_desig return {"success": True, "enqueued": True, "total": len(employees_data)} +@frappe.whitelist() +def get_amas_import_status(asan_login_name): + """Return whether an ƏMAS import is currently running for this Asan Login.""" + running = frappe.db.get_value("Asan Login", asan_login_name, "amas_import_running") + return {"running": bool(running)} + + +@frappe.whitelist() +def cancel_amas_import(asan_login_name): + """Request cancellation of the running ƏMAS import. + + The worker checks this flag between iterations; cancellation is best-effort + (the worker finishes its current employee before bailing out). + """ + if not frappe.db.get_value("Asan Login", asan_login_name, "amas_import_running"): + return {"success": False, "message": "No import is currently running."} + + frappe.db.set_value("Asan Login", asan_login_name, "amas_import_cancel_requested", 1) + frappe.db.commit() + return {"success": True, "message": "Cancellation requested. The import will stop shortly."} + + +BULK_IMPORT_PARALLELISM = 10 + + +def _fetch_employee_detail_for_bulk(asan_login_name, emp_data, state): + """Worker: fetch detail for one employee using stateless ƏMAS calls. + + Returns dict {emp_data, detail_data, error, csrf_error}. + """ + doc_oid = emp_data.get("doc_oid") + if not doc_oid: + return {"emp_data": emp_data, "detail_data": None, "error": None, "csrf_error": False} + + try: + result = get_employee_detail( + asan_login_name=asan_login_name, + doc_oid=doc_oid, + doc_no=emp_data.get("doc_no"), + doc_type=emp_data.get("doc_type", "docType_47"), + _state=state, + ) + if result.get("csrf_error"): + return {"emp_data": emp_data, "detail_data": None, "error": None, "csrf_error": True} + if not result.get("success"): + return { + "emp_data": emp_data, + "detail_data": None, + "error": result.get("message", "fetch failed"), + "csrf_error": False, + } + return { + "emp_data": emp_data, + "detail_data": result.get("data"), + "error": None, + "csrf_error": False, + } + except Exception as e: + return {"emp_data": emp_data, "detail_data": None, "error": str(e), "csrf_error": False} + + +def _is_cancel_requested(asan_login_name): + """Fresh-from-DB check of the cancel flag (no Frappe doc cache).""" + return bool( + frappe.db.get_value("Asan Login", asan_login_name, "amas_import_cancel_requested") + ) + + +def _clear_import_lock(asan_login_name): + """Clear running + cancel flags. Called from the worker's finally block.""" + frappe.db.set_value("Asan Login", asan_login_name, "amas_import_running", 0) + frappe.db.set_value("Asan Login", asan_login_name, "amas_import_cancel_requested", 0) + frappe.db.commit() + + def _process_bulk_employees_import(asan_login_name, employees, company, create_designation, user): - """Background job: import each employee with realtime progress.""" + """Background job: import employees with realtime progress. + + Phase A — parallel network: fetch get_employee_detail() for all employees + concurrently using BULK_IMPORT_PARALLELISM threads. Workers share an + in-memory snapshot of the ƏMAS session and never touch the DB. If any + worker hits a CSRF error (HTTP 419 or response-level CSRF code), the + whole import aborts with a hint to lower parallelism. + + Phase B — sequential DB writes: pass the prefetched detail into + create_single_employee_from_amas so it skips its own per-employee fetch. + + The Asan Login.amas_import_running flag is cleared in `finally` no matter + how the worker exits. The user can request cancellation by flipping + Asan Login.amas_import_cancel_requested; we check it between iterations + of both phases. + """ frappe.set_user(user) total = len(employees) @@ -2671,44 +2875,185 @@ def _process_bulk_employees_import(asan_login_name, employees, company, create_d updated_count = 0 errors = [] - for idx, emp_data in enumerate(employees): - emp_name = emp_data.get("full_name") or emp_data.get("identification_number") or "Employee" + try: + # Refresh the plaintext CSRF token from /core.dashboard once before going + # parallel. The DB may have a stale (or previously-corrupted) token; the + # parallel workers can't refresh on their own (no DB access), so we ensure + # the snapshot starts from a known-good plaintext value. + refreshed_session, refreshed_csrf = refresh_csrf_token_internal(asan_login_name) + if not refreshed_session: + frappe.publish_realtime( + "amas_import_complete", + { + "total": total, + "created": 0, + "updated": 0, + "errors": [{"employee": "", "error": "ƏMAS session not found. Please reconnect."}], + "aborted": True, + }, + user=user, + ) + return - try: - result = create_single_employee_from_amas( - asan_login_name=asan_login_name, - emp_data=emp_data, - company=company, - create_designation=create_designation, + asan_doc = frappe.get_doc("Asan Login", asan_login_name) + state = { + "session_data": refreshed_session, + "csrf_token": refreshed_csrf or asan_doc.amas_csrf_token or refreshed_session.get("xsrf_token", ""), + "cookies": dict(refreshed_session.get("all_cookies", {})), + } + + # Phase A: parallel detail fetches + detail_by_idx = {} + aborted_csrf = False + cancelled = False + completed = 0 + + with ThreadPoolExecutor(max_workers=BULK_IMPORT_PARALLELISM) as executor: + futures = { + executor.submit(_fetch_employee_detail_for_bulk, asan_login_name, emp, state): idx + for idx, emp in enumerate(employees) + } + + for fut in as_completed(futures): + idx = futures[fut] + try: + outcome = fut.result() + except Exception as e: + outcome = { + "emp_data": employees[idx], + "detail_data": None, + "error": str(e), + "csrf_error": False, + } + + if outcome["csrf_error"]: + aborted_csrf = True + for f in futures: + if not f.done(): + f.cancel() + break + + if _is_cancel_requested(asan_login_name): + cancelled = True + for f in futures: + if not f.done(): + f.cancel() + break + + detail_by_idx[idx] = outcome + completed += 1 + emp_name = ( + outcome["emp_data"].get("full_name") + or outcome["emp_data"].get("identification_number") + or "Employee" + ) + frappe.publish_realtime( + "amas_import_progress", + {"current": completed, "total": total, "employee_name": emp_name, "phase": "fetch"}, + user=user, + ) + + if aborted_csrf: + frappe.log_error( + "Bulk import aborted: ƏMAS rejected a CSRF token during parallel fetch. " + f"Parallelism was {BULK_IMPORT_PARALLELISM}.", + "ƏMAS Bulk Import Abort" + ) + frappe.publish_realtime( + "amas_import_complete", + { + "total": total, + "created": 0, + "updated": 0, + "errors": [{ + "employee": "", + "error": ( + f"ƏMAS rejected one of the parallel requests. The import was stopped to " + f"avoid a partial state. This usually means the parallelism level " + f"({BULK_IMPORT_PARALLELISM}) is too high — try lowering " + f"BULK_IMPORT_PARALLELISM in invoice_az/amas_api.py to 5 and retry." + ), + }], + "aborted": True, + }, + user=user, + ) + return + + if cancelled: + frappe.publish_realtime( + "amas_import_complete", + { + "total": total, + "created": 0, + "updated": 0, + "errors": [{"employee": "", "error": "Import cancelled by user."}], + "aborted": True, + "cancelled": True, + }, + user=user, + ) + return + + # Phase B: sequential DB writes (Frappe ORM is not thread-safe). + for idx in range(total): + if _is_cancel_requested(asan_login_name): + cancelled = True + break + + outcome = detail_by_idx.get(idx) + if outcome is None: + continue + emp_data = outcome["emp_data"] + emp_name = emp_data.get("full_name") or emp_data.get("identification_number") or "Employee" + + if outcome["error"]: + errors.append({"employee": emp_name, "error": outcome["error"]}) + frappe.publish_realtime( + "amas_import_progress", + {"current": idx + 1, "total": total, "employee_name": emp_name, "phase": "save"}, + user=user, + ) + continue + + try: + result = create_single_employee_from_amas( + asan_login_name=asan_login_name, + emp_data=emp_data, + company=company, + create_designation=create_designation, + prefetched_detail=outcome["detail_data"], + ) + if result and result.get("success"): + if result.get("action") == "created": + created_count += 1 + elif result.get("action") == "updated": + updated_count += 1 + else: + error_msg = result.get("message", "Unknown error") if result else "Empty result" + errors.append({"employee": emp_name, "error": error_msg}) + except Exception as e: + errors.append({"employee": emp_name, "error": str(e)}) + + frappe.publish_realtime( + "amas_import_progress", + {"current": idx + 1, "total": total, "employee_name": emp_name, "phase": "save"}, + user=user, ) - if result and result.get("success"): - if result.get("action") == "created": - created_count += 1 - elif result.get("action") == "updated": - updated_count += 1 - else: - error_msg = result.get("message", "Unknown error") if result else "Empty result" - errors.append({"employee": emp_name, "error": error_msg}) - - except Exception as e: - errors.append({"employee": emp_name, "error": str(e)}) + frappe.db.commit() frappe.publish_realtime( - "amas_import_progress", - {"current": idx + 1, "total": total, "employee_name": emp_name}, + "amas_import_complete", + { + "total": total, + "created": created_count, + "updated": updated_count, + "errors": errors, + "cancelled": cancelled, + }, user=user, ) - frappe.db.commit() - - frappe.publish_realtime( - "amas_import_complete", - { - "total": total, - "created": created_count, - "updated": updated_count, - "errors": errors, - }, - user=user, - ) + finally: + _clear_import_lock(asan_login_name) diff --git a/invoice_az/client/employee.js b/invoice_az/client/employee.js index 99e7be3..86f010d 100644 --- a/invoice_az/client/employee.js +++ b/invoice_az/client/employee.js @@ -273,17 +273,33 @@ function populate_amas_integration_tab(data) { $('#amas_integration_data').html(html); } +// Bootstrap 4.6 stacked-modal fix (twbs/bootstrap#4182): when any modal hides, +// BS4 unconditionally strips `modal-open` from , which orphans the +// backdrop of any modal still open underneath (e.g. progress + nested confirm). +// Re-apply the class if at least one modal is still visible. Idempotent: the +// flag prevents duplicate listeners if this file is re-loaded. +if (!window._amas_stacked_modal_fix_installed) { + window._amas_stacked_modal_fix_installed = true; + $(document).on('hidden.bs.modal', '.modal', function () { + if ($('.modal.show').length) { + $('body').addClass('modal-open'); + } + }); +} + // List view — add "Load from ƏMAS" button if (frappe.listview_settings['Employee']) { const orig_onload = frappe.listview_settings['Employee'].onload; frappe.listview_settings['Employee'].onload = function(listview) { if (orig_onload) orig_onload(listview); add_amas_load_button(listview); + reattach_amas_import_if_running(listview); }; } else { frappe.listview_settings['Employee'] = { onload: function(listview) { add_amas_load_button(listview); + reattach_amas_import_if_running(listview); } }; } @@ -294,6 +310,39 @@ function add_amas_load_button(listview) { }); } +// On listview mount, check if a worker is currently running an import for any +// of the user's connected Asan Logins. If yes, re-register the realtime +// listeners so the user can see progress + Cancel on the bar even after a +// page reload. +function reattach_amas_import_if_running(listview) { + frappe.call({ + method: 'invoice_az.amas_api.get_connected_asan_logins', + callback: function(r) { + if (!r.message || !r.message.success) return; + const asan_logins = r.message.asan_logins || []; + if (!asan_logins.length) return; + + // Probe each login until we find one with an active job. + const tryNext = function(idx) { + if (idx >= asan_logins.length) return; + const name = asan_logins[idx].name; + frappe.call({ + method: 'invoice_az.amas_api.get_amas_import_status', + args: { asan_login_name: name }, + callback: function(r2) { + if (r2.message && r2.message.running) { + attach_amas_import_listeners(listview, name, 0); + } else { + tryNext(idx + 1); + } + } + }); + }; + tryNext(0); + } + }); +} + // Step 1: Get default connected Asan Login → fetch employees from ƏMAS API function start_amas_employee_import(listview) { frappe.call({ @@ -597,21 +646,82 @@ function show_employee_selection(listview, asan_login_name, organization_name, e } // Step 5: Create employees via backend - Socket.IO bulk import -function create_employees(listview, asan_login_name, selected_employees, company, create_designation) { - const total = selected_employees.length; +// Wire realtime listeners for an in-flight ƏMAS import. Used both when the +// user starts a new import and when reattaching to an already-running one +// after a page reload. +function attach_amas_import_listeners(listview, asan_login_name, total_hint) { + const total = total_hint || 0; - frappe.show_progress( - __('Loading from ƏMAS'), 0, total, - __('Starting import of {0} employees...', [total]) + // Reused single confirm dialog. frappe.confirm creates a fresh dialog on + // each call and never removes the old element from DOM, which causes + // Bootstrap's modal-backdrop stack to glitch on repeat use. + let cancel_confirm_dialog = null; + + const ensure_cancel_button = function(dialog) { + if (!dialog || dialog._amas_cancel_attached) return; + dialog._amas_cancel_attached = true; + dialog.set_primary_action(__('Cancel Import'), function() { + if (!cancel_confirm_dialog) { + cancel_confirm_dialog = new frappe.ui.Dialog({ + title: __('Confirm'), + primary_action_label: __('Yes, cancel'), + primary_action: function() { + cancel_confirm_dialog.hide(); + frappe.call({ + method: 'invoice_az.amas_api.cancel_amas_import', + args: { asan_login_name: asan_login_name }, + callback: function() { + frappe.show_alert({ + message: __('Cancellation requested. Finishing current employee...'), + indicator: 'orange' + }, 5); + } + }); + }, + secondary_action_label: __('No, keep importing'), + secondary_action: function() { + cancel_confirm_dialog.hide(); + } + }); + cancel_confirm_dialog.$body.append( + '

' + __('Cancel the running ƏMAS import? Already-saved employees will stay; the rest will be skipped.') + '

' + ); + } + cancel_confirm_dialog.show(); + // Progress dialog forces z-index 2000; lift the confirm above it. + cancel_confirm_dialog.$wrapper.css('z-index', 2100); + }); + }; + + // Hand the reusable confirm dialog over to the import-complete handler so + // it can fully remove the leftover element from DOM when the import ends. + const cleanup_cancel_dialog = function() { + if (cancel_confirm_dialog) { + cancel_confirm_dialog.hide(); + cancel_confirm_dialog.$wrapper.remove(); + cancel_confirm_dialog = null; + } + }; + + let dialog = frappe.show_progress( + __('Importing from ƏMAS'), 0, total, + total + ? __('Starting import of {0} employees...', [total]) + : __('Reconnecting to import in progress... waiting for next update') ); + ensure_cancel_button(dialog); frappe.realtime.off('amas_import_progress'); frappe.realtime.on('amas_import_progress', function(data) { - let detail = __('Processing employee {0} of {1}', [data.current, data.total]); - if (data.employee_name) { - detail = __('Processing: ') + data.employee_name; - } - frappe.show_progress(__('Loading from ƏMAS'), data.current, data.total, detail); + const is_save = data.phase === 'save'; + const verb = is_save ? __('Saving: ') : __('Fetching: '); + const detail = data.employee_name + ? verb + data.employee_name + : verb + __('{0} of {1}', [data.current, data.total]); + const d = frappe.show_progress( + __('Importing from ƏMAS'), data.current, data.total, detail + ); + ensure_cancel_button(d); }); frappe.realtime.off('amas_import_complete'); @@ -619,6 +729,32 @@ function create_employees(listview, asan_login_name, selected_employees, company frappe.realtime.off('amas_import_progress'); frappe.realtime.off('amas_import_complete'); frappe.hide_progress(); + cleanup_cancel_dialog(); + + if (data.cancelled) { + frappe.msgprint({ + title: __('Import Cancelled'), + indicator: 'orange', + message: __( + 'Cancelled. Created: {0}, updated: {1}.', + [data.created || 0, data.updated || 0] + ) + }); + listview.refresh(); + return; + } + + if (data.aborted) { + const reason = (data.errors && data.errors[0] && data.errors[0].error) + || __('Import was aborted.'); + frappe.msgprint({ + title: __('Import Aborted'), + indicator: 'red', + message: reason + }); + listview.refresh(); + return; + } const error_messages = (data.errors || []).map(function(err) { return (err.employee || 'Unknown') + ': ' + (err.error || 'Unknown error'); @@ -629,6 +765,12 @@ function create_employees(listview, asan_login_name, selected_employees, company listview.refresh(); }); +} + +function create_employees(listview, asan_login_name, selected_employees, company, create_designation) { + const total = selected_employees.length; + + attach_amas_import_listeners(listview, asan_login_name, total); frappe.call({ method: 'invoice_az.amas_api.import_bulk_employees', @@ -639,6 +781,17 @@ function create_employees(listview, asan_login_name, selected_employees, company create_designation: create_designation ? 1 : 0 }, callback: function(r) { + if (r.message && r.message.already_running) { + frappe.realtime.off('amas_import_progress'); + frappe.realtime.off('amas_import_complete'); + frappe.hide_progress(); + frappe.msgprint({ + title: __('Import Already Running'), + indicator: 'orange', + message: r.message.message + }); + return; + } if (!r.message || !r.message.enqueued) { frappe.realtime.off('amas_import_progress'); frappe.realtime.off('amas_import_complete'); diff --git a/invoice_az/invoice_az/doctype/asan_login/asan_login.json b/invoice_az/invoice_az/doctype/asan_login/asan_login.json index 1adc8bb..12a2cb6 100644 --- a/invoice_az/invoice_az/doctype/asan_login/asan_login.json +++ b/invoice_az/invoice_az/doctype/asan_login/asan_login.json @@ -33,7 +33,9 @@ "amas_account_oid", "amas_last_activity", "amas_session", - "amas_csrf_token" + "amas_csrf_token", + "amas_import_running", + "amas_import_cancel_requested" ], "fields": [ { @@ -205,6 +207,20 @@ "fieldtype": "Small Text", "hidden": 1, "label": "\u018fMAS CSRF Token" + }, + { + "default": "0", + "fieldname": "amas_import_running", + "fieldtype": "Check", + "hidden": 1, + "label": "\u018fMAS Import Running" + }, + { + "default": "0", + "fieldname": "amas_import_cancel_requested", + "fieldtype": "Check", + "hidden": 1, + "label": "\u018fMAS Import Cancel Requested" } ], "index_web_pages_for_search": 1,