713 lines
24 KiB
Python
713 lines
24 KiB
Python
# ======= COMPANY E-TAXES DATA LOADING =======
|
|
|
|
import frappe
|
|
import requests
|
|
from frappe.utils import getdate
|
|
|
|
from invoice_az.auth import DEFAULT_HEADERS, get_default_asan_login, record_etaxes_activity
|
|
|
|
BASE_URL = "https://new.e-taxes.gov.az"
|
|
OBJECT_LIST_URL = f"{BASE_URL}/api/po/profile/public/v1/object/list"
|
|
CASH_REGISTER_LIST_URL = f"{BASE_URL}/api/po/profile/public/v2/profile/cashRegister/list"
|
|
POS_TERMINAL_LIST_URL = f"{BASE_URL}/api/po/profile/public/v2/profile/posTerminal/list"
|
|
BANK_ACCOUNT_LIST_URL = f"{BASE_URL}/api/po/profile/public/v1/profile/bankAccount/list"
|
|
PRESENTED_CERT_LIST_URL = f"{BASE_URL}/api/po/edi-legal/public/v1/application/contractor/presented-accepted-e-certificates"
|
|
|
|
PAGE_SIZE = 200
|
|
MAX_PAGES = 50 # safety cap (≤ 10 000 records per import)
|
|
|
|
|
|
# ======= HELPERS =======
|
|
|
|
def _get(d, *path):
|
|
cur = d
|
|
for key in path:
|
|
if not isinstance(cur, dict):
|
|
return None
|
|
cur = cur.get(key)
|
|
return cur
|
|
|
|
|
|
def _safe_date(value):
|
|
if not value:
|
|
return None
|
|
try:
|
|
return getdate(value)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _unauthorized():
|
|
return {
|
|
"error": "unauthorized",
|
|
"message": "Authentication required. Please login again.",
|
|
}
|
|
|
|
|
|
def _server_error():
|
|
return {
|
|
"error": "server_error",
|
|
"message": "Service temporarily unavailable. Please try again in a few minutes.",
|
|
"status_code": 500,
|
|
}
|
|
|
|
|
|
def _auth_headers():
|
|
"""Return (headers, error_response_or_None). If err is truthy, bail out."""
|
|
asan_login = get_default_asan_login()
|
|
if not asan_login.get("found") or not asan_login.get("main_token"):
|
|
return None, _unauthorized()
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {asan_login['main_token']}"
|
|
return headers, None
|
|
|
|
|
|
def _etaxes_request(method, url, headers, **kwargs):
|
|
"""Make an E-Taxes request with the §10 401-retry-once pattern.
|
|
|
|
Returns (response, error_dict) — if error_dict is truthy, caller must return it.
|
|
"""
|
|
try:
|
|
response = requests.request(method, url, headers=headers, timeout=60, **kwargs)
|
|
|
|
if response.status_code == 401:
|
|
fresh_login = get_default_asan_login()
|
|
if fresh_login.get("found") and fresh_login.get("main_token"):
|
|
headers["x-authorization"] = f"Bearer {fresh_login['main_token']}"
|
|
response = requests.request(method, url, headers=headers, timeout=60, **kwargs)
|
|
if response.status_code == 401:
|
|
return None, _unauthorized()
|
|
|
|
if response.status_code == 500:
|
|
return None, _server_error()
|
|
|
|
if response.status_code == 404:
|
|
return None, {"success": False, "message": "Resource not found."}
|
|
|
|
response.raise_for_status()
|
|
return response, None
|
|
|
|
except requests.exceptions.HTTPError as e:
|
|
code = e.response.status_code if e.response is not None else None
|
|
if code == 401:
|
|
return None, _unauthorized()
|
|
if code == 500:
|
|
return None, _server_error()
|
|
frappe.log_error(f"HTTP error {code}: {e}", "E-Taxes Company Load")
|
|
return None, {"error": "http_error", "message": f"HTTP error {code}: {e}"}
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(f"Network error: {e}", "E-Taxes Company Load")
|
|
return None, {"error": "connection_error", "message": f"Network error: {e}"}
|
|
|
|
|
|
# ======= OBJECTS =======
|
|
|
|
@frappe.whitelist()
|
|
def load_company_objects(company):
|
|
"""Load taxpayer objects (registered locations) from E-Taxes into E-Taxes Object doctype."""
|
|
try:
|
|
record_etaxes_activity()
|
|
|
|
if not company or not frappe.db.exists("Company", company):
|
|
return {"success": False, "message": f"Company '{company}' not found."}
|
|
|
|
headers, err = _auth_headers()
|
|
if err:
|
|
return err
|
|
|
|
response, err = _etaxes_request("GET", OBJECT_LIST_URL, headers)
|
|
if err:
|
|
return err
|
|
|
|
data = response.json() or {}
|
|
objects = data.get("objectInfoShortList") or []
|
|
|
|
created, updated = 0, 0
|
|
for obj in objects:
|
|
code = obj.get("objectCode")
|
|
if not code:
|
|
continue
|
|
|
|
values = {
|
|
"object_name": (obj.get("objectName") or "").strip(),
|
|
"object_status": obj.get("status") or "active",
|
|
"address": obj.get("address"),
|
|
"main_activity_code": _get(obj, "mainActivity", "code"),
|
|
"main_activity_name": _get(obj, "mainActivity", "name", "az"),
|
|
"main_class_code": _get(obj, "mainClass", "code"),
|
|
"main_class_name": _get(obj, "mainClass", "name", "az"),
|
|
"middle_class_code": _get(obj, "middleClass", "code"),
|
|
"middle_class_name": _get(obj, "middleClass", "name", "az"),
|
|
"sub_class_code": _get(obj, "subClass", "code"),
|
|
"sub_class_name": _get(obj, "subClass", "name", "az"),
|
|
"company": company,
|
|
}
|
|
|
|
if frappe.db.exists("E-Taxes Object", code):
|
|
frappe.db.set_value("E-Taxes Object", code, values, update_modified=False)
|
|
updated += 1
|
|
else:
|
|
doc = frappe.get_doc({
|
|
"doctype": "E-Taxes Object",
|
|
"object_code": code,
|
|
**values,
|
|
})
|
|
doc.insert(ignore_permissions=True)
|
|
created += 1
|
|
|
|
frappe.db.commit()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": f"Objects loaded: {created} created, {updated} updated",
|
|
"created": created,
|
|
"updated": updated,
|
|
"total": len(objects),
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error: {e}\n{frappe.get_traceback()}",
|
|
"E-Taxes Object Load",
|
|
)
|
|
return {"success": False, "message": f"Error: {e}"}
|
|
|
|
|
|
# ======= CASH REGISTERS =======
|
|
|
|
@frappe.whitelist()
|
|
def load_company_cash_registers(company):
|
|
"""Load cash registers from E-Taxes into E-Taxes Cash Register doctype.
|
|
|
|
Uses hasMore pagination (CLAUDE.md §10 401-retry-once applies per page).
|
|
"""
|
|
try:
|
|
record_etaxes_activity()
|
|
|
|
if not company or not frappe.db.exists("Company", company):
|
|
return {"success": False, "message": f"Company '{company}' not found."}
|
|
|
|
headers, err = _auth_headers()
|
|
if err:
|
|
return err
|
|
|
|
created, updated, total_fetched = 0, 0, 0
|
|
offset = 0
|
|
|
|
for page in range(MAX_PAGES):
|
|
payload = {"maxCount": PAGE_SIZE, "offset": offset}
|
|
response, err = _etaxes_request(
|
|
"POST", CASH_REGISTER_LIST_URL, headers, json=payload
|
|
)
|
|
if err:
|
|
return err
|
|
|
|
data = response.json() or {}
|
|
registers = data.get("cashRegisters") or []
|
|
has_more = bool(data.get("hasMore"))
|
|
|
|
if not registers:
|
|
break
|
|
|
|
for reg in registers:
|
|
token = reg.get("token")
|
|
if not token:
|
|
continue
|
|
|
|
values = {
|
|
"number": reg.get("number"),
|
|
"model": reg.get("model"),
|
|
"ecash_status": reg.get("ecashStatus"),
|
|
"status_code": reg.get("status"),
|
|
"object": reg.get("object") if frappe.db.exists("E-Taxes Object", reg.get("object")) else None,
|
|
"certificate_type": reg.get("certificateType"),
|
|
"installed_at": _safe_date(reg.get("installedAt")),
|
|
"certificate_from": _safe_date(reg.get("certificateFrom")),
|
|
"certificate_to": _safe_date(reg.get("certificateTo")),
|
|
"replacement_date": _safe_date(reg.get("replacementDate")),
|
|
"operator_tin": reg.get("operatorTin"),
|
|
"operator_name": reg.get("operatorName"),
|
|
"company": company,
|
|
}
|
|
|
|
if frappe.db.exists("E-Taxes Cash Register", token):
|
|
frappe.db.set_value(
|
|
"E-Taxes Cash Register", token, values, update_modified=False
|
|
)
|
|
updated += 1
|
|
else:
|
|
doc = frappe.get_doc({
|
|
"doctype": "E-Taxes Cash Register",
|
|
"token": token,
|
|
**values,
|
|
})
|
|
doc.insert(ignore_permissions=True)
|
|
created += 1
|
|
|
|
total_fetched += len(registers)
|
|
offset += PAGE_SIZE
|
|
|
|
if not has_more:
|
|
break
|
|
else:
|
|
frappe.log_error(
|
|
f"Cash register pagination hit safety cap of {MAX_PAGES} pages "
|
|
f"for company {company}",
|
|
"E-Taxes Cash Register Load",
|
|
)
|
|
|
|
frappe.db.commit()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": f"Cash registers loaded: {created} created, {updated} updated",
|
|
"created": created,
|
|
"updated": updated,
|
|
"total": total_fetched,
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error: {e}\n{frappe.get_traceback()}",
|
|
"E-Taxes Cash Register Load",
|
|
)
|
|
return {"success": False, "message": f"Error: {e}"}
|
|
|
|
|
|
# ======= POS TERMINALS =======
|
|
|
|
def _object_from_registration_number(registration_number):
|
|
"""Registration numbers look like "4800111741-84001P02" — strip the trailing Pxx."""
|
|
if not registration_number:
|
|
return None
|
|
code = registration_number.rsplit("P", 1)[0]
|
|
if frappe.db.exists("E-Taxes Object", code):
|
|
return code
|
|
return None
|
|
|
|
|
|
@frappe.whitelist()
|
|
def load_company_pos_terminals(company):
|
|
"""Load POS terminals from E-Taxes into E-Taxes POS Terminal doctype.
|
|
|
|
Uses hasMore pagination (CLAUDE.md §10 401-retry-once applies per page).
|
|
"""
|
|
try:
|
|
record_etaxes_activity()
|
|
|
|
if not company or not frappe.db.exists("Company", company):
|
|
return {"success": False, "message": f"Company '{company}' not found."}
|
|
|
|
headers, err = _auth_headers()
|
|
if err:
|
|
return err
|
|
|
|
created, updated, total_fetched = 0, 0, 0
|
|
offset = 0
|
|
|
|
for page in range(MAX_PAGES):
|
|
payload = {"maxCount": PAGE_SIZE, "offset": offset}
|
|
response, err = _etaxes_request(
|
|
"POST", POS_TERMINAL_LIST_URL, headers, json=payload
|
|
)
|
|
if err:
|
|
return err
|
|
|
|
data = response.json() or {}
|
|
terminals = data.get("posTerminals") or []
|
|
has_more = bool(data.get("hasMore"))
|
|
|
|
if not terminals:
|
|
break
|
|
|
|
for term in terminals:
|
|
registration_number = term.get("registrationNumber")
|
|
if not registration_number:
|
|
continue
|
|
|
|
values = {
|
|
"serial_number": term.get("serialNumber"),
|
|
"status": term.get("status"),
|
|
"registration_date": _safe_date(term.get("registrationDate")),
|
|
"bank_name": term.get("bankName"),
|
|
"object": _object_from_registration_number(registration_number),
|
|
"company": company,
|
|
}
|
|
|
|
if frappe.db.exists("E-Taxes POS Terminal", registration_number):
|
|
frappe.db.set_value(
|
|
"E-Taxes POS Terminal", registration_number, values,
|
|
update_modified=False,
|
|
)
|
|
updated += 1
|
|
else:
|
|
doc = frappe.get_doc({
|
|
"doctype": "E-Taxes POS Terminal",
|
|
"registration_number": registration_number,
|
|
**values,
|
|
})
|
|
doc.insert(ignore_permissions=True)
|
|
created += 1
|
|
|
|
total_fetched += len(terminals)
|
|
offset += PAGE_SIZE
|
|
|
|
if not has_more:
|
|
break
|
|
else:
|
|
frappe.log_error(
|
|
f"POS terminal pagination hit safety cap of {MAX_PAGES} pages "
|
|
f"for company {company}",
|
|
"E-Taxes POS Terminal Load",
|
|
)
|
|
|
|
frappe.db.commit()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": f"POS terminals loaded: {created} created, {updated} updated",
|
|
"created": created,
|
|
"updated": updated,
|
|
"total": total_fetched,
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error: {e}\n{frappe.get_traceback()}",
|
|
"E-Taxes POS Terminal Load",
|
|
)
|
|
return {"success": False, "message": f"Error: {e}"}
|
|
|
|
|
|
# ======= COMPACT DATE/DATETIME PARSERS =======
|
|
|
|
def _parse_compact_date(value):
|
|
"""Parse "yyyyMMdd" → Date string "yyyy-MM-dd". Also accepts standard Date."""
|
|
if not value:
|
|
return None
|
|
s = str(value).strip()
|
|
if len(s) == 8 and s.isdigit():
|
|
s = f"{s[0:4]}-{s[4:6]}-{s[6:8]}"
|
|
return _safe_date(s)
|
|
|
|
|
|
def _parse_compact_datetime(value):
|
|
"""Parse "yyyyMMddHHmmss" → Datetime string. Also accepts standard Datetime."""
|
|
if not value:
|
|
return None
|
|
s = str(value).strip()
|
|
if len(s) == 14 and s.isdigit():
|
|
return f"{s[0:4]}-{s[4:6]}-{s[6:8]} {s[8:10]}:{s[10:12]}:{s[12:14]}"
|
|
return s
|
|
|
|
|
|
def _upsert(doctype, primary_key_value, values, primary_field):
|
|
"""Upsert by primary field. Returns (created, updated) deltas as (0,1) or (1,0)."""
|
|
if frappe.db.exists(doctype, primary_key_value):
|
|
frappe.db.set_value(doctype, primary_key_value, values, update_modified=False)
|
|
return 0, 1
|
|
doc = frappe.get_doc({
|
|
"doctype": doctype,
|
|
primary_field: primary_key_value,
|
|
**values,
|
|
})
|
|
doc.insert(ignore_permissions=True)
|
|
return 1, 0
|
|
|
|
|
|
def _summarize(created, updated, total, label):
|
|
return {
|
|
"success": True,
|
|
"message": f"{label}: {created} created, {updated} updated",
|
|
"created": created,
|
|
"updated": updated,
|
|
"total": total,
|
|
}
|
|
|
|
|
|
# ======= BANK ACCOUNTS =======
|
|
|
|
@frappe.whitelist()
|
|
def load_company_bank_accounts(company):
|
|
"""Load bank accounts from E-Taxes into E-Taxes Bank Account doctype."""
|
|
try:
|
|
record_etaxes_activity()
|
|
|
|
if not company or not frappe.db.exists("Company", company):
|
|
return {"success": False, "message": f"Company '{company}' not found."}
|
|
|
|
headers, err = _auth_headers()
|
|
if err:
|
|
return err
|
|
|
|
response, err = _etaxes_request("GET", BANK_ACCOUNT_LIST_URL, headers)
|
|
if err:
|
|
return err
|
|
|
|
data = response.json() or {}
|
|
accounts = data.get("bankAccounts") or []
|
|
|
|
created, updated = 0, 0
|
|
for acc in accounts:
|
|
duplicate_number = acc.get("duplicateNumber")
|
|
if not duplicate_number:
|
|
continue
|
|
values = {
|
|
"number": acc.get("number"),
|
|
"currency": acc.get("currency"),
|
|
"status": acc.get("status"),
|
|
"account_type": acc.get("type"),
|
|
"registration_date": _safe_date(acc.get("registrationDate")),
|
|
"duplicate_date": _safe_date(acc.get("duplicateDate")),
|
|
"bank_name": acc.get("bankName"),
|
|
"bank_tin": acc.get("bankTin"),
|
|
"bank_code": acc.get("bankCode"),
|
|
"main_bank_code": acc.get("mainBankCode"),
|
|
"settlement_account": acc.get("settlementAccount"),
|
|
"company": company,
|
|
}
|
|
c, u = _upsert("E-Taxes Bank Account", duplicate_number, values, "duplicate_number")
|
|
created += c
|
|
updated += u
|
|
|
|
frappe.db.commit()
|
|
return _summarize(created, updated, len(accounts), "Bank accounts loaded")
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error: {e}\n{frappe.get_traceback()}",
|
|
"E-Taxes Bank Account Load",
|
|
)
|
|
return {"success": False, "message": f"Error: {e}"}
|
|
|
|
|
|
# ======= OBLIGATION PACTS (sub-contractor obl-pact-list) =======
|
|
|
|
@frappe.whitelist()
|
|
# ======= PRESENTED / ACCEPTED E-CERTIFICATES =======
|
|
|
|
def _map_certificate(cert, company):
|
|
return {
|
|
"cert_no": cert.get("certNo"),
|
|
"cert_category": cert.get("certCategory"),
|
|
"cert_state": cert.get("certState"),
|
|
"operation_date": _parse_compact_datetime(cert.get("operationDate")),
|
|
"taxpayer_full_name": cert.get("taxPayerFullName"),
|
|
"voen": cert.get("voen"),
|
|
"sazis_name": cert.get("sazisName"),
|
|
"sazis_code": cert.get("sazisKodu") or None,
|
|
"contract_start_date": _parse_compact_date(cert.get("contractStartDate")),
|
|
"contract_end_date": _parse_compact_date(cert.get("contractEndDate")),
|
|
"cert_start_date": _parse_compact_date(cert.get("certStartDate")),
|
|
"cert_end_date": _parse_compact_date(cert.get("certEndDate")),
|
|
"fk_doc_oid": cert.get("fkDocOid"),
|
|
"company": company,
|
|
}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def load_company_presented_certificates(company):
|
|
"""Load presented/accepted e-certificates from E-Taxes."""
|
|
try:
|
|
record_etaxes_activity()
|
|
|
|
if not company or not frappe.db.exists("Company", company):
|
|
return {"success": False, "message": f"Company '{company}' not found."}
|
|
|
|
headers, err = _auth_headers()
|
|
if err:
|
|
return err
|
|
|
|
response, err = _etaxes_request("GET", PRESENTED_CERT_LIST_URL, headers)
|
|
if err:
|
|
return err
|
|
|
|
data = response.json() or {}
|
|
certs = data.get("certificateList") or []
|
|
|
|
created, updated = 0, 0
|
|
for cert in certs:
|
|
oid = cert.get("oid")
|
|
if not oid:
|
|
continue
|
|
values = _map_certificate(cert, company)
|
|
c, u = _upsert("E-Taxes Presented Certificate", oid, values, "oid")
|
|
created += c
|
|
updated += u
|
|
|
|
frappe.db.commit()
|
|
return _summarize(created, updated, len(certs), "Presented certificates loaded")
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error: {e}\n{frappe.get_traceback()}",
|
|
"E-Taxes Presented Certificate Load",
|
|
)
|
|
return {"success": False, "message": f"Error: {e}"}
|
|
|
|
|
|
|
|
|
|
# ======= BULK ORCHESTRATOR (Socket.IO progress) =======
|
|
|
|
PROGRESS_EVENT = "company_data_loading_progress"
|
|
COMPLETE_EVENT = "company_data_loading_complete"
|
|
|
|
|
|
@frappe.whitelist()
|
|
def load_company_data_bulk(company, selections):
|
|
"""Enqueue a background job that runs the selected loaders sequentially
|
|
and broadcasts progress/summary over Socket.IO.
|
|
|
|
`selections` is either a JSON string or a list of loader keys:
|
|
["load_objects", "load_cash_registers", ...]
|
|
"""
|
|
import json as _json
|
|
|
|
if isinstance(selections, str):
|
|
try:
|
|
selections = _json.loads(selections)
|
|
except ValueError:
|
|
selections = [s.strip() for s in selections.split(",") if s.strip()]
|
|
if not isinstance(selections, list):
|
|
selections = []
|
|
|
|
if not company or not frappe.db.exists("Company", company):
|
|
return {"success": False, "message": f"Company '{company}' not found."}
|
|
if not selections:
|
|
return {"success": False, "message": "No loaders selected."}
|
|
|
|
frappe.enqueue(
|
|
_process_company_data_bulk,
|
|
company=company,
|
|
selections=selections,
|
|
user=frappe.session.user,
|
|
queue="default",
|
|
timeout=1800,
|
|
)
|
|
return {"success": True, "enqueued": True, "total": len(selections)}
|
|
|
|
|
|
def _process_company_data_bulk(company, selections, user):
|
|
"""Background worker — runs each loader in turn, publishing realtime updates."""
|
|
frappe.set_user(user)
|
|
record_etaxes_activity()
|
|
|
|
loaders = {
|
|
"load_objects": (load_company_objects, "Objects"),
|
|
"load_cash_registers": (load_company_cash_registers, "Cash Registers"),
|
|
"load_pos_terminals": (load_company_pos_terminals, "POS Terminals"),
|
|
"load_bank_accounts": (load_company_bank_accounts, "Bank Accounts"),
|
|
"load_presented_certificates": (load_company_presented_certificates, "Presented Certificates"),
|
|
}
|
|
|
|
total = len(selections)
|
|
summary = []
|
|
|
|
frappe.publish_realtime(
|
|
PROGRESS_EVENT,
|
|
{"current": 0, "total": total, "label": "", "message": "Starting..."},
|
|
user=user,
|
|
)
|
|
|
|
for idx, key in enumerate(selections):
|
|
entry = loaders.get(key)
|
|
if not entry:
|
|
summary.append({"key": key, "label": key, "result": {
|
|
"success": False, "message": f"Unknown loader: {key}"
|
|
}})
|
|
continue
|
|
loader_func, label = entry
|
|
|
|
frappe.publish_realtime(
|
|
PROGRESS_EVENT,
|
|
{"current": idx, "total": total, "label": label,
|
|
"message": f"Loading {label}..."},
|
|
user=user,
|
|
)
|
|
|
|
try:
|
|
result = loader_func(company)
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Bulk loader '{key}' crashed: {e}\n{frappe.get_traceback()}",
|
|
"Company Data Bulk Load",
|
|
)
|
|
result = {"success": False, "message": f"{label}: {e}"}
|
|
|
|
summary.append({"key": key, "label": label, "result": result})
|
|
|
|
# Abort early on auth failure — frontend will re-auth and retry
|
|
if isinstance(result, dict) and result.get("error") == "unauthorized":
|
|
frappe.publish_realtime(
|
|
COMPLETE_EVENT,
|
|
{"summary": summary, "total": total, "completed": idx,
|
|
"error": "unauthorized",
|
|
"message": "Authentication required. Please login again."},
|
|
user=user,
|
|
)
|
|
return
|
|
|
|
frappe.publish_realtime(
|
|
PROGRESS_EVENT,
|
|
{"current": total, "total": total, "label": "", "message": "Done"},
|
|
user=user,
|
|
)
|
|
frappe.publish_realtime(
|
|
COMPLETE_EVENT,
|
|
{"summary": summary, "total": total, "completed": total},
|
|
user=user,
|
|
)
|
|
|
|
|
|
# ======= REFERENCE LIST (for Tax Policy tab sections) =======
|
|
|
|
_REFERENCE_LIST_FIELDS = {
|
|
"E-Taxes Object": ["name", "object_code", "object_name", "object_status"],
|
|
"E-Taxes Cash Register": ["name", "number", "model", "ecash_status", "object", "installed_at"],
|
|
"E-Taxes POS Terminal": ["name", "registration_number", "serial_number", "status", "registration_date"],
|
|
"E-Taxes Bank Account": ["name", "number", "currency", "status", "bank_name"],
|
|
"E-Taxes Presented Certificate": ["name", "cert_no", "taxpayer_full_name", "cert_category"],
|
|
}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_company_reference_list(doctype, company, limit=50):
|
|
"""Return first `limit` records of an E-Taxes reference doctype, filtered by company.
|
|
|
|
Only doctypes in the allow-list above are accessible via this endpoint.
|
|
"""
|
|
try:
|
|
if doctype not in _REFERENCE_LIST_FIELDS:
|
|
return {"success": False, "message": f"Doctype '{doctype}' not allowed."}
|
|
|
|
if not company or not frappe.db.exists("Company", company):
|
|
return {"success": False, "message": f"Company '{company}' not found."}
|
|
|
|
try:
|
|
limit = int(limit)
|
|
except (TypeError, ValueError):
|
|
limit = 50
|
|
limit = max(1, min(limit, 200))
|
|
|
|
fields = _REFERENCE_LIST_FIELDS[doctype]
|
|
filters = {"company": company}
|
|
|
|
data = frappe.get_all(
|
|
doctype,
|
|
filters=filters,
|
|
fields=fields,
|
|
limit=limit,
|
|
order_by="modified desc",
|
|
)
|
|
total_count = frappe.db.count(doctype, filters=filters)
|
|
|
|
return {"success": True, "data": data, "total_count": total_count}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Reference list load failed for {doctype}: {e}\n{frappe.get_traceback()}",
|
|
"Company Reference List",
|
|
)
|
|
return {"success": False, "message": f"Error: {e}"}
|