1516 lines
48 KiB
Python
1516 lines
48 KiB
Python
import frappe
|
|
import requests
|
|
import json
|
|
import uuid
|
|
from frappe.utils import nowdate, now_datetime, getdate
|
|
from invoice_az.auth import record_etaxes_activity, get_default_asan_login
|
|
|
|
# Constants
|
|
BASE_URL = "https://new.e-taxes.gov.az"
|
|
SERIAL_URL_ACT = f"{BASE_URL}/api/po/invoice/public/v1/act/generateSerialNumber/agriculturalProductsAct"
|
|
ACT_URL = f"{BASE_URL}/api/po/invoice/public/v1/act"
|
|
SIGN_URL = f"{BASE_URL}/api/po/invoice/public/v1/invoice/sign/withAsanImza"
|
|
REMOVE_DRAFT_URL = f"{BASE_URL}/api/po/invoice/public/v1/common/removeDrafts"
|
|
|
|
DEFAULT_HEADERS = {
|
|
"Accept": "application/json, text/plain, */*",
|
|
"Accept-Language": "en-US,en;q=0.9",
|
|
"Content-Type": "application/json",
|
|
"Cache-Control": "no-cache",
|
|
"Pragma": "no-cache"
|
|
}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def send_purchase_invoice_to_etaxes(purchase_invoice_name):
|
|
"""
|
|
Main function to create draft Purchase Act on E-Taxes (Step 1 of 2)
|
|
|
|
Workflow:
|
|
1. Validate invoice
|
|
2. Generate serial number
|
|
3. Build payload
|
|
4. Create draft act on E-Taxes (NOT signed yet)
|
|
5. Create tracking record
|
|
6. Update Purchase Invoice with status "Created, not signed"
|
|
|
|
Note: User must use "Sign Act" button to complete the process (Step 2)
|
|
|
|
Args:
|
|
purchase_invoice_name: Name of Purchase Invoice to send
|
|
|
|
Returns:
|
|
dict: Success/error response with details
|
|
"""
|
|
try:
|
|
# Record activity
|
|
record_etaxes_activity()
|
|
|
|
# Get Purchase Invoice document
|
|
doc = frappe.get_doc("Purchase Invoice", purchase_invoice_name)
|
|
|
|
# Validate invoice
|
|
validation_result = validate_invoice_for_sending(doc)
|
|
if not validation_result.get("valid"):
|
|
return {
|
|
"success": False,
|
|
"message": validation_result.get("message")
|
|
}
|
|
|
|
# Get authentication token
|
|
asan_login = get_default_asan_login()
|
|
if not asan_login.get("found") or not asan_login.get("main_token"):
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes authentication not found. Please login via ASAN."
|
|
}
|
|
|
|
token = asan_login.get("main_token")
|
|
|
|
# Step 1: Generate serial number based on act kind
|
|
if doc.act_kind == "Metal Scrap Reception Act":
|
|
serial_result = generate_metal_scrap_serial_number(token)
|
|
elif doc.act_kind == "Tire Products For Disposal Act":
|
|
serial_result = generate_tire_disposal_serial_number(token)
|
|
elif doc.act_kind == "Plastic Products For Disposal Act":
|
|
serial_result = generate_plastic_disposal_serial_number(token)
|
|
elif doc.act_kind == "Rawhide Supply":
|
|
serial_result = generate_rawhide_supply_serial_number(token)
|
|
elif doc.act_kind == "Other Product Receipt Act":
|
|
serial_result = generate_other_product_receipt_serial_number(token)
|
|
elif doc.act_kind == "Agricultural Products Act":
|
|
serial_result = generate_act_serial_number(token, doc.act_kind)
|
|
else:
|
|
return {
|
|
"success": False,
|
|
"message": f"Unknown act kind: {doc.act_kind}. Please select a valid act type."
|
|
}
|
|
|
|
if not serial_result.get("success"):
|
|
return {
|
|
"success": False,
|
|
"message": f"Failed to generate serial number: {serial_result.get('message')}"
|
|
}
|
|
|
|
serial_number = serial_result.get("serial_number")
|
|
|
|
# Step 2: Build act payload
|
|
payload = build_act_payload(doc, serial_number)
|
|
|
|
# Step 3: Create act on E-Taxes
|
|
create_result = create_act_on_etaxes(token, payload, serial_number)
|
|
if not create_result.get("success"):
|
|
return {
|
|
"success": False,
|
|
"message": f"Failed to create act: {create_result.get('message')}"
|
|
}
|
|
|
|
act_id = create_result.get("act_id")
|
|
|
|
# Step 4: Set status as draft (signing will be done separately via "Sign Act" button)
|
|
verification_code = None
|
|
final_status = "Created, not signed"
|
|
|
|
# Step 5: Create tracking record
|
|
create_outbox_record(
|
|
purchase_invoice_name,
|
|
act_id,
|
|
serial_number,
|
|
verification_code,
|
|
final_status,
|
|
None # No error message for draft creation
|
|
)
|
|
|
|
# Step 6: Update Purchase Invoice
|
|
|
|
# Use db.set_value to avoid triggering validation
|
|
frappe.db.set_value("Purchase Invoice", purchase_invoice_name, {
|
|
"etaxes_purchase_id": act_id,
|
|
"etaxes_purchase_serial_number": serial_number,
|
|
"etaxes_purchase_verification_code": verification_code,
|
|
"etaxes_purchase_send_status": final_status
|
|
})
|
|
frappe.db.commit()
|
|
|
|
success_message = f"Draft purchase act created successfully. Serial: {serial_number}"
|
|
success_message += " Please use 'Sign Act with ASAN Imza' button to complete the process."
|
|
|
|
return {
|
|
"success": True,
|
|
"message": success_message,
|
|
"act_id": act_id,
|
|
"serial_number": serial_number,
|
|
"verification_code": verification_code,
|
|
"status": final_status
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error sending Purchase Invoice to E-Taxes: {str(e)}\n{frappe.get_traceback()}",
|
|
"Send Purchase Act Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"An unexpected error occurred: {str(e)}"
|
|
}
|
|
|
|
|
|
def validate_invoice_for_sending(doc):
|
|
"""
|
|
Validate Purchase Invoice before sending to E-Taxes
|
|
|
|
Checks:
|
|
- Invoice is submitted
|
|
- Supplier type is Individual
|
|
- Supplier has all required Individual fields (FIN, passport, DOB, names)
|
|
- act_kind is either "Agricultural Products Act" or "Metal Scrap Reception Act"
|
|
- All items have tax_type (Tax Free or Taxable)
|
|
- For Metal Scrap, all items must be Taxable (no Tax Free items allowed)
|
|
- All items have product_group_code
|
|
- All items have UOM
|
|
|
|
Args:
|
|
doc: Purchase Invoice document
|
|
|
|
Returns:
|
|
dict: {"valid": bool, "message": str}
|
|
"""
|
|
try:
|
|
# Check if submitted
|
|
if doc.docstatus != 1:
|
|
return {
|
|
"valid": False,
|
|
"message": "Purchase Invoice must be submitted before sending to E-Taxes"
|
|
}
|
|
|
|
# Check supplier type
|
|
supplier = frappe.get_doc("Supplier", doc.supplier)
|
|
if supplier.supplier_type != "Individual":
|
|
return {
|
|
"valid": False,
|
|
"message": "Only Individual suppliers (physical persons) can be used for Purchase Acts"
|
|
}
|
|
|
|
# Check supplier FIN
|
|
if not supplier.fin:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Supplier '{doc.supplier}' does not have a FIN. Please update supplier details."
|
|
}
|
|
|
|
# Check supplier passport
|
|
if not supplier.passport_serial_number:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Supplier '{doc.supplier}' does not have a passport serial number. Please update supplier details."
|
|
}
|
|
|
|
# Check supplier date of birth
|
|
if not supplier.date_of_birth:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Supplier '{doc.supplier}' does not have a date of birth. Please update supplier details."
|
|
}
|
|
|
|
# Check supplier first name
|
|
if not supplier.first_name_individual:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Supplier '{doc.supplier}' does not have a first name. Please update supplier details."
|
|
}
|
|
|
|
# Check supplier last name
|
|
if not supplier.last_name_individual:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Supplier '{doc.supplier}' does not have a last name. Please update supplier details."
|
|
}
|
|
|
|
# Check act kind - must be one of the supported types
|
|
valid_act_kinds = ["Agricultural Products Act", "Metal Scrap Reception Act", "Tire Products For Disposal Act", "Plastic Products For Disposal Act", "Rawhide Supply", "Other Product Receipt Act"]
|
|
if not doc.act_kind:
|
|
return {
|
|
"valid": False,
|
|
"message": "Act Kind is required. Please select either 'Agricultural Products Act' or 'Metal Scrap Reception Act'."
|
|
}
|
|
|
|
if doc.act_kind not in valid_act_kinds:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Invalid Act Kind '{doc.act_kind}'. Please select a valid act type from the dropdown."
|
|
}
|
|
|
|
# Validate items
|
|
if not doc.items:
|
|
return {
|
|
"valid": False,
|
|
"message": "Purchase Invoice has no items"
|
|
}
|
|
|
|
missing_tax_types = []
|
|
missing_product_codes = []
|
|
missing_units = []
|
|
|
|
for item in doc.items:
|
|
# Check Tax Type (only validate if purchase_type is checked)
|
|
if doc.purchase_type:
|
|
if not item.tax_type:
|
|
missing_tax_types.append(item.item_code)
|
|
|
|
# Check product code
|
|
item_master = frappe.get_doc("Item", item.item_code)
|
|
if not item_master.product_group_code:
|
|
missing_product_codes.append(item.item_code)
|
|
|
|
# Check UOM
|
|
if not item.uom:
|
|
missing_units.append(item.item_code)
|
|
|
|
if missing_tax_types:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Items missing Tax Type: {', '.join(missing_tax_types)}. Please select 'Tax Free' or 'Taxable' for each item."
|
|
}
|
|
|
|
if missing_product_codes:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Items missing product codes: {', '.join(missing_product_codes)}"
|
|
}
|
|
|
|
if missing_units:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Items missing units (UOM): {', '.join(missing_units)}"
|
|
}
|
|
|
|
# For metal scrap, items cannot be tax-free (all must be taxable)
|
|
if doc.act_kind == "Metal Scrap Reception Act":
|
|
tax_free_items = []
|
|
for item in doc.items:
|
|
if item.tax_type == "Tax Free":
|
|
tax_free_items.append(item.item_code or item.item_name)
|
|
|
|
if tax_free_items:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Metal scrap items cannot be tax-free. Please change Tax Type to 'Taxable' for items: {', '.join(tax_free_items)}"
|
|
}
|
|
|
|
# For tire disposal, items cannot be tax-free (same as metal scrap)
|
|
if doc.act_kind == "Tire Products For Disposal Act":
|
|
tax_free_items = []
|
|
for item in doc.items:
|
|
if item.tax_type == "Tax Free":
|
|
tax_free_items.append(item.item_code or item.item_name)
|
|
|
|
if tax_free_items:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Tire disposal items cannot be tax-free. Please change Tax Type to 'Taxable' for items: {', '.join(tax_free_items)}"
|
|
}
|
|
|
|
# For plastic disposal, items cannot be tax-free (same as metal scrap)
|
|
if doc.act_kind == "Plastic Products For Disposal Act":
|
|
tax_free_items = []
|
|
for item in doc.items:
|
|
if item.tax_type == "Tax Free":
|
|
tax_free_items.append(item.item_code or item.item_name)
|
|
|
|
if tax_free_items:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Plastic disposal items cannot be tax-free. Please change Tax Type to 'Taxable' for items: {', '.join(tax_free_items)}"
|
|
}
|
|
|
|
# For rawhide supply, items cannot be tax-free (same as metal scrap)
|
|
if doc.act_kind == "Rawhide Supply":
|
|
tax_free_items = []
|
|
for item in doc.items:
|
|
if item.tax_type == "Tax Free":
|
|
tax_free_items.append(item.item_code or item.item_name)
|
|
|
|
if tax_free_items:
|
|
return {
|
|
"valid": False,
|
|
"message": f"Rawhide supply items cannot be tax-free. Please change Tax Type to 'Taxable' for items: {', '.join(tax_free_items)}"
|
|
}
|
|
|
|
# Note: Other Product Receipt Act allows both Tax Free and Taxable (like Agricultural)
|
|
# No validation needed for Other Product Receipt Act
|
|
|
|
return {"valid": True}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error validating invoice: {str(e)}\n{frappe.get_traceback()}",
|
|
"Purchase Invoice Validation Error"
|
|
)
|
|
return {
|
|
"valid": False,
|
|
"message": f"Validation error: {str(e)}"
|
|
}
|
|
|
|
|
|
def generate_act_serial_number(token, act_kind):
|
|
"""
|
|
Generate serial number for purchase act from E-Taxes API
|
|
|
|
Args:
|
|
token: ASAN authentication token
|
|
act_kind: Kind of act ("Agricultural Products Act")
|
|
|
|
Returns:
|
|
dict: {"success": bool, "serial_number": str, "message": str}
|
|
"""
|
|
try:
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {token}"
|
|
|
|
# Use Agricultural Products Act URL
|
|
url = SERIAL_URL_ACT
|
|
|
|
response = requests.get(url, headers=headers, timeout=30)
|
|
|
|
if response.status_code == 401:
|
|
return {
|
|
"success": False,
|
|
"message": "Authentication expired. Please login again."
|
|
}
|
|
|
|
if response.status_code == 500:
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes service error. Please try again later."
|
|
}
|
|
|
|
response.raise_for_status()
|
|
|
|
# Response: {"serialNumber":"AA25011234****"}
|
|
result = response.json()
|
|
serial_number = result.get("serialNumber")
|
|
|
|
if not serial_number:
|
|
# Fallback: try plain text response
|
|
serial_number = response.text.strip().strip('"')
|
|
|
|
return {
|
|
"success": True,
|
|
"serial_number": serial_number
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(
|
|
f"Error generating act serial number: {str(e)}",
|
|
"Act Serial Number Generation Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error generating serial: {str(e)}\n{frappe.get_traceback()}",
|
|
"Act Serial Number Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Error: {str(e)}"
|
|
}
|
|
|
|
|
|
def generate_metal_scrap_serial_number(token):
|
|
"""
|
|
Generate serial number for metal scrap reception act from E-Taxes API
|
|
|
|
Args:
|
|
token: Authentication token
|
|
|
|
Returns:
|
|
dict: {"success": True, "serial_number": "MS25012912345"} or error
|
|
"""
|
|
try:
|
|
url = f"{BASE_URL}/api/po/invoice/public/v1/act/generateSerialNumber/metalProductsAct"
|
|
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {token}"
|
|
|
|
response = requests.get(url, headers=headers, timeout=60)
|
|
|
|
# Handle common errors
|
|
if response.status_code == 401:
|
|
return {
|
|
"success": False,
|
|
"message": "Authentication expired. Please login again."
|
|
}
|
|
|
|
if response.status_code == 500:
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes service error. Please try again later."
|
|
}
|
|
|
|
if response.status_code == 404:
|
|
return {
|
|
"success": False,
|
|
"message": "Metal scrap serial number generation endpoint not found."
|
|
}
|
|
|
|
response.raise_for_status()
|
|
|
|
# Response: {"serialNumber":"MS25011234****"}
|
|
result = response.json()
|
|
serial_number = result.get("serialNumber")
|
|
|
|
if not serial_number:
|
|
# Fallback: try plain text response
|
|
serial_number = response.text.strip().strip('"')
|
|
|
|
return {
|
|
"success": True,
|
|
"serial_number": serial_number
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(
|
|
f"Error generating metal scrap serial number: {str(e)}",
|
|
"Metal Scrap Serial Number Generation Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error generating metal scrap serial: {str(e)}\n{frappe.get_traceback()}",
|
|
"Metal Scrap Serial Number Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Error: {str(e)}"
|
|
}
|
|
|
|
|
|
def generate_tire_disposal_serial_number(token):
|
|
"""
|
|
Generate serial number for tire products disposal act from E-Taxes API
|
|
|
|
Args:
|
|
token: Authentication token
|
|
|
|
Returns:
|
|
dict: {"success": True, "serial_number": "TP25012912345"} or error
|
|
"""
|
|
try:
|
|
url = f"{BASE_URL}/api/po/invoice/public/v1/act/generateSerialNumber/tireProductsForDisposalAct"
|
|
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {token}"
|
|
|
|
response = requests.get(url, headers=headers, timeout=60)
|
|
|
|
# Handle errors (401, 500, 404)
|
|
if response.status_code == 401:
|
|
return {
|
|
"success": False,
|
|
"message": "Authentication expired. Please login again."
|
|
}
|
|
|
|
if response.status_code == 500:
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes service error. Please try again later."
|
|
}
|
|
|
|
if response.status_code == 404:
|
|
return {
|
|
"success": False,
|
|
"message": "Tire disposal serial number generation endpoint not found."
|
|
}
|
|
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
serial_number = data.get("serialNumber")
|
|
|
|
if not serial_number:
|
|
# Some endpoints return plain text, not JSON
|
|
serial_number = response.text.strip().strip('"')
|
|
|
|
if not serial_number:
|
|
return {
|
|
"success": False,
|
|
"message": "No serial number returned from E-Taxes"
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"serial_number": serial_number
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(
|
|
f"Tire disposal serial generation error: {str(e)}",
|
|
"E-Taxes API Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error in generate_tire_disposal_serial_number: {str(e)}\n{frappe.get_traceback()}",
|
|
"E-Taxes API Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Error: {str(e)}"
|
|
}
|
|
|
|
|
|
def generate_plastic_disposal_serial_number(token):
|
|
"""
|
|
Generate serial number for plastic products disposal act from E-Taxes API
|
|
|
|
Args:
|
|
token: Authentication token
|
|
|
|
Returns:
|
|
dict: {"success": True, "serial_number": "PP25012912345"} or error
|
|
"""
|
|
try:
|
|
url = f"{BASE_URL}/api/po/invoice/public/v1/act/generateSerialNumber/plasticProductsForDisposalAct"
|
|
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {token}"
|
|
|
|
response = requests.get(url, headers=headers, timeout=60)
|
|
|
|
# Handle errors (401, 500, 404)
|
|
if response.status_code == 401:
|
|
return {
|
|
"success": False,
|
|
"message": "Authentication expired. Please login again."
|
|
}
|
|
|
|
if response.status_code == 500:
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes service error. Please try again later."
|
|
}
|
|
|
|
if response.status_code == 404:
|
|
return {
|
|
"success": False,
|
|
"message": "Plastic disposal serial number generation endpoint not found."
|
|
}
|
|
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
serial_number = data.get("serialNumber")
|
|
|
|
if not serial_number:
|
|
# Some endpoints return plain text, not JSON
|
|
serial_number = response.text.strip().strip('"')
|
|
|
|
if not serial_number:
|
|
return {
|
|
"success": False,
|
|
"message": "No serial number returned from E-Taxes"
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"serial_number": serial_number
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(
|
|
f"Plastic disposal serial generation error: {str(e)}",
|
|
"E-Taxes API Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error in generate_plastic_disposal_serial_number: {str(e)}\n{frappe.get_traceback()}",
|
|
"E-Taxes API Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Error: {str(e)}"
|
|
}
|
|
|
|
|
|
def generate_rawhide_supply_serial_number(token):
|
|
"""
|
|
Generate serial number for rawhide supply act from E-Taxes API
|
|
|
|
Args:
|
|
token: Authentication token
|
|
|
|
Returns:
|
|
dict: {"success": True, "serial_number": "RS25012912345"} or error
|
|
"""
|
|
try:
|
|
url = f"{BASE_URL}/api/po/invoice/public/v1/act/generateSerialNumber/rawhideSupply"
|
|
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {token}"
|
|
|
|
response = requests.get(url, headers=headers, timeout=60)
|
|
|
|
# Handle errors (401, 500, 404)
|
|
if response.status_code == 401:
|
|
return {
|
|
"success": False,
|
|
"message": "Authentication expired. Please login again."
|
|
}
|
|
|
|
if response.status_code == 500:
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes service error. Please try again later."
|
|
}
|
|
|
|
if response.status_code == 404:
|
|
return {
|
|
"success": False,
|
|
"message": "Rawhide supply serial number generation endpoint not found."
|
|
}
|
|
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
serial_number = data.get("serialNumber")
|
|
|
|
if not serial_number:
|
|
# Some endpoints return plain text, not JSON
|
|
serial_number = response.text.strip().strip('"')
|
|
|
|
if not serial_number:
|
|
return {
|
|
"success": False,
|
|
"message": "No serial number returned from E-Taxes"
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"serial_number": serial_number
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(
|
|
f"Rawhide supply serial generation error: {str(e)}",
|
|
"E-Taxes API Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error in generate_rawhide_supply_serial_number: {str(e)}\n{frappe.get_traceback()}",
|
|
"E-Taxes API Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Error: {str(e)}"
|
|
}
|
|
|
|
|
|
def generate_other_product_receipt_serial_number(token):
|
|
"""
|
|
Generate serial number for other product receipt act from E-Taxes API
|
|
|
|
Args:
|
|
token: Authentication token
|
|
|
|
Returns:
|
|
dict: {"success": True, "serial_number": "OP25012912345"} or error
|
|
"""
|
|
try:
|
|
url = f"{BASE_URL}/api/po/invoice/public/v1/act/generateSerialNumber/otherProductReceiptAct"
|
|
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {token}"
|
|
|
|
response = requests.get(url, headers=headers, timeout=60)
|
|
|
|
# Handle errors (401, 500, 404)
|
|
if response.status_code == 401:
|
|
return {
|
|
"success": False,
|
|
"message": "Authentication expired. Please login again."
|
|
}
|
|
|
|
if response.status_code == 500:
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes service error. Please try again later."
|
|
}
|
|
|
|
if response.status_code == 404:
|
|
return {
|
|
"success": False,
|
|
"message": "Other product receipt serial number generation endpoint not found."
|
|
}
|
|
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
serial_number = data.get("serialNumber")
|
|
|
|
if not serial_number:
|
|
# Some endpoints return plain text, not JSON
|
|
serial_number = response.text.strip().strip('"')
|
|
|
|
if not serial_number:
|
|
return {
|
|
"success": False,
|
|
"message": "No serial number returned from E-Taxes"
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"serial_number": serial_number
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(
|
|
f"Other product receipt serial generation error: {str(e)}",
|
|
"E-Taxes API Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error in generate_other_product_receipt_serial_number: {str(e)}\n{frappe.get_traceback()}",
|
|
"E-Taxes API Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Error: {str(e)}"
|
|
}
|
|
|
|
|
|
def get_etaxes_unit_name(erp_uom):
|
|
"""
|
|
Map ERPNext UOM to E-Taxes unit name
|
|
|
|
Priority:
|
|
1. Check E-Taxes Unit Mapping for manual mapping
|
|
2. Check E-Taxes Unit for direct match
|
|
3. Use UOM as is
|
|
|
|
Args:
|
|
erp_uom: ERPNext UOM name (e.g., "Nos", "Kg")
|
|
|
|
Returns:
|
|
str: E-Taxes unit name (e.g., "ədəd", "kq")
|
|
|
|
Note: unitCode field is NOT used - per user confirmation,
|
|
it should always be empty string. Only 'unit' field is populated.
|
|
"""
|
|
try:
|
|
# Step 1: Check E-Taxes Unit Mapping
|
|
mapping = frappe.db.get_value(
|
|
"E-Taxes Unit Mapping",
|
|
{"erp_unit": erp_uom},
|
|
"etaxes_unit_name"
|
|
)
|
|
|
|
if mapping:
|
|
return mapping
|
|
|
|
# Step 2: Check E-Taxes Unit for direct match
|
|
etaxes_unit = frappe.db.get_value(
|
|
"E-Taxes Unit",
|
|
{"name": erp_uom},
|
|
"etaxes_unit_name"
|
|
)
|
|
|
|
if etaxes_unit:
|
|
return etaxes_unit
|
|
|
|
# Step 3: Fallback to UOM as is
|
|
return erp_uom
|
|
|
|
except Exception as e:
|
|
return erp_uom
|
|
|
|
|
|
def get_tax_rate_from_type(tax_type):
|
|
"""
|
|
Map Tax Type field to E-Taxes taxRate value for purchase acts
|
|
|
|
Args:
|
|
tax_type: Tax Type field value ("Tax Free" or "Taxable")
|
|
|
|
Returns:
|
|
str: E-Taxes tax rate code ("taxFree" or "tax2")
|
|
|
|
Mapping:
|
|
- "Tax Free" → "taxFree" (0% VAT, азад от НДС)
|
|
- "Taxable" → "tax2" (5% tax, облагаемый)
|
|
- Empty/None → "taxFree" (default)
|
|
"""
|
|
if not tax_type:
|
|
return "taxFree"
|
|
|
|
mapping = {
|
|
"Tax Free": "taxFree",
|
|
"Taxable": "tax2"
|
|
}
|
|
|
|
return mapping.get(tax_type, "taxFree")
|
|
|
|
|
|
def calculate_tax_amount(cost, tax_rate):
|
|
"""
|
|
Calculate tax amount based on taxRate for purchase acts
|
|
|
|
Rules (per user confirmation):
|
|
- "taxFree": 0 (azad - освобождение от НДС)
|
|
- "tax2": cost * 0.05 (cəlb - облагаемый, 5% tax)
|
|
|
|
Args:
|
|
cost: Item cost (amount)
|
|
tax_rate: Tax rate code from get_tax_rate_from_template()
|
|
|
|
Returns:
|
|
float: Calculated tax amount
|
|
"""
|
|
if tax_rate == "tax2":
|
|
return round(cost * 0.05, 2)
|
|
return 0.0 # taxFree and others
|
|
|
|
|
|
def format_date(date_val, format_str):
|
|
"""
|
|
Format dates for E-Taxes API
|
|
|
|
Formats:
|
|
- "YYYY-MM-DD" → "2003-05-10" (for date of birth)
|
|
- "DD.MM.YYYY" → "20.01.2026" (for note field)
|
|
|
|
Args:
|
|
date_val: Date value (date object, string, etc.)
|
|
format_str: Target format string
|
|
|
|
Returns:
|
|
str: Formatted date string
|
|
"""
|
|
if not date_val:
|
|
return ""
|
|
|
|
date_obj = getdate(date_val)
|
|
|
|
if format_str == "YYYY-MM-DD":
|
|
return date_obj.strftime("%Y-%m-%d")
|
|
elif format_str == "DD.MM.YYYY":
|
|
return date_obj.strftime("%d.%m.%Y")
|
|
else:
|
|
return str(date_val)
|
|
|
|
|
|
def build_act_payload(doc, serial_number):
|
|
"""
|
|
Build JSON payload for E-Taxes purchase act creation
|
|
|
|
Args:
|
|
doc: Purchase Invoice document
|
|
serial_number: Generated serial number
|
|
|
|
Returns:
|
|
dict: JSON payload for E-Taxes API
|
|
"""
|
|
try:
|
|
# Get supplier details
|
|
supplier = frappe.get_doc("Supplier", doc.supplier)
|
|
|
|
# Build full name from individual fields (first name + patronymic + last name)
|
|
# E-Taxes API requires exact match with their database
|
|
full_name_parts = []
|
|
if supplier.first_name_individual:
|
|
full_name_parts.append(supplier.first_name_individual)
|
|
if supplier.patronimyc:
|
|
full_name_parts.append(supplier.patronimyc)
|
|
if supplier.last_name_individual:
|
|
full_name_parts.append(supplier.last_name_individual)
|
|
|
|
full_name = " ".join(full_name_parts) if full_name_parts else supplier.supplier_name
|
|
|
|
# Build seller object (physical person)
|
|
# IMPORTANT: Only include required fields (fin, passportSerialNumber, fullName, address)
|
|
# Extra fields (dateOfBirth, firstname, lastname, phoneNumber) cause API rejection
|
|
seller = {
|
|
"fin": supplier.fin,
|
|
"passportSerialNumber": supplier.passport_serial_number,
|
|
"fullName": full_name,
|
|
"address": "" # Empty string instead of None
|
|
}
|
|
|
|
# Build items array
|
|
items = []
|
|
|
|
for item in doc.items:
|
|
item_master = frappe.get_doc("Item", item.item_code)
|
|
|
|
# Get tax rate from Tax Type field
|
|
tax_rate = get_tax_rate_from_type(item.tax_type)
|
|
|
|
# Calculate tax amount based on tax rate
|
|
tax_amount = calculate_tax_amount(item.amount, tax_rate)
|
|
|
|
# Map ERPNext UOM to E-Taxes unit name
|
|
etaxes_unit = get_etaxes_unit_name(item.uom)
|
|
|
|
item_payload = {
|
|
"id": "", # Required field (empty string)
|
|
"gtin": None,
|
|
"unit": etaxes_unit,
|
|
"unitCode": "", # Always empty string per user confirmation
|
|
"note": format_date(doc.posting_date, "DD.MM.YYYY"),
|
|
"cost": item.amount,
|
|
"taxRate": tax_rate,
|
|
"quantity": item.qty,
|
|
"pricePerUnit": item.rate,
|
|
"productGroupCode": item_master.product_group_code,
|
|
"productName": item.item_name,
|
|
"taxAmount": tax_amount # Calculated based on tax rate
|
|
}
|
|
|
|
items.append(item_payload)
|
|
|
|
# Determine kind based on act_kind field
|
|
if doc.act_kind == "Metal Scrap Reception Act":
|
|
kind = "metalProductsAct"
|
|
elif doc.act_kind == "Tire Products For Disposal Act":
|
|
kind = "tireProductsForDisposalAct"
|
|
elif doc.act_kind == "Plastic Products For Disposal Act":
|
|
kind = "plasticProductsForDisposalAct"
|
|
elif doc.act_kind == "Rawhide Supply":
|
|
kind = "rawhideSupply"
|
|
elif doc.act_kind == "Other Product Receipt Act":
|
|
kind = "otherProductReceiptAct"
|
|
elif doc.act_kind == "Agricultural Products Act":
|
|
kind = "agriculturalProductsAct"
|
|
else:
|
|
# Fallback to agricultural for backward compatibility
|
|
kind = "agriculturalProductsAct"
|
|
|
|
# Build complete payload
|
|
# CRITICAL: Use "ministeryBlank" for signingMethod (not "byPhone")
|
|
# CRITICAL: Use empty strings ("") instead of None for optional fields
|
|
payload = {
|
|
"type": "current",
|
|
"serialNumber": serial_number,
|
|
"kind": kind,
|
|
"corrected": False,
|
|
"correctedActSerialNumber": "", # Empty string instead of None
|
|
"ministeryBlank": {
|
|
"serial": None,
|
|
"number": None,
|
|
"date": None
|
|
},
|
|
"signedBlankFileId": "", # Empty string instead of None
|
|
"signingMethod": "ministeryBlank", # Changed from "byPhone"
|
|
"seller": seller,
|
|
"items": items
|
|
}
|
|
|
|
return payload
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error building act payload: {str(e)}\n{frappe.get_traceback()}",
|
|
"Build Act Payload Error"
|
|
)
|
|
raise
|
|
|
|
|
|
def create_act_on_etaxes(token, payload, serial_number):
|
|
"""
|
|
Create purchase act on E-Taxes system
|
|
|
|
Args:
|
|
token: ASAN authentication token
|
|
payload: Act JSON payload
|
|
serial_number: Serial number for logging
|
|
|
|
Returns:
|
|
dict: {"success": bool, "act_id": str, "message": str}
|
|
"""
|
|
try:
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {token}"
|
|
|
|
response = requests.post(
|
|
ACT_URL,
|
|
data=json.dumps(payload),
|
|
headers=headers,
|
|
timeout=60
|
|
)
|
|
|
|
if response.status_code == 401:
|
|
return {
|
|
"success": False,
|
|
"message": "Authentication expired. Please login again."
|
|
}
|
|
|
|
if response.status_code == 500:
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes service error. Please try again later."
|
|
}
|
|
|
|
# Handle errors with E-Taxes message format
|
|
# Accept both 200 (OK) and 201 (Created) as success
|
|
if response.status_code not in [200, 201]:
|
|
try:
|
|
error_data = response.json()
|
|
# Try to get Azərbaycan message first
|
|
if isinstance(error_data, dict) and "message" in error_data:
|
|
if isinstance(error_data["message"], dict) and "az" in error_data["message"]:
|
|
return {
|
|
"success": False,
|
|
"message": error_data["message"]["az"]
|
|
}
|
|
elif isinstance(error_data["message"], str):
|
|
return {
|
|
"success": False,
|
|
"message": error_data["message"]
|
|
}
|
|
except:
|
|
pass
|
|
|
|
return {
|
|
"success": False,
|
|
"message": f"E-Taxes error (status {response.status_code})"
|
|
}
|
|
|
|
response.raise_for_status()
|
|
|
|
result = response.json()
|
|
act_id = result.get("id")
|
|
|
|
if not act_id:
|
|
return {
|
|
"success": False,
|
|
"message": "No act ID returned from E-Taxes"
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"act_id": act_id
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(
|
|
f"Error creating act on E-Taxes: {str(e)}\nPayload: {json.dumps(payload, indent=2)}",
|
|
"Create Act Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error creating act: {str(e)}\n{frappe.get_traceback()}",
|
|
"Create Act Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Error: {str(e)}"
|
|
}
|
|
|
|
|
|
def sign_act_on_etaxes(token, act_id):
|
|
"""
|
|
Sign purchase act with ASAN Imza
|
|
|
|
Args:
|
|
token: ASAN authentication token
|
|
act_id: Act ID to sign
|
|
|
|
Returns:
|
|
dict: {"success": bool, "verification_code": str, "message": str}
|
|
"""
|
|
try:
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {token}"
|
|
|
|
payload = {
|
|
"type": "send",
|
|
"invoiceIds": [act_id]
|
|
}
|
|
|
|
response = requests.post(
|
|
SIGN_URL,
|
|
data=json.dumps(payload),
|
|
headers=headers,
|
|
timeout=60
|
|
)
|
|
|
|
if response.status_code == 401:
|
|
return {
|
|
"success": False,
|
|
"message": "Authentication expired during signing"
|
|
}
|
|
|
|
if response.status_code == 500:
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes service error during signing"
|
|
}
|
|
|
|
response.raise_for_status()
|
|
|
|
result = response.json()
|
|
|
|
if result.get("status") == "ok":
|
|
verification_code = result.get("verificationCode")
|
|
return {
|
|
"success": True,
|
|
"verification_code": verification_code
|
|
}
|
|
else:
|
|
error_msg = result.get("error", "Unknown signing error")
|
|
return {
|
|
"success": False,
|
|
"message": f"Signing failed: {error_msg}"
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(
|
|
f"Error signing act {act_id}: {str(e)}",
|
|
"Sign Act Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Network error during signing: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error signing act: {str(e)}\n{frappe.get_traceback()}",
|
|
"Sign Act Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Signing error: {str(e)}"
|
|
}
|
|
|
|
|
|
def create_outbox_record(purchase_invoice_name, act_id, serial_number,
|
|
verification_code, status, error_message=None):
|
|
"""
|
|
Create E-Taxes Purchase Outbox tracking record
|
|
|
|
Args:
|
|
purchase_invoice_name: Purchase Invoice reference
|
|
act_id: E-Taxes act ID
|
|
serial_number: Serial number
|
|
verification_code: Verification code (if signed)
|
|
status: Send status
|
|
error_message: Error message if any
|
|
"""
|
|
try:
|
|
# Check if record already exists
|
|
existing = frappe.db.exists(
|
|
"E-Taxes Purchase Outbox",
|
|
{"purchase_invoice": purchase_invoice_name}
|
|
)
|
|
|
|
if existing:
|
|
# Update existing record
|
|
doc = frappe.get_doc("E-Taxes Purchase Outbox", existing)
|
|
doc.etaxes_act_id = act_id
|
|
doc.serial_number = serial_number
|
|
doc.verification_code = verification_code
|
|
doc.status = status
|
|
doc.error_message = error_message
|
|
doc.save()
|
|
else:
|
|
# Create new record
|
|
doc = frappe.new_doc("E-Taxes Purchase Outbox")
|
|
doc.purchase_invoice = purchase_invoice_name
|
|
doc.etaxes_act_id = act_id
|
|
doc.serial_number = serial_number
|
|
doc.verification_code = verification_code
|
|
doc.send_date = now_datetime()
|
|
doc.status = status
|
|
doc.error_message = error_message
|
|
doc.insert()
|
|
|
|
frappe.db.commit()
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error creating outbox record: {str(e)}\n{frappe.get_traceback()}",
|
|
"Outbox Record Error"
|
|
)
|
|
# Don't fail the whole process if outbox creation fails
|
|
|
|
|
|
@frappe.whitelist()
|
|
def retry_signing_act(purchase_invoice_name):
|
|
"""
|
|
Sign an act that was created as draft (Step 2 of 2)
|
|
|
|
This is the second step of the workflow, called when user clicks "Sign Act" button.
|
|
First step is send_purchase_invoice_to_etaxes() which creates the draft.
|
|
|
|
Args:
|
|
purchase_invoice_name: Purchase Invoice name
|
|
|
|
Returns:
|
|
dict: Success/error response with verification code
|
|
"""
|
|
try:
|
|
# Record activity
|
|
record_etaxes_activity()
|
|
|
|
# Get Purchase Invoice
|
|
doc = frappe.get_doc("Purchase Invoice", purchase_invoice_name)
|
|
|
|
# Check status
|
|
if doc.etaxes_purchase_send_status != "Created, not signed":
|
|
return {
|
|
"success": False,
|
|
"message": "Act is not in 'Created, not signed' status"
|
|
}
|
|
|
|
if not doc.etaxes_purchase_id:
|
|
return {
|
|
"success": False,
|
|
"message": "No E-Taxes act ID found"
|
|
}
|
|
|
|
# Get token
|
|
asan_login = get_default_asan_login()
|
|
if not asan_login.get("found") or not asan_login.get("main_token"):
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes authentication not found"
|
|
}
|
|
|
|
token = asan_login.get("main_token")
|
|
|
|
# Try signing
|
|
sign_result = sign_act_on_etaxes(token, doc.etaxes_purchase_id)
|
|
|
|
if sign_result.get("success"):
|
|
verification_code = sign_result.get("verification_code")
|
|
|
|
# Update Purchase Invoice
|
|
frappe.db.set_value("Purchase Invoice", purchase_invoice_name, {
|
|
"etaxes_purchase_verification_code": verification_code,
|
|
"etaxes_purchase_send_status": "Sent and Signed"
|
|
})
|
|
|
|
# Update outbox record
|
|
outbox = frappe.db.get_value(
|
|
"E-Taxes Purchase Outbox",
|
|
{"purchase_invoice": purchase_invoice_name},
|
|
"name"
|
|
)
|
|
if outbox:
|
|
frappe.db.set_value("E-Taxes Purchase Outbox", outbox, {
|
|
"verification_code": verification_code,
|
|
"status": "Sent and Signed",
|
|
"error_message": None
|
|
})
|
|
|
|
frappe.db.commit()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Purchase act signed successfully",
|
|
"verification_code": verification_code
|
|
}
|
|
else:
|
|
return {
|
|
"success": False,
|
|
"message": sign_result.get("message")
|
|
}
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error retrying signing: {str(e)}\n{frappe.get_traceback()}",
|
|
"Retry Signing Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Error: {str(e)}"
|
|
}
|
|
|
|
|
|
@frappe.whitelist()
|
|
def cancel_act_on_etaxes(purchase_invoice_name):
|
|
"""
|
|
Cancel/Remove draft act on E-Taxes system
|
|
|
|
This function removes a draft act from E-Taxes using the removeDrafts API.
|
|
Can be called manually or automatically during Purchase Invoice cancellation.
|
|
|
|
Args:
|
|
purchase_invoice_name: Name of Purchase Invoice to cancel on E-Taxes
|
|
|
|
Returns:
|
|
dict: Success/error response
|
|
"""
|
|
try:
|
|
# Record activity
|
|
record_etaxes_activity()
|
|
|
|
# Get Purchase Invoice
|
|
doc = frappe.get_doc("Purchase Invoice", purchase_invoice_name)
|
|
|
|
# Check if act has E-Taxes ID
|
|
if not doc.etaxes_purchase_id:
|
|
return {
|
|
"success": False,
|
|
"message": "This act was not sent to E-Taxes or has no E-Taxes Act ID"
|
|
}
|
|
|
|
# Get authentication token
|
|
asan_login = get_default_asan_login()
|
|
if not asan_login.get("found") or not asan_login.get("main_token"):
|
|
return {
|
|
"success": False,
|
|
"message": "E-Taxes authentication not found. Please login via ASAN."
|
|
}
|
|
|
|
token = asan_login.get("main_token")
|
|
act_id = doc.etaxes_purchase_id
|
|
|
|
# Build headers
|
|
headers = DEFAULT_HEADERS.copy()
|
|
headers["x-authorization"] = f"Bearer {token}"
|
|
|
|
# Build payload - array with act ID (API expects array of IDs)
|
|
payload = [act_id]
|
|
|
|
# Make DELETE request
|
|
response = requests.delete(
|
|
REMOVE_DRAFT_URL,
|
|
data=json.dumps(payload),
|
|
headers=headers,
|
|
timeout=30
|
|
)
|
|
|
|
# Check response
|
|
if response.status_code == 204:
|
|
# Success - act removed
|
|
# Clear E-Taxes fields in Purchase Invoice
|
|
frappe.db.set_value("Purchase Invoice", purchase_invoice_name, {
|
|
"etaxes_purchase_id": None,
|
|
"etaxes_purchase_serial_number": None,
|
|
"etaxes_purchase_verification_code": None,
|
|
"etaxes_purchase_send_status": "Not Sent"
|
|
})
|
|
|
|
# Delete outbox record if exists
|
|
outbox_name = frappe.db.get_value(
|
|
"E-Taxes Purchase Outbox",
|
|
{"purchase_invoice": purchase_invoice_name},
|
|
"name"
|
|
)
|
|
if outbox_name:
|
|
frappe.delete_doc("E-Taxes Purchase Outbox", outbox_name, force=True)
|
|
|
|
frappe.db.commit()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Purchase act successfully removed from E-Taxes"
|
|
}
|
|
|
|
elif response.status_code == 401:
|
|
return {
|
|
"success": False,
|
|
"message": "Authentication expired. Please login again."
|
|
}
|
|
|
|
elif response.status_code == 404:
|
|
# Act not found on E-Taxes - might be already deleted
|
|
# Clear fields anyway
|
|
frappe.db.set_value("Purchase Invoice", purchase_invoice_name, {
|
|
"etaxes_purchase_id": None,
|
|
"etaxes_purchase_serial_number": None,
|
|
"etaxes_purchase_verification_code": None,
|
|
"etaxes_purchase_send_status": "Not Sent"
|
|
})
|
|
frappe.db.commit()
|
|
|
|
return {
|
|
"success": True,
|
|
"message": "Act not found on E-Taxes (may be already deleted). Local fields cleared."
|
|
}
|
|
|
|
else:
|
|
# Other error - try to get Azərbaycan message
|
|
error_msg = f"E-Taxes returned status {response.status_code}"
|
|
try:
|
|
error_data = response.json()
|
|
if isinstance(error_data, dict) and "message" in error_data:
|
|
if isinstance(error_data["message"], dict) and "az" in error_data["message"]:
|
|
error_msg = error_data["message"]["az"]
|
|
elif isinstance(error_data["message"], str):
|
|
error_msg = error_data["message"]
|
|
except:
|
|
pass
|
|
|
|
return {
|
|
"success": False,
|
|
"message": f"Failed to remove act from E-Taxes: {error_msg}"
|
|
}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
frappe.log_error(
|
|
f"Error cancelling act on E-Taxes: {str(e)}",
|
|
"Cancel Act Network Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Network error: {str(e)}"
|
|
}
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Unexpected error cancelling act on E-Taxes: {str(e)}\n{frappe.get_traceback()}",
|
|
"Cancel Act Error"
|
|
)
|
|
return {
|
|
"success": False,
|
|
"message": f"Error: {str(e)}"
|
|
}
|
|
|
|
|
|
def on_delete_purchase_invoice(doc, method):
|
|
"""Deletes related E-Taxes Purchase Outbox record when Purchase Invoice is deleted or cancelled"""
|
|
try:
|
|
# Find all E-Taxes Purchase Outbox records linked to this Purchase Invoice
|
|
outbox_records = frappe.get_all(
|
|
"E-Taxes Purchase Outbox",
|
|
filters={"purchase_invoice": doc.name},
|
|
fields=["name"]
|
|
)
|
|
|
|
if outbox_records:
|
|
for record in outbox_records:
|
|
frappe.delete_doc('E-Taxes Purchase Outbox', record.name, ignore_permissions=True, force=True)
|
|
|
|
frappe.db.commit()
|
|
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
f"Error deleting E-Taxes Purchase Outbox for Purchase Invoice {doc.name}: {str(e)}\n{frappe.get_traceback()}",
|
|
"Purchase Invoice Delete Error"
|
|
)
|