Chaim
9402c7302f
Backend follow-up to INV-UI9 (X6 §"אודיט-נלווה"): make server-side
datetimes explicit so flipping the container to TZ=Asia/Jerusalem cannot
silently skew stored or displayed timestamps. Storage stays UTC; human
display is Israel time.
- audit.py: naive datetime.utcnow() → datetime.now(timezone.utc) for the
TIMESTAMPTZ audit_log.created_at. Under TZ=Asia/Jerusalem the naive UTC
wall-clock would be read as Israel-local and stored ~3h off. (§6: no
silent skew.)
- calibrate_effort.py: _ts() report stamp now Israel-time via
ZoneInfo("Asia/Jerusalem") with a filename-safe "-IL" suffix (script runs
on the UTC host); added a labeled "שעון ישראל" line to the report header
so the chair-facing timestamp is unambiguous. --self-test still ALL PASS.
- research_md.py: display fields made deterministic Israel-tz —
update_chair_position timestamp (datetime.now(IL_TZ)) and parse() mtime
(datetime.fromtimestamp(..., tz=IL_TZ)) — correct regardless of process TZ.
Left intentionally Israel-local-day date.today()/year-cutoff call sites
untouched (drafting/party_claims_summary/block_writer/court_fetch_supreme)
— a Jerusalem committee wants the Israel calendar day; they resolve
correctly once TZ flips.
Invariants: INV-UI9 (store-UTC / display-Israel, server side); §6
(timezone-aware datetimes, no silent ~3h skew).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
"""Audit log — תיעוד כל פעולה במערכת.
|
|
|
|
כל פעולה מתועדת: מי, מתי, מה, על איזה תיק.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from uuid import UUID, uuid4
|
|
|
|
from legal_mcp.services import db
|
|
|
|
logger = logging.getLogger("audit")
|
|
|
|
|
|
async def log_action(
|
|
action: str,
|
|
case_id: UUID | None = None,
|
|
document_id: UUID | None = None,
|
|
details: dict | None = None,
|
|
user: str = "system",
|
|
) -> None:
|
|
"""רישום פעולה ב-audit log.
|
|
|
|
Args:
|
|
action: סוג הפעולה (upload, classify, extract_claims, set_outcome, write_block, export, etc.)
|
|
case_id: מזהה תיק (אם רלוונטי)
|
|
document_id: מזהה מסמך (אם רלוונטי)
|
|
details: פרטים נוספים
|
|
user: מזהה המשתמש
|
|
"""
|
|
pool = await db.get_pool()
|
|
async with pool.acquire() as conn:
|
|
await conn.execute(
|
|
"""INSERT INTO audit_log (id, action, case_id, document_id, details, actor, created_at)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7)""",
|
|
uuid4(), action, case_id, document_id,
|
|
json.dumps(details or {}, ensure_ascii=False, default=str),
|
|
user, datetime.now(timezone.utc),
|
|
)
|
|
logger.info("AUDIT: %s | case=%s | user=%s | %s", action, case_id, user,
|
|
json.dumps(details or {}, ensure_ascii=False)[:200])
|
|
|
|
|
|
async def log_action_safe(
|
|
action: str,
|
|
case_id: "UUID | None" = None,
|
|
document_id: "UUID | None" = None,
|
|
details: dict | None = None,
|
|
user: str = "system",
|
|
) -> None:
|
|
"""Non-fatal audit: never let an audit-log failure break the caller's action.
|
|
|
|
The authoritative integrity trail is git (X5 §2.1); audit_log is the
|
|
'who/what/when' observability layer, so a write failure is logged as a
|
|
warning and swallowed.
|
|
"""
|
|
try:
|
|
await log_action(action, case_id=case_id, document_id=document_id,
|
|
details=details, user=user)
|
|
except Exception as e: # noqa: BLE001 — observability must not break the op
|
|
logger.warning("audit log_action failed (non-fatal) for %s: %s", action, e)
|
|
|
|
|
|
async def get_audit_log(
|
|
case_id: UUID | None = None,
|
|
action: str | None = None,
|
|
limit: int = 50,
|
|
) -> list[dict]:
|
|
"""שליפת audit log."""
|
|
pool = await db.get_pool()
|
|
conditions = []
|
|
params: list = []
|
|
idx = 1
|
|
|
|
if case_id:
|
|
conditions.append(f"case_id = ${idx}")
|
|
params.append(case_id)
|
|
idx += 1
|
|
if action:
|
|
conditions.append(f"action = ${idx}")
|
|
params.append(action)
|
|
idx += 1
|
|
|
|
where = f"WHERE {' AND '.join(conditions)}" if conditions else ""
|
|
params.append(limit)
|
|
|
|
async with pool.acquire() as conn:
|
|
rows = await conn.fetch(
|
|
f"SELECT * FROM audit_log {where} ORDER BY created_at DESC LIMIT ${idx}",
|
|
*params,
|
|
)
|
|
|
|
return [dict(r) for r in rows]
|