path: root/jimbrella/database.py

from datetime import datetime, timedelta
from .csv_table import CsvTable
from .admin_log import AdminLog
from .utils import human_datetime, human_timedelta
from .config import DUE_HOURS, ADMIN_LOG_PATH
from .exceptions import *

STATUSES = ["available", "lent", "overdue", "maintenance", "withheld", "unknown"]


class Database(CsvTable):
    def __init__(self, path):
        """A database of all umbrellas and their current state.

        Currently, the data are represented in a csv file.

        Explanation of data types:
        - id        | a unique numerical identity of unsigned integer type.
        - uint      | whatever fits the regex /[0-9]+/. despite carrying a numerical value, it is
                    | internally represented as a string.
        - string    | any UTF-8 string.
        - date      | an ISO 8601 date of format "YYYY-MM-DDThh:mm:ss+08:00".

        The status of an umbrella consists of:
        - serial        | (id) unique identifier for the umbrella.
        - alias         | (string) future compatibility. a human readable (preferably cute) name
                        | for a particular umbrella
        - status        | (string) one of ("available", "lent", "overdue", "maintenance",
                        | "withheld", "unknown")
                        | available     : is in service on the stand
                        | lent          : is in temporary possession of a user
                        | overdue       : is in overly prolonged possession of a user
                        | maintenance   : is on the stand but not in service
                        | withheld      : is not on the stand but rather in the possession of an
                        |                 administrator
                        | unknown       : none of the above
                        | NOTE: the values above are subject to changes.
        - tenant_name   | (string) the person in temporary possession of the umbrella.
        - tenant_id     | (uint) student or faculty ID.
        - tenant_phone  | (uint) phone number via which to contact tenant when the lease is due.
        - tenant_email  | (string) future compatibility. empty for the time being.
        - lent_at       | (date) if status is "lent", lent_at is the submission time of the user's
                        | jForm answer sheet. is blank otherwise.
        """
        super().__init__(
            path,
            [
                {"name": "serial", "deserializer": int},
                {"name": "alias"},
                {"name": "status"},
                {"name": "tenant_name"},
                {"name": "tenant_id"},
                {"name": "tenant_phone"},
                {"name": "tenant_email"},
                {
                    "name": "lent_at",
                    "serializer": lambda d: d.isoformat(timespec="seconds")
                    if d
                    else "",
                    "deserializer": lambda d: datetime.fromisoformat(d) if d else None,
                },
            ],
        )
        self.admin_log = AdminLog(ADMIN_LOG_PATH)

    def _find_by_serial(self, serial: int) -> dict:
        """Given a serial number, returns an umbrella with such serial.
        If there is no such umbrella, returns None.
        """
        umbrellas = self._read()
        for umb in umbrellas:
            if umb["serial"] == serial:
                return umb

    def read(self) -> list:
        """An interface to `_read()` with supplemental data included.

        All exposed methods with a return value should use this method instead of `_read()`.

        Supplemental data:
        - lent_at_str:          string representation for lent_at.
        - lent_time_ago:        time since umbrella was taken away by tenant. if umbrella is not
                                taken away, its value is None.
        - lent_time_ago_str:    string representation for lent_time_ago.
        """
        umbrellas = self._read()
        now = datetime.now()
        for idx, umb in enumerate(umbrellas):
            if umb["status"] in ("lent", "overdue"):
                umbrellas[idx]["lent_at_str"] = human_datetime(umb["lent_at"])
                lent_time_ago = now - umb["lent_at"]
                umbrellas[idx]["lent_time_ago"] = lent_time_ago
                umbrellas[idx]["lent_time_ago_str"] = human_timedelta(lent_time_ago)
        return umbrellas

    def update(self, umb) -> list:
        """An interface to `_update()` with added convenience and safeguards.

        Convenience:    Not all fields in an umbrella dict need to be present in `umb`. If an
                        optional field is not there, its value is left untouched. Every field other
                        than `serial` is optional.

                        `serial` may be an int or a str that evaluates to the correct serial when
                        int() is applied to it.

                        `tenant_id` and `tenant_phone` may be a str or an int.

                        `lent_at` in `umb` may either be a datetime.datetime object, or an ISO 8601
                        formatted string.

        Safeguards:     Invalid values are rejected as an Exception, in the default case, an
                        UmbrellaValueError.

        Both:           When a value is set, values of unneeded fields are automatically discarded
                        if applicable. For example, when `status` is set to "available", `tenant_*`
                        and `lent_at` are no longer needed; in fact, their existence is dangerous
                        as it implies the status of the umbrella is "lent" or "overdue", which is
                        not the case. Therefore, if `status` is set to "available", the unnecessary
                        fields mentioned above are made blank, and these fields supplied in `umb`
                        are ignored.

        Returns updated database (same as `_update()`).
        """
        # `serial` must be specified.
        if "serial" not in umb:
            raise UmbrellaValueError("serial")

        # check if umbrella #<serial> exists in database
        umb["serial"] = int(umb["serial"])
        umb_in_db = self._find_by_serial(umb["serial"])
        if umb_in_db is None:
            raise UmbrellaNotFoundError(umb["serial"])

        status = umb_in_db["status"]
        if "status" in umb and umb["status"]:
            if status not in STATUSES:
                raise UmbrellaValueError("status")  # invalid

            # admin specifies a (perhaps different) status
            status = umb["status"]

        if "lent_at" in umb and umb["lent_at"]:
            # check if `umb` has valid date (`lent_at`) if any
            if isinstance(umb["lent_at"], datetime):
                lent_at = umb["lent_at"]
            elif isinstance(umb["lent_at"], str):
                try:
                    umb["lent_at"] = datetime.fromisoformat(umb["lent_at"])
                except ValueError:
                    raise UmbrellaValueError("lent_at")
            else:
                raise UmbrellaValueError("lent_at")

        # we will now check the validity of `umb` based on `status`
        if status == "available":
            # discard unneeded fields
            for key in ["tenant_name", "tenant_id", "tenant_phone", "tenant_email"]:
                umb[key] = ""
            umb["lent_at"] = None
        elif status in ("lent", "overdue"):
            # copy values in database into unfilled or non-existent fields of `umb`
            for key in [
                "tenant_name",
                "tenant_id",
                "tenant_phone",
                "tenant_email",
                "lent_at",
            ]:
                umb[key] = umb[key] if (key in umb and umb[key]) else umb_in_db[key]

            if not umb["lent_at"]:
                raise UmbrellaValueError("lent_at")

        return self._update(umb)

    @staticmethod
    def group_by_status(umbrellas) -> dict:
        """(static method) Returns umbrellas grouped into a dict by their status."""
        keys = set([umb["status"] for umb in umbrellas])
        # initiate statuses: each status is []
        statuses = dict.fromkeys(STATUSES, [])
        for key in keys:
            statuses[key] = [umb for umb in umbrellas if umb["status"] == key]
        return statuses

    @staticmethod
    def find_overdue(umbrellas) -> list:
        """(static method) Returns umbrellas in possession of their tenant for too long."""
        now = datetime.now()
        return [
            umb
            for umb in umbrellas
            if umb["lent_at"] is not None
            and now - umb["lent_at"] > timedelta(hours=DUE_HOURS)
        ]

    def take_away(
        self, serial, date, tenant_name, tenant_id, tenant_phone="", tenant_email=""
    ) -> None:
        """When a user has borrowed an umbrella."""
        umb = self._find_by_serial(serial)
        if umb is None:
            raise UmbrellaNotFoundError(serial)
        elif umb["status"] != "available":
            raise UmbrellaStatusError
        umb["status"] = "lent"
        umb["tenant_name"] = tenant_name
        umb["tenant_id"] = tenant_id
        umb["tenant_phone"] = tenant_phone
        umb["tenant_email"] = tenant_email
        umb["lent_at"] = date
        self._update(umb)
        self.admin_log.log("TAKEAWAY", umb)

    def give_back(self, serial) -> None:
        """When a user has returned an umbrella."""
        umb = self._find_by_serial(serial)
        if umb is None:
            raise UmbrellaNotFoundError(serial)
        elif umb["status"] not in ("lent", "overdue"):
            raise UmbrellaStatusError
        self.admin_log.log("GIVEBACK", umb)
        umb["status"] = "available"
        for key in ["tenant_name", "tenant_id", "tenant_phone", "tenant_email"]:
            umb[key] = ""
        umb["lent_at"] = None
        self._update(umb)

    def mark_overdue(self, serial) -> None:
        """When an umbrella is overdue, change its status to "overdue"."""
        umb = self._find_by_serial(serial)
        if umb is None:
            raise UmbrellaNotFoundError(serial)
        elif umb["status"] != "lent":
            raise UmbrellaStatusError
        umb["status"] = "overdue"
        self._update(umb)
        self.admin_log.log("OVERDUE", umb)