42 lines
1.1 KiB
Python
42 lines
1.1 KiB
Python
import json
|
|
import logging
|
|
from database import get_db
|
|
|
|
logger = logging.getLogger("manufacturing.audit")
|
|
|
|
|
|
async def log_action(
|
|
admin_user: str,
|
|
action: str,
|
|
serial_number: str | None = None,
|
|
detail: dict | None = None,
|
|
):
|
|
"""Write a manufacturing audit entry to SQLite.
|
|
|
|
action examples: batch_created, device_flashed, device_assigned, status_updated
|
|
"""
|
|
try:
|
|
db = await get_db()
|
|
await db.execute(
|
|
"""INSERT INTO mfg_audit_log (admin_user, action, serial_number, detail)
|
|
VALUES (?, ?, ?, ?)""",
|
|
(
|
|
admin_user,
|
|
action,
|
|
serial_number,
|
|
json.dumps(detail) if detail else None,
|
|
),
|
|
)
|
|
await db.commit()
|
|
except Exception as e:
|
|
logger.error(f"Failed to write audit log: {e}")
|
|
|
|
|
|
async def get_recent(limit: int = 20) -> list[dict]:
|
|
db = await get_db()
|
|
rows = await db.execute_fetchall(
|
|
"SELECT * FROM mfg_audit_log ORDER BY timestamp DESC LIMIT ?",
|
|
(limit,),
|
|
)
|
|
return [dict(r) for r in rows]
|