feat - b2c doctype migration with related code

frappe_mpsa_payments/frappe_mpsa_payments/api/base_class.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+import frappe
+
+from ...utils.helpers import update_integration_request
+
+
+class ConnectorAbstractClass(ABC):
+    """Abstract Base class for Connector Classes"""
+
+    @abstractmethod
+    def attach(self, observer: Observer) -> None:
+        """Attach Observers
+
+        Args:
+            observer (Observer): The observer to attach
+        """
+
+    @abstractmethod
+    def notify(self) -> None:
+        """Notify all registered observers"""
+
+
+class ConnectorBaseClass(ConnectorAbstractClass):
+    """Base class for Connector Classes"""
+
+    def __init__(self) -> None:
+        self.error: str | Exception | None = None
+        self.integration_request: str | None = None
+
+        self._observers: list[Observer] = []
+
+    def attach(self, observer: Observer) -> None:
+        """Attach Observers
+
+        Args:
+            observer (Observer): The observer to attach
+        """
+        self._observers.append(observer)
+
+    def notify(self) -> None:
+        """Notify all registered observers"""
+        for observer in self._observers:
+            observer.update(self)
+
+
+class Observer(ABC):
+    """Observer Abstract Class"""
+
+    @abstractmethod
+    def update(self, notifier: ConnectorBaseClass) -> None:
+        """Method that reacts to specific state in the notifier when called
+
+        Args:
+            notifier (ConnectorBaseClass): The Notifier (calling class)
+        """
+
+
+class ErrorObserver(Observer):
+    """Error Observer concrete class"""
+
+    def update(self, notifier: ConnectorBaseClass) -> None:
+        if notifier.error:
+            frappe.log_error(
+                title="HTTPError",
+                message=notifier.error,
+            )
+            update_integration_request(
+                notifier.integration_request,
+                status="Failed",
+                error=str(notifier.error),
+            )
+            frappe.throw(
+                str(notifier.error),
+                frappe.DataError,
+                title="HTTPError",
+            )

frappe_mpsa_payments/frappe_mpsa_payments/api/m_pesa_api.py

@@ -147,8 +147,6 @@ def submit_mpesa_payment(mpesa_payment, customer):
 def submit_instant_mpesa_payment():
     mpesa_payment = frappe.form_dict.get("mpesa_payment")
     customer = frappe.form_dict.get("customer")
-    # pos_profile = frappe.form_dict.get("pos_profile")
-    # mode_of_payment = get_payment_method(pos_profile)
 
     try:
         process_mpesa_payment(mpesa_payment, customer, submit_payment=False)
@@ -160,7 +158,6 @@ def process_mpesa_payment(mpesa_payment, customer, submit_payment=False):
     try:
         doc = frappe.get_doc("Mpesa C2B Payment Register", mpesa_payment)
         doc.customer = customer
-        # doc.mode_of_payment = mode_of_payment
         #TODO: after testing, mode of payment
         doc.mode_of_payment = get_mode_of_payment(doc)
         doc.submit_payment=submit_payment

frappe_mpsa_payments/frappe_mpsa_payments/api/mpsa_b2c.py

@@ -0,0 +1,145 @@
+from datetime import datetime, timedelta
+from enum import Enum
+from urllib.parse import urlparse
+
+import requests
+from requests.auth import HTTPBasicAuth
+import traceback
+
+import frappe
+from frappe.integrations.utils import create_request_log
+from frappe.utils import get_request_site_address
+from frappe.utils.password import get_decrypted_password
+import json
+from ...utils.definitions import B2CRequestDefinition
+from .base_class import ConnectorBaseClass, ErrorObserver
+from ...utils.helpers import update_integration_request
+
+
+class URLS(Enum):
+    """URLs Constant Exporting class"""
+
+    SANDBOX = "https://sandbox.safaricom.co.ke"
+    PRODUCTION = "https://api.safaricom.co.ke"
+
+
+class MpesaB2CConnector(ConnectorBaseClass):
+    """MPesa B2C Connector Class"""
+
+    def __init__(self, env="sandbox", app_key=None, app_secret=None):
+        """Setup configuration for Mpesa connector and generate new access token."""
+        super().__init__()
+
+        self.authentication_token = None
+        self.expires_in = None
+
+        self.env = env
+        self.app_key = app_key
+        self.app_secret = app_secret
+
+        self.base_url = URLS.SANDBOX.value if env == "sandbox" else URLS.PRODUCTION.value
+
+        self.attach(ErrorObserver())
+
+    def authenticate(self) -> str:
+        """Fetch a new access token from MPesa API."""
+        authenticate_uri = "/oauth/v1/generate?grant_type=client_credentials"
+        authenticate_url = f"{self.base_url}{authenticate_uri}"
+
+        # Use the app credentials to fetch the token
+        response = requests.get(
+            authenticate_url,
+            auth=HTTPBasicAuth(self.app_key, self.app_secret),
+            timeout=60,
+        )
+
+        # Handle API response
+        if response.status_code == 200:
+            data = response.json()
+            self.authentication_token = data["access_token"]
+            self.expires_in = datetime.now() + timedelta(seconds=int(data["expires_in"]))
+            return self.authentication_token
+        else:
+            error_msg = f"Failed to authenticate: {response.status_code} - {response.text}"
+            frappe.throw(error_msg)
+
+    def make_b2c_payment_request(self, request_data: B2CRequestDefinition) -> dict:
+        """Make a B2C Payment Request."""
+        # Ensure token is valid or fetch a new one
+        if not self.authentication_token or datetime.now() >= self.expires_in:
+            self.app_key = request_data.ConsumerKey
+            self.app_secret = request_data.ConsumerSecret
+            self.authentication_token = self.authenticate()
+        saf_url = f"{self.base_url}/mpesa/b2c/v3/paymentrequest"
+        callback_url = (
+            f"https://{urlparse(get_request_site_address(full_address=True)).hostname}"
+            "/api/method/frappe_mpsa_payments.frappe_mpsa_payments.api.mpsa_b2c.results_callback_url"
+        )
+
+        payload_dict = {
+        **request_data.to_dict(), 
+        "QueueTimeOutURL": callback_url,
+        "ResultURL": callback_url,
+    }
+
+        headers = {
+            "Authorization": f"Bearer {self.authentication_token}",
+            "Content-Type": "application/json",
+        }
+
+        # Log the integration request
+        integration_request_name = create_request_log(
+            url=saf_url,
+            is_remote_request=1,
+            data=payload_dict,
+            service_name="Mpesa B2C",
+            name=request_data.OriginatorConversationID,
+            error=None,
+            request_headers=headers,
+        ).name
+        try:
+            response = requests.post(
+                saf_url,
+                json=payload_dict,
+                headers=headers,
+                timeout=60,
+            )
+            response.raise_for_status()
+            return response.json()
+        except requests.HTTPError as e:
+            self.error = e
+            # self.notify()
+            error_msg = f"HTTP error during B2C request: {e.response.status_code} - {e.response.text}"
+            frappe.log_error(error_msg, "Error")
+            frappe.throw(error_msg)
+        except Exception as e:
+            error_msg = (
+                f"Unexpected error: {str(e)}\n"
+                f"Traceback: {traceback.format_exc()}"
+            )
+            frappe.log_error(error_msg, "Mpesa B2C UnexpectedError")
+            frappe.throw("An unexpected error occurred during the B2C request. Please check the error log.")
+
+
+@frappe.whitelist(allow_guest=True)
+def results_callback_url(**kwargs) -> dict:
+    """Handle the callback from MPesa API."""
+    result = frappe._dict(kwargs["Result"])
+
+    result_json = json.dumps(result)
+    if result.get("ResultCode") != 0:
+        update_integration_request(
+            result.get("OriginatorConversationID"),
+            "Failed",
+            output=result_json,
+            error=result.get("ResultDesc"),
+        )
+        frappe.log_error(f"B2C Request failed: {result.ResultDesc}", "Mpesa B2C Error")
+    else:
+        print(str(result))
+        update_integration_request(
+            result.get("OriginatorConversationID"),
+            "Completed",
+            output=result_json,
+        )
+    return "Success"

frappe_mpsa_payments/frappe_mpsa_payments/doctype/__init__.py

@@ -0,0 +1,6 @@
+
+import frappe
+from frappe.utils import logger
+
+logger.set_log_level("DEBUG")
+app_logger = frappe.logger("api", allow_site=True, file_count=50)

frappe_mpsa_payments/frappe_mpsa_payments/doctype/custom_exceptions.py

@@ -0,0 +1,40 @@
+"""Custom Exceptions and Errors raised by modules in the MPesa B2C application"""
+
+
+class InvalidReceiverMobileNumberError(Exception):
+    """Raised when receiver's mobile number fails validation"""
+
+
+class InsufficientPaymentAmountError(Exception):
+    """Raised when the payment amount is less than the required KShs. 10"""
+
+
+class IncorrectStatusError(Exception):
+    """Raised when status is Errored but no errod description or error code has been supplied"""
+
+
+class InvalidTokenExpiryTimeError(Exception):
+    """
+    Raised when the access token's expiry time is earlier
+    or the same as the access token's fetch time.
+    It should always be 1 hour after the fetch time.
+    """
+
+
+class InvalidURLError(Exception):
+    """Raised when URLs fail validation"""
+
+
+class InvalidAuthenticationCertificateFileError(Exception):
+    """Raised when an invalid certificate file, i.e. not a .cer or .pem, is uploaded"""
+
+
+class UnExistentB2CPaymentRecordError(Exception):
+    """Raised when referencing a B2C Payment that does not exist"""
+
+
+class InformationMismatchError(Exception):
+    """
+    Raised when there's a mismatch in any of the B2C Payment's records
+    and the corresponding B2C Payments Transaction's records
+    """

frappe_mpsa_payments/frappe_mpsa_payments/doctype/mpesa_b2c_employee_payment_item/mpesa_b2c_employee_payment_item.json

@@ -0,0 +1,92 @@
+{
+ "actions": [],
+ "allow_rename": 1,
+ "autoname": "hash",
+ "creation": "2024-05-10 10:19:22.013333",
+ "doctype": "DocType",
+ "editable_grid": 1,
+ "engine": "InnoDB",
+ "field_order": [
+  "originator_conversation_id",
+  "section_break_erqv",
+  "reference_doctype",
+  "record",
+  "record_amount",
+  "column_break_guwr",
+  "receiver_name",
+  "partyb",
+  "amount"
+ ],
+ "fields": [
+  {
+   "fieldname": "originator_conversation_id",
+   "fieldtype": "Data",
+   "label": "Originator Conversation ID",
+   "no_copy": 1,
+   "read_only": 1
+  },
+  {
+   "fieldname": "section_break_erqv",
+   "fieldtype": "Section Break"
+  },
+  {
+   "fieldname": "reference_doctype",
+   "fieldtype": "Link",
+   "in_list_view": 1,
+   "label": "Reference doctype",
+   "options": "DocType"
+  },
+  {
+   "fieldname": "record",
+   "fieldtype": "Dynamic Link",
+   "in_list_view": 1,
+   "label": "Record",
+   "options": "reference_doctype"
+  },
+  {
+   "fieldname": "record_amount",
+   "fieldtype": "Float",
+   "in_list_view": 1,
+   "label": "Record Amount",
+   "precision": "2"
+  },
+  {
+   "fieldname": "column_break_guwr",
+   "fieldtype": "Column Break"
+  },
+  {
+   "fieldname": "receiver_name",
+   "fieldtype": "Data",
+   "in_list_view": 1,
+   "label": "Receiver Name"
+  },
+  {
+   "fieldname": "partyb",
+   "fieldtype": "Data",
+   "in_list_view": 1,
+   "label": "Contacts"
+  },
+  {
+   "fieldname": "amount",
+   "fieldtype": "Float",
+   "in_list_view": 1,
+   "label": "Payment Amount",
+   "non_negative": 1,
+   "precision": "2",
+   "reqd": 1
+  }
+ ],
+ "index_web_pages_for_search": 1,
+ "istable": 1,
+ "links": [],
+ "modified": "2024-11-20 09:22:40.196307",
+ "modified_by": "Administrator",
+ "module": "Frappe Mpsa Payments",
+ "name": "MPesa B2C Employee Payment Item",
+ "naming_rule": "Random",
+ "owner": "Administrator",
+ "permissions": [],
+ "sort_field": "modified",
+ "sort_order": "DESC",
+ "states": []
+}