Sync with Main

.gitignore

@@ -137,3 +137,6 @@ jobs/sftp-*/archive
 # ruff
 **/.ruff_cache
 
+# Delta restore run artifacts
+data-tool/scripts/generated/delta_restore/
+

colin-api/src/colin_api/models/__init__.py

@@ -1,6 +1,8 @@
 """Model imports."""
+
 from .address import Address
 from .business import Business
+from .business_snapshot import BusinessSnapshot
 from .cont_out import ContOut
 from .corp_involved import CorpInvolved
 from .corp_name import CorpName

colin-api/src/colin_api/models/business.py

@@ -81,6 +81,13 @@ class CorpStateTypes(Enum):
              TypeCodes.ULC_CONTINUE_IN.value,
              TypeCodes.CCC_CONTINUE_IN.value]
 
+    # Corp types that auth may sync auth info (including the corp password) for.
+    # These are the COLIN businesses that can be affiliated in the business registry
+    # dashboard while not loaded in LEAR.
+    AUTH_INFO_TYPES = [TypeCodes.BC_COMP.value,
+                       TypeCodes.ULC_COMP.value,
+                       TypeCodes.CCC_COMP.value]
+
     NUMBERED_CORP_NAME_SUFFIX = {
         TypeCodes.BCOMP.value: 'B.C. LTD.',
         TypeCodes.BC_COMP.value: 'B.C. LTD.',
@@ -116,6 +123,16 @@ class CorpStateTypes(Enum):
     def __init__(self):
         """Initialize with all values None."""
 
+    @property
+    def lear_state(self) -> str:
+        """Return the LEAR-style business state: only op-state ACT is active, everything else is historical.
+
+        Same rule as the tombstone migration (data-tool tombstone_utils) and the search
+        importers - keyed off CORP_OP_STATE.OP_STATE_TYP_CD (ACT/HIS), not the
+        fine-grained corp state code.
+        """
+        return 'ACTIVE' if self.corp_state_class == 'ACT' else 'HISTORICAL'
+
     def as_dict(self) -> Dict:
         """Return dict version of self."""
         return {
@@ -374,6 +391,50 @@ def find_by_identifier(cls,  # pylint: disable=too-many-statements
             # pass through exception to caller
             raise err
 
+    @classmethod
+    def get_auth_info(cls, identifier: str) -> Dict:
+        """Return the info needed by auth to create/refresh an entity for a COLIN business.
+
+        The response contains the corporation password, so callers must be service
+        authenticated and must never log the result or pass it on to a browser.
+        Restricted to the corp types that can be affiliated while not loaded in LEAR.
+        """
+        # auth holds the prefixed identifier, COLIN stores the bare corp num. BC/ULC/CC all
+        # use the BC prefix, so it can be stripped without knowing the legal type first.
+        if identifier.startswith('BC'):
+            identifier = identifier[2:]
+
+        # acquire a single session from the pool and reuse it for both lookups
+        con = DB.connection
+        business = cls.find_by_identifier(identifier, corp_types=cls.AUTH_INFO_TYPES, con=con)
+
+        cursor = con.cursor()
+        cursor.execute(
+            """
+            select corp_password
+            from CORPORATION
+            where corp_num=:corp_num
+            """,
+            corp_num=identifier
+        )
+        result = cursor.fetchone()
+        pass_code = result[0] if result else None
+
+        return {
+            'identifier': business.corp_num,
+            'legalName': business.corp_name,
+            'legalType': business.corp_type,
+            # LEAR-style state, not COLIN's free-text status description - auth stores it
+            # verbatim and the dashboard filters on ACTIVE/HISTORICAL
+            'status': business.lear_state,
+            'goodStanding': business.good_standing,
+            'businessNumber': business.business_number,
+            # find_by_identifier returns the COLIN 'True'/'False' string; normalize for callers
+            'adminFreeze': business.admin_freeze == 'True',
+            'email': business.email,
+            'passCode': pass_code
+        }
+
     @classmethod
     def create_corporation(cls, con, filing_info: Dict):
         """Insert a new business from an incorporation filing."""

colin-api/src/colin_api/models/business_snapshot.py

@@ -0,0 +1,199 @@
+# Copyright © 2026 Province of British Columbia
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Snapshot of a COLIN business, normalized to LEAR structure."""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Dict, Optional
+
+import pycountry
+from flask import current_app
+
+from colin_api.exceptions import PartiesNotFoundException
+from colin_api.models.business import Business
+from colin_api.models.corp_party import Party
+from colin_api.models.office import Office
+from colin_api.models.shares import ShareObject
+from colin_api.resources.db import DB
+
+
+class BusinessSnapshot:  # pylint: disable=too-few-public-methods
+    """Builds the LEAR-structured snapshot dict for a COLIN business."""
+
+    # snapshot offices are limited to the two the amalgamation flow prepopulates
+    OFFICE_TYPES = ('registeredOffice', 'recordsOffice')
+
+    @classmethod
+    def get_snapshot(cls, orig_identifier: str) -> Dict:
+        """Return the business/parties/offices/shareClasses/resolutions snapshot."""
+        identifier = orig_identifier
+        if identifier.startswith('BC'):
+            identifier = identifier[2:]
+
+        con = DB.connection
+        business = Business.find_by_identifier(identifier, con=con)
+        cursor = con.cursor()
+
+        try:
+            parties = Party.get_current(cursor, identifier)
+        except PartiesNotFoundException:
+            # no current directors on file is bad data but not an error state here
+            parties = []
+
+        offices = Office.convert_obj_list(Office.get_current(cursor, identifier)) or {}
+
+        # mirror the /sharestructure resource: current structure is the one with no end event
+        share_structs = ShareObject.get_all(cursor, identifier) or []
+        share_struct = next((x for x in share_structs if not x.end_event_id), None)
+        share_classes = share_struct.to_dict()['shareClasses'] if share_struct else []
+
+        resolutions = Business.get_resolutions(cursor, identifier)
+
+        return {
+            'business': cls._business_dict(business, orig_identifier, cursor),
+            'parties': [cls._normalize_party(party) for party in parties],
+            'offices': {
+                office_type: cls._normalize_office(office)
+                for office_type, office in offices.items() if office_type in cls.OFFICE_TYPES
+            },
+            'shareClasses': [cls._normalize_share_class(share_class) for share_class in share_classes],
+            'resolutions': [{'date': date} for date in resolutions],
+        }
+
+    @classmethod
+    def _business_dict(cls, business: Business, orig_identifier: str, cursor) -> Dict:
+        """Return the business section in the shape of LEAR's slim business json."""
+        return {
+            'identifier': orig_identifier,
+            'legalName': business.corp_name,
+            'legalType': business.corp_type,
+            'state': business.lear_state,
+            # tri-state: None when COLIN can't compute good standing for the corp
+            'goodStanding': business.good_standing,
+            # find_by_identifier returns the COLIN 'True'/'False' string
+            'adminFreeze': business.admin_freeze == 'True',
+            'foundingDate': cls._to_iso_datetime(business.founding_date),
+            'taxId': business.business_number,
+            'hasFutureEffectiveFiling': cls._has_future_effective_filing(cursor, business.corp_num),
+        }
+
+    @staticmethod
+    def _has_future_effective_filing(cursor, corp_num: str) -> bool:
+        """Return whether any filing has an effective date still in the future."""
+        current_date = datetime.now(timezone.utc).strftime('%Y-%m-%d')
+        cursor.execute(
+            """
+            select count(*)
+            from event join filing on event.event_id = filing.event_id
+            where event.corp_num=:corp_num
+            and filing.effective_dt > TO_DATE(:current_date, 'YYYY-mm-dd')
+            """,
+            corp_num=corp_num,
+            current_date=current_date
+        )
+        return cursor.fetchone()[0] > 0
+
+    @classmethod
+    def _normalize_party(cls, party: Party) -> Dict:
+        """Return a party in the structure of LEAR's /parties items."""
+        raw = party.as_dict()
+        return {
+            'officer': {**raw['officer'], 'id': raw['id'], 'email': None},
+            'deliveryAddress': cls._normalize_address(raw['deliveryAddress']),
+            'mailingAddress': cls._normalize_address(raw['mailingAddress']),
+            'roles': raw['roles'] or [],
+        }
+
+    @classmethod
+    def _normalize_office(cls, office: Dict) -> Dict:
+        """Return an office's addresses in LEAR structure."""
+        return {
+            'deliveryAddress': cls._normalize_address(office.get('deliveryAddress')),
+            'mailingAddress': cls._normalize_address(office.get('mailingAddress')),
+        }
+
+    @classmethod
+    def _normalize_address(cls, address: Optional[Dict]) -> Optional[Dict]:
+        """Return an Address.as_dict in the structure of LEAR's address json."""
+        if not address:
+            return None
+        return {
+            'id': address.get('addressId'),
+            'streetAddress': address.get('streetAddress'),
+            'streetAddressAdditional': address.get('streetAddressAdditional'),
+            'addressCity': address.get('addressCity'),
+            'addressRegion': address.get('addressRegion'),
+            'addressCountry': cls._country_to_alpha2(address.get('addressCountry')),
+            'postalCode': address.get('postalCode'),
+            'deliveryInstructions': address.get('deliveryInstructions'),
+        }
+
+    @classmethod
+    def _normalize_share_class(cls, share_class: Dict) -> Dict:
+        """Return a share class in the structure of LEAR's /share-classes items."""
+        return {
+            'id': share_class['id'],
+            'name': share_class['name'],
+            # COLIN never stores a priority - the class id preserves creation order
+            'priority': share_class['displayOrder'],
+            'hasMaximumShares': share_class['hasMaximumShares'],
+            'maxNumberOfShares': cls._to_int(share_class['maxNumberOfShares']),
+            'hasParValue': share_class['hasParValue'],
+            'parValue': float(share_class['parValue']) if share_class['parValue'] is not None else None,
+            'currency': share_class['currency'],
+            'currencyAdditional': share_class['currencyAdditional'],
+            'hasRightsOrRestrictions': share_class['hasRightsOrRestrictions'],
+            'series': [cls._normalize_share_series(series) for series in share_class['series']],
+        }
+
+    @classmethod
+    def _normalize_share_series(cls, series: Dict) -> Dict:
+        """Return a share series in the structure of LEAR's series json."""
+        return {
+            'id': series['id'],
+            'name': series['name'],
+            'priority': series['displayOrder'],
+            'hasMaximumShares': series['hasMaximumShares'],
+            'maxNumberOfShares': cls._to_int(series['maxNumberOfShares']),
+            'hasRightsOrRestrictions': series['hasRightsOrRestrictions'],
+        }
+
+    _country_cache: Dict[str, str] = {}
+
+    @classmethod
+    def _country_to_alpha2(cls, country: Optional[str]) -> Optional[str]:
+        """Map COLIN's country description to the alpha-2 code LEAR stores."""
+        if not country:
+            return country
+        if len(country) == 2:  # already a code
+            return country
+        if country not in cls._country_cache:
+            try:
+                cls._country_cache[country] = pycountry.countries.search_fuzzy(country)[0].alpha_2
+            except LookupError:
+                current_app.logger.error('Could not map COLIN country %s to an alpha-2 code', country)
+                cls._country_cache[country] = country
+        return cls._country_cache[country]
+
+    @staticmethod
+    def _to_int(value) -> Optional[int]:
+        """Coerce an Oracle NUMBER to int, preserving None."""
+        return int(value) if value is not None else None
+
+    @staticmethod
+    def _to_iso_datetime(value: Optional[str]) -> Optional[str]:
+        """Rewrite convert_to_json_datetime's '-00:00' suffix to the ISO '+00:00' LEAR uses."""
+        if value and value.endswith('-00:00'):
+            return value[:-6] + '+00:00'
+        return value

colin-api/src/colin_api/models/filing.py

@@ -1298,7 +1298,7 @@ def get_future_effective_filings(cls, business: Business) -> List:
         """Get the list of all future effective filings for a business."""
         try:
             future_effective_filings = []
-            current_date = datetime.datetime.utcnow().strftime('%Y-%m-%d')
+            current_date = datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d')
             cursor = DB.connection.cursor()
             cursor.execute(
                 """

colin-api/src/colin_api/models/shares.py

@@ -58,6 +58,7 @@ class ShareClass(Share):  # pylint: disable=too-many-instance-attributes;
 
     # pylint: disable=too-few-public-methods
     currency_type = None
+    other_currency = None
     has_par_value = None
     par_value_amt = None
     series = None
@@ -77,6 +78,7 @@ def to_dict(self):
             'maxNumberOfShares': self.max_number_shares,
             'parValue': self.par_value_amt,
             'currency': self.currency_type,
+            'currencyAdditional': self.other_currency,
             'hasMaximumShares': self.has_max_shares == 'N' or False,
             'hasParValue': self.has_par_value == 'Y' or False,
             'hasRightsOrRestrictions': self.has_special_rights == 'Y' or False,
@@ -140,6 +142,7 @@ def _get_share_classes(cls, cursor, event_id, corp_num):
                 row = dict(zip([x[0].lower() for x in description], row))
                 share_class = ShareClass()
                 share_class.currency_type = row['currency_typ_cd']
+                share_class.other_currency = row['other_currency']
                 share_class.has_max_shares = row['max_share_ind']
                 share_class.has_special_rights = row['spec_rights_ind']
                 share_class.has_par_value = row['par_value_ind']