""" ƏMAS (e-social.gov.az) Integration API This module provides functions to interact with ƏMAS (Employment Management System) of Azərbaycan's Ministry of Labour and Social Protection. ƏMAS uses MyGovID/ASAN authentication. The flow is: 1. Get MyGovID JWT token via ASAN Sign (same as existing MyGovID Login) 2. Use JWT token to get authorization code for ƏMAS via MyGovID OAuth API 3. Exchange authorization code for ƏMAS session via SSO callback 4. Use session cookies for ƏMAS API calls """ import http.client import json import random import string from concurrent.futures import ThreadPoolExecutor, as_completed from urllib.parse import unquote import frappe import requests # Increase header limit - AMAS returns many Set-Cookie headers http.client._MAXHEADERS = 1000 # AMAS Base URL AMAS_BASE_URL = "https://eroom.e-social.gov.az" # AMAS API Endpoints AMAS_SSO_CALLBACK = "/ssoAsan/v2" AMAS_CHANGE_ACCOUNT = "/service/aas.changeAccount" AMAS_EXECUTE_REPORT = "/service/magusbi.executeReport" AMAS_GET_DOMAINS = "/service/aas.domains" AMAS_GET_ACCOUNTS = "/request/accounts" # MyGovID OAuth API MYGOVID_API_URL = "https://api.mygovid.gov.az" MYGOVID_AUTH_CODES_URL = f"{MYGOVID_API_URL}/ssoauth/oauth2/auth/codes" # AMAS OAuth Client ID (from HAR analysis) AMAS_CLIENT_ID = "64942e33ec8d49059333a1e0ebad7fe2" AMAS_REDIRECT_URI = f"{AMAS_BASE_URL}{AMAS_SSO_CALLBACK}" # AMAS request headers AMAS_HEADERS = { "Accept": "text/plain, */*; q=0.01", "Accept-Language": "en-US,en;q=0.9", "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", "X-Requested-With": "XMLHttpRequest", "Origin": AMAS_BASE_URL, "Referer": f"{AMAS_BASE_URL}/", "X-MAGUS-RECAPTCHA": "null", } MYGOVID_HEADERS = { "Accept": "application/json, text/plain, */*", "Accept-Language": "az", "Content-Type": "application/json", "Origin": "https://mygovid.gov.az", "Referer": "https://mygovid.gov.az/", } def generate_request_number(): """Generate a unique request number for AMAS API calls""" # Format: 1XXXXXXXXXXXXXXX (alphanumeric) chars = string.ascii_uppercase + string.digits return "1" + "".join(random.choices(chars, k=13)) def generate_state(): """Generate a unique state for SSO flow""" # Format: XXXXXXXXXXXXX:eroom:emas chars = string.hexdigits.lower()[:16] return "".join(random.choices(chars, k=13)) + ":eroom:emas" def parse_cookies_from_response(response): """Extract cookies from response and return as dict""" cookies = {} for cookie in response.cookies: cookies[cookie.name] = cookie.value return cookies AMAS_GENERIC_UNAVAILABLE_MESSAGE = ( "ƏMAS service is currently unavailable. Please try again in a few minutes." ) 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. ƏMAS returns codes like 'PERMISSION_ERROR' which are useless for end users. Returns a dict that callers can spread into their failure response — it always contains 'message', and sets 'permission_error': True for the permission case so the frontend can offer a Reconnect action. """ code_str = (str(code) if code else "").strip().lower() msg_str = (str(message) if message else "").strip().lower() if "permission" in code_str or "permission" in msg_str: return { "message": ( "The selected ƏMAS certificate doesn't have permission for this " "operation. You can reconnect to ƏMAS and choose a different " "organization (certificate)." ), "permission_error": True, } return {"message": AMAS_GENERIC_UNAVAILABLE_MESSAGE} @frappe.whitelist() def connect_amas(asan_login_name): """ Connect to AMAS using MyGovID JWT token. This is the main entry point for AMAS connection. If MyGovID token exists - uses it to get AMAS session automatically. If not - returns error asking user to login with MyGovID first. """ try: doc = frappe.get_doc("Asan Login", asan_login_name) # Check if we have MyGovID token if not doc.mygovid_token: return { "success": False, "need_mygovid_login": True, "message": "MyGovID token not found. Please login with MyGovID first." } # Step 1: Get authorization code from MyGovID auth_code_result = get_amas_auth_code(doc.mygovid_token) if not auth_code_result.get("success"): # Token might be expired if "401" in str(auth_code_result.get("message", "")): return { "success": False, "need_mygovid_login": True, "message": "MyGovID token expired. Please login with MyGovID again." } return auth_code_result auth_code = auth_code_result.get("code") state = auth_code_result.get("state") # Step 2: Exchange authorization code for EMAS session session_result = exchange_code_for_session(auth_code, state) if not session_result.get("success"): return session_result # Step 3: Save session data (but don't save account yet - user must select) session_data = session_result.get("session_data") doc.amas_session = json.dumps(session_data) doc.amas_csrf_token = session_data.get("xsrf_token") doc.amas_auth_status = "Connected" doc.amas_last_activity = frappe.utils.now_datetime() # Clear any previous account selection - user must choose doc.amas_account_oid = None doc.amas_account_name = None doc.amas_account_number = None doc.save(ignore_permissions=True) frappe.db.commit() # Step 4: Fetch available accounts from /request/accounts accounts_result = get_amas_accounts(asan_login_name) accounts = accounts_result.get("accounts", []) if accounts_result.get("success") else [] return { "success": True, "message": "ƏMAS connected! Please select an organization.", "accounts": accounts, "has_accounts": len(accounts) > 0, "need_account_selection": True } except Exception as e: frappe.log_error( f"EMAS connect error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} def get_amas_auth_code(mygovid_token): """ Get EMAS authorization code using MyGovID JWT token. This calls the MyGovID OAuth API to get an authorization code that can be exchanged for EMAS session. """ try: state = generate_state() payload = { "client_id": AMAS_CLIENT_ID, "state": state, "response_type": "code", "redirect_uri": AMAS_REDIRECT_URI } headers = MYGOVID_HEADERS.copy() headers["Authorization"] = mygovid_token response = requests.post( MYGOVID_AUTH_CODES_URL, json=payload, headers=headers, timeout=30 ) if response.status_code == 201: data = response.json() return { "success": True, "code": data.get("code"), "state": data.get("state", state) } elif response.status_code == 401: return { "success": False, "message": "MyGovID token expired (401)" } else: return { "success": False, "message": f"Failed to get auth code: status {response.status_code}" } except requests.exceptions.RequestException as e: return {"success": False, "message": f"Network error: {str(e)}"} except Exception as e: return {"success": False, "message": f"Error: {str(e)}"} def parse_config_from_html(html_text): """ Parse the 'var config = {...}' JavaScript object from EMAS HTML page. Returns a dict with account info. """ import re config = {} # Extract the var config = {...} block # Pattern matches: var config = { ... }; config_match = re.search(r'var\s+config\s*=\s*\{([^}]+(?:\{[^}]*\}[^}]*)*)\}', html_text) if not config_match: return config config_text = config_match.group(1) # Parse individual fields using pattern: 'key' : 'value' or 'key' : value # Fields we want to extract fields = [ 'token', 'accountOid', 'accountName', 'accountNumber', 'firstName', 'lastName', 'middleName', 'userId', 'userLevel', 'userBranchId', 'userBranchName', 'userPosId', 'customerOid', 'domain', 'loginType', 'entityType' ] for field in fields: # Try 'field' : 'value' pattern (string value) pattern = rf"'{field}'\s*:\s*'([^']*)'" match = re.search(pattern, config_text) if match: config[field] = match.group(1) continue # Try 'field' : value pattern (non-string value) pattern = rf"'{field}'\s*:\s*([^,\n]+)" match = re.search(pattern, config_text) if match: value = match.group(1).strip().strip("'\"") config[field] = value return config def exchange_code_for_session(code, state): """ Exchange authorization code for EMAS session. Makes request to EMAS SSO callback endpoint and captures session cookies. Also parses the account info from the HTML response config. """ try: # Build SSO callback URL callback_url = f"{AMAS_REDIRECT_URI}?code={code}&state={state}" # Create session to capture cookies session = requests.Session() response = session.get( callback_url, allow_redirects=True, timeout=30 ) # Extract cookies cookies = {} for cookie in session.cookies: cookies[cookie.name] = cookie.value # Look for required cookies session_cookie = cookies.get("emek_ve_mesgulluq_alt_sistemi_session") xsrf_token_raw = cookies.get("XSRF-TOKEN") # Also check for _s cookie (Laravel session with data) s_cookie = cookies.get("emek_ve_mesgulluq_alt_sistemi_s") if not session_cookie: return { "success": False, "message": "Failed to get ƏMAS session cookie" } # IMPORTANT: Laravel's XSRF-TOKEN cookie is URL-encoded, must decode it! # The cookie value contains %3D instead of = etc. xsrf_token = unquote(xsrf_token_raw) if xsrf_token_raw else None # Try to extract CSRF token from HTML meta tag (preferred - already decoded) import re meta_csrf_token = None match = re.search(r'name="csrf-token" content="([^"]+)"', response.text) if match: meta_csrf_token = match.group(1) # Prefer meta tag token over cookie (meta tag is already decoded) csrf_token = meta_csrf_token or xsrf_token # Parse token and account info from HTML config html_config = parse_config_from_html(response.text) token = html_config.get("token") # Extract account info from the config account_info = None if html_config.get("accountOid"): account_info = { "accountOid": html_config.get("accountOid"), "accountName": html_config.get("accountName", ""), "accountNumber": html_config.get("accountNumber", ""), "userId": html_config.get("userId", ""), "userLevel": html_config.get("userLevel", "owner"), "branchId": html_config.get("userBranchId", "0000"), "positionId": html_config.get("userPosId", "0000"), } session_data = { "session_cookie": session_cookie, "xsrf_token": csrf_token, "s_cookie": s_cookie, "token": token, "all_cookies": cookies, "account_info": account_info, "html_config": html_config } return { "success": True, "session_data": session_data } except requests.exceptions.RequestException as e: return {"success": False, "message": f"Network error: {str(e)}"} except Exception as e: return {"success": False, "message": f"Error: {str(e)}"} def get_amas_session_data(asan_login_name): """Get EMAS session data from document""" doc = frappe.get_doc("Asan Login", asan_login_name) if not doc.amas_session: return None try: return json.loads(doc.amas_session) except (json.JSONDecodeError, TypeError): return None def refresh_csrf_token_internal(asan_login_name): """ Refresh CSRF token by making a GET request to EMAS dashboard. EMAS requires fresh CSRF token for EVERY operation. Returns updated session_data and csrf_token. """ import re session_data = get_amas_session_data(asan_login_name) if not session_data: return None, None doc = frappe.get_doc("Asan Login", asan_login_name) cookies = session_data.get("all_cookies", {}) try: # Make GET request to dashboard to get fresh CSRF token response = requests.get( f"{AMAS_BASE_URL}/core.dashboard", cookies=cookies, timeout=30 ) if response.status_code != 200: return session_data, doc.amas_csrf_token # Update cookies from response for cookie in response.cookies: cookies[cookie.name] = cookie.value session_data["all_cookies"] = cookies # Extract CSRF token from HTML meta tag (preferred) csrf_token = None match = re.search(r'name="csrf-token" content="([^"]+)"', response.text) if match: csrf_token = match.group(1) # Also try from cookie if not csrf_token and "XSRF-TOKEN" in cookies: csrf_token = unquote(cookies["XSRF-TOKEN"]) if csrf_token: doc.amas_csrf_token = csrf_token session_data["xsrf_token"] = csrf_token # Update token from page config if available html_config = parse_config_from_html(response.text) if html_config.get("token"): session_data["token"] = html_config.get("token") # Save updated session doc.amas_session = json.dumps(session_data) doc.amas_last_activity = frappe.utils.now_datetime() doc.save(ignore_permissions=True) frappe.db.commit() return session_data, csrf_token except Exception as e: frappe.log_error(f"CSRF refresh error: {str(e)}", "ƏMAS API Error") return session_data, doc.amas_csrf_token def make_amas_request(asan_login_name, endpoint, data, _retried=False, _state=None): """Make an authenticated request to ƏMAS API. 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 { "success": False, "message": "ƏMAS session not found. Please connect to ƏMAS first." } doc = frappe.get_doc("Asan Login", asan_login_name) csrf_token = doc.amas_csrf_token or session_data.get("xsrf_token", "") cookies = session_data.get("all_cookies", {}) result = _post_amas(session_data, csrf_token, cookies, endpoint, data) 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.save(ignore_permissions=True) frappe.db.commit() return {"success": True, "data": api_data} 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 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": result.get("message", AMAS_GENERIC_UNAVAILABLE_MESSAGE)} @frappe.whitelist() def get_amas_accounts(asan_login_name): """ Get list of organizations/accounts user has access to in EMAS. Fetches from /request/accounts endpoint. """ try: session_data = get_amas_session_data(asan_login_name) if not session_data: return {"success": False, "accounts": [], "message": "No ƏMAS session"} 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() # Build cookies (updated by refresh) cookies = session_data.get("all_cookies", {}) # Build request data with fresh token data = { "requestNumber": headers["MAGUS-REQUEST-NUMBER"], "_token": csrf_token, } # Add currentTabToken token = session_data.get("token") if token: data["currentTabToken"] = token # Make request to /request/accounts response = requests.post( f"{AMAS_BASE_URL}{AMAS_GET_ACCOUNTS}", data=data, headers=headers, cookies=cookies, timeout=30 ) # Update cookies from response new_cookies = parse_cookies_from_response(response) if new_cookies: cookies.update(new_cookies) session_data["all_cookies"] = cookies # Update CSRF token if changed 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 doc.amas_session = json.dumps(session_data) doc.save(ignore_permissions=True) frappe.db.commit() if response.status_code == 200: try: accounts = response.json() # The response is a direct array of accounts if isinstance(accounts, list): return {"success": True, "accounts": accounts} else: return {"success": False, "accounts": [], "message": "Unexpected response format"} except json.JSONDecodeError: return {"success": False, "accounts": [], "message": "Invalid JSON response"} else: return {"success": False, "accounts": [], "message": f"HTTP {response.status_code}"} except Exception as e: frappe.log_error( f"EMAS get accounts error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}", "accounts": []} @frappe.whitelist() def change_amas_account(asan_login_name, account_oid, account_name, account_number, role_name="chairman"): """ Select/switch to a specific organization account in EMAS. """ try: data = { "accountOid": account_oid, "accountName": account_name, "accountNumber": account_number, "roleName": role_name, "branchId": "0000", "positionId": "0000", "domain": "emas" } result = make_amas_request( asan_login_name, AMAS_CHANGE_ACCOUNT, data ) if not result.get("success"): return result response_data = result.get("data", {}) # Check response code response_info = response_data.get("response", {}) if response_info.get("code") != "0": raw_code = response_info.get("code") raw_msg = response_info.get("message") frappe.log_error( f"changeAccount failed: code={raw_code}, message={raw_msg}", "ƏMAS Account Switch Error" ) return {"success": False, **humanize_amas_error(raw_code, raw_msg)} # Store selected account info doc = frappe.get_doc("Asan Login", asan_login_name) doc.amas_account_oid = account_oid doc.amas_account_name = account_name doc.amas_account_number = account_number doc.amas_auth_status = "Connected" doc.amas_last_activity = frappe.utils.now_datetime() # Update session with token from response token = response_data.get("token") if token: session_data = get_amas_session_data(asan_login_name) if session_data: session_data["token"] = token doc.amas_session = json.dumps(session_data) doc.save(ignore_permissions=True) frappe.db.commit() return { "success": True, "message": f"Switched to account: {account_name}" } except Exception as e: frappe.log_error( f"EMAS change account error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} @frappe.whitelist() def get_employees_report(asan_login_name, offset=0, limit=100): """ Get list of employees from EMAS. Uses the emasValidEmploymentContractOnline report. """ try: doc = frappe.get_doc("Asan Login", asan_login_name) if not doc.amas_account_oid: return { "success": False, "message": "No ƏMAS account selected. Please select an organization first." } data = { "reportId": "emasValidEmploymentContractOnline", "pagination[offset]": str(offset), "pagination[limit]": str(limit), "attributes[newExcel]": "true" } result = make_amas_request(asan_login_name, AMAS_EXECUTE_REPORT, data) if not result.get("success"): return result response_data = result.get("data", {}) # Check response code response_info = response_data.get("response", {}) if response_info.get("code") != "0": raw_code = response_info.get("code") raw_msg = response_info.get("message") frappe.log_error( f"getEmployeesReport failed: code={raw_code}, message={raw_msg}", "ƏMAS Employees Report Error" ) return {"success": False, **humanize_amas_error(raw_code, raw_msg)} # Extract employees from result employees = response_data.get("result", []) has_more = response_data.get("hasMore", False) return { "success": True, "employees": employees, "has_more": has_more, "offset": offset, "limit": limit } except Exception as e: frappe.log_error( f"EMAS employees report error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} @frappe.whitelist() def get_contract_stats(asan_login_name): """ Get employment contract statistics from EMAS. """ try: doc = frappe.get_doc("Asan Login", asan_login_name) if not doc.amas_account_oid: return { "success": False, "message": "No ƏMAS account selected. Please select an organization first." } data = { "reportId": "graphContractStatsEmployer", "pagination[offset]": "0", "pagination[limit]": "100" } result = make_amas_request(asan_login_name, AMAS_EXECUTE_REPORT, data) if not result.get("success"): return result response_data = result.get("data", {}) response_info = response_data.get("response", {}) if response_info.get("code") != "0": raw_code = response_info.get("code") raw_msg = response_info.get("message") frappe.log_error( f"getContractStats failed: code={raw_code}, message={raw_msg}", "ƏMAS Contract Stats Error" ) return {"success": False, **humanize_amas_error(raw_code, raw_msg)} stats = response_data.get("result", []) return { "success": True, "stats": stats[0] if stats else {} } except Exception as e: frappe.log_error( f"EMAS contract stats error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} @frappe.whitelist() def check_amas_connection(asan_login_name): """ Check if EMAS connection is active. """ try: doc = frappe.get_doc("Asan Login", asan_login_name) return { "success": True, "amas_auth_status": doc.amas_auth_status or "Not Connected", "amas_account_name": doc.amas_account_name, "amas_account_number": doc.amas_account_number, "amas_last_activity": str(doc.amas_last_activity) if doc.amas_last_activity else None, "has_session": bool(doc.amas_session), "has_mygovid_token": bool(doc.mygovid_token) } except Exception as e: frappe.log_error( f"EMAS check connection error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} @frappe.whitelist() def refresh_amas_session(asan_login_name): """ Refresh EMAS session by making a GET request to get fresh CSRF token. Use this when you get CSRF_TOKEN_MISMATCH errors. """ try: doc = frappe.get_doc("Asan Login", asan_login_name) session_data = get_amas_session_data(asan_login_name) if not session_data: return { "success": False, "message": "No ƏMAS session found. Please connect first." } cookies = session_data.get("all_cookies", {}) # Make GET request to EMAS dashboard to get fresh CSRF token response = requests.get( f"{AMAS_BASE_URL}/core.dashboard", cookies=cookies, timeout=30 ) if response.status_code != 200: return { "success": False, "message": f"Failed to refresh session: status {response.status_code}" } # Update cookies for cookie in response.cookies: cookies[cookie.name] = cookie.value session_data["all_cookies"] = cookies # Extract new CSRF token from HTML import re match = re.search(r'name="csrf-token" content="([^"]+)"', response.text) if match: new_csrf_token = match.group(1) doc.amas_csrf_token = new_csrf_token session_data["xsrf_token"] = new_csrf_token # Also update from cookie if available if "XSRF-TOKEN" in cookies: decoded_token = unquote(cookies["XSRF-TOKEN"]) # Prefer meta tag token, but update session data if not match: doc.amas_csrf_token = decoded_token session_data["xsrf_token"] = decoded_token # Parse config for updated token html_config = parse_config_from_html(response.text) if html_config.get("token"): session_data["token"] = html_config.get("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, "message": "ƏMAS session refreshed successfully" } except Exception as e: frappe.log_error( f"EMAS refresh session error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} @frappe.whitelist() def disconnect_amas(asan_login_name): """ Disconnect from EMAS by clearing session data. """ try: doc = frappe.get_doc("Asan Login", asan_login_name) doc.amas_session = None doc.amas_csrf_token = None doc.amas_account_oid = None doc.amas_account_name = None doc.amas_account_number = None doc.amas_auth_status = "Not Connected" doc.amas_last_activity = None doc.save(ignore_permissions=True) frappe.db.commit() return { "success": True, "message": "Disconnected from ƏMAS" } except Exception as e: frappe.log_error( f"EMAS disconnect error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} @frappe.whitelist() def get_connected_asan_logins(): """ Get list of Asan Logins with active ƏMAS connection. Prioritizes is_default = 1, similar to e-taxes. """ try: # First try to get default Asan Login with ƏMAS connection default_login = frappe.get_all( "Asan Login", filters={ "is_default": 1, "amas_auth_status": "Connected", "amas_account_oid": ["is", "set"] }, fields=["name", "amas_account_name", "amas_account_number", "amas_account_oid"], limit=1 ) if default_login: # Use default login asan_logins = default_login else: # Fall back to any connected login asan_logins = frappe.get_all( "Asan Login", filters={ "amas_auth_status": "Connected", "amas_account_oid": ["is", "set"] }, fields=["name", "amas_account_name", "amas_account_number", "amas_account_oid"], limit=1 ) result = [] for al in asan_logins: result.append({ "name": al.name, "organization_name": al.amas_account_name or "Unknown", "organization_voen": al.amas_account_number or "" }) return { "success": True, "asan_logins": result } except Exception as e: frappe.log_error( f"Get connected Asan Logins error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}", "asan_logins": []} # Exact list of API fields that map 1:1 to Əmas Employees DocType fields AMAS_EMPLOYEE_FIELDS = [ "full_name", "identification_number", "ssn", "birthday", "work_position_text", "monthly_salary", "doc_no", "doc_date", "doc_type", "contract_type", "contract_status", "begin_date", "end_date", "main_contract_begin_date", "emp_job_start_date", "sign_type", "is_valid", "dur_state", "operation", "after_exec_label", "workplace_type", "form_size", "flow", "flow_alias", "tree_path_name", "view_type", "doc_oid", ] @frappe.whitelist() def import_employees(asan_login_name, employees): """ Import employees from ƏMAS to Əmas Employees doctype. NOW FETCHES COMPLETE DETAIL DATA for each employee. Args: asan_login_name: Name of the Asan Login document employees: List of employee data from ƏMAS API (list report) """ try: if isinstance(employees, str): employees = json.loads(employees) doc = frappe.get_doc("Asan Login", asan_login_name) organization_name = doc.amas_account_name organization_voen = doc.amas_account_number imported_count = 0 updated_count = 0 error_count = 0 errors = [] for emp_data in employees: try: doc_oid = emp_data.get('doc_oid') doc_no = emp_data.get('doc_no') doc_type = emp_data.get('doc_type', 'docType_47') if not doc_oid: # Can't fetch details without doc_oid, skip name_val = emp_data.get('full_name', '?') errors.append(f"Employee {name_val}: No doc_oid") error_count += 1 continue # FETCH COMPLETE DETAIL DATA detail_result = get_employee_detail(asan_login_name, doc_oid, doc_no, doc_type) if not detail_result.get('success'): # Detail fetch failed, use list data as fallback name_val = emp_data.get('full_name', '?') frappe.log_error( f"Failed to fetch detail for {doc_oid}: {detail_result.get('message')}", "EMAS Employee Detail Fetch Error" ) errors.append(f"{name_val}: Detail fetch failed, using list data") # Fallback to list data fin = emp_data.get("identification_number") if not fin: error_count += 1 continue fin = str(fin).strip() existing = frappe.db.exists( "Amas Employees", {"identification_number": fin, "organization_voen": organization_voen} ) if existing: employee_doc = frappe.get_doc("Amas Employees", existing) update_employee_from_data(employee_doc, emp_data) employee_doc.save(ignore_permissions=True) updated_count += 1 else: employee_doc = frappe.new_doc("Amas Employees") employee_doc.asan_login = asan_login_name employee_doc.organization_name = organization_name employee_doc.organization_voen = organization_voen update_employee_from_data(employee_doc, emp_data) employee_doc.insert(ignore_permissions=True) imported_count += 1 continue detail_data = detail_result.get('data') # Get FIN from detail data (more reliable than list) fin = None if detail_data and 'EmasContractsEnt' in detail_data and isinstance(detail_data.get('EmasContractsEnt'), list) and len(detail_data['EmasContractsEnt']) > 0: fin = detail_data['EmasContractsEnt'][0].get('pin') if not fin: name_val = emp_data.get('full_name', '?') errors.append(f"{name_val}: No FIN in detail data") error_count += 1 continue fin = str(fin).strip() # Check if exists existing = frappe.db.exists( "Amas Employees", { "identification_number": fin, "organization_voen": organization_voen } ) if existing: employee_doc = frappe.get_doc("Amas Employees", existing) # First populate basic fields from list data update_employee_from_data(employee_doc, emp_data) # Then overlay with detailed data update_employee_from_detail_data(employee_doc, detail_data) # Ensure FIN is set (fallback to verified FIN) if not employee_doc.identification_number: employee_doc.identification_number = fin employee_doc.save(ignore_permissions=True) updated_count += 1 else: employee_doc = frappe.new_doc("Amas Employees") employee_doc.asan_login = asan_login_name employee_doc.organization_name = organization_name employee_doc.organization_voen = organization_voen # Set FIN first (required field) employee_doc.identification_number = fin # First populate basic fields from list data update_employee_from_data(employee_doc, emp_data) # Then overlay with detailed data update_employee_from_detail_data(employee_doc, detail_data) # Ensure FIN is still set after mapping if not employee_doc.identification_number: employee_doc.identification_number = fin employee_doc.insert(ignore_permissions=True) imported_count += 1 except Exception as e: name_val = emp_data.get('full_name', '?') error_msg = f"{name_val}: {str(e)}" frappe.log_error( f"{error_msg}\n{frappe.get_traceback()}", "EMAS Employee Import Error" ) errors.append(error_msg) error_count += 1 frappe.db.commit() return { "success": True, "imported_count": imported_count, "updated_count": updated_count, "error_count": error_count, "errors": errors[:10], "total": len(employees) } except Exception as e: frappe.log_error( f"Import employees error: {str(e)}\n{frappe.get_traceback()}", "EMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} def update_employee_from_data(employee_doc, emp_data): """ Update employee document fields from ƏMAS API data. Hard 1:1 mapping - API field names match DocType field names exactly. """ # Map all API fields directly to DocType fields for field in AMAS_EMPLOYEE_FIELDS: value = emp_data.get(field) if value is not None: setattr(employee_doc, field, str(value)) else: setattr(employee_doc, field, None) # Store raw data for reference employee_doc.raw_data = json.dumps(emp_data, ensure_ascii=False, indent=2) # Import timestamp employee_doc.import_date = frappe.utils.now_datetime() def parse_amas_date(date_str): """ Parse date string from ƏMAS into YYYY-MM-DD format for Frappe Date fields. ƏMAS dates can be in various formats. """ if not date_str or not str(date_str).strip(): return None date_str = str(date_str).strip() from datetime import datetime for fmt in ("%d.%m.%Y", "%Y-%m-%d", "%d/%m/%Y", "%Y-%m-%dT%H:%M:%S"): try: return datetime.strptime(date_str, fmt).strftime("%Y-%m-%d") except (ValueError, TypeError): continue return None def map_gender(amas_gender): """ Transform AMAS gender to ERPNext gender. AMAS values: "Kişi" (Male), "Qadın" (Female), "Male", "Female" ERPNext values: "Male", "Female", "Other" """ if not amas_gender: return "Male" # Default fallback gender_str = str(amas_gender).strip().lower() # Map Azərbaycan values if gender_str in ("kişi", "kisi", "male", "m", "erkək", "erkek"): return "Male" elif gender_str in ("qadın", "qadin", "female", "f", "qız", "qiz"): return "Female" return "Male" # Default fallback def map_marital_status(amas_status): """ Transform AMAS marital status to ERPNext marital status. AMAS values: "Evli", "Subay", "Married", "Single", etc. ERPNext values: "Single", "Married", "Divorced", "Widowed" """ if not amas_status: return None status_str = str(amas_status).strip().lower() # Map Azərbaycan values if status_str in ("evli", "married", "nikahli", "nikаhlı"): return "Married" elif status_str in ("subay", "subаy", "single", "bekar", "bekаr"): return "Single" elif status_str in ("boşanmış", "bosanmis", "divorced"): return "Divorced" elif status_str in ("dul", "widowed", "widow"): return "Widowed" return None def map_blood_group(amas_blood): """ Transform AMAS blood type to ERPNext blood group. AMAS format: "A(II) Rh+", "O(I) Rh-", etc. ERPNext values: "A+", "A-", "B+", "B-", "AB+", "AB-", "O+", "O-" """ if not amas_blood: return None blood_str = str(amas_blood).strip().upper() # Extract blood type and Rh factor # Examples: "A(II) Rh+", "AB(IV) Rh-", "O(I) Rh+" # Map blood group letter if "AB" in blood_str or "IV" in blood_str: blood_type = "AB" elif "A" in blood_str or "II" in blood_str: blood_type = "A" elif "B" in blood_str or "III" in blood_str: blood_type = "B" elif "O" in blood_str or "I" in blood_str or "0" in blood_str: blood_type = "O" else: return None # Map Rh factor if "+" in blood_str or "POSITIVE" in blood_str or "POZİTİV" in blood_str: rh_factor = "+" elif "-" in blood_str or "NEGATIVE" in blood_str or "NEQATİV" in blood_str: rh_factor = "-" else: rh_factor = "+" # Default to positive return f"{blood_type}{rh_factor}" def build_full_address(address, city, district): """ Combine AMAS address fields into single address string. """ parts = [] if address: parts.append(str(address).strip()) if district: parts.append(str(district).strip()) if city: parts.append(str(city).strip()) return ", ".join(parts) if parts else None def resolve_designation(position_text, create_if_missing): """ Find or create a Designation from ƏMAS position text. Returns designation name or None. """ if not position_text or not str(position_text).strip(): return None position_text = str(position_text).strip() # Search for existing designation by name existing = frappe.db.exists("Designation", position_text) if existing: return existing # Try case-insensitive search by designation_name field result = frappe.db.get_value( "Designation", filters={"designation_name": ["like", position_text]}, fieldname="name" ) if result: return result # Not found — create if allowed if create_if_missing: designation = frappe.new_doc("Designation") designation.designation_name = position_text designation.insert(ignore_permissions=True) return designation.name return None @frappe.whitelist() def create_single_employee_from_amas(asan_login_name, employee_data, company, create_designation=0): """ Create/update single Employee record from ƏMAS data. Used for sequential processing with frontend progress updates. """ try: if isinstance(employee_data, str): employee_data = json.loads(employee_data) create_designation = int(create_designation) if not company: return {"success": False, "message": "Company is required"} doc = frappe.get_doc("Asan Login", asan_login_name) organization_name = doc.amas_account_name organization_voen = doc.amas_account_number # Status mapping status_map = { "qüvvədədir": "Active", "aktiv": "Active", "active": "Active", "ləğv edilib": "Left", "xitam": "Left", "terminated": "Left", "dayandırılıb": "Suspended", "suspended": "Suspended", } doc_oid = employee_data.get("doc_oid") doc_no = employee_data.get("doc_no") doc_type = employee_data.get("doc_type", "docType_47") # Fetch detail data detail_data = None full_name = str(employee_data.get("full_name") or "").strip() if 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') # Get FIN fin = None if detail_data and 'EmasContractsEnt' in detail_data and isinstance(detail_data.get('EmasContractsEnt'), list) and len(detail_data['EmasContractsEnt']) > 0: fin = detail_data['EmasContractsEnt'][0].get('pin') if 'iamasBean' in detail_data and detail_data.get('iamasBean'): person = detail_data['iamasBean'] name = person.get('name', '') surname = person.get('surname', '') full_name = f"{name} {surname}".strip() if not fin: fin = employee_data.get("identification_number") if not fin: return {"success": False, "message": f"{full_name}: No identification_number"} fin = str(fin).strip() # Save/update Amas Employees record amas_existing = None if doc_oid: amas_existing = frappe.db.exists("Amas Employees", {"doc_oid": str(doc_oid)}) if not amas_existing: amas_existing = frappe.db.exists( "Amas Employees", {"identification_number": fin, "organization_voen": organization_voen} ) if amas_existing: amas_doc = frappe.get_doc("Amas Employees", amas_existing) update_employee_from_data(amas_doc, employee_data) if detail_data: update_employee_from_detail_data(amas_doc, detail_data) if not amas_doc.identification_number: amas_doc.identification_number = fin amas_doc.save(ignore_permissions=True) else: amas_doc = frappe.new_doc("Amas Employees") amas_doc.asan_login = asan_login_name amas_doc.organization_name = organization_name amas_doc.organization_voen = organization_voen amas_doc.identification_number = fin update_employee_from_data(amas_doc, employee_data) if detail_data: update_employee_from_detail_data(amas_doc, detail_data) if not amas_doc.identification_number: amas_doc.identification_number = fin amas_doc.insert(ignore_permissions=True) # Create/update Employee record existing_employee = frappe.db.exists("Employee", {"passport_number": fin}) # Split full_name name_parts = full_name.split() if full_name else [] first_name = name_parts[0] if name_parts else "?" last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else "" # Parse dates dob = None doj = None contract_end = None if detail_data: if 'iamasBean' in detail_data and detail_data.get('iamasBean'): dob = parse_amas_date(detail_data['iamasBean'].get('birthDate')) if 'EmasContractDetailsEnt' in detail_data and isinstance(detail_data.get('EmasContractDetailsEnt'), list) and len(detail_data['EmasContractDetailsEnt']) > 0: details = detail_data['EmasContractDetailsEnt'][0] doj = parse_amas_date(details.get('beginDate')) contract_end = parse_amas_date(details.get('endDate')) else: dob = parse_amas_date(employee_data.get("birthday")) doj = parse_amas_date(employee_data.get("begin_date")) contract_end = parse_amas_date(employee_data.get("end_date")) # Map status raw_status = "" if detail_data and 'EmasContractsEnt' in detail_data and isinstance(detail_data.get('EmasContractsEnt'), list) and len(detail_data['EmasContractsEnt']) > 0: raw_status = str(detail_data['EmasContractsEnt'][0].get('contractStatus') or "").strip().lower() else: raw_status = str(employee_data.get("contract_status") or "").strip().lower() status = status_map.get(raw_status, "Active") # Parse salary salary = None if detail_data and 'EmasContractDetailsEnt' in detail_data and isinstance(detail_data.get('EmasContractDetailsEnt'), list) and len(detail_data['EmasContractDetailsEnt']) > 0: raw_salary = detail_data['EmasContractDetailsEnt'][0].get('monthlySalary') else: raw_salary = employee_data.get("monthly_salary") if raw_salary: try: salary = float(str(raw_salary).replace(",", ".").replace(" ", "")) except (ValueError, TypeError): pass # Resolve designation position_text = None if detail_data and 'staffData' in detail_data and detail_data.get('staffData'): position_text = detail_data['staffData'].get('positionText') else: position_text = employee_data.get("work_position_text") designation = resolve_designation(position_text, create_designation) # Extract additional fields from amas_doc (already populated with detail data) middle_name = amas_doc.father_name if hasattr(amas_doc, 'father_name') else None gender = map_gender(amas_doc.gender if hasattr(amas_doc, 'gender') else None) cell_number = amas_doc.contact_phone if hasattr(amas_doc, 'contact_phone') else None personal_email = amas_doc.contact_email if hasattr(amas_doc, 'contact_email') else None marital_status = map_marital_status(amas_doc.marital_status if hasattr(amas_doc, 'marital_status') else None) blood_group = map_blood_group(amas_doc.blood_type if hasattr(amas_doc, 'blood_type') else None) date_of_issue = parse_amas_date(amas_doc.document_issue_date if hasattr(amas_doc, 'document_issue_date') else None) valid_upto = parse_amas_date(amas_doc.document_expiry_date if hasattr(amas_doc, 'document_expiry_date') else None) place_of_issue = amas_doc.document_issued_by if hasattr(amas_doc, 'document_issued_by') else None current_address = build_full_address( amas_doc.personal_address if hasattr(amas_doc, 'personal_address') else None, amas_doc.personal_city if hasattr(amas_doc, 'personal_city') else None, amas_doc.personal_district if hasattr(amas_doc, 'personal_district') else None ) if existing_employee: emp = frappe.get_doc("Employee", existing_employee) emp.first_name = first_name if middle_name: emp.middle_name = middle_name emp.last_name = last_name emp.employee_name = full_name or first_name emp.gender = gender if dob: emp.date_of_birth = dob if doj: emp.date_of_joining = doj if contract_end: emp.contract_end_date = contract_end if salary is not None: emp.ctc = salary if designation: emp.designation = designation if cell_number: emp.cell_number = cell_number if personal_email: emp.personal_email = personal_email if marital_status: emp.marital_status = marital_status if blood_group: emp.blood_group = blood_group if date_of_issue: emp.date_of_issue = date_of_issue if valid_upto: emp.valid_upto = valid_upto if place_of_issue: emp.place_of_issue = place_of_issue if current_address: emp.current_address = current_address # Link to AMAS Employees record emp.custom_amas_employee = amas_doc.name emp.save(ignore_permissions=True) # Link Amas Employees → Employee amas_doc.employee = emp.name amas_doc.save(ignore_permissions=True) frappe.db.commit() return { "success": True, "updated": True, "created": False, "employee_name": emp.name } else: emp = frappe.new_doc("Employee") emp.first_name = first_name if middle_name: emp.middle_name = middle_name emp.last_name = last_name emp.company = company emp.passport_number = fin emp.employee_number = fin emp.date_of_birth = dob or "2000-01-01" emp.date_of_joining = doj or frappe.utils.today() emp.gender = gender emp.status = status if contract_end: emp.contract_end_date = contract_end if salary is not None: emp.ctc = salary if designation: emp.designation = designation if cell_number: emp.cell_number = cell_number if personal_email: emp.personal_email = personal_email if marital_status: emp.marital_status = marital_status if blood_group: emp.blood_group = blood_group if date_of_issue: emp.date_of_issue = date_of_issue if valid_upto: emp.valid_upto = valid_upto if place_of_issue: emp.place_of_issue = place_of_issue if current_address: emp.current_address = current_address # Link to AMAS Employees record emp.custom_amas_employee = amas_doc.name emp.insert(ignore_permissions=True) # Link Amas Employees → Employee amas_doc.employee = emp.name amas_doc.save(ignore_permissions=True) frappe.db.commit() return { "success": True, "created": True, "updated": False, "employee_name": emp.name } except Exception as e: frappe.log_error( f"Single employee create error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS Employee Create Error" ) return { "success": False, "message": str(e) } @frappe.whitelist() 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. Args: asan_login_name: Asan Login name 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: { "success": True/False, "action": "created"/"updated"/"error", "employee_name": "...", "full_name": "...", "message": "..." } """ try: if isinstance(emp_data, str): emp_data = json.loads(emp_data) create_designation = int(create_designation) if not company: return {"success": False, "message": "Company is required"} doc = frappe.get_doc("Asan Login", asan_login_name) organization_name = doc.amas_account_name organization_voen = doc.amas_account_number # Status mapping status_map = { "qüvvədədir": "Active", "aktiv": "Active", "active": "Active", "ləğv edilib": "Left", "xitam": "Left", "terminated": "Left", "dayandırılıb": "Suspended", "suspended": "Suspended", } doc_oid = emp_data.get("doc_oid") doc_no = emp_data.get("doc_no") doc_type = emp_data.get("doc_type", "docType_47") # 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 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') # Get FIN from detail or list data fin = None if detail_data and 'EmasContractsEnt' in detail_data and isinstance(detail_data.get('EmasContractsEnt'), list) and len(detail_data['EmasContractsEnt']) > 0: fin = detail_data['EmasContractsEnt'][0].get('pin') # Also update full_name from detail data if available if 'iamasBean' in detail_data and detail_data.get('iamasBean'): person = detail_data['iamasBean'] name = person.get('name', '') surname = person.get('surname', '') full_name = f"{name} {surname}".strip() if not fin: fin = emp_data.get("identification_number") if not fin: return { "success": False, "action": "error", "full_name": full_name, "message": f"{full_name}: No identification_number" } fin = str(fin).strip() # 1. Save/update Amas Employees record amas_existing = None if doc_oid: amas_existing = frappe.db.exists("Amas Employees", {"doc_oid": str(doc_oid)}) if not amas_existing: amas_existing = frappe.db.exists( "Amas Employees", {"identification_number": fin, "organization_voen": organization_voen} ) if amas_existing: amas_doc = frappe.get_doc("Amas Employees", amas_existing) update_employee_from_data(amas_doc, emp_data) if detail_data: update_employee_from_detail_data(amas_doc, detail_data) if not amas_doc.identification_number: amas_doc.identification_number = fin amas_doc.save(ignore_permissions=True) else: amas_doc = frappe.new_doc("Amas Employees") amas_doc.asan_login = asan_login_name amas_doc.organization_name = organization_name amas_doc.organization_voen = organization_voen amas_doc.identification_number = fin update_employee_from_data(amas_doc, emp_data) if detail_data: update_employee_from_detail_data(amas_doc, detail_data) if not amas_doc.identification_number: amas_doc.identification_number = fin amas_doc.insert(ignore_permissions=True) # 2. Create/update Employee record existing_employee = frappe.db.exists("Employee", {"passport_number": fin}) # Split full_name into first/last name_parts = full_name.split() if full_name else [] first_name = name_parts[0] if name_parts else "?" last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else "" # Parse dates from detail data if available, otherwise from list data dob = None doj = None contract_end = None if detail_data: if 'iamasBean' in detail_data and detail_data.get('iamasBean'): dob = parse_amas_date(detail_data['iamasBean'].get('birthDate')) if 'EmasContractDetailsEnt' in detail_data and isinstance(detail_data.get('EmasContractDetailsEnt'), list) and len(detail_data['EmasContractDetailsEnt']) > 0: details = detail_data['EmasContractDetailsEnt'][0] doj = parse_amas_date(details.get('beginDate')) contract_end = parse_amas_date(details.get('endDate')) else: dob = parse_amas_date(emp_data.get("birthday")) doj = parse_amas_date(emp_data.get("begin_date")) contract_end = parse_amas_date(emp_data.get("end_date")) # Map status raw_status = "" if detail_data and 'EmasContractsEnt' in detail_data and isinstance(detail_data.get('EmasContractsEnt'), list) and len(detail_data['EmasContractsEnt']) > 0: raw_status = str(detail_data['EmasContractsEnt'][0].get('contractStatus') or "").strip().lower() else: raw_status = str(emp_data.get("contract_status") or "").strip().lower() status = status_map.get(raw_status, "Active") # Parse salary salary = None if detail_data and 'EmasContractDetailsEnt' in detail_data and isinstance(detail_data.get('EmasContractDetailsEnt'), list) and len(detail_data['EmasContractDetailsEnt']) > 0: raw_salary = detail_data['EmasContractDetailsEnt'][0].get('monthlySalary') else: raw_salary = emp_data.get("monthly_salary") if raw_salary: try: salary = float(str(raw_salary).replace(",", ".").replace(" ", "")) except (ValueError, TypeError): pass # Resolve designation position_text = None if detail_data and 'staffData' in detail_data and detail_data.get('staffData'): position_text = detail_data['staffData'].get('positionText') else: position_text = emp_data.get("work_position_text") designation = resolve_designation(position_text, create_designation) # Extract additional fields from amas_doc middle_name = amas_doc.father_name if hasattr(amas_doc, 'father_name') else None gender = map_gender(amas_doc.gender if hasattr(amas_doc, 'gender') else None) cell_number = amas_doc.contact_phone if hasattr(amas_doc, 'contact_phone') else None personal_email = amas_doc.contact_email if hasattr(amas_doc, 'contact_email') else None marital_status_mapped = map_marital_status(amas_doc.marital_status if hasattr(amas_doc, 'marital_status') else None) blood_group_mapped = map_blood_group(amas_doc.blood_type if hasattr(amas_doc, 'blood_type') else None) date_of_issue = parse_amas_date(amas_doc.document_issue_date if hasattr(amas_doc, 'document_issue_date') else None) valid_upto = parse_amas_date(amas_doc.document_expiry_date if hasattr(amas_doc, 'document_expiry_date') else None) place_of_issue = amas_doc.document_issued_by if hasattr(amas_doc, 'document_issued_by') else None current_address = build_full_address( amas_doc.personal_address if hasattr(amas_doc, 'personal_address') else None, amas_doc.personal_city if hasattr(amas_doc, 'personal_city') else None, amas_doc.personal_district if hasattr(amas_doc, 'personal_district') else None ) if existing_employee: emp = frappe.get_doc("Employee", existing_employee) emp.first_name = first_name if middle_name: emp.middle_name = middle_name emp.last_name = last_name emp.employee_name = full_name or first_name emp.gender = gender if dob: emp.date_of_birth = dob if doj: emp.date_of_joining = doj if contract_end: emp.contract_end_date = contract_end if salary is not None: emp.ctc = salary if designation: emp.designation = designation if cell_number: emp.cell_number = cell_number if personal_email: emp.personal_email = personal_email if marital_status_mapped: emp.marital_status = marital_status_mapped if blood_group_mapped: emp.blood_group = blood_group_mapped if date_of_issue: emp.date_of_issue = date_of_issue if valid_upto: emp.valid_upto = valid_upto if place_of_issue: emp.place_of_issue = place_of_issue if current_address: emp.current_address = current_address emp.save(ignore_permissions=True) action = "updated" else: emp = frappe.new_doc("Employee") emp.first_name = first_name if middle_name: emp.middle_name = middle_name emp.last_name = last_name emp.company = company emp.passport_number = fin emp.employee_number = fin emp.date_of_birth = dob or "2000-01-01" emp.date_of_joining = doj or frappe.utils.today() emp.gender = gender emp.status = status if contract_end: emp.contract_end_date = contract_end if salary is not None: emp.ctc = salary if designation: emp.designation = designation if cell_number: emp.cell_number = cell_number if personal_email: emp.personal_email = personal_email if marital_status_mapped: emp.marital_status = marital_status_mapped if blood_group_mapped: emp.blood_group = blood_group_mapped if date_of_issue: emp.date_of_issue = date_of_issue if valid_upto: emp.valid_upto = valid_upto if place_of_issue: emp.place_of_issue = place_of_issue if current_address: emp.current_address = current_address emp.insert(ignore_permissions=True) action = "created" # Link Amas Employees → Employee amas_doc.employee = emp.name amas_doc.save(ignore_permissions=True) frappe.db.commit() return { "success": True, "action": action, "employee_name": emp.name, "full_name": full_name, "message": f"{action.capitalize()}: {full_name}" } except Exception as e: fin_val = emp_data.get("identification_number", "?") if isinstance(emp_data, dict) else "?" name_val = emp_data.get("full_name", "?") if isinstance(emp_data, dict) else "?" error_msg = f"{name_val} ({fin_val}): {str(e)}" frappe.log_error( f"{error_msg}\n{frappe.get_traceback()}", "EMAS Single Employee Create Error" ) return { "success": False, "action": "error", "full_name": name_val, "message": error_msg } @frappe.whitelist() def create_employees_from_amas(asan_login_name, employees, company, create_designation=0): """ Create Employee records directly from ƏMAS API data. Receives raw employee data from the ƏMAS API (same format as get_employees_report returns). Also saves to Emas Employees doctype for record keeping. Field mapping (ƏMAS → Employee): full_name → first_name / last_name (split by space) identification_number → passport_number birthday → date_of_birth begin_date → date_of_joining end_date → contract_end_date monthly_salary → ctc contract_status → status work_position_text → designation (Link, searched/created) """ try: if isinstance(employees, str): employees = json.loads(employees) create_designation = int(create_designation) if not company: return {"success": False, "message": "Company is required"} doc = frappe.get_doc("Asan Login", asan_login_name) organization_name = doc.amas_account_name organization_voen = doc.amas_account_number created_count = 0 updated_count = 0 error_count = 0 errors = [] # Status mapping: ƏMAS contract_status → Employee status status_map = { "qüvvədədir": "Active", "aktiv": "Active", "active": "Active", "ləğv edilib": "Left", "xitam": "Left", "terminated": "Left", "dayandırılıb": "Suspended", "suspended": "Suspended", } total_employees = len(employees) for idx, emp_data in enumerate(employees, start=1): try: # Publish progress frappe.publish_progress( percent=(idx / total_employees) * 100, title=f"Loading employee {idx} of {total_employees}", description=f"Processing {emp_data.get('full_name', 'employee')}..." ) doc_oid = emp_data.get("doc_oid") 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 full_name = str(emp_data.get("full_name") or "").strip() if 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') # Get FIN from detail or list data fin = None if detail_data and 'EmasContractsEnt' in detail_data and isinstance(detail_data.get('EmasContractsEnt'), list) and len(detail_data['EmasContractsEnt']) > 0: fin = detail_data['EmasContractsEnt'][0].get('pin') # Also update full_name from detail data if available if 'iamasBean' in detail_data and detail_data.get('iamasBean'): person = detail_data['iamasBean'] name = person.get('name', '') surname = person.get('surname', '') full_name = f"{name} {surname}".strip() if not fin: fin = emp_data.get("identification_number") if not fin: errors.append(f"{full_name}: No identification_number") error_count += 1 continue fin = str(fin).strip() # 1. Save/update Emas Employees record amas_existing = None if doc_oid: amas_existing = frappe.db.exists("Amas Employees", {"doc_oid": str(doc_oid)}) if not amas_existing: amas_existing = frappe.db.exists( "Amas Employees", {"identification_number": fin, "organization_voen": organization_voen} ) if amas_existing: amas_doc = frappe.get_doc("Amas Employees", amas_existing) # Always populate basic fields from list data first update_employee_from_data(amas_doc, emp_data) # Then overlay with detail data if available if detail_data: update_employee_from_detail_data(amas_doc, detail_data) # Ensure FIN is set (required field) if not amas_doc.identification_number: amas_doc.identification_number = fin amas_doc.save(ignore_permissions=True) else: amas_doc = frappe.new_doc("Amas Employees") amas_doc.asan_login = asan_login_name amas_doc.organization_name = organization_name amas_doc.organization_voen = organization_voen # Set FIN first (required field) amas_doc.identification_number = fin # Always populate basic fields from list data first update_employee_from_data(amas_doc, emp_data) # Then overlay with detail data if available if detail_data: update_employee_from_detail_data(amas_doc, detail_data) # Ensure FIN is still set after mapping if not amas_doc.identification_number: amas_doc.identification_number = fin amas_doc.insert(ignore_permissions=True) # 2. Create/update Employee record existing_employee = frappe.db.exists("Employee", {"passport_number": fin}) # Split full_name into first/last name_parts = full_name.split() if full_name else [] first_name = name_parts[0] if name_parts else "?" last_name = " ".join(name_parts[1:]) if len(name_parts) > 1 else "" # Parse dates from detail data if available, otherwise from list data dob = None doj = None contract_end = None if detail_data: # Get from detail data if 'iamasBean' in detail_data and detail_data.get('iamasBean'): dob = parse_amas_date(detail_data['iamasBean'].get('birthDate')) if 'EmasContractDetailsEnt' in detail_data and isinstance(detail_data.get('EmasContractDetailsEnt'), list) and len(detail_data['EmasContractDetailsEnt']) > 0: details = detail_data['EmasContractDetailsEnt'][0] doj = parse_amas_date(details.get('beginDate')) contract_end = parse_amas_date(details.get('endDate')) else: # Fallback to list data dob = parse_amas_date(emp_data.get("birthday")) doj = parse_amas_date(emp_data.get("begin_date")) contract_end = parse_amas_date(emp_data.get("end_date")) # Map status raw_status = "" if detail_data and 'EmasContractsEnt' in detail_data and isinstance(detail_data.get('EmasContractsEnt'), list) and len(detail_data['EmasContractsEnt']) > 0: raw_status = str(detail_data['EmasContractsEnt'][0].get('contractStatus') or "").strip().lower() else: raw_status = str(emp_data.get("contract_status") or "").strip().lower() status = status_map.get(raw_status, "Active") # Parse salary salary = None if detail_data and 'EmasContractDetailsEnt' in detail_data and isinstance(detail_data.get('EmasContractDetailsEnt'), list) and len(detail_data['EmasContractDetailsEnt']) > 0: raw_salary = detail_data['EmasContractDetailsEnt'][0].get('monthlySalary') else: raw_salary = emp_data.get("monthly_salary") if raw_salary: try: salary = float(str(raw_salary).replace(",", ".").replace(" ", "")) except (ValueError, TypeError): pass # Resolve designation position_text = None if detail_data and 'staffData' in detail_data and detail_data.get('staffData'): position_text = detail_data['staffData'].get('positionText') else: position_text = emp_data.get("work_position_text") designation = resolve_designation(position_text, create_designation) # Extract additional fields from amas_doc (already populated with detail data) middle_name = amas_doc.father_name if hasattr(amas_doc, 'father_name') else None gender = map_gender(amas_doc.gender if hasattr(amas_doc, 'gender') else None) cell_number = amas_doc.contact_phone if hasattr(amas_doc, 'contact_phone') else None personal_email = amas_doc.contact_email if hasattr(amas_doc, 'contact_email') else None marital_status_mapped = map_marital_status(amas_doc.marital_status if hasattr(amas_doc, 'marital_status') else None) blood_group_mapped = map_blood_group(amas_doc.blood_type if hasattr(amas_doc, 'blood_type') else None) date_of_issue = parse_amas_date(amas_doc.document_issue_date if hasattr(amas_doc, 'document_issue_date') else None) valid_upto = parse_amas_date(amas_doc.document_expiry_date if hasattr(amas_doc, 'document_expiry_date') else None) place_of_issue = amas_doc.document_issued_by if hasattr(amas_doc, 'document_issued_by') else None current_address = build_full_address( amas_doc.personal_address if hasattr(amas_doc, 'personal_address') else None, amas_doc.personal_city if hasattr(amas_doc, 'personal_city') else None, amas_doc.personal_district if hasattr(amas_doc, 'personal_district') else None ) if existing_employee: emp = frappe.get_doc("Employee", existing_employee) emp.first_name = first_name if middle_name: emp.middle_name = middle_name emp.last_name = last_name emp.employee_name = full_name or first_name emp.gender = gender if dob: emp.date_of_birth = dob if doj: emp.date_of_joining = doj if contract_end: emp.contract_end_date = contract_end if salary is not None: emp.ctc = salary if designation: emp.designation = designation if cell_number: emp.cell_number = cell_number if personal_email: emp.personal_email = personal_email if marital_status_mapped: emp.marital_status = marital_status_mapped if blood_group_mapped: emp.blood_group = blood_group_mapped if date_of_issue: emp.date_of_issue = date_of_issue if valid_upto: emp.valid_upto = valid_upto if place_of_issue: emp.place_of_issue = place_of_issue if current_address: emp.current_address = current_address emp.save(ignore_permissions=True) updated_count += 1 else: emp = frappe.new_doc("Employee") emp.first_name = first_name if middle_name: emp.middle_name = middle_name emp.last_name = last_name emp.company = company emp.passport_number = fin emp.employee_number = fin emp.date_of_birth = dob or "2000-01-01" emp.date_of_joining = doj or frappe.utils.today() emp.gender = gender emp.status = status if contract_end: emp.contract_end_date = contract_end if salary is not None: emp.ctc = salary if designation: emp.designation = designation if cell_number: emp.cell_number = cell_number if personal_email: emp.personal_email = personal_email if marital_status_mapped: emp.marital_status = marital_status_mapped if blood_group_mapped: emp.blood_group = blood_group_mapped if date_of_issue: emp.date_of_issue = date_of_issue if valid_upto: emp.valid_upto = valid_upto if place_of_issue: emp.place_of_issue = place_of_issue if current_address: emp.current_address = current_address emp.insert(ignore_permissions=True) created_count += 1 # Link Amas Employees → Employee amas_doc.employee = emp.name amas_doc.save(ignore_permissions=True) except Exception as e: fin_val = emp_data.get("identification_number", "?") name_val = emp_data.get("full_name", "?") error_msg = f"{name_val} ({fin_val}): {str(e)}" frappe.log_error(error_msg, "EMAS Employee Create Error") errors.append(error_msg) error_count += 1 frappe.db.commit() return { "success": True, "created_count": created_count, "updated_count": updated_count, "error_count": error_count, "errors": errors[:10] } except Exception as e: frappe.log_error( f"Create employees from EMAS error: {str(e)}\n{frappe.get_traceback()}", "EMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} # ==================== EMPLOYEE DETAIL API FUNCTIONS ==================== # These functions fetch complete employee details from ƏMAS # (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, _state=None): """ Call eroom.getEditForm to get contract form data. Returns: EmasContractsEnt, EmasContractDetailsEnt, EmasContractTextDetailsEnt """ data = { "serviceName": "AppEmploymentContractOnline", "docOid": doc_oid, "docNo": doc_no, "docType": doc_type } return make_amas_request(asan_login_name, "/service/eroom.getEditForm", data, _state=_state) 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, _state=_state) 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, _state=_state) 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, _state=_state) 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, _state=_state) 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. """ data = { "docOid[]": doc_oid, "attachType": attach_type } return make_amas_request(asan_login_name, "/service/eroom.getAttachListByType", data, _state=_state) 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, _state=_state) @frappe.whitelist() def get_employee_detail(asan_login_name, doc_oid, doc_no, doc_type="docType_47", _state=None): """ Fetch complete employee detail from ƏMAS. 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 - staffData, iamasBean, addressBean, workAddress - docMainData, contractHtml """ combined_data = {} try: # 1. Get contract form data 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: result_data = data.get('resultData', {}) if result_data: combined_data.update(result_data) # 2. Get staff data (need staff_oid from contract) 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, _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: 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, _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: combined_data['iamasBean'] = data.get('iamasBean') combined_data['addressBean'] = data.get('addressBean') # 4. Get work address (need address_oid from contract details) 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, _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: 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, _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: combined_data['docMainData'] = { 'docOid': data.get('docOid'), 'docNo': data.get('docNo'), 'docDate': data.get('docDate'), 'docType': data.get('docType'), 'docState': data.get('docState'), 'listDocInfo': data.get('listDocInfo', []) } # 6. Get contract document (HTML) 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: file_id = attach_list[0].get('fileId') if 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: combined_data['contractHtml'] = file_data.get('fileContent') return { "success": True, "data": combined_data } except Exception as e: frappe.log_error( f"Get employee detail error: {str(e)}\n{frappe.get_traceback()}", "ƏMAS API Error" ) return {"success": False, "message": f"Error: {str(e)}"} def update_employee_from_detail_data(employee_doc, detail_data): """ Map all 115 fields from detail API responses to Emas Employees document. Handles nested objects from 6 different API calls. """ if not detail_data: return # Map EmasContractsEnt fields (contract info) if 'EmasContractsEnt' in detail_data and isinstance(detail_data.get('EmasContractsEnt'), list) and len(detail_data['EmasContractsEnt']) > 0: contract = detail_data['EmasContractsEnt'][0] employee_doc.contract_number = contract.get('contractNumber') employee_doc.ssn = contract.get('ssn') employee_doc.contract_type = contract.get('contractType') employee_doc.contract_status = contract.get('contractStatus') # Only set identification_number if it exists in detail data pin = contract.get('pin') if pin: employee_doc.identification_number = pin employee_doc.employer_tin = contract.get('employerTin') employee_doc.employer_pin = contract.get('employerPin') employee_doc.prev_contract = contract.get('prevContract') employee_doc.is_signed = 1 if contract.get('isSigned') else 0 employee_doc.contract_validity = contract.get('validity') employee_doc.workplace_type = contract.get('workplaceType') employee_doc.staff_oid = contract.get('staff') employee_doc.structure_unit_oid = contract.get('structureUnitOid') employee_doc.entity_oid = contract.get('entity') employee_doc.entity_details_oid = contract.get('entityDetails') employee_doc.from_migration = 1 if contract.get('fromMigration') else 0 employee_doc.doc_oid = contract.get('fkDocOid') # Map EmasContractDetailsEnt fields (salary, schedule, vacation) if 'EmasContractDetailsEnt' in detail_data and isinstance(detail_data.get('EmasContractDetailsEnt'), list) and len(detail_data['EmasContractDetailsEnt']) > 0: details = detail_data['EmasContractDetailsEnt'][0] employee_doc.base_salary = details.get('salary') employee_doc.monthly_salary = str(details.get('monthlySalary') or '') employee_doc.salary_add = details.get('salaryAdd') employee_doc.salary_add_hw = details.get('salaryAddHw') employee_doc.award_amount = details.get('awardAmount') employee_doc.award_type = details.get('awardType') employee_doc.overtime_amount = details.get('overTimeAmount') employee_doc.bad_condition_amount = details.get('badCondAmount') employee_doc.bad_condition_id = details.get('badCondId') employee_doc.currency_type = details.get('currencyType') employee_doc.payment_type = details.get('paymentType') employee_doc.payment_first_day = details.get('paymentFirstDay') employee_doc.bank_oid = details.get('bank') # Work schedule employee_doc.work_mode_type = details.get('workModeType') employee_doc.is_full_time = details.get('isFullTime') employee_doc.first_shift_start = details.get('firstShiftStart') employee_doc.first_shift_end = details.get('firstShiftEnd') employee_doc.second_shift_start = details.get('secondShiftStart') employee_doc.second_shift_end = details.get('secondShiftEnd') employee_doc.lunch_start = details.get('lunchStart') employee_doc.lunch_end = details.get('lunchEnd') employee_doc.work_time = details.get('workTime') employee_doc.working_time_cumulative = 1 if details.get('workingTimeCumulative') else 0 # Vacation employee_doc.vacation_days_count = details.get('vacationDaysCount') employee_doc.vacation_general_duration = details.get('vacGenDur') employee_doc.vacation_main_duration = details.get('vacMainDur') employee_doc.vacation_add_dur_one = details.get('vacAddDurOne') employee_doc.vacation_add_dur_two = details.get('vacAddDurTwo') employee_doc.vacation_add_dur_four = details.get('vacAddDurFour') employee_doc.vacation_amount = details.get('vacAmount') employee_doc.profession_employee_for_vacation = details.get('professionEmployeeForVacation') # Education employee_doc.education = details.get('education') employee_doc.scientific_degree = details.get('scientificDegree') employee_doc.speciality_degree = details.get('specialityDegree') employee_doc.state_speciality_degree = details.get('stateSpecialityDegree') employee_doc.qualification_degree_salary_add = details.get('qualificationDegreeSalaryAdd') # Dates employee_doc.main_contract_begin_date = str(details.get('mainContractBeginDate') or '') employee_doc.contract_begin_date = parse_amas_date(details.get('beginDate')) employee_doc.contract_end_date = parse_amas_date(details.get('endDate')) employee_doc.emp_job_start_date = str(details.get('empJobStartDate') or '') employee_doc.begin_date = str(details.get('beginDate') or '') employee_doc.end_date = str(details.get('endDate') or '') # Other fields employee_doc.position_id = details.get('positionId') employee_doc.activity_code = details.get('activityCode') employee_doc.work_address_oid = details.get('address') employee_doc.chairman_pin = details.get('chairmanPin') employee_doc.contract_reason = details.get('contractReason') employee_doc.workplace_sector = details.get('workplaceSector') # Map EmasContractTextDetailsEnt fields if 'EmasContractTextDetailsEnt' in detail_data and isinstance(detail_data.get('EmasContractTextDetailsEnt'), list) and len(detail_data['EmasContractTextDetailsEnt']) > 0: text_details = detail_data['EmasContractTextDetailsEnt'][0] if text_details: employee_doc.work_position = text_details.get('workPosition') employee_doc.activity_name_detail = text_details.get('activityName') employee_doc.labour_function = text_details.get('labourFunction') employee_doc.protection_tools = text_details.get('protectionTools') employee_doc.payment_other_cond = text_details.get('paymentOtherCond') employee_doc.rule_text = text_details.get('ruleText') # Map staffData fields if 'staffData' in detail_data and detail_data.get('staffData'): staff = detail_data['staffData'] employee_doc.position_text = staff.get('positionText') employee_doc.tree_path_name = staff.get('treePathName') employee_doc.staff_unit = staff.get('staffUnit') employee_doc.structure_path = staff.get('treePathName') employee_doc.work_position_text = staff.get('positionText') # Map iamasBean (personal data) if 'iamasBean' in detail_data and detail_data.get('iamasBean'): person = detail_data['iamasBean'] # Construct full name name = person.get('name', '') surname = person.get('surname', '') employee_doc.full_name = f"{name} {surname}".strip() employee_doc.father_name = person.get('fatherName') employee_doc.gender = person.get('gender') employee_doc.birthday = str(person.get('birthDate') or '') employee_doc.nationality = person.get('nationality') employee_doc.blood_type = person.get('bloodType') employee_doc.marital_status = person.get('maritalStatus') employee_doc.military_status = person.get('militaryStatus') employee_doc.document_number = person.get('documentNumber') employee_doc.document_issue_date = parse_amas_date(person.get('issueDate')) employee_doc.document_expiry_date = parse_amas_date(person.get('expireDate')) employee_doc.document_issued_by = person.get('issuedBy') employee_doc.personal_address = person.get('address') employee_doc.personal_city = person.get('city') employee_doc.personal_district = person.get('district') # Map addressBean (contact info) if 'addressBean' in detail_data and detail_data.get('addressBean'): address = detail_data['addressBean'] employee_doc.contact_phone = address.get('mobilePhoneNumber') employee_doc.contact_email = address.get('email') # Map workAddress if 'workAddress' in detail_data and detail_data.get('workAddress'): work_addr = detail_data['workAddress'] employee_doc.work_address_street = work_addr.get('street') employee_doc.work_address_district = work_addr.get('district') # Map docMainData if 'docMainData' in detail_data and detail_data.get('docMainData'): doc_main = detail_data['docMainData'] employee_doc.doc_no = doc_main.get('docNo') employee_doc.doc_date = str(doc_main.get('docDate') or '') employee_doc.doc_type = doc_main.get('docType') employee_doc.doc_state = doc_main.get('docState') employee_doc.flow = doc_main.get('flow') employee_doc.flow_alias = doc_main.get('flowAlias') if 'listDocInfo' in doc_main and isinstance(doc_main['listDocInfo'], list) and len(doc_main['listDocInfo']) > 0: employee_doc.after_exec_label = doc_main['listDocInfo'][0].get('afterExecLabel') created_date_str = doc_main['listDocInfo'][0].get('createDate') if created_date_str: employee_doc.created_date = created_date_str # Map contract HTML document if 'contractHtml' in detail_data and detail_data.get('contractHtml'): employee_doc.contract_html = detail_data.get('contractHtml') # Store complete raw data as JSON (only if detail_data has meaningful content) # Check if detail_data has any of the expected keys has_detail_content = any(key in detail_data for key in [ 'EmasContractsEnt', 'EmasContractDetailsEnt', 'EmasContractTextDetailsEnt', 'staffData', 'iamasBean', 'addressBean', 'workAddress', 'docMainData', 'contractHtml' ]) if has_detail_content: employee_doc.raw_data = json.dumps(detail_data, ensure_ascii=False, indent=2) employee_doc.import_date = frappe.utils.now_datetime() def on_delete_employee(doc, method): """ When an Employee is deleted, also delete linked Emas Employees records. """ amas_records = frappe.get_all( "Amas Employees", filters={"employee": doc.name}, fields=["name"] ) for rec in amas_records: frappe.delete_doc("Amas Employees", rec.name, ignore_permissions=True) @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. 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) try: create_designation = int(create_designation) 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( _process_bulk_employees_import, asan_login_name=asan_login_name, employees=employees_data, company=company, create_designation=create_designation, user=user, queue="default", timeout=1200, ) 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 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) created_count = 0 updated_count = 0 errors = [] 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 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, ) frappe.db.commit() frappe.publish_realtime( "amas_import_complete", { "total": total, "created": created_count, "updated": updated_count, "errors": errors, "cancelled": cancelled, }, user=user, ) finally: _clear_import_lock(asan_login_name)