111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
"""Shared API helpers: DB dependency, serialization, filename sanitization."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from datetime import date, datetime
|
|
|
|
from fastapi import Depends, HTTPException
|
|
from sqlalchemy.orm import Session as OrmSession
|
|
|
|
from ..db.database import get_db
|
|
from ..db import models
|
|
|
|
|
|
def db_dep() -> OrmSession: # thin alias so routes read cleanly
|
|
yield from get_db()
|
|
|
|
|
|
def get_session_or_404(session_id: int, db: OrmSession) -> models.Session:
|
|
s = db.get(models.Session, session_id)
|
|
if s is None:
|
|
raise HTTPException(status_code=404, detail=f"Session {session_id} not found")
|
|
return s
|
|
|
|
|
|
def is_blocked(s: models.Session) -> bool:
|
|
"""A month-end control failed with error severity — no figure may be published."""
|
|
return bool(getattr(s, "blocked_reason", ""))
|
|
|
|
|
|
def blocked_payload(s: models.Session) -> dict:
|
|
"""Read-endpoint response for a blocked close: the reason, never a number."""
|
|
return {
|
|
"available": False,
|
|
"blocked": True,
|
|
"blocked_reason": s.blocked_reason or "",
|
|
"detail": ("A month-end control failed, so no receivable figure is published for this "
|
|
"closing. Resolve the failed control on the Controls tab and re-run it."),
|
|
}
|
|
|
|
|
|
def ensure_not_blocked(s: models.Session) -> None:
|
|
"""Guard for actions that would put an unverified number into someone's hands."""
|
|
if is_blocked(s):
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail=f"This closing is blocked by a failed month-end control — "
|
|
f"{s.blocked_reason}",
|
|
)
|
|
|
|
|
|
def ensure_editable(s: models.Session) -> None:
|
|
"""Guard every mutating endpoint: a completed closing is locked history.
|
|
|
|
Its figures were signed off and possibly booked — silently editing them would make the
|
|
record disagree with what was published. Corrections go through an explicit reopen
|
|
(POST /sessions/{id}/reopen), which is visible and deliberate."""
|
|
if s.status == "completed":
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="This closing is completed and locked (read-only). "
|
|
"Reopen it first if a correction is genuinely needed.",
|
|
)
|
|
|
|
|
|
_SAFE = re.compile(r"[^A-Za-z0-9 ._,()\-]+")
|
|
|
|
|
|
def sanitize_filename(name: str) -> str:
|
|
name = name.replace("\\", "/").split("/")[-1] # strip any path
|
|
name = _SAFE.sub("_", name).strip() or "upload.xlsx"
|
|
return name[:200]
|
|
|
|
|
|
def to_dict(obj, fields: list[str]) -> dict:
|
|
out = {}
|
|
for f in fields:
|
|
v = getattr(obj, f, None)
|
|
if isinstance(v, (date, datetime)):
|
|
v = v.isoformat()
|
|
out[f] = v
|
|
return out
|
|
|
|
|
|
def session_dict(s: models.Session) -> dict:
|
|
d = to_dict(s, [
|
|
"id", "name", "reporting_month", "month_end_date", "reporting_currency",
|
|
"clearing_lag_days", "rounding_tolerance", "allowance_for_returns",
|
|
"manual_adjustment", "manual_adjustment_note", "status", "progress_stage",
|
|
"progress_pct", "progress_rows_done", "progress_rows_total", "eta_seconds",
|
|
"opening_mode", "opening_source_session_id",
|
|
"error", "created_at", "updated_at",
|
|
])
|
|
d["blocked"] = is_blocked(s)
|
|
d["blocked_reason"] = getattr(s, "blocked_reason", "") or ""
|
|
d["payout_mode"] = getattr(s, "payout_mode", "auto") or "auto"
|
|
d["needs_reprocess"] = bool(getattr(s, "needs_reprocess", False))
|
|
return d
|
|
|
|
|
|
def file_dict(f: models.SessionFile) -> dict:
|
|
d = to_dict(f, [
|
|
"id", "filename", "size_bytes", "sha256", "data_sheet", "imported_rows",
|
|
"min_date", "max_date", "currency", "marketplace", "status", "message",
|
|
])
|
|
try:
|
|
d["worksheets"] = json.loads(f.worksheets) if f.worksheets else []
|
|
except Exception:
|
|
d["worksheets"] = []
|
|
return d
|