dknog-membership/admidio/model.py

"""High-levle api."""

import logging
from datetime import date
from dataclasses import dataclass
from functools import cached_property

from typing import Final
from .db import AdmidioDB


MEMBER_ROLE = "Member"
FIELD_PAID_DATE = "PMB_PAID"
FIELD_FEE = "PMB_FEE"
FIELD_DUEDATE = "PMB_DUEDATE"
FIELD_EMAIL = "SYS_EMAIL"

LOG = logging.getLogger(__name__)


@dataclass
class AdmidioDBSettings:
    """Database settings class."""

    user: str
    password: str
    host: str
    database: str
    port: int = 3306


class AdmidioMemberFee:
    """Admidio member fee manager."""

    def __init__(self, user_id: int, connection: AdmidioDBSettings) -> None:
        """Create member fee manager class."""
        self.db: AdmidioDB = AdmidioDB(
            connection.user,
            connection.password,
            connection.host,
            connection.database,
            connection.port,
        )
        self.user_id: Final = user_id
        self._user_data: dict[str, str] | None = None

    @property
    def is_member(self) -> bool:
        """Check if user is a member."""
        role_ids = self.db.get_user_roles(self.user_id)
        role_map = self.db.get_roles()
        roles = [role_map[role_id] for role_id in role_ids]
        return MEMBER_ROLE in roles

    @property
    def member_fee(self) -> int:
        """Get membership fee."""
        roles = self.db.get_roles()
        member_role_id = next(
            iter(
                [
                    role_id
                    for role_id, role_name in roles.items()
                    if role_name == MEMBER_ROLE
                ]
            )
        )
        role_payments = self.db.get_role_payments()
        fee = role_payments.get(member_role_id)
        if not fee:
            raise RuntimeError("Error: No membership cost set on Member role.")
        return int(fee)

    @property
    def user_data(self) -> dict[str, str]:
        """Get user data."""
        if not self._user_data:
            self._user_data = self._get_user_data()
        return self._user_data

    def _get_user_data(self) -> dict[str, str]:
        """Get user data."""
        user_fields = self.db.get_custom_fields()
        user_field_ids = {value: key for key, value in user_fields.items()}
        adm_user_data = self.db.get_adm_user_data(self.user_id)
        user_data: dict[str, str] = {}
        for field_id, data in adm_user_data.items():
            if field_name := user_field_ids.get(field_id):
                user_data[field_name] = data
            else:
                LOG.warning("Could not find a field definition for %s", field_id)
        return user_data

    def _lookup_user_field(self, name: str) -> int:
        """Lookup the ID of a user field."""
        user_fields = self.db.get_custom_fields()
        field_id = user_fields.get(name)
        if not field_id:
            raise RuntimeError(f"Unable to find field ID for field {name}")
        return field_id

    @property
    def last_paid(self) -> date | None:
        """Get the date of last payment."""
        paid_data = self.user_data.get(FIELD_PAID_DATE)
        if not paid_data:
            return None
        paid_date = date.fromisoformat(paid_data)
        return paid_date

    @property
    def has_paid(self) -> bool:
        """Check if a user has paid."""
        if not self.last_paid:
            return False
        paid_days = date.today() - self.last_paid
        if paid_days.days > 365:
            return False
        return True

    @property
    def amount_due(self) -> int:
        """Get amount due."""
        if fee := self.user_data.get(FIELD_FEE):
            return int(fee)
        if not self.is_member:
            raise RuntimeError(
                "User does not seem to be a member, and has no due fees."
            )
        return self.member_fee

    def register_payment(self, amount_paid: int | None = None) -> None:
        """Register payment,"""
        date_field = self._lookup_user_field(FIELD_PAID_DATE)
        amount_field = self._lookup_user_field(FIELD_FEE)
        date_paid = date.today().isoformat()
        if not amount_paid:
            amount_paid = self.member_fee
        self.db.create_user_data(self.user_id, date_field, date_paid)
        self.db.create_user_data(self.user_id, amount_field, str(amount_paid))

    @classmethod
    def lookup_email(
        cls, email: str, settings: AdmidioDBSettings
    ) -> "AdmidioMemberFee | None":
        """Create instance by lookup up user email address."""
        db = AdmidioDB(
            settings.user,
            settings.password,
            settings.host,
            settings.database,
            settings.port,
        )
        fields = db.get_custom_fields()
        email_field_id = fields.get(FIELD_EMAIL)
        if not email_field_id:
            raise RuntimeError("Could not resolve email address field in database.")
        if user_id := db.get_user_id_by_field(email_field_id, email):
            return cls(user_id, settings)

        return None