#24319 Add business info in the business query from COLIN-API in Enti…

…ties (#1610)

* add business information in the business query from COLIN-API in Entities

* update version

* consider retrieve attornys for an expo company

* query nature of business info

* business info

* update colin-api response

api/config.py

@@ -79,10 +79,6 @@ class Config(object):
 
     JWT_OIDC_AUDIENCE = os.getenv('JWT_OIDC_AUDIENCE')
 
-    COLIN_SVC_AUTH_URL = os.getenv('KEYCLOAK_AUTH_TOKEN_URL', '')
-    COLIN_SVC_AUTH_CLIENT_ID = os.getenv('NAME_REQUEST_SERVICE_ACCOUNT_CLIENT_ID', '')
-    COLIN_SVC_CLIENT_SECRET = os.getenv('NAME_REQUEST_SERVICE_ACCOUNT_CLIENT_SECRET', '')
-
     ENTITY_SVC_AUTH_URL = os.getenv('KEYCLOAK_AUTH_TOKEN_URL', '')
     ENTITY_SERVICE_ACCOUNT_CLIENT_ID = os.getenv('ENTITY_SERVICE_ACCOUNT_CLIENT_ID', '')
     ENTITY_SERVICE_ACCOUNT_CLIENT_SECRET = os.getenv('ENTITY_SERVICE_ACCOUNT_CLIENT_SECRET', '')

api/namex/VERSION.py

@@ -1 +1 @@
-__version__ = '1.2.9'
+__version__ = '1.2.10'

api/namex/exceptions/__init__.py

@@ -27,4 +27,33 @@ def wrapped(*args, **kwargs):
                     logger.exception(e)
                 raise
         return wrapped
-    return decorated
+    return decorated
+
+class EntityServiceException(Exception):
+    """get business info"""
+
+    def __init__(self, wrapped_err=None, message="Entity API exception.", status_code=500):
+        self.err = wrapped_err
+        self.colin_error_code = None
+        self.status_code = status_code
+
+        if wrapped_err and hasattr(wrapped_err, 'status'):
+            # Map HTTP status if the wrapped error has an HTTP status code
+            self.status_code = wrapped_err.status if wrapped_err.status else status_code
+
+        if wrapped_err and hasattr(wrapped_err, 'error_code'):
+            # Map COLIN error code if the wrapped error has a COLIN error code
+            self.error_code = int(wrapped_err.error_code)
+
+        if wrapped_err and hasattr(wrapped_err, 'internal_error_code'):
+            # Map COLIN error code if the wrapped error has a COLIN error code
+            self.colin_error_code = int(wrapped_err.internal_error_code)
+
+        if self.colin_error_code is not None:
+            self.message = message if message else str(self.colin_error_code) + ': ' + wrapped_err['internal_error_message']
+        elif wrapped_err:
+            self.message = '{msg}\r\n\r\n{desc}'.format(msg=message, desc=str(wrapped_err))
+        else:
+            self.message = message
+
+        super().__init__(self.message)

api/namex/models/request.py

@@ -644,6 +644,15 @@ def include_designations_in_name(cls, special_characters_name, any_designation_l
                     r'\W*({0})?\W*({1})?$'.format(any_designation_alternators, end_designation_alternators)
         return full_name
 
+    @classmethod
+    def get_nature_business_info(cls, corp_num):
+        """
+        Retrieve the nature_business_info for the given corp_num.
+        """
+        subquery = db.session.query(Name.nrId).filter(Name.corpNum == corp_num).subquery()
+        result = db.session.query(Request.natureBusinessInfo).filter(Request.id == subquery).first()
+        return result[0] if result else None
+
 
 @event.listens_for(Request, 'after_insert')
 @event.listens_for(Request, 'after_update')

api/namex/resources/colin.py

@@ -1,12 +1,16 @@
+from http import HTTPStatus
 import json
 
 import requests
 from flask import current_app, jsonify, make_response
 from flask_restx import Namespace, Resource, cors
+from namex.exceptions import EntityServiceException
+from namex.models.request import Request
 
 from namex.utils.api_resource import handle_exception
-from namex.utils.auth import MSG_CLIENT_CREDENTIALS_REQ_FAILED, cors_preflight, get_client_credentials
+from namex.utils.auth import MSG_CLIENT_CREDENTIALS_REQ_FAILED, cors_preflight
 from namex.utils.logging import setup_logging
+from .utils import EntityUtils
 
 
 setup_logging()  # Important to do this first
@@ -17,91 +21,186 @@
 MSG_COULD_NOT_FIND_CORP = 'Error: Could not find corporation details'
 
 
-class ColinServiceException(Exception):
-    def __init__(self, wrapped_err=None, message="COLIN API exception.", status_code=500):
-        self.err = wrapped_err
-        self.colin_error_code = None
-        self.status_code = status_code
-
-        if wrapped_err and hasattr(wrapped_err, 'status'):
-            # Map HTTP status if the wrapped error has an HTTP status code
-            self.status_code = wrapped_err.status if wrapped_err.status else status_code
-
-        if wrapped_err and hasattr(wrapped_err, 'error_code'):
-            # Map COLIN error code if the wrapped error has a COLIN error code
-            self.error_code = int(wrapped_err.error_code)
-
-        if wrapped_err and hasattr(wrapped_err, 'internal_error_code'):
-            # Map COLIN error code if the wrapped error has a COLIN error code
-            self.colin_error_code = int(wrapped_err.internal_error_code)
-
-        if self.colin_error_code is not None:
-            self.message = message if message else str(self.colin_error_code) + ': ' + wrapped_err['internal_error_message']
-        elif wrapped_err:
-            self.message = '{msg}\r\n\r\n{desc}'.format(msg=message, desc=str(wrapped_err))
-        else:
-            self.message = message
-
-        super().__init__(self.message)
-
-
 # Register a local namespace for the NR reserve
 colin_api = Namespace('colin', description='COLIN API')
 
-
 @cors_preflight('GET')
 @colin_api.route('/<string:corp_num>', strict_slashes=False, methods=['GET', 'OPTIONS'])
 @colin_api.doc(params={
     'corp_num': 'Incorporation Number - This field is required'
 })
 class ColinApi(Resource):
     def get(self, corp_num):
-        colin_url = f'{current_app.config.get("COLIN_SVC_URL")}/businesses/{corp_num}/public'
-        response = _init(colin_url)
-
-        business_info = response.get('business', {})
-
-        response_dict = {
-            'identifier': business_info.get('identifier', corp_num),
-            'legalName': business_info.get('legalName'),
-            'legalType': business_info.get('legalType'),
-            'corpState': business_info.get('corpStateClass'),
-            'status': business_info.get('status'),
-            'jurisdiction': business_info.get('jurisdiction'),
-            'homeIdentifier': business_info.get('homeJurisdictionNumber')
-        }
-
-        return make_response(jsonify(response_dict), 200)
-
-
-def _init(colin_url):
-    try:
-        authenticated, token = get_client_credentials(
-            current_app.config.get('COLIN_SVC_AUTH_URL'),
-            current_app.config.get('COLIN_SVC_AUTH_CLIENT_ID'),
-            current_app.config.get('COLIN_SVC_CLIENT_SECRET'),
-        )
-        if not authenticated:
-            raise ColinServiceException(message=MSG_CLIENT_CREDENTIALS_REQ_FAILED)
-
-        # Get the profile
-        headers = {
-            'Authorization': 'Bearer ' + token
-        }
-
-        response = requests.get(
-            colin_url,
-            headers=headers
-        )
-
-        if response.status_code != 200:
-            current_app.logger.debug(
-                'Error fetching data from URL: %s, Response: %d - %s',
-                colin_url, response.status_code, response.text
-            )
-            raise ColinServiceException(message="Failed to fetch data", status_code=response.status_code)
-        return response.json()
-    except ColinServiceException as err:
-        return handle_exception(err, err.message, err.status_code)
-    except Exception as err:
-        return handle_exception(err, 'Internal Server Error', 500)
+        """
+        Handles GET requests to retrieve business, office, and parties data for a given corporation number.
+
+        Args:
+            corp_num (str): The corporation number.
+
+        Returns:
+            Response: A Flask response object with the business data in JSON format.
+        """
+        try:
+            # Build the Colin public endpoint URL
+            colin_url = f'{current_app.config.get("COLIN_SVC_URL")}/businesses/{corp_num}/public'
+            # Fetch the business information
+            response = EntityUtils.make_authenticated_request(colin_url)
+
+            if response.status_code != HTTPStatus.OK:
+                error_message = f"Error retrieving {corp_num}: "
+
+                # If response is JSON, modify the JSON object
+                response_body = response.json()
+                if isinstance(response_body, dict):
+                    response_body['error'] = error_message + response_body.get('message', 'Unknown error')
+                else:
+                    # Handle cases where the JSON response is not a dictionary
+                    response_body = {'error': error_message + str(response_body)}
+                return make_response(jsonify(response_body), response.status_code)
+
+            business_info = response.json().get('business', {})
+            legal_type = business_info.get('legalType')
+
+            # Fetch office and parties information
+            office_info = self._get_office_data(corp_num, legal_type)
+            parties_info = self._get_parties_data(corp_num, legal_type)
+
+            # Fetch nature business info
+            nature_business_info = self._get_nature_of_business(corp_num)
+
+            # Construct the response dictionary
+            response_dict = {
+                'identifier': business_info.get('identifier', corp_num),
+                'incorporated': business_info.get('foundingDate'),
+                'legalName': business_info.get('legalName'),
+                'legalType': legal_type,
+                'corpState': business_info.get('corpStateClass'),
+                'status': business_info.get('status'),
+                'jurisdiction': business_info.get('jurisdiction'),
+                'homeIdentifier': business_info.get('homeJurisdictionNumber'),
+                'registered office delivery address': office_info.get('registered', []),
+                'records office delivery address': office_info.get('records', []),
+                'directors': parties_info.get('directorNames'),
+                'attorney names': parties_info.get('attorneyNames'),
+                'nature of business': nature_business_info,
+            }
+
+            # Return the response as JSON
+            return make_response(jsonify(response_dict), 200)
+
+        except ValueError as ve:
+            current_app.logger.error(f"ValueError: {ve}")
+            return handle_exception(ve, 'Invalid Request', 400)
+
+        except Exception as err:
+            current_app.logger.error(f"Unexpected error: {err}")
+            return handle_exception(err, 'Internal Server Error', 500)
+
+
+    def _get_office_data(self, corp_num, legal_type):
+        """
+        Process the office data for a given corporation number.
+
+        Args:
+            corp_num (str): The corporation number.
+            legal_type (str): the legal type
+
+        Returns:
+            dict: A dictionary containing address components for registered and records offices.
+                If an office type is missing, its value will be an empty list.
+        """
+        try:
+            # Build the Colin service endpoint
+            colin_service_url = f'{current_app.config.get("COLIN_SVC_URL")}'
+            office_endpoint = f'{colin_service_url}/businesses/{legal_type}/{corp_num}/office'
+
+            # Make the authenticated request
+            response = EntityUtils.make_authenticated_request(office_endpoint)
+            if response.status_code != HTTPStatus.OK:
+                office_data = {}
+            else:
+                office_data = response.json()
+
+            # Helper function to extract and format address
+            def extract_address(office_type):
+                address = office_data.get(office_type, {}).get("deliveryAddress", {})
+                return [
+                    address.get("streetAddress", "").strip(),
+                    address.get("streetAddressAdditional", "").strip(),
+                    address.get("addressCity", "").strip(),
+                    address.get("addressRegion", "").strip(),
+                    address.get("postalCode", "").strip(),
+                    address.get("addressCountry", "").strip()
+                ] if address else []
+
+            # Return formatted addresses
+            return {
+                'registered': extract_address("registeredOffice"),
+                'records': extract_address("recordsOffice"),
+                # 'head': extract_address('headOffice') -- extract head office for expro companies
+            }
+
+        except Exception as e:
+            current_app.logger.error(f"Error while processing office data for {corp_num}: {e}")
+            raise ValueError(f"Failed to retrieve or process office data for {corp_num}: {e}")
+
+    def _get_parties_data(self, corp_num, legal_type):
+        """
+        Process the parties data for a given corporation number.
+
+        Args:
+            corp_num (str): The corporation number.
+            legal_type (str): The legal type.
+
+        Returns:
+            list: A list of director names in the format "FirstName MiddleInitial LastName".
+        """
+        try:
+            # Build the Colin service endpoint
+            colin_service_url = f'{current_app.config.get("COLIN_SVC_URL")}'
+            parties_endpoint = f"{colin_service_url}/businesses/{legal_type}/{corp_num}/parties/all"
+
+            # Make the authenticated request
+            response = EntityUtils.make_authenticated_request(parties_endpoint)
+            parties_data = {}
+            if response.status_code == HTTPStatus.OK:
+                parties_data = response.json()
+
+            # Log the received data for debugging
+            current_app.logger.debug(f"Processing parties data for corporation number {corp_num}: {parties_data}")
+
+            # Helper function to extract names for a specific role
+            def extract_names(party_list, role_type):
+                # Return an empty list immediately if the party_list is empty
+                if not party_list:
+                    return []
+
+                names = []
+                for party in party_list:
+                    roles = party.get("roles", [])
+                    if any(role.get("roleType") == role_type for role in roles):
+                        officer = party.get("officer", {})
+                        first_name = officer.get("firstName", "").strip()
+                        middle_initial = officer.get("middleInitial", "").strip()
+                        last_name = officer.get("lastName", "").strip()
+                        full_name = f"{first_name} {middle_initial} {last_name}".strip()
+                        names.append(full_name)
+                return names
+
+            # Extract names for the directoires, if it is a BC company
+            role = 'Director'
+            party_list = parties_data.get('parties', [])
+            director_names = extract_names(party_list, role)
+
+            # extract names for Attorneys, if it is a xpro corp
+            role = 'Attorney'
+            attorney_names = extract_names(party_list, role)
+
+            return {'directorNames': director_names, 'attorneyNames': attorney_names}
+        except Exception as e:
+            current_app.logger.error(f"Error while processing parties data for {corp_num}: {e}")
+            raise ValueError(f"Failed to retrieve or process parties data for {corp_num}: {e}")
+
+    def _get_nature_of_business(self, corp_num):
+        return Request.get_nature_business_info(corp_num)
+