Source code for netsgiro.objects

"""The higher-level objects API."""

import datetime
from collections import OrderedDict
from decimal import Decimal
from typing import TYPE_CHECKING, Callable, ClassVar, Iterable, List, Optional
from typing import OrderedDict as OrderedDictType
from typing import TypeVar, Union, cast

from attrs import Factory, define, field
from attrs.validators import instance_of, optional

from netsgiro.converters import (
    to_assignment_type,
    to_avtalegiro_registration_type,
    to_service_code,
    to_transaction_type,
)
from netsgiro.enums import AssignmentType, ServiceCode, TransactionType
from netsgiro.records import (
    AssignmentEnd,
    AssignmentStart,
    AvtaleGiroAgreement,
    TransactionAmountItem1,
    TransactionAmountItem2,
    TransactionAmountItem3,
    TransactionRecord,
    TransactionSpecification,
    TransmissionEnd,
    TransmissionStart,
)
from netsgiro.records import parse as records_parse
from netsgiro.validators import str_of_length
from netsgiro.validators import validate_due_date as _validate_due_date

if TYPE_CHECKING:
    from netsgiro.enums import AvtaleGiroRegistrationType
    from netsgiro.records import Record

__all__: List[str] = [
    'Transmission',
    'Assignment',
    'Agreement',
    'PaymentRequest',
    'Transaction',
    'parse',
]

TransactionAmountItems = Union[
    TransactionAmountItem1, TransactionAmountItem2, TransactionAmountItem3
]

# Record or Record subclasses
R = TypeVar('R', bound='Record')


[docs]@define class Transmission: """Transmission is the top-level object. An OCR file contains a single transmission. The transmission can contain multiple :class:`~netsgiro.Assignment` objects of various types. """ #: Data transmitters unique enumeration of the transmission. String of 7 digits. number: str = field(validator=str_of_length(7)) #: Data transmitter's Nets ID. String of 8 digits. data_transmitter: str = field(validator=str_of_length(8)) #: Data recipient's Nets ID. String of 8 digits. data_recipient: str = field(validator=str_of_length(8)) #: For OCR Giro files from Nets, this is Nets' processing date. #: #: For AvtaleGiro payment request, the earliest due date in the #: transmission is automatically used. date: Optional[datetime.date] = field( default=None, validator=optional(instance_of(datetime.date)) ) #: List of assignments. assignments: List['Assignment'] = field(default=Factory(list), repr=False)
[docs] @classmethod def from_records(cls, records: List[R]) -> 'Transmission': """Build a Transmission object from a list of record objects.""" if len(records) < 2: raise ValueError(f'At least 2 records required, got {len(records)}') start, body, end = records[0], records[1:-1], records[-1] assert isinstance(start, TransmissionStart) assert isinstance(end, TransmissionEnd) return cls( number=start.transmission_number, data_transmitter=start.data_transmitter, data_recipient=start.data_recipient, date=end.nets_date, assignments=cls._get_assignments(body), )
@staticmethod def _get_assignments(records: List[R]) -> List['Assignment']: assignments: OrderedDictType[str, List[R]] = OrderedDict() current_assignment_number = None for record in records: if isinstance(record, AssignmentStart): current_assignment_number = record.assignment_number assignments[current_assignment_number] = [] if current_assignment_number is None: raise ValueError(f'Expected AssignmentStart record, got {record!r}') assignments[current_assignment_number].append(record) if isinstance(record, AssignmentEnd): current_assignment_number = None return [Assignment.from_records(rs) for rs in assignments.values()]
[docs] def to_ocr(self) -> str: """Convert the transmission to an OCR string.""" lines = [record.to_ocr() for record in self.to_records()] return '\n'.join(lines)
[docs] def to_records(self) -> Iterable['Record']: """Convert the transmission to a list of records.""" yield self._get_start_record() for assignment in self.assignments: yield from assignment.to_records() yield self._get_end_record()
def _get_start_record(self) -> 'Record': return TransmissionStart( service_code=ServiceCode.NONE, transmission_number=self.number, data_transmitter=self.data_transmitter, data_recipient=self.data_recipient, ) def _get_end_record(self) -> 'Record': avtalegiro_payment_request = all( assignment.service_code == ServiceCode.AVTALEGIRO and assignment.type in ( AssignmentType.TRANSACTIONS, AssignmentType.AVTALEGIRO_CANCELLATIONS, ) for assignment in self.assignments ) if self.assignments and avtalegiro_payment_request: date: Optional[datetime.date] = None for assignment in self.assignments: earliest_transaction_date = assignment.get_earliest_transaction_date() if date is None or ( isinstance(earliest_transaction_date, datetime.date) and earliest_transaction_date < date ): date = earliest_transaction_date else: date = self.date assert isinstance(date, datetime.date) return TransmissionEnd( service_code=ServiceCode.NONE, num_transactions=self.get_num_transactions(), num_records=self.get_num_records(), total_amount=int(self.get_total_amount() * 100), nets_date=date, )
[docs] def add_assignment( self, *, service_code: ServiceCode, assignment_type: AssignmentType, agreement_id: Optional[str] = None, number: str, account: str, date: Optional[datetime.date] = None, ) -> 'Assignment': """Add an assignment to the transmission.""" assignment = Assignment( service_code=service_code, type=assignment_type, agreement_id=agreement_id, number=number, account=account, date=date, ) self.assignments.append(assignment) return assignment
[docs] def get_num_transactions(self) -> int: """Get number of transactions in the transmission.""" return sum(assignment.get_num_transactions() for assignment in self.assignments)
[docs] def get_num_records(self) -> int: """Get number of records in the transmission. Includes the transmission's start and end record. """ return 2 + sum(assignment.get_num_records() for assignment in self.assignments)
[docs] def get_total_amount(self) -> Decimal: """Get the total amount from all transactions in the transmission.""" return Decimal(sum(assignment.get_total_amount() for assignment in self.assignments))
# Assigment transactions TS = List[Union['Transaction', 'Agreement', 'PaymentRequest']] # TransactionRecord and subclasses TR = TypeVar('TR', bound=TransactionRecord)
[docs]@define class Assignment: """An Assignment groups multiple transactions within a transmission. Use :meth:`netsgiro.Transmission.add_assignment` to create assignments. """ #: The service code. One of :class:`~netsgiro.ServiceCode`. service_code: ServiceCode = field(converter=to_service_code) #: The transaction type. One of :class:`~TransactionType`. type: AssignmentType = field(converter=to_assignment_type) #: The assignment number. String of 7 digits. number: str = field(validator=str_of_length(7)) #: The payee's bank account. String of 11 digits. account: str = field(validator=str_of_length(11)) #: Used for OCR Giro. #: #: The payee's agreement ID with Nets. String of 9 digits. agreement_id: Optional[str] = field(default=None, validator=optional(str_of_length(9))) #: Used for OCR Giro. #: #: The date the assignment was generated by Nets. date: Optional[datetime.date] = field( default=None, validator=optional(instance_of(datetime.date)) ) #: List of transaction objects, like :class:`~netsgiro.Agreement`, #: :class:`~netsgiro.PaymentRequest`, :class:`~netsgiro.Transaction`. transactions: TS = field(default=Factory(list), repr=False) _next_transaction_number: int = field(default=1, init=False)
[docs] @classmethod def from_records(cls, records: List[R]) -> 'Assignment': """Build an Assignment object from a list of record objects.""" if len(records) < 2: raise ValueError(f'At least 2 records required, got {len(records)}') start, body, end = records[0], records[1:-1], records[-1] assert isinstance(start, AssignmentStart) assert isinstance(end, AssignmentEnd) sc = start.service_code at = start.assignment_type cb: Callable if sc == ServiceCode.OCR_GIRO: # -> List[Transaction] cb = cls._get_transactions elif sc == ServiceCode.AVTALEGIRO: if at == AssignmentType.AVTALEGIRO_AGREEMENTS: # -> List[Agreement] cb = cls._get_agreements else: # -> List[PaymentRequest] cb = cls._get_payment_requests else: # pragma: no cover raise ValueError(f'Unknown service code: {start.service_code}') transactions: TS = cb(body) return cls( service_code=start.service_code, type=start.assignment_type, agreement_id=start.agreement_id, number=start.assignment_number, account=start.assignment_account, date=end.nets_date, transactions=transactions, )
@staticmethod def _get_agreements(records: List[AvtaleGiroAgreement]) -> List['Agreement']: return [Agreement.from_records([r]) for r in records] @classmethod def _get_payment_requests(cls, records: List[TR]) -> List['PaymentRequest']: transactions = cls._group_by_transaction_number(records) return [PaymentRequest.from_records(rs) for rs in transactions.values()] @classmethod def _get_transactions(cls, records: List[TR]) -> List['Transaction']: transactions = cls._group_by_transaction_number(records) return [Transaction.from_records(rs) for rs in transactions.values()] @staticmethod def _group_by_transaction_number(records: List[TR]) -> OrderedDictType[int, List[TR]]: transactions: OrderedDictType[int, List[TR]] = OrderedDict() for transaction_record in records: tr = transaction_record.transaction_number if tr not in transactions: transactions[tr] = [] transactions[tr].append(transaction_record) return transactions
[docs] def to_records(self) -> Iterable[Union[AssignmentStart, TransactionAmountItems, AssignmentEnd]]: """Convert the assignment to a list of records.""" yield self._get_start_record() for transaction in self.transactions: yield from transaction.to_records() yield self._get_end_record()
def _get_start_record(self) -> AssignmentStart: return AssignmentStart( service_code=self.service_code, assignment_type=self.type, assignment_number=self.number, assignment_account=self.account, agreement_id=self.agreement_id, ) def _get_end_record(self) -> AssignmentEnd: if self.service_code == ServiceCode.OCR_GIRO: dates = { 'nets_date_1': self.date, 'nets_date_2': self.get_earliest_transaction_date(), 'nets_date_3': self.get_latest_transaction_date(), } elif self.service_code == ServiceCode.AVTALEGIRO: dates = { 'nets_date_1': self.get_earliest_transaction_date(), 'nets_date_2': self.get_latest_transaction_date(), } else: # pragma: no cover raise ValueError(f'Unhandled service code: {self.service_code}') return AssignmentEnd( service_code=self.service_code, assignment_type=self.type, num_transactions=self.get_num_transactions(), num_records=self.get_num_records(), total_amount=int(self.get_total_amount() * 100), **dates, )
[docs] def add_payment_request( self, *, kid: str, due_date: datetime.date, amount: Decimal, reference: Optional[str] = None, payer_name: Optional[str] = None, bank_notification: Union[bool, str] = False, validate_due_date: bool = False, ) -> 'PaymentRequest': """Add an AvtaleGiro payment request to the assignment. The assignment must have service code :attr:`~ServiceCode.AVTALEGIRO` and assignment type :attr:`~AssignmentType.TRANSACTIONS`. """ assert ( self.service_code == ServiceCode.AVTALEGIRO ), 'Can only add payment requests to AvtaleGiro assignments' assert ( self.type == AssignmentType.TRANSACTIONS ), 'Can only add payment requests to transaction assignments' if validate_due_date: # Make sure we're not passing invalid due dates _validate_due_date(due_date) if bank_notification: transaction_type = TransactionType.AVTALEGIRO_WITH_BANK_NOTIFICATION else: transaction_type = TransactionType.AVTALEGIRO_WITH_PAYEE_NOTIFICATION return self._add_avtalegiro_transaction( transaction_type=transaction_type, kid=kid, due_date=due_date, amount=amount, reference=reference, payer_name=payer_name, bank_notification=bank_notification, )
[docs] def add_payment_cancellation( self, *, kid: str, due_date: datetime.date, amount: Decimal, reference: Optional[str] = None, payer_name: Optional[str] = None, bank_notification: Union[bool, str] = False, ) -> 'PaymentRequest': """Add an AvtaleGiro cancellation to the assignment. The assignment must have service code :attr:`~netsgiro.ServiceCode.AVTALEGIRO` and assignment type :attr:`~AssignmentType.AVTALEGIRO_CANCELLATIONS`. Otherwise, the cancellation must be identical to the payment request it is cancelling. """ assert ( self.service_code == ServiceCode.AVTALEGIRO ), 'Can only add cancellation to AvtaleGiro assignments' assert ( self.type == AssignmentType.AVTALEGIRO_CANCELLATIONS ), 'Can only add cancellation to cancellation assignments' return self._add_avtalegiro_transaction( transaction_type=TransactionType.AVTALEGIRO_CANCELLATION, kid=kid, due_date=due_date, amount=amount, reference=reference, payer_name=payer_name, bank_notification=bank_notification, )
def _add_avtalegiro_transaction( self, *, transaction_type: TransactionType, kid: str, due_date: datetime.date, amount: Decimal, reference: Optional[str] = None, payer_name: Optional[str] = None, bank_notification: Union[str, bool] = False, ) -> 'PaymentRequest': text = bank_notification if isinstance(bank_notification, str) else '' number = self._next_transaction_number self._next_transaction_number += 1 transaction = PaymentRequest( service_code=self.service_code, type=transaction_type, number=number, date=due_date, amount=amount, kid=kid, reference=reference, text=text, payer_name=payer_name, ) self.transactions.append(transaction) return transaction
[docs] def get_num_transactions(self) -> int: """Get number of transactions in the assignment.""" return len(self.transactions)
[docs] def get_num_records(self) -> int: """Get number of records in the assignment. Includes the assignment's start and end record. """ return 2 + sum(len(list(transaction.to_records())) for transaction in self.transactions)
[docs] def get_total_amount(self) -> Decimal: """Get the total amount from all transactions in the assignment.""" total = Decimal(0) for t in self.transactions: iter_amount = getattr(t, 'amount', None) if iter_amount: total += iter_amount return total
def _get_earliest_or_latest_transaction_date(self, latest: bool) -> Optional[datetime.date]: """Get earliest or latest date from the assignment's transactions.""" date_: Optional[datetime.date] = None for t in self.transactions: iter_date = getattr(t, 'date', None) if iter_date and ( date_ is None or (iter_date > date_ if latest else iter_date < date_) ): date_ = iter_date return date_
[docs] def get_earliest_transaction_date(self) -> Optional[datetime.date]: """Get earliest date from the assignment's transactions.""" return self._get_earliest_or_latest_transaction_date(latest=False)
[docs] def get_latest_transaction_date(self) -> Optional[datetime.date]: """Get latest date from the assignment's transactions.""" return self._get_earliest_or_latest_transaction_date(latest=True)
[docs]@define class Agreement: """Agreement contains an AvtaleGiro agreement update. Agreements are only found in assignments of the :attr:`~AssignmentType.AVTALEGIRO_AGREEMENTS` type, which are only created by Nets. """ #: The service code. One of :class:`~netsgiro.ServiceCode`. service_code: ServiceCode = field(converter=to_service_code) #: Transaction number. Unique and ordered within an assignment. number: int = field(validator=instance_of(int)) #: Type of agreement registration update. #: One of :class:`~AvtaleGiroRegistrationType`. registration_type: 'AvtaleGiroRegistrationType' = field( converter=to_avtalegiro_registration_type ) #: KID number to identify the customer and invoice. kid: Optional[str] = field(validator=optional(instance_of(str))) #: Whether the payer wants notification about payment requests. notify: bool = field(validator=instance_of(bool)) TRANSACTION_TYPE: ClassVar[TransactionType] = TransactionType.AVTALEGIRO_AGREEMENT
[docs] @classmethod def from_records(cls, records: List[AvtaleGiroAgreement]) -> 'Agreement': """Build an Agreement object from a list of record objects.""" assert len(records) == 1 record = records[0] assert isinstance(record, AvtaleGiroAgreement) assert record.transaction_type == TransactionType.AVTALEGIRO_AGREEMENT return cls( service_code=record.service_code, number=record.transaction_number, registration_type=record.registration_type, kid=record.kid, notify=record.notify, )
[docs] def to_records(self) -> Iterable[AvtaleGiroAgreement]: """Convert the agreement to a list of records.""" yield AvtaleGiroAgreement( service_code=self.service_code, transaction_type=self.TRANSACTION_TYPE, transaction_number=self.number, registration_type=self.registration_type, kid=self.kid, notify=self.notify, )
[docs]@define class PaymentRequest: """PaymentRequest contains an AvtaleGiro payment request or cancellation. To create a transaction, you will normally use the helper methods on :class:`~netsgiro.Assignment`: :meth:`~netsgiro.Assignment.add_payment_request` and :meth:`~netsgiro.Assignment.add_payment_cancellation`. """ #: The service code. One of :class:`~netsgiro.ServiceCode`. service_code: ServiceCode = field(converter=to_service_code) #: The transaction type. One of :class:`~TransactionType`. type: TransactionType = field(converter=to_transaction_type) #: Transaction number. Unique and ordered within an assignment. number: int = field(validator=instance_of(int)) #: The due date. date: datetime.date = field(validator=instance_of(datetime.date)) #: Transaction amount in NOK with two decimals. amount: Decimal = field(converter=Decimal) #: KID number to identify the customer and invoice. kid: Optional[str] = field(validator=optional(instance_of(str))) #: This is a specification line that will, if set, be displayed on the #: payers account statement. Alphanumeric, max 25 chars. reference: Optional[str] = field(validator=optional(instance_of(str))) #: This is up to 42 lines of 80 chars each of free text used by the bank to #: notify the payer about the payment request. It is not used if the payee #: is responsible for notifying the payer. text: Optional[str] = field(validator=optional(instance_of(str))) #: The value is only used to help the payee cross-reference reports from #: Nets with their own records. It is not vi.attr.ible to the payer. payer_name: Optional[str] = field(validator=optional(instance_of(str))) @property def amount_in_cents(self) -> int: """Transaction amount in NOK cents.""" return int(self.amount * 100)
[docs] @classmethod def from_records(cls, records: List[TR]) -> 'PaymentRequest': """Build a Transaction object from a list of record objects.""" amount_item_1 = records.pop(0) assert isinstance(amount_item_1, TransactionAmountItem1) amount_item_2 = records.pop(0) assert isinstance(amount_item_2, TransactionAmountItem2) text = TransactionSpecification.to_text(cast(List['TransactionSpecification'], records)) return cls( service_code=amount_item_1.service_code, type=amount_item_1.transaction_type, number=amount_item_1.transaction_number, date=amount_item_1.nets_date, amount=Decimal(amount_item_1.amount) / 100, kid=amount_item_1.kid, reference=amount_item_2.reference, text=text, payer_name=amount_item_2.payer_name, )
[docs] def to_records( self, ) -> Iterable[Union[TransactionAmountItem1, TransactionAmountItem2, TransactionSpecification]]: """Convert the transaction to a list of records.""" yield TransactionAmountItem1( service_code=self.service_code, transaction_type=self.type, transaction_number=self.number, nets_date=self.date, amount=self.amount_in_cents, kid=self.kid, ) yield TransactionAmountItem2( service_code=self.service_code, transaction_type=self.type, transaction_number=self.number, reference=self.reference, payer_name=self.payer_name, ) if self.type == TransactionType.AVTALEGIRO_WITH_BANK_NOTIFICATION: yield from TransactionSpecification.from_text( service_code=self.service_code, transaction_type=self.type, transaction_number=self.number, text=self.text, )
[docs]@define class Transaction: """Transaction contains an OCR Giro transaction. Transactions are found in assignments with the service code :attr:`~netsgiro.ServiceCode.OCR_GIRO` type, which are only created by Nets. """ #: The service code. One of :class:`~netsgiro.ServiceCode`. service_code: ServiceCode = field(converter=to_service_code) #: The transaction type. One of :class:`~TransactionType`. type: TransactionType = field(converter=to_transaction_type) #: Transaction number. Unique and ordered within an assignment. number: int = field(validator=instance_of(int)) #: Nets' processing date. date: datetime.date = field(validator=instance_of(datetime.date)) #: Transaction amount in NOK with two decimals. amount: Decimal = field(converter=Decimal) #: KID number to identify the customer and invoice. kid: Optional[str] = field(validator=optional(instance_of(str))) #: The value depends on the payment method. reference: Optional[str] = field(validator=optional(instance_of(str))) #: Up to 40 chars of free text from the payment terminal. text: Optional[str] = field(validator=optional(instance_of(str))) #: Used for OCR Giro. centre_id: Optional[str] = field(validator=optional(str_of_length(2))) #: Used for OCR Giro. day_code: Optional[int] = field(validator=optional(instance_of(int))) #: Used for OCR Giro. partial_settlement_number: Optional[int] = field(validator=optional(instance_of(int))) #: Used for OCR Giro. partial_settlement_serial_number: Optional[str] = field(validator=optional(str_of_length(5))) #: Used for OCR Giro. sign: Optional[str] = field(validator=optional(str_of_length(1))) #: Used for OCR Giro. form_number: Optional[str] = field(validator=optional(str_of_length(10))) #: Used for OCR Giro. bank_date: Optional[datetime.date] = field(validator=optional(instance_of(datetime.date))) #: Used for OCR Giro. debit_account: Optional[str] = field(validator=optional(str_of_length(11))) _filler: Optional[str] = field(validator=optional(str_of_length(7))) @property def amount_in_cents(self) -> int: """Transaction amount in NOK cents.""" return int(self.amount * 100)
[docs] @classmethod def from_records(cls, records: List[TR]) -> 'Transaction': """Build a Transaction object from a list of record objects.""" amount_item_1 = records.pop(0) assert isinstance(amount_item_1, TransactionAmountItem1) amount_item_2 = records.pop(0) assert isinstance(amount_item_2, TransactionAmountItem2) if len(records) == 1 and isinstance(records[0], TransactionAmountItem3): text = records[0].text else: text = None return cls( service_code=amount_item_1.service_code, type=amount_item_1.transaction_type, number=amount_item_1.transaction_number, date=amount_item_1.nets_date, amount=Decimal(amount_item_1.amount) / 100, kid=amount_item_1.kid, reference=amount_item_2.reference, text=text, centre_id=amount_item_1.centre_id, day_code=amount_item_1.day_code, partial_settlement_number=amount_item_1.partial_settlement_number, partial_settlement_serial_number=amount_item_1.partial_settlement_serial_number, sign=amount_item_1.sign, form_number=amount_item_2.form_number, bank_date=amount_item_2.bank_date, debit_account=amount_item_2.debit_account, filler=amount_item_2._filler, )
[docs] def to_records(self) -> Iterable[TransactionAmountItems]: """Convert the transaction to a list of records.""" yield TransactionAmountItem1( service_code=self.service_code, transaction_type=self.type, transaction_number=self.number, nets_date=self.date, amount=self.amount_in_cents, kid=self.kid, centre_id=self.centre_id, day_code=self.day_code, partial_settlement_number=self.partial_settlement_number, partial_settlement_serial_number=self.partial_settlement_serial_number, sign=self.sign, ) yield TransactionAmountItem2( service_code=self.service_code, transaction_type=self.type, transaction_number=self.number, reference=self.reference, form_number=self.form_number, bank_date=self.bank_date, debit_account=self.debit_account, filler=self._filler, ) if self.type in ( TransactionType.REVERSING_WITH_TEXT, TransactionType.PURCHASE_WITH_TEXT, ): yield TransactionAmountItem3( service_code=self.service_code, transaction_type=self.type, transaction_number=self.number, text=self.text, )
[docs]def parse(data: str) -> Transmission: """Parse an OCR file into a Transmission object.""" return Transmission.from_records(records_parse(data))