summaryrefslogtreecommitdiff
path: root/jimbrella/admin_log.py
diff options
context:
space:
mode:
Diffstat (limited to 'jimbrella/admin_log.py')
-rw-r--r--jimbrella/admin_log.py257
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()
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-15 02:23:32 +0000