fp_data_parsing.py

from pydantic import BaseModel, Field, ValidationError, model_validator, field_validator
from datetime import datetime

from utils import find_bank_by_account_number
from input_data_parsing import IDClass



class FPAddress(BaseModel):
    address_type: str = Field(default="VUA")
    # Street name and house number
    address: str = Field(max_length=100)
    city: str = Field(max_length=255)
    # Sec. 5.14
    country_code: str


class FPAddresses(BaseModel):
    address: list[FPAddress]


class FPPhone(BaseModel):
    # Sec. 5.9
    tph_contact_type: str = Field(default="TELE")
    # Sec. 5.10
    tph_communication_type: str = Field(default="TELE")
    tph_number: str = Field(max_length=50)


class FPPhones(BaseModel):
    phone: list[FPPhone]


class FPEmails(BaseModel):
    email: list[str] = Field(max_length=255)


class FPTransactionPerson(BaseModel):
    # Names, no special characters, e.g. comma and numbers, allowed.
    first_name: str = Field(max_length=100)
    last_name: str = Field(max_length=100)
    birthdate: str
    # Only for Swedish customers. Format YYYYMMDDnnnn
    ssn: str | None = Field(default=None, pattern=r"^\d{12}$")
    # sec 4.6. NOTE: nullable
    phones: FPPhones | None
    # Sec. 4.5. NOTE: nullable
    addresses: FPAddresses | None
    # NOTE: nullable
    emails: FPEmails | None
    # private attributes
    _country_code: str

    @field_validator("birthdate", mode="before")
    def validate_my_datetime(value):
        return value.isoformat(timespec="seconds")

    @model_validator(mode="before")
    def validate_ssn(cls, values):
        # no ssn non-SE customers
        if values["_country_code"] != "SE" and values["ssn"] is not None:
            raise ValueError("ssn provided for non-SE customer, not allowed")
        return values


class FPTransactionEntity(BaseModel):
    # NOTE: only used for entity related to account_to
    name: str = Field(max_length=255)
    # Företagsform, sec. 5.11
    incorporation_legal_form: str = Field(default="AB")
    # Organisationsnummer
    incorporation_number: str = Field(max_length=50)
    incorporation_country_code: str = Field(default="SE")

    @classmethod
    def create_from_country_code(cls, country_code: str):
        if country_code == "SE":
            return cls(name="Anyfin Finance 1 AB", incorporation_number="5592349962")
        elif country_code in ("FI", "DE"):
            return cls(name="Anyfin Finance 2 AB", incorporation_number="5593971806")
        return 
    

class FPAccountRelatedPerson(BaseModel):
    # NOTE: only used for person related to account_from
    # Sec. 4.3.1
    t_person: FPTransactionPerson
    # sec. 5.15
    role: str = Field(default="OKANT")


class FPRelatedPerson(BaseModel):
    account_related_person: list[FPAccountRelatedPerson]


class FPTransactionAccount(BaseModel):
    # NOTE: only used for account_to
    # Leverantör av konto, kort, etc. (KKE). Provider of the account, card, etc.
    institution_name: str = Field(max_length=255)
    # Swift/BIC för leverantör av KKE
    # FIXME
    swift: str = Field(default="OKANTOKANT", max_length=11)
    # Reference number of the account, card, etc
    account: str = Field(max_length=255) 
    # Organisations relation till kontot, sec. 4.2.1
    t_entity: list[FPTransactionEntity]
    # private attributes
    _country_code: str

    @model_validator(mode="before")
    #@classmethod
    def set_t_entity(cls, values):
        country_code = values["_country_code"]
        values["t_entity"] = [FPTransactionEntity.create_from_country_code(country_code)]
        return values


class FPTransactionAccountMyClient(BaseModel):
    # NOTE: only used for account_from
    # Leverantör av konto, kort, etc. (KKE). Provider of the account, card, etc.
    institution_name: str = Field(max_length=255)
    # Swift/BIC för leverantör av KKE
    swift: str = Field(default="OKANTOKANT", max_length=11)
    # Reference number of the account, card, etc
    account: str = Field(max_length=255) 
    # # The currency the account is kept in, sec. 5.13
    currency_code: str
    # # Sec. 5.3
    account_type: str
    # Person(s) with access to the account, sec. 4.22.2
    # NOTE: should be list
    related_persons: FPRelatedPerson
    # Account opened date
    opened: str = Field(default=datetime(1900,1,2).isoformat(timespec="seconds"))
    # Account status when transaction was reported, sec. 5.4
    status_code: str = Field(default="OKANT")


class FPTransactionForeignCurrency(BaseModel):
    # sec. 5.13
    foreign_currency_code: str
    foreign_amount: float
    # FIXME: nullable?
    #foreign_exchange_rate: float 


class FPTransactionFrom(BaseModel):
    # Carrier of transaction, sec. 5.2
    from_funds_code: str = Field(default="IRREL")
    # If transaction conducted in foreign currency, sec 4.7 
    from_foreign_currency: FPTransactionForeignCurrency | None
    # Sec. 4.1.1
    from_account: FPTransactionAccountMyClient
    # # Sec. 5.14
    from_country: str

    @model_validator(mode="before")
    def check_foreign_currency(cls, values):
        if values["from_country"] != "SE" and values.get("from_foreign_currency") is None:
            raise ValueError("foreign currency info required when non-SE customer")
        elif values["from_country"] == "SE":
            values["from_foreign_currency"] = None
        return values


class FPTransactionTo(BaseModel):
    # Carrier of transaction, sec. 5.2
    to_funds_code: str = Field(default="IRREL")
    # Sec. 4.1.1
    to_account: FPTransactionAccount
    # # Sec. 5.14
    to_country: str = Field(default="SE")


class FPTransaction(BaseModel):
    # The unique reference number of a transaction within a reporting entity.
    transactionnumber: str = Field(max_length=100)
    # # Transaction date and time YYYY-MM-DDTHH:MM:SS
    date_transaction: str
    # Transaction channel, sec. 5.6
    transmode_code: str
    # Amount in SEK, sec. 4.24, 
    # FIXME: sql_decimal
    amount_local: float
    # # # Sec. 3.4.2
    t_from: FPTransactionFrom
    # # # Sec. 3.5.2
    t_to: FPTransactionTo
    # private attributes used to calculate other fields
    _reference_ag: str | None
    _reference_bg: str | None
    _iban_from: str | None
    _iban_to: str | None
    _country_code: str
    _currency_code_local: str

    @model_validator(mode="before")
    def validate_transaction_reference(cls, values):
        reference_ag = values.get("reference_ag")
        reference_bg = values.get("reference_bg")
        iban_from = values.get("iban_from")
        iban_to = values.get("iban_from")
        # Ensure correct payment information provided for SE
        # NOTE: Assume AG/BG only used by SE, IBAN only used by FI/DE
        if (country_code:=values["_country_code"])=="SE":
            if (reference_ag is None and reference_bg is None) or (reference_ag is not None and reference_bg is not None):
                raise ValueError("Exactly one of 'reference_ag' or 'reference_bg' must be provided for SE customer.")
            if iban_from is not None or iban_to is not None:
                raise ValueError("Iban information provided for SE customer, not allowed.")
        # Ensure iban from and to provided for FI and DE
        elif country_code in ("FI", "DE"):
            if iban_from is None or iban_to is None:
                raise ValueError("Iban information missing for FI/DE customer.")
            if reference_ag is not None or reference_bg is not None:
                raise ValueError("Autogiro or bankgiro information provided for FI/DE customer, not allowed.")
        return values

    @field_validator("date_transaction", mode="before")
    def validate_my_datetime(value):
        """Enforce ISO 8601 format: YYYY-MM-DDTHH:MM:SS"""
        return value.isoformat(timespec="seconds")

    @model_validator(mode="before")
    def set_transactionnumber_and_channel(cls, values):
        if (reference_ag:=values.get("reference_ag")) is not None:
            values["transactionnumber"] = reference_ag
            values["transmode_code"] = "KOVER"
        elif (reference_bg:=values.get("reference_bg")) is not None:
            values["transactionnumber"] = reference_bg
            values["transmode_code"] = "BANKG"
        elif (iban_from:=values.get("iban_from")) is not None:
            values["transactionnumber"] = iban_from
            values["transmode_code"] = "KOVER"
        return values
    
    @model_validator(mode="before")
    def set_transaction_country_code(cls, values):
        # FIXME: Assuming transaction from customer country?
        values["t_from"]["from_country"] = values["_country_code"]
        return values
    
    @model_validator(mode="before")
    def set_account_institution_name_and_swift(cls, values):
        # transaction from: institution name and swift
        if (reference_ag:=values.get("reference_ag")) is not None:
            values["t_from"]["from_account"]["institution_name"] = find_bank_by_account_number(bankkontonummer=reference_ag)
            values["t_from"]["from_account"]["swift"] = "SWEDSESS"
        elif values.get("reference_bg") is not None or values.get("iban_from") is not None:
            values["t_from"]["from_account"]["institution_name"] = "OKANT"
            # FIXME
            values["t_from"]["from_account"]["swift"] = "OKANTOKANT"
        # transaction to: institution name and swift
        values["t_to"]["to_account"]["institution_name"] = "SEB"
        values["t_to"]["to_account"]["swift"] = "ESESSSES"
        return values
    
    @model_validator(mode="before")
    def set_account_reference(cls, values):
        # transaction from: account reference number
        if (reference_ag:=values.get("reference_ag")) is not None:
            values["t_from"]["from_account"]["account"] = reference_ag.replace("-", "")
            values["t_from"]["from_account"]["account_type"] = "TRANS"
        elif (reference_bg:=values.get("reference_bg")) is not None:
            values["t_from"]["from_account"]["account"] = reference_bg
            values["t_from"]["from_account"]["account_type"] = "TRANS"
        elif (iban_from:=values.get("iban_from")) is not None:
            values["t_from"]["from_account"]["account"] = iban_from
            values["t_from"]["from_account"]["account_type"] = "IBAN"
        # transaction to: account reference number, dependent on market
        if reference_ag is not None or reference_bg is not None:
            values["t_to"]["to_account"]["account"] = "5231000001180770"
        elif iban_from is not None:
            if values["_country_code"]=="FI":
                values["t_to"]["to_account"]["account"] = "33010001178896"
            elif values["_country_code"]=="DE":
                values["t_to"]["to_account"]["account"] = "71807006"
        return values
    
    @model_validator(mode="before")
    def set_account_currency(cls, values):
        # FIXME: assume currency based on market?
        if (country_code:=values["_country_code"])=="SE":
            values["t_from"]["from_account"]["currency_code"] = "SEK"
        elif country_code in ("FI", "DE"):
            values["t_from"]["from_account"]["currency_code"] = "EUR"
        elif country_code == "NO":
            values["t_from"]["from_account"]["currency_code"] = "NOK"
        return values


class FPPersonRegistationInReport(BaseModel):
    first_name: str | None = Field(default="Emma")
    last_name: str | None = Field(default="Hedrén")
    email: str | None = Field(default="emma.hedren@anyfin.com")


class FPIndicators(BaseModel):
    indicator: list[str]


class FPReport(BaseModel):
    # Verksamhetsutövare ID, the number is provided by the FIU on request.
    rentity_id: int = Field(ge=1)
    # Rapporteringskod, sec. 5.1
    submission_code: str = Field(default="E")
    # sec. 5.8
    report_code: str = Field(default="STR")
    # Reporting entities internal report reference, customer_id
    entity_reference: str = Field(max_length=255)
    report_date: str
    # sec. 5.13
    currency_code_local: str
    # reporting person sec. 4.3.3
    reporting_person: FPPersonRegistationInReport
    # Anledning till rapport
    reason: str = Field(max_length=8000)
    # Vidtagna eller planerade åtgärder
    action: str = Field(max_length=8000)
    # sec. 3.2
    transaction: list[FPTransaction]
    # # sec. 5.17
    # # FIXME:
    report_indicators: FPIndicators
    # private attributes
    _type: str
    _country_code: str 

    @model_validator(mode="before")
    def set_rentity_id(cls, values):
        country_code = values["_country_code"]
        if country_code in ("SE", "FI", "NO"):
            values["rentity_id"] = 720
        elif country_code == "DE":
            values["rentity_id"] = 1247
        return values
    
    @field_validator("report_date", mode="before")
    def validate_my_datetime(value):
        """Enforce ISO 8601 format: YYYY-MM-DDTHH:MM:SS"""
        return value.isoformat(timespec="seconds")

    @model_validator(mode="before")
    def set_report_indicators(cls, values):
        # FIXME: correct?
        type = values["_type"]
        if type == "OP":
            values["report_indicators"] = {
                "indicator": [
                    "AMTPT", "KDK01"
                ]
            }
        elif type == "OB":
            values["report_indicators"] = {
                "indicator": [
                    "AMTPT", "KDK01", "KDK02"
                ]
            }
        return values

    @model_validator(mode="before")
    def preprocess_country_and_currency_codes(cls, values):
        country_code = values["_country_code"]
        
        # 1. set currency_code_local based on country_code
        if country_code=="SE":
            currency_code_local = "SEK"
        elif country_code in ("FI", "DE"):
            currency_code_local = "EUR"
        elif country_code=="NO":
            currency_code_local = "NOK"
        values["currency_code_local"] = currency_code_local

        # 2. pass currency_code_local and country_code to nested dicts 
        for transaction in values.get("transaction", []):
            transaction["_country_code"] = country_code
            transaction["_currency_code_local"] = currency_code_local
            if country_code != "SE":
                transaction["t_from"]["from_foreign_currency"]["foreign_currency_code"] = currency_code_local
            for related_person in transaction["t_from"]["from_account"]["related_persons"]["account_related_person"]:
                related_person["t_person"]["_country_code"] = country_code
            transaction["t_to"]["to_account"]["_country_code"] = country_code

        return values


class FPClass(BaseModel):
    report: FPReport

    @classmethod
    def format_datetime(cls, value: datetime) -> str:
        """Enforce ISO 8601 format: YYYY-MM-DDTHH:MM:SS"""
        return value.isoformat(timespec="seconds")



def create_fp_report(input: IDClass):
    transaction_list = []
    for transaction in input.transactions:
        transaction_dict = {
            "reference_ag": transaction.reference_ag,
            "reference_bg": transaction.reference_bg,
            "iban_from": transaction.iban_from,
            "iban_to": transaction.iban_to,
            "date_transaction": transaction.transaction_date,
            "amount_local": transaction.local_amount,
            "t_from": {
                "from_foreign_currency": {
                    "foreign_amount": transaction.foreign_amount,
                },
                "from_account": {
                    "related_persons": {
                        "account_related_person": [
                            {
                                "t_person": {
                                    "first_name": transaction.first_name,
                                    "last_name": transaction.last_name,
                                    "birthdate": transaction.birthdate,
                                    "ssn": transaction.social_number,
                                    "addresses": {
                                        "address": [addr.model_dump() for addr in transaction.address],
                                    },
                                    "phones": {
                                        "phone": [{"tph_number": nbr} for nbr in transaction.tph_number]
                                    },
                                    "emails": {
                                        "email": transaction.email,
                                    },
                                },
                            },
                        ],
                    },
                },
            },
            "t_to": {
                "to_account": {
                }
            },
        }

        transaction_list.append(transaction_dict)

    formatted_input = {
        "report": 
            {
                "entity_reference": input.customer_id,
                "report_date": input.report_date,
                "reason": input.reason,
                "action": input.action,
                "transaction": transaction_list,
                "reporting_person": {},
                "_type": input.type,
                "_country_code": input.market,
            }
    }

    report = FPClass(**formatted_input)

    return report.model_dump(exclude_none=True)