Migrate AdminLog to sqlite

1 files changed, 93 insertions, 164 deletions

diff --git a/jimbrella/admin_log.py b/jimbrella/admin_log.py
index 3a390a1..f576f42 100644
--- a/jimbrella/admin_log.py
+++ b/jimbrella/admin_log.py
@@ -1,178 +1,107 @@
-import csv
-import os
-import logging
+import sqlite3
 from datetime import datetime
-from .lockfile import Lockfile
-from .utils import human_datetime
+from typing import Union
+from .config import config
 
 
 class AdminLog:
-    """Logs JImbrella-specific events into a file.
-
-    The file is intended to be read, deserialized, and represented in a user-friendly format to an
-    admin on the web console. The file format is csv, but the number and meanings of columns may
-    not be uniform for all rows. This is why it cannot be a subclass of CsvTable, which expects the
-    columns to be uniform.
-
-    For each row, there are a minimum of three columns. The first column is called the "event".
-    It describes the log entry by and large. What other columns in the row represent depends on
-    the event. The second column is always the date and time, in ISO8601 format. The last column
-    is called the "note", an optional textual note in natural language.
-
-    Here we list possible events, what information will follow them, and the scenario for each one.
-    Words surrounded by angle brackets (< >) are defined in JForm.get_unread, and those in square
-    brackets ([ ]) are defined here in AdminLog.
-
-    TAKEAWAY,[date],<key>,<name>,<id>,<phone>,<email>,[note]
-        A user borrows an umbrella normally.
-    GIVEBACK,[date],<key>,<name>,<id>,<phone>,<email>,[note]
-        A user returns an umbrella normally.
-    OVERDUE,[date],<key>,<name>,<id>,<phone>,<email>,[note]
-        An umbrella is judged overdue by JImbrella's routine process.
-    ADMIN_MODIFY_DB,[date],[admin_name],<key>,[column],[past_value],[new_value],[note]
-        An admin makes modifications to one cell of the database via the web console or API.
-        If multiple cells are modified, the same number of log entries are written, although
-        they can be multiplexed in one HTTP request.
+    """Logs admin operations in a database so they can be recalled and possibly undone.
+
+    Every time an admin modifies the Umbrellas table via the web console, the information below are
+    kept for archival and/or rollback.
+
+    Once created, the log cannot be changed.
+
+    - serial    | incremental serial number for each modification
+    - date      | modification time, same format as Umbrellas table
+    - actor     | admin username
+    - id        | umbrella id
+    - (status|tenant_(name|id|phone|email)|lent_at)_a:
+        respective values in database pertaining to umbrella #<id> before modification
+    - (status|tenant_(name|id|phone|email)|lent_at)_b:
+        same as above but after modification
+    - note      | optional note describing the modification
+
+    Schema:
+    CREATE TABLE AdminLog(
+        serial INT PRIMARY KEY,
+        date TEXT,
+        actor TEXT,
+        id INT,
+        status_a TEXT,
+        status_b TEXT,
+        tenant_name_a TEXT,
+        tenant_name_b TEXT,
+        tenant_id_a TEXT,
+        tenant_id_b TEXT,
+        tenant_phone_a TEXT,
+        tenant_phone_b TEXT,
+        tenant_email_a TEXT,
+        tenant_email_b TEXT,
+        lent_at_a TEXT,
+        lent_at_b TEXT,
+        note TEXT
+    );
     """
 
     def __init__(self, path: str):
         self.path = path
-        self.lockfile = Lockfile(self.path)
-        self.lockfile.unlock()
-
-    def _read(self) -> list:
-        """Deserialize admin log."""
-        # Create log if it does not yet exist
-        try:
-            f = open(self.path, "x")
-            f.close()
-        except FileExistsError:
-            pass
-
-        with open(self.path) as f:
-            reader = csv.reader(f)
-            logs = []
-            for ln, row in enumerate(reader):
-                try:
-                    event = row[0]
-                    entry = {
-                        "event": event,
-                        "date": datetime.fromisoformat(row[1]),
-                        "note": row[-1],
-                    }
-                    info = {}
-                    if event in ("TAKEAWAY", "GIVEBACK", "OVERDUE"):
-                        info = {
-                            "key": int(row[2]),
-                            "name": row[3],
-                            "id": row[4],
-                            "phone": row[5],
-                            "email": row[6],
-                        }
-                    elif event == "ADMIN_MODIFY_DB":
-                        info = {
-                            "admin_name": row[2],
-                            "id": int(row[3]),
-                            "column": row[4],
-                            "past_value": row[5],
-                            "new_value": row[6],
-                        }
-                    else:
-                        logging.warning(
-                            "%s:%d is not a recognized event. Skip.", self.path, ln
-                        )
-                        continue
-
-                    entry.update(info)
-                    logs.append(entry)
-                except Exception:
-                    logging.warning("%s:%d cannot be read. Skip.", self.path, ln)
 
-            f.close()
-            return logs
-
-    def read(self) -> list:
-        """Human-friendly representation of each of the admin log entries."""
-        logs = self._read()
-        friendly_logs = []
-        for entry in logs:
-            event = entry["event"]
-            tenant_info = "(ID: {id}, phone: {phone})"
-            if event == "TAKEAWAY":
-                description = "{name} borrowed umbrella #{key}. " + tenant_info
-            elif event == "GIVEBACK":
-                description = "{name} returned umbrella #{key}. " + tenant_info
-            elif event == "OVERDUE":
-                description = (
-                    "{name} missed the due for umbrella #{key}. " + tenant_info
-                )
-            elif event == "ADMIN_MODIFY_DB":
-                if not entry["past_value"]:
-                    description = (
-                        "{admin_name} set {column} of umbrella #{id} to {new_value}."
-                    )
-                elif not entry["new_value"]:
-                    description = "{admin_name} cleared {column} of umbrella #{id} (was {past_value})."
-                else:
-                    description = (
-                        "{admin_name} changed {column} of umbrella #{id} "
-                        "from {past_value} to {new_value}."
-                    )
-
-            friendly_logs.append(
-                {
-                    "date_str": human_datetime(entry["date"]),
-                    "event": event,
-                    "description": description.format(**entry),
-                    "note": entry["note"],
-                }
+    def read(self, maxlogs=20, serial=None) -> Union[dict, list]:
+        db = sqlite3.connect(self.path)
+        db.row_factory = sqlite3.Row
+        if serial is None:
+            data = db.execute("SELECT * FROM AdminLog ORDER BY serial DESC").fetchmany(
+                maxlogs
             )
+        else:
+            data = db.execute(
+                "SELECT * FROM AdminLog WHERE serial = ?", (serial,)
+            ).fetchone()
 
-        return friendly_logs
+        db.close()
+        return data
 
-    def log(self, event: str, entry: dict, date=None, note="") -> None:
-        """Serialize a log, and append it to the end of the log file.
-
-        Arguments:
-        event   | one of the events defined in the class docstring.
-        entry   | information pertaining to the log entry.
-        date    | if is None, defaults to `datetime.now()`.
-        note    | optional note.
-        """
-        self.lockfile.lock()
+    def log(
+        self, date: str, actor: str, id: int, umb_a: dict, umb_b: dict, note=""
+    ) -> None:
+        """Add an entry to log table.
 
-        with open(self.path, "a") as f:  # append only
-            writer = csv.writer(f)
-            date = date or datetime.now()
-            line = [event, date.isoformat()]
-            info = []
-            if event in ("TAKEAWAY", "GIVEBACK", "OVERDUE"):
-                info = [
-                    entry[col] if col in entry else ""
-                    for col in [
-                        "key",
-                        "name",
-                        "id",
-                        "phone",
-                        "email",
-                    ]
-                ]
-            elif event == "ADMIN_MODIFY_DB":
-                info = [
-                    entry[col] if col in entry else ""
-                    for col in [
-                        "admin_name",
-                        "id",
-                        "column",
-                        "past_value",
-                        "new_value",
-                    ]
-                ]
+        Does no authentication whatsoever. Authenticity must be ensured before calling this method.
 
-            line.extend(info)
-            line.append(note)
-            writer.writerow(line)
-
-            f.close()
-
-        self.lockfile.unlock()
+        `date` must be an ISO 8601 string.
+        """
+        db = sqlite3.connect(self.path)
+
+        # get serial of most recent log, add one
+        latest = db.execute(
+            "SELECT serial FROM AdminLog ORDER BY serial DESC"
+        ).fetchone()
+        if latest is None:
+            serial = 1
+        else:
+            serial = latest[0] + 1
+
+        db.execute(
+            "INSERT INTO AdminLog VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                serial,
+                date,
+                actor,
+                id,
+                umb_a.get("status", ""),
+                umb_b.get("status", ""),
+                umb_a.get("tenant_name", ""),
+                umb_b.get("tenant_name", ""),
+                umb_a.get("tenant_id", ""),
+                umb_b.get("tenant_id", ""),
+                umb_a.get("tenant_phone", ""),
+                umb_b.get("tenant_phone", ""),
+                umb_a.get("tenant_email", ""),
+                umb_b.get("tenant_email", ""),
+                umb_a.get("lent_at", ""),
+                umb_b.get("lent_at", ""),
+                note,
+            ),
+        )
+        db.close()