You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
322 lines
11 KiB
Python
322 lines
11 KiB
Python
import frappe
|
|
from typing import List, Dict, Tuple, Optional
|
|
|
|
def check_customer_if_exists(data: dict) -> Tuple[bool, Optional[str]]:
|
|
"""Check if customer exists."""
|
|
order_header = data.get("Header", {}).get("OrderHeader", {})
|
|
trading_partner_id = order_header.get("TradingPartnerId")
|
|
if trading_partner_id:
|
|
customer = frappe.db.get_value("Customer", {"custom_trading_partner_id": trading_partner_id}, "name")
|
|
return bool(customer), customer
|
|
return False, None
|
|
|
|
def check_address_exists_or_not(address: dict) -> Optional[str]:
|
|
"""Check if address exists and return its name."""
|
|
return frappe.db.exists("Address", {"address_line1": address.get("Address1")})
|
|
|
|
def check_postal_code_exists_or_not(address_name: str) -> bool:
|
|
"""Check if postal code exists for a given address."""
|
|
return bool(frappe.db.get_value("Address", address_name, "pincode"))
|
|
|
|
def create_or_update_address(address: dict, address_type: str, customer: str) -> str:
|
|
"""Create or update an address."""
|
|
address_name = check_address_exists_or_not(address)
|
|
postal_code = address.get("PostalCode")
|
|
|
|
if address_name:
|
|
# If address exists, ensure postal code is present only for shipping address
|
|
if address_type == "Shipping" and not check_postal_code_exists_or_not(address_name):
|
|
assert postal_code, f"Postal code is required for existing shipping address: {address_name}"
|
|
# Update address if postal code is provided in the payload
|
|
frappe.db.set_value("Address", address_name, "pincode", postal_code)
|
|
frappe.db.commit()
|
|
return address_name
|
|
|
|
# If address does not exist, ensure postal code is provided only for shipping address
|
|
if address_type == "Shipping":
|
|
assert postal_code, "Postal code is required for a new shipping address."
|
|
|
|
address_doc = frappe.new_doc("Address")
|
|
country = "United States" if address.get("Country") in ["US", "USA", "United States"] else address.get("Country")
|
|
|
|
|
|
address_doc.update({
|
|
"title": address.get("AddressName"),
|
|
"address_line1": address.get("Address1"),
|
|
"address_line2": address.get("Address2"),
|
|
"city": address.get("City"),
|
|
"state": address.get("State"),
|
|
"pincode": postal_code,
|
|
"country": country,
|
|
"address_type": address_type,
|
|
})
|
|
|
|
address_doc.append("links", {"link_doctype": "Customer", "link_name": customer})
|
|
address_doc.save()
|
|
frappe.db.commit()
|
|
return address_doc.name
|
|
|
|
def create_address_contact(data: dict, customer: str) -> Tuple[str, str]:
|
|
"""Create or retrieve billing and shipping addresses for the customer."""
|
|
addresses = data.get("Header", {}).get("Address", [])
|
|
billing_address_name, shipping_address_name = None, None
|
|
|
|
for address in addresses:
|
|
address_type = address.get("AddressTypeCode")
|
|
if address_type == "BT":
|
|
billing_address_name = create_or_update_address(address, "Billing", customer)
|
|
elif address_type == "ST":
|
|
shipping_address_name = create_or_update_address(address, "Shipping", customer)
|
|
|
|
return billing_address_name, shipping_address_name
|
|
|
|
def get_sps_item_code_if_exists(order_uom: str, description: str, sku: str, setting_doc: str) -> str:
|
|
"""Get or create item code."""
|
|
if frappe.db.exists("Item", {"item_code": sku}):
|
|
return sku
|
|
|
|
# item_doc = frappe.new_doc("Item")
|
|
# item_doc.update({
|
|
# "item_code": sku,
|
|
# "item_name": description,
|
|
# "stock_uom": get_uom(order_uom) if order_uom != "EA" and order_uom != None else frappe.get_value("SPS Integration Settings", setting_doc, "default_uom"),
|
|
# "item_group": frappe.get_value("SPS Integration Settings", setting_doc, "default_item_group"),
|
|
# "custom_item_shipping_category": frappe.get_value("SPS Integration Settings", setting_doc, "shipping_category")
|
|
# })
|
|
|
|
# try:
|
|
# item_doc.save()
|
|
# frappe.db.commit()
|
|
# except:
|
|
# frappe.log_error(
|
|
# title="Item creation",
|
|
# message=f"{frappe.get_traceback()}"
|
|
# )
|
|
|
|
# return item_doc.name
|
|
|
|
def get_uom(uom: str) -> str:
|
|
"""Get or create UOM."""
|
|
if frappe.db.exists("UOM", uom):
|
|
return uom
|
|
|
|
new_uom = frappe.new_doc("UOM")
|
|
new_uom.uom_name = uom
|
|
new_uom.enabled = 1
|
|
new_uom.save()
|
|
frappe.db.commit()
|
|
return uom
|
|
|
|
def create_sps_sales_order_item_row(sku: str, qty: int, amount: float, uom: str, description: str, setting_doc: str) -> Dict:
|
|
"""Create sales order item row."""
|
|
return {
|
|
"item_code": get_sps_item_code_if_exists(uom, description, sku, setting_doc),
|
|
"qty": qty,
|
|
"rate": amount,
|
|
"custom_sku": sku,
|
|
}
|
|
|
|
|
|
def get_sps_market_place_order_ID(market_place_order_id: str, setting_doc: str) -> str:
|
|
"""Get or create Marketplace Order ID."""
|
|
|
|
# Skip if market_place_order_id is empty or None
|
|
if not market_place_order_id:
|
|
frappe.log_error(
|
|
title="Missing Marketplace Order ID",
|
|
message=f"Marketplace Order ID is missing for setting_doc: {setting_doc}. Skipping creation."
|
|
)
|
|
return ""
|
|
|
|
# Check if the Marketplace Order ID already exists
|
|
if frappe.db.exists("Marketplace Order ID", market_place_order_id):
|
|
return market_place_order_id
|
|
|
|
# Create new Marketplace Order ID
|
|
try:
|
|
doc = frappe.new_doc("Marketplace Order ID")
|
|
doc.update({
|
|
"marketplace_order_id": market_place_order_id,
|
|
"marketplace": frappe.get_value("SPS Integration Settings", setting_doc, "marketplace"),
|
|
})
|
|
doc.save()
|
|
frappe.db.commit()
|
|
return doc.name
|
|
|
|
except Exception as e:
|
|
frappe.log_error(title="Error Creating Marketplace Order ID", message=frappe.get_traceback())
|
|
return ""
|
|
|
|
|
|
|
|
def get_link_row(docname: str, link_name: str) -> Dict:
|
|
"""Creates link doctype table row"""
|
|
row = {}
|
|
row["link_doctype"] = docname
|
|
row["link_name"] = link_name
|
|
return row
|
|
|
|
|
|
# def check_and_create_contact(data, setting_doc):
|
|
# """Check if contacts exist, else create new contacts"""
|
|
|
|
# addresses = data.get('Header', {}).get('Address', [])
|
|
# contacts = None
|
|
|
|
|
|
# for address in addresses:
|
|
# if address.get("AddressTypeCode") == "BT" and address.get("Contacts"):
|
|
# contacts = address["Contacts"]
|
|
# break # Stop once billing contacts are found
|
|
|
|
# if not contacts:
|
|
# for address in addresses:
|
|
# if address.get("AddressTypeCode") == "ST" and address.get("Contacts"):
|
|
# contacts = address["Contacts"]
|
|
# break # Stop once shipping contacts are found
|
|
|
|
# if not contacts:
|
|
# return None
|
|
|
|
# # Check if customer exists
|
|
# customer_exists, customer_name = check_customer_if_exists(data)
|
|
|
|
# if customer_exists:
|
|
# # Check if the contact email already exists
|
|
# email_id = contacts[0].get("PrimaryEmail", "")
|
|
# if frappe.db.exists("Contact Email", {"email_id": email_id}):
|
|
# contact_name = frappe.get_value(
|
|
# "Contact Email", {"email_id": email_id}, "parent"
|
|
# )
|
|
# else:
|
|
# doc = frappe.new_doc("Contact")
|
|
# doc.first_name = customer_name
|
|
# if email_id:
|
|
# email_row = {
|
|
# "email_id": email_id,
|
|
# "is_primary": 1
|
|
# }
|
|
# doc.append("email_ids", email_row)
|
|
|
|
# phone_number = contacts[0].get("PrimaryPhone", "")
|
|
# if phone_number:
|
|
# num_row = {
|
|
# "phone": phone_number,
|
|
# "is_primary_phone": 1
|
|
# }
|
|
# doc.append("phone_nos", num_row)
|
|
|
|
# link_row = get_link_row("Customer", customer_name)
|
|
# doc.append("links", link_row)
|
|
|
|
# doc.save()
|
|
# frappe.db.commit()
|
|
# contact_name = doc.name
|
|
|
|
# return contact_name
|
|
|
|
# Made the fixes related to assignment error of contact_name variable
|
|
def check_and_create_contact(data, setting_doc):
|
|
"""Check if contacts exist, else create new contacts"""
|
|
|
|
addresses = data.get('Header', {}).get('Address', [])
|
|
contacts = None
|
|
|
|
for address in addresses:
|
|
if address.get("AddressTypeCode") == "BT" and address.get("Contacts"):
|
|
contacts = address["Contacts"]
|
|
break
|
|
|
|
if not contacts:
|
|
for address in addresses:
|
|
if address.get("AddressTypeCode") == "ST" and address.get("Contacts"):
|
|
contacts = address["Contacts"]
|
|
break
|
|
|
|
if not contacts or len(contacts) == 0:
|
|
return None
|
|
|
|
customer_exists, customer_name = check_customer_if_exists(data)
|
|
if not customer_exists:
|
|
return None
|
|
|
|
contact_name = None
|
|
first_contact = contacts[0]
|
|
email_id = first_contact.get("PrimaryEmail", "")
|
|
|
|
if frappe.db.exists("Contact Email", {"email_id": email_id}):
|
|
contact_name = frappe.get_value(
|
|
"Contact Email", {"email_id": email_id}, "parent"
|
|
)
|
|
else:
|
|
doc = frappe.new_doc("Contact")
|
|
doc.first_name = customer_name
|
|
|
|
if email_id:
|
|
doc.append("email_ids", {
|
|
"email_id": email_id,
|
|
"is_primary": 1
|
|
})
|
|
|
|
phone_number = first_contact.get("PrimaryPhone", "")
|
|
if phone_number:
|
|
doc.append("phone_nos", {
|
|
"phone": phone_number,
|
|
"is_primary_phone": 1
|
|
})
|
|
|
|
doc.append("links", get_link_row("Customer", customer_name))
|
|
doc.save()
|
|
frappe.db.commit()
|
|
contact_name = doc.name
|
|
|
|
return contact_name
|
|
|
|
|
|
def get_path_to_skip():
|
|
"""Function to get all paths to skip"""
|
|
return frappe.get_all("Paths to Skip", pluck="file_path")
|
|
|
|
def exclude_skipped_file_paths(paths: List[str]) -> List[str]:
|
|
"""
|
|
Excludes file paths that are marked as skipped in the 'File Skip List' doctype.
|
|
|
|
Args:
|
|
paths (List[str]): List of file paths to filter.
|
|
|
|
Returns:
|
|
List[str]: Filtered list with skipped paths removed.
|
|
"""
|
|
try:
|
|
skipped_set = set(get_path_to_skip())
|
|
filtered_paths = [path for path in paths if path not in skipped_set]
|
|
frappe.log_error(
|
|
title="Exclude skipped file paths",
|
|
message=f"Skipped Paths: {skipped_set}\nFiltered Paths: {filtered_paths}"
|
|
)
|
|
return filtered_paths
|
|
except Exception as e:
|
|
frappe.log_error(
|
|
title="Error while excluding skipped file paths",
|
|
message=frappe.get_traceback()
|
|
)
|
|
return []
|
|
|
|
|
|
def get_unsyced_sps_orders_urls(unsyced_path: list) -> list:
|
|
"""Returns list of sps unsyced sales order's urls"""
|
|
syced_sps_orders_path = get_path_to_skip()
|
|
cleaned_unsyced_path = [item["path"].split("/")[2] for item in unsyced_path]
|
|
unsyced_urls = []
|
|
count = 0
|
|
for path in cleaned_unsyced_path:
|
|
if path not in syced_sps_orders_path:
|
|
count += 1
|
|
unsyced_urls.append(path)
|
|
|
|
print(f"Total unsynced sales orders: {count}")
|
|
frappe.log_error(
|
|
title="Unsyced SPS SO URLs",
|
|
message=f"Total unsyced sales order: {count}\\n Paths:{unsyced_urls}"
|
|
)
|
|
|