diff --git a/legal-api/src/legal_api/resources/v2/business/business_filings/business_filings.py b/legal-api/src/legal_api/resources/v2/business/business_filings/business_filings.py index 105ea33f5c..3217810179 100644 --- a/legal-api/src/legal_api/resources/v2/business/business_filings/business_filings.py +++ b/legal-api/src/legal_api/resources/v2/business/business_filings/business_filings.py @@ -66,7 +66,7 @@ from legal_api.services.authz import is_allowed from legal_api.services.event_publisher import publish_to_queue from legal_api.services.filings import validate -from legal_api.services.permissions import PermissionService +from legal_api.services.permissions import ListActionsPermissionsAllowed, PermissionService from legal_api.services.utils import get_str from legal_api.utils import datetime from legal_api.utils.auth import jwt @@ -684,6 +684,7 @@ def save_filing(client_request: LocalProxy, # noqa: PLR0912 } """ json_input = client_request.get_json() + previous_filing_json = copy.deepcopy(filing.filing_json) if filing and filing.filing_json else None ListFilingResource.modify_filing_json(json_input, filing) if business_identifier.startswith("T"): @@ -732,6 +733,9 @@ def save_filing(client_request: LocalProxy, # noqa: PLR0912 ListFilingResource.link_now_and_withdrawn_filing(filing) if business_identifier.startswith("T"): filing.temp_reg = None + + ListFilingResource._maybe_record_staff_completing_party_override(filing, user, previous_filing_json) + filing.save() except BusinessException as err: return None, None, {"error": err.error}, err.status_code @@ -779,6 +783,90 @@ def sanitize_html_fields(filing_json): filing_json["filing"]["correction"]["resolution"] = Sanitizer().sanitize(resolution_content) return filing_json + @staticmethod + def _normalize_str(value: Optional[str]) -> str: + """Normalize a string for comparisons (strip + uppercase).""" + return (value or "").strip().upper() + + @staticmethod + def _normalize_completing_party_snapshot(snapshot: Optional[dict]) -> dict: + """Normalize a completing party snapshot dict for stable comparisons.""" + snapshot = snapshot or {} + return { + "firstName": ListFilingResource._normalize_str(snapshot.get("firstName")), + "middleName": ListFilingResource._normalize_str(snapshot.get("middleName")), + "lastName": ListFilingResource._normalize_str(snapshot.get("lastName")), + "email": ListFilingResource._normalize_str(snapshot.get("email")) + } + + @staticmethod + def _get_completing_party_snapshot_from_filing_json(filing_json: dict) -> Optional[dict]: + """Extract completing party (first/middle/last/email) from a filing JSON.""" + if not filing_json: + return None + + try: + filing_type = filing_json["filing"]["header"]["name"] + parties = filing_json.get("filing", {}).get(filing_type, {}).get("parties", []) + except (TypeError, KeyError): + return None + + for party in parties or []: + roles = party.get("roles", []) or [] + if any((role.get("roleType") or "").lower().replace(" ", "_") == "completing_party" for role in roles): + officer = party.get("officer", {}) or {} + return { + "firstName": str(officer.get("firstName") or "").strip(), + "middleName": str(officer.get("middleName") or officer.get("middleInitial") or "").strip(), + "lastName": str(officer.get("lastName") or "").strip(), + "email": str(officer.get("email") or "").strip() + } + + return None + + @staticmethod + def _user_can_edit_completing_party() -> bool: + """Return True if the current caller is permitted to edit completing party fields.""" + err = PermissionService.check_user_permission( + ListActionsPermissionsAllowed.EDITABLE_COMPLETING_PARTY.value, + message="Permission Denied: You do not have permission to edit the completing party." + ) + return not bool(err) + + @staticmethod + def _maybe_record_staff_completing_party_override( + filing: Filing, + user: User, + previous_filing_json: Optional[dict] + ) -> None: + """Persist a staff-authorized completing party snapshot in meta_data when staff edits it.""" + if not filing or not user: + return + + new_snapshot = ListFilingResource._get_completing_party_snapshot_from_filing_json(filing.filing_json) + if not new_snapshot: + return + + old_snapshot = ListFilingResource._get_completing_party_snapshot_from_filing_json(previous_filing_json) \ + if previous_filing_json else None + + if (ListFilingResource._normalize_completing_party_snapshot(old_snapshot) == + ListFilingResource._normalize_completing_party_snapshot(new_snapshot)): + return + + if not ListFilingResource._user_can_edit_completing_party(): + return + + meta_data = copy.deepcopy(filing.meta_data) if filing.meta_data else {} + staff_overrides = meta_data.setdefault("staffOverrides", {}) + staff_overrides["completingParty"] = { + "version": 1, + "updatedAt": datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).isoformat(), + "updatedBy": user.id, + "snapshot": new_snapshot + } + filing.meta_data = meta_data + @staticmethod def get_filing_types(business: Business, filing_json: dict): # noqa: PLR0912 """Get the filing type fee codes for the filing. diff --git a/legal-api/src/legal_api/services/filings/validations/common_validations.py b/legal-api/src/legal_api/services/filings/validations/common_validations.py index 74db190e03..38021615c6 100644 --- a/legal-api/src/legal_api/services/filings/validations/common_validations.py +++ b/legal-api/src/legal_api/services/filings/validations/common_validations.py @@ -24,7 +24,7 @@ from flask_babel import _ from legal_api.errors import Error -from legal_api.models import Address, Business, PartyRole +from legal_api.models import Address, Business, Filing, PartyRole from legal_api.models.configuration import EMAIL_PATTERN from legal_api.services import MinioService, colin, flags, namex from legal_api.services.bootstrap import AccountService @@ -983,19 +983,93 @@ def validate_party_role_firms(parties: list, filing_type: str) -> list: return msg +def _get_completing_party_officer_and_mailing_address( + filing_json: dict, + filing_type: str +) -> tuple[Optional[dict], Optional[dict]]: + """Return (officer, mailingAddress) for the completing party in the filing JSON.""" + try: + parties = filing_json.get("filing", {}).get(filing_type, {}).get("parties", []) + except AttributeError: + return None, None + + for party in parties or []: + roles = party.get("roles", []) or [] + if any( + (role.get("roleType") or "").lower().replace(" ", "_") == + PartyRole.RoleTypes.COMPLETING_PARTY.value.lower() + for role in roles + ): + return party.get("officer") or None, party.get("mailingAddress") or {} + + return None, None + + +def _get_filing_id_from_filing_json(filing_json: dict) -> Optional[int]: + """Best-effort filing id lookup (payload first, then Flask route params).""" + filing_id = filing_json.get("filing", {}).get("header", {}).get("filingId") + if filing_id: + try: + return int(filing_id) + except (TypeError, ValueError): + pass + + # Some requests may not include filingId in the payload; attempt to read it from the Flask route params. + try: + if get_request_context() and hasattr(request, "view_args") and request.view_args: + view_filing_id = request.view_args.get("filing_id") or request.view_args.get("filingId") + if view_filing_id: + return int(view_filing_id) + except (TypeError, ValueError): + pass + + return None + + +def _get_staff_completing_party_snapshot(filing_id: int) -> Optional[dict]: + """Return the staff-authorized completing party snapshot from Filing.meta_data, if present.""" + filing = Filing.find_by_id(filing_id) + if not filing or not filing.meta_data: + return None + + snapshot = ( + filing.meta_data + .get("staffOverrides", {}) + .get("completingParty", {}) + .get("snapshot") + ) + return snapshot if isinstance(snapshot, dict) else None + + +def _get_completing_party_baseline_from_contact(contact: dict, filing_json: dict) -> dict: + """Baseline completing party fields from account contact, overridden by staff snapshot if present.""" + baseline = { + "firstName": contact.get("firstName", ""), + "middleName": contact.get("middleName", "") or contact.get("middleInitial", ""), + "lastName": contact.get("lastName", ""), + "email": contact.get("email", "") + } + + filing_id = _get_filing_id_from_filing_json(filing_json) + if not filing_id: + return baseline + + staff_snapshot = _get_staff_completing_party_snapshot(filing_id) + if not staff_snapshot: + return baseline + + baseline["firstName"] = staff_snapshot.get("firstName", baseline["firstName"]) + baseline["middleName"] = staff_snapshot.get("middleName", baseline["middleName"]) + baseline["lastName"] = staff_snapshot.get("lastName", baseline["lastName"]) + baseline["email"] = staff_snapshot.get("email", baseline["email"]) + return baseline + + def validate_completing_party(filing_json: dict, filing_type: str, org_id: int) -> dict: """Validate completing party edited.""" msg = [] - parties = filing_json["filing"][filing_type].get("parties", {}) - officer = None - mailing_address = None - for party in parties: - roles = party.get("roles", []) - if any(role.get("roleType").lower().replace(" ", "_") == PartyRole.RoleTypes.COMPLETING_PARTY.value.lower() for role in roles): - officer = party.get("officer", {}) - mailing_address = party.get("mailingAddress", {}) - break + officer, mailing_address = _get_completing_party_officer_and_mailing_address(filing_json, filing_type) if not officer: return { "error":[{ @@ -1009,6 +1083,7 @@ def validate_completing_party(filing_json: dict, filing_type: str, org_id: int) filing_completing_party_mailing_address = mailing_address or {} filing_firstname = officer.get("firstName") + filing_middlename = officer.get("middleName", None) or officer.get("middleInitial", None) filing_lastname = officer.get("lastName") filing_email = officer.get("email") @@ -1034,30 +1109,31 @@ def validate_completing_party(filing_json: dict, filing_type: str, org_id: int) "deliveryInstructions": contact.get("deliveryInstructions", ""), "streetAddressAdditional": contact.get("streetAdditional", "") } - existing_firstname = contact.get("firstName", "") - existing_lastname = contact.get("lastName", "") - existing_email = contact.get("email", "") + + baseline = _get_completing_party_baseline_from_contact(contact, filing_json) address_changed = False if filing_completing_party_mailing_address: address_changed = not is_address_changed(existing_cp_mailing_address, filing_completing_party_mailing_address) existing_name = { - "firstName": existing_firstname, - "lastName": existing_lastname + "firstName": baseline.get("firstName"), + "middleName": baseline.get("middleName"), + "lastName": baseline.get("lastName") } filing_name = { "firstName": filing_firstname, + "middleName": filing_middlename, "lastName": filing_lastname } name_changed = False - if filing_firstname or filing_lastname: + if filing_firstname or filing_middlename or filing_lastname: name_changed = is_name_changed(existing_name, filing_name) email_changed = False if filing_email: - email_changed = not is_same_str(existing_email, filing_email) + email_changed = not is_same_str(baseline.get("email"), filing_email) return { "error": msg,