feat: Add integration to shiprocket, delhivery, aramex, envia, shippo

erpnext_shipping/erpnext_shipping/aramex/aramex.py

@@ -0,0 +1,146 @@
+import json
+
+import frappe
+import requests
+from frappe import _
+from requests.exceptions import HTTPError
+
+from erpnext_shipping.erpnext_shipping.aramex.constants import ARAMEX_BASE_URL, ARAMEX_VERSION
+from erpnext_shipping.erpnext_shipping.utils import (
+	get_shipping_provider,
+	handle_shipping_error,
+	save_shipment_label,
+)
+
+ARAMEX_PROVIDER = "Aramex"
+
+
+class AramexUtils:
+	def __init__(self, company):
+		doc = get_shipping_provider(company, ARAMEX_PROVIDER)
+		self.name = doc["name"]
+		settings = frappe.get_doc("Shipping Provider", self.name)
+		self.service_provider = settings.service_provider
+		self.company = settings.company
+		self.user = settings.user_key
+		self.password = settings.get_password("user_secret")
+		self.account = settings.account_number
+		self.account_pin = settings.get_password("account_pin")
+		self.account_entity = settings.account_entity
+		self.account_country_code = settings.account_country_entity
+		self.version = settings.version
+		self.source = settings.source
+		self.is_preferred = settings.get("is_preferred")
+
+	def make_request(self, method, endpoint, payload=None):
+		url = f"{ARAMEX_BASE_URL}/{ARAMEX_VERSION}/{endpoint}"
+		headers = {"Content-Type": "application/json", "Accept": "application/json"}
+
+		try:
+			response = requests.request(
+				method, url, headers=headers, data=json.dumps(payload) if payload else None
+			)
+			response.raise_for_status()
+			return response.json()
+		except HTTPError as http_err:
+			handle_shipping_error(
+				self.name, ARAMEX_PROVIDER, f"HTTP error occurred on {url} Aramex", http_err
+			)
+		except Exception as err:
+			handle_shipping_error(self.name, ARAMEX_PROVIDER, f"An error occurred on {url} Aramex", err)
+		return None
+
+	def get_available_services(self, delivery_address, pickup_address, parcels, weight):
+		payload = self.get_rate_payload(delivery_address, pickup_address, parcels, weight)
+		response_data = self.make_request(
+			"POST", "RateCalculator/Service_1_0.svc/json/CalculateRate", payload
+		)
+		available_services = []
+
+		if response_data and not response_data.get("HasErrors"):
+			rate_details = response_data.get("RateCalculatorResponse", [])
+			for rate in rate_details:
+				available_services.append(self.get_service_dict(rate.get("TotalAmount")))
+
+		return available_services
+
+	def create_shipment(self, shipment_details):
+		payload = self.get_shipment_payload(shipment_details)
+		response_data = self.make_request(
+			"POST", "shippingapi.v2/shipping/service_1_0.svc/json/CreateShipments", payload
+		)
+
+		if response_data:
+			shipments = response_data.get("Shipments", [{}])[0]
+			save_shipment_label(response_data, shipments.get("LabelURL", {}))
+			return {
+				"service_provider": "Aramex",
+				"shipment_id": shipments.get("ID", ""),
+				"carrier": "Aramex",
+				"carrier_service": "Priority Express",
+				"shipment_amount": shipments.get("ShipmentDetails", {}).get("Charges", {}).get("Value", 0),
+				"awb_number": shipments.get("ID", ""),
+			}
+
+	def get_tracking_data(self, shipment_id):
+		payload = {
+			"ClientInfo": self.get_client_info(),
+			"GetLastTrackingUpdateOnly": True,
+			"Shipments": [shipment_id],
+		}
+		response_data = self.make_request("GET", "Tracking/Service_1_0.svc/json/TrackShipments", payload)
+
+		if response_data:
+			tracking_results = response_data.get("TrackingResults", [{}])[0]
+			return {
+				"awb_number": tracking_results.get("WaybillNumber", ""),
+				"tracking_status": tracking_results.get("UpdateDescription", ""),
+				"tracking_status_info": tracking_results.get("Comments", ""),
+				"tracking_url": "",
+			}
+
+	def get_service_dict(self, service):
+		return {
+			"service_provider": "Aramex",
+			"carrier": "Aramex",
+			"service_name": "Express",
+			"total_price": service[0].get("Value", 0.0),
+			"currency": "OMR",
+			"service_id": "Express",
+			"is_preferred": self.is_preferred,
+		}
+
+	def get_rate_payload(self, delivery_address, pickup_address, parcels, weight):
+		return {
+			"ClientInfo": self.get_client_info(),
+			"DestinationAddress": self.get_address_payload(delivery_address),
+			"OriginAddress": self.get_address_payload(pickup_address),
+			"PreferredCurrencyCode": "OMR",
+			"ShipmentDetails": {
+				"ActualWeight": {"Value": weight, "Unit": "KG"},
+				"ProductGroup": "DOM",
+				"ProductType": "OND",
+			},
+			"Transaction": {"Reference1": "", "Reference2": ""},
+		}
+
+	def get_shipment_payload(self, shipment_details):
+		return {"ClientInfo": self.get_client_info(), "ShipmentDetails": shipment_details}
+
+	def get_address_payload(self, address):
+		return {
+			"City": address.city,
+			"PostCode": address.pincode,
+			"CountryCode": address.country_code.upper(),
+		}
+
+	def get_client_info(self):
+		return {
+			"UserName": self.user,
+			"Password": self.password,
+			"AccountNumber": self.account,
+			"AccountPin": self.account_pin,
+			"AccountEntity": self.account_entity,
+			"AccountCountryCode": self.account_country_code,
+			"Version": self.version,
+		}

erpnext_shipping/erpnext_shipping/aramex/constants.py

@@ -0,0 +1,2 @@
+ARAMEX_BASE_URL = "https://ws.sbx.aramex.net"
+ARAMEX_VERSION = "ShippingAPI.V2"

erpnext_shipping/erpnext_shipping/constants.py

@@ -0,0 +1,39 @@
+state_codes = {
+	"tamilnadu": "TN",
+	"andhrapradesh": "AP",
+	"arunachalpradesh": "AR",
+	"assam": "AS",
+	"bihar": "BR",
+	"chhattisgarh": "CG",
+	"goa": "GA",
+	"gujarat": "GJ",
+	"haryana": "HR",
+	"himachalpradesh": "HP",
+	"jharkhand": "JH",
+	"karnataka": "KA",
+	"kerala": "KL",
+	"madhyapradesh": "MP",
+	"maharashtra": "MH",
+	"manipur": "MN",
+	"meghalaya": "ML",
+	"mizoram": "MZ",
+	"nagaland": "NL",
+	"odisha": "OD",
+	"punjab": "PB",
+	"rajasthan": "RJ",
+	"sikkim": "SK",
+	"telangana": "TG",
+	"tripura": "TR",
+	"uttarpradesh": "UP",
+	"uttarakhand": "UK",
+	"westbengal": "WB",
+	"andamanandnicobar": "AN",
+	"chandigarh": "CH",
+	"dadraandnagarhaveli": "DN",
+	"damananddiu": "DD",
+	"lakshadweep": "LD",
+	"delhi": "DL",
+	"puducherry": "PY",
+	"ladakh": "LA",
+	"jammukashmir": "JK",
+}

erpnext_shipping/erpnext_shipping/delhivery_one/constants.py

@@ -0,0 +1 @@
+DELHIVERY_API_BASE_URL = "https://track.delhivery.com"

erpnext_shipping/erpnext_shipping/delhivery_one/delhivery_one.py

@@ -0,0 +1,176 @@
+import json
+
+import frappe
+import requests
+from frappe import _
+from requests.exceptions import HTTPError
+
+from erpnext_shipping.erpnext_shipping.delhivery_one.constants import DELHIVERY_API_BASE_URL
+from erpnext_shipping.erpnext_shipping.utils import get_shipping_provider, handle_shipping_error
+
+DELHIVERY_PROVIDER = "Delhiveryone"
+
+
+class DelhiveryOneUtils:
+	def __init__(self, company: str):
+		settings = get_shipping_provider(company, DELHIVERY_PROVIDER)
+		self.company = settings.get("company")
+		self.api_key = frappe.get_doc("Shipping Provider", settings.get("name")).get_password("api_key")
+		self.enable = settings.get("enable")
+		self.name = settings.get("name")
+		self.is_preferred = settings.get("is_preferred")
+
+	def _make_request(self, method: str, endpoint: str, params=None, data=None) -> dict:
+		url = f"{DELHIVERY_API_BASE_URL}{endpoint}"
+		headers = {"Content-Type": "application/json", "Authorization": f"Token {self.api_key}"}
+
+		try:
+			response = requests.request(method, url, headers=headers, params=params, data=data)
+			response.raise_for_status()
+			return response.json()
+		except HTTPError as http_err:
+			handle_shipping_error(
+				self.name, DELHIVERY_PROVIDER, f"HTTP error in {method} request to {endpoint}", str(http_err)
+			)
+		except Exception as err:
+			handle_shipping_error(
+				self.name, DELHIVERY_PROVIDER, f"Exception in {method} request to {endpoint}", str(err)
+			)
+		return {}
+
+	def get_availability(self, pickup_code: str) -> bool:
+		params = {"token": self.api_key, "filter_codes": pickup_code}
+		response = self._make_request("GET", "/c/api/pin-codes/json/", params=params)
+		return bool(response.get("delivery_codes"))
+
+	def get_available_services(self, delivery_address, pickup_address, weight):
+		if not self.enable or not self.api_key:
+			return []
+
+		if not self.get_availability(pickup_address.pincode):
+			return []
+
+		services = []
+		for mode in ["S", "E"]:
+			params = {
+				"md": mode,
+				"ss": "Delivered",
+				"d_pin": delivery_address.pincode,
+				"o_pin": pickup_address.pincode,
+				"cgm": int(weight) * 1000,
+			}
+			response = self._make_request("GET", "/api/kinko/v1/invoice/charges/.json", params=params)
+			services.append({mode: response})
+
+		return [self.get_service_dict(service) for service in services]
+
+	def create_shipment(self, **kwargs):
+		pickup_phone = frappe.db.get_value("Address", kwargs["pickup_address_name"], "phone")
+
+		shipments = [
+			self.get_parcel_dict(
+				kwargs["shipment"],
+				parcel,
+				i,
+				kwargs["delivery_address"],
+				kwargs["delivery_contact"],
+				kwargs["service_info"],
+			)
+			for i, parcel in enumerate(json.loads(kwargs["shipment_parcel"]), start=1)
+		]
+
+		payload = {
+			"data": {
+				"pickup_location": {
+					"add": kwargs["pickup_address"].address_title,
+					"country": kwargs["pickup_address"].country_code.upper(),
+					"pin": kwargs["pickup_address"].pincode,
+					"phone": pickup_phone,
+					"city": kwargs["pickup_address"].city,
+					"name": kwargs["pickup_address_name"],
+				},
+				"shipments": shipments,
+			}
+		}
+		json_data = json.dumps(payload["data"])
+		formatted_payload = f"format=json&data={json_data}"
+		response = self._make_request("POST", "/api/cmu/create.json", data=formatted_payload)
+		if response and response.get("success"):
+			awb_numbers = [pkg["waybill"] for pkg in response.get("packages", [])]
+			return {
+				"service_provider": "Delhiveryone",
+				"shipment_id": ", ".join(awb_numbers),
+				"carrier": "Delhiveryone",
+				"carrier_service": kwargs["service_info"].get("service_name"),
+				"shipment_amount": response.get("cod_amount", 0),
+				"awb_number": ", ".join(awb_numbers),
+			}
+
+		return {}
+
+	def get_label(self, shipment_id):
+		shipment_ids = shipment_id.split(", ")
+		label_urls = []
+		for ship_id in shipment_ids:
+			params = {"wbns": ship_id, "pdf": "true"}
+			response = self._make_request("GET", "/api/p/packing_slip", params=params)
+			if response and response.get("packages"):
+				label_urls.append(response["packages"][0]["pdf_download_link"])
+		return label_urls
+
+	def get_tracking_data(self, shipment_id):
+		shipment_ids = shipment_id.split(", ")
+		awb_numbers, tracking_statuses, tracking_info = [], [], []
+		for ship_id in shipment_ids:
+			params = {"token": self.api_key, "waybill": ship_id}
+			response = self._make_request("GET", "/api/v1/packages/json", params=params)
+			if response.get("ShipmentData"):
+				shipment = response["ShipmentData"][0]["Shipment"]
+				awb_numbers.append(shipment.get("AWB", "N/A"))
+				tracking_statuses.append(shipment["Status"]["Status"])
+				tracking_info.append(shipment["Status"]["Instructions"])
+		return {
+			"awb_number": ", ".join(awb_numbers),
+			"tracking_status": ", ".join(tracking_statuses),
+			"tracking_status_info": ", ".join(tracking_info),
+			"tracking_url": "",
+		}
+
+	def get_service_dict(self, service):
+		service_type = next(iter(service))
+		service_details = service[service_type][0]
+		return frappe._dict(
+			service_provider="Delhiveryone",
+			carrier="Delhiveryone",
+			service_name="Surface" if service_type == "S" else "Express",
+			total_price=service_details.get("total_amount", 0.0),
+			currency="INR",
+			is_preferred=self.is_preferred,
+			service_id=service_type,
+		)
+
+	def get_parcel_dict(
+		self,
+		shipment,
+		parcel,
+		index,
+		delivery_address,
+		delivery_contact,
+		service_info,
+	):
+		return {
+			"name": f"{delivery_contact.first_name} {delivery_contact.last_name}",
+			"country": delivery_address.country,
+			"city": delivery_address.city,
+			"add": delivery_address.address_line1,
+			"pin": delivery_address.pincode,
+			"phone": delivery_contact.phone,
+			"payment_mode": "Prepaid",
+			"cod_amount": 0,
+			"quantity": parcel.get("count"),
+			"order": f"{shipment}-{index}",
+			"shipment_width": parcel.get("width"),
+			"shipment_height": parcel.get("height"),
+			"weight": parcel.get("weight"),
+			"shipping_mode": service_info.get("service_name"),
+		}