|
|
@@ -17,11 +17,28 @@ import threading
|
|
|
import time
|
|
|
import traceback
|
|
|
from datetime import datetime, timezone
|
|
|
+from logging.handlers import RotatingFileHandler
|
|
|
|
|
|
from pps_client import PPSClient, PPSError
|
|
|
|
|
|
log = logging.getLogger("pps.worker")
|
|
|
|
|
|
+# Dedicated operations/audit log — one line per message acted on. Written to its own
|
|
|
+# file (see _setup_ops_log) and NOT propagated to the root/stdout logger.
|
|
|
+ops_log = logging.getLogger("pps.ops")
|
|
|
+ops_log.propagate = False
|
|
|
+
|
|
|
+
|
|
|
+def _setup_ops_log(path: str) -> None:
|
|
|
+ """Attach a rotating file handler to the ops logger (idempotent)."""
|
|
|
+ for h in ops_log.handlers:
|
|
|
+ if isinstance(h, RotatingFileHandler) and getattr(h, "baseFilename", "").endswith(path.split("/")[-1]):
|
|
|
+ return # already configured
|
|
|
+ handler = RotatingFileHandler(path, maxBytes=5_000_000, backupCount=5, encoding="utf-8")
|
|
|
+ handler.setFormatter(logging.Formatter("%(asctime)s %(message)s"))
|
|
|
+ ops_log.addHandler(handler)
|
|
|
+ ops_log.setLevel(logging.INFO)
|
|
|
+
|
|
|
# Actions understood from the frontend.
|
|
|
VALID_ACTIONS = {"release", "report_release", "delete", "move"}
|
|
|
|
|
|
@@ -33,7 +50,8 @@ def _now() -> str:
|
|
|
|
|
|
|
|
|
class JobQueue:
|
|
|
- def __init__(self, db_path: str, pps: PPSClient, qconfig: dict):
|
|
|
+ def __init__(self, db_path: str, pps: PPSClient, qconfig: dict,
|
|
|
+ ops_log_path: str = "worker.log"):
|
|
|
self.pps = pps
|
|
|
self.cfg = qconfig
|
|
|
self.chunk_size = int(qconfig.get("chunk_size", 25))
|
|
|
@@ -43,6 +61,8 @@ class JobQueue:
|
|
|
self._init_db()
|
|
|
self._recover()
|
|
|
self._stop = threading.Event()
|
|
|
+ _setup_ops_log(ops_log_path)
|
|
|
+ self.ops_log_path = ops_log_path
|
|
|
|
|
|
# -------------------------------------------------------------- schema/setup
|
|
|
|
|
|
@@ -67,12 +87,20 @@ class JobQueue:
|
|
|
localguid TEXT NOT NULL,
|
|
|
guid TEXT,
|
|
|
folder TEXT NOT NULL,
|
|
|
+ subject TEXT,
|
|
|
+ sender TEXT,
|
|
|
+ recipient TEXT,
|
|
|
FOREIGN KEY (job_id) REFERENCES jobs(id)
|
|
|
);
|
|
|
CREATE INDEX IF NOT EXISTS idx_job_items_folder ON job_items(folder);
|
|
|
CREATE INDEX IF NOT EXISTS idx_job_items_job ON job_items(job_id);
|
|
|
"""
|
|
|
)
|
|
|
+ # Migrate older DBs that lack the display columns.
|
|
|
+ existing = {r["name"] for r in self._conn.execute("PRAGMA table_info(job_items)")}
|
|
|
+ for col in ("subject", "sender", "recipient"):
|
|
|
+ if col not in existing:
|
|
|
+ self._conn.execute(f"ALTER TABLE job_items ADD COLUMN {col} TEXT")
|
|
|
self._conn.commit()
|
|
|
|
|
|
def _recover(self) -> None:
|
|
|
@@ -89,7 +117,8 @@ class JobQueue:
|
|
|
# -------------------------------------------------------------- public API
|
|
|
|
|
|
def enqueue(self, action: str, folder: str, items: list[dict], extra: dict) -> int:
|
|
|
- """Insert a job. `items` is a list of {'localguid':..., 'guid':...} dicts."""
|
|
|
+ """Insert a job. `items` is a list of dicts with keys:
|
|
|
+ localguid (required), guid, subject, sender, recipient (all optional)."""
|
|
|
if action not in VALID_ACTIONS:
|
|
|
raise ValueError(f"unknown action: {action}")
|
|
|
now = _now()
|
|
|
@@ -101,10 +130,16 @@ class JobQueue:
|
|
|
)
|
|
|
job_id = cur.lastrowid
|
|
|
self._conn.executemany(
|
|
|
- "INSERT INTO job_items (job_id, localguid, guid, folder) VALUES (?,?,?,?)",
|
|
|
- [(job_id, it["localguid"], it.get("guid"), folder) for it in items],
|
|
|
+ "INSERT INTO job_items (job_id, localguid, guid, folder, subject, sender, recipient)"
|
|
|
+ " VALUES (?,?,?,?,?,?,?)",
|
|
|
+ [
|
|
|
+ (job_id, it["localguid"], it.get("guid"), folder,
|
|
|
+ it.get("subject"), it.get("sender"), it.get("recipient"))
|
|
|
+ for it in items
|
|
|
+ ],
|
|
|
)
|
|
|
self._conn.commit()
|
|
|
+ log.info("enqueued job #%d: action=%s folder=%r items=%d", job_id, action, folder, len(items))
|
|
|
return job_id
|
|
|
|
|
|
def acted_localguids(self, folder: str) -> set[str]:
|
|
|
@@ -177,6 +212,7 @@ class JobQueue:
|
|
|
"job #%d start: action=%s folder=%r total=%d", job_id, action, folder, len(items)
|
|
|
)
|
|
|
|
|
|
+ dst = self._destination(action, extra)
|
|
|
errors: list[str] = []
|
|
|
processed = 0
|
|
|
for chunk in _chunks(items, self.chunk_size):
|
|
|
@@ -184,12 +220,15 @@ class JobQueue:
|
|
|
self._run_action(action, folder, chunk, extra)
|
|
|
processed += len(chunk)
|
|
|
self._set_processed(job_id, processed)
|
|
|
+ self._log_ops(job_id, action, folder, dst, chunk, "ok", None)
|
|
|
except PPSError as exc:
|
|
|
log.error("job #%d chunk failed: %s", job_id, exc)
|
|
|
errors.append(f"chunk failed: {exc}")
|
|
|
+ self._log_ops(job_id, action, folder, dst, chunk, "FAILED", str(exc))
|
|
|
except Exception as exc: # noqa: BLE001
|
|
|
log.exception("job #%d chunk raised", job_id)
|
|
|
errors.append(f"chunk failed: {exc}")
|
|
|
+ self._log_ops(job_id, action, folder, dst, chunk, "FAILED", str(exc))
|
|
|
|
|
|
if errors:
|
|
|
log.error("job #%d FAILED: %d/%d processed, %d chunk error(s)",
|
|
|
@@ -199,6 +238,36 @@ class JobQueue:
|
|
|
log.info("job #%d done: %d/%d processed", job_id, processed, len(items))
|
|
|
self._mark_done(job_id)
|
|
|
|
|
|
+ def _destination(self, action: str, extra: dict) -> str | None:
|
|
|
+ """The folder messages end up in, for the ops log."""
|
|
|
+ if action == "move":
|
|
|
+ return extra.get("targetfolder")
|
|
|
+ if action == "report_release":
|
|
|
+ return self.cfg.get("report_release_folder")
|
|
|
+ if action in ("release", "delete"):
|
|
|
+ return self.cfg.get("deleted_folder")
|
|
|
+ return None
|
|
|
+
|
|
|
+ def _log_ops(self, job_id: int, action: str, src: str, dst: str | None,
|
|
|
+ chunk: list[dict], result: str, error: str | None) -> None:
|
|
|
+ """Write one audit line per message in the chunk to the operations log."""
|
|
|
+ for it in chunk:
|
|
|
+ fields = [
|
|
|
+ f"job=#{job_id}",
|
|
|
+ f"action={action}",
|
|
|
+ f"result={result}",
|
|
|
+ f"src={src!r}",
|
|
|
+ f"dst={dst!r}",
|
|
|
+ f"localguid={it.get('localguid')}",
|
|
|
+ f"guid={it.get('guid')}",
|
|
|
+ f"from={_clip(it.get('sender'))!r}",
|
|
|
+ f"rcpt={_clip(it.get('recipient'))!r}",
|
|
|
+ f"subject={_clip(it.get('subject'))!r}",
|
|
|
+ ]
|
|
|
+ if error:
|
|
|
+ fields.append(f"error={_clip(error, 300)!r}")
|
|
|
+ ops_log.info(" ".join(fields))
|
|
|
+
|
|
|
# -------------------------------------------------------------- action logic
|
|
|
|
|
|
def _run_action(self, action: str, folder: str, chunk: list[dict], extra: dict) -> None:
|
|
|
@@ -272,7 +341,8 @@ class JobQueue:
|
|
|
def _items_for(self, job_id: int) -> list[dict]:
|
|
|
with self._lock:
|
|
|
rows = self._conn.execute(
|
|
|
- "SELECT localguid, guid FROM job_items WHERE job_id=?", (job_id,)
|
|
|
+ "SELECT localguid, guid, subject, sender, recipient"
|
|
|
+ " FROM job_items WHERE job_id=?", (job_id,)
|
|
|
).fetchall()
|
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
@@ -304,3 +374,9 @@ class JobQueue:
|
|
|
def _chunks(seq: list, size: int):
|
|
|
for i in range(0, len(seq), size):
|
|
|
yield seq[i : i + size]
|
|
|
+
|
|
|
+
|
|
|
+def _clip(value, length: int = 120) -> str:
|
|
|
+ """Collapse newlines and truncate a value for a single-line log field."""
|
|
|
+ s = " ".join(str(value or "").split())
|
|
|
+ return s[:length] + ("…" if len(s) > length else "")
|
