dodil/erp-brick-expenses

DODIL ERP brick: expenses

src/brick_expenses/domain/receipts.py

"""The receipt image (EXP-01, EXP-16): evidence kept for the retention period under a content-addressed key.

Plain Python. The key is raw/expenses/<report_id>/<line_no>/<sha256>.<ext>: the same bytes are the same key (a second upload is the first), a
different image of the same line is a different key, and nothing already stored is ever overwritten. What is accepted is checked by its first
bytes, not by the name or the type the caller claims."""
import base64
import binascii
import hashlib
from datetime import date

from brick_expenses.domain.errors import RuleBroken

# The platform refuses a request body over 5120 KiB (erp-platform bricks.py), and base64 is 4/3 of the bytes: 3 MiB of image is 4 MiB on the wire. A bigger
# photo is the camera's to shrink; a direct-to-object-store upload (a presigned PUT) is the way past it, and a gap until the platform offers one.
MAX_BYTES = 3 * 1024 * 1024
# content type -> (extension, the bytes it starts with)
KINDS = {"image/jpeg": ("jpg", b"\xff\xd8\xff"), "image/png": ("png", b"\x89PNG\r\n\x1a\n"), "application/pdf": ("pdf", b"%PDF-")}


def decode(content_type: str, content_base64: str) -> bytes:
    """The image's bytes, or the rule it breaks: valid base64, a supported type whose first bytes agree, not empty, not over 3 MiB."""
    if content_type not in KINDS:
        raise RuleBroken(f"a receipt is {', '.join(KINDS)}, not {content_type}")
    try:
        body = base64.b64decode(content_base64, validate=True)
    except (binascii.Error, ValueError):
        raise RuleBroken("the receipt is not valid base64") from None
    if not body:
        raise RuleBroken("the receipt is empty")
    if len(body) > MAX_BYTES:
        raise RuleBroken(f"a receipt is at most {MAX_BYTES // (1024 * 1024)} MiB ({len(body)} bytes given)")
    if not body.startswith(KINDS[content_type][1]):
        raise RuleBroken(f"the bytes are not a {content_type} (they do not start like one)")
    return body


def sha256(body: bytes) -> str:
    return hashlib.sha256(body).hexdigest()


def object_key(report_id: str, line_no: int, digest: str, content_type: str) -> str:
    return f"raw/expenses/{report_id}/{line_no}/{digest}.{KINDS[content_type][0]}"


def parse_key(key: str) -> tuple:
    """(report_id, line_no, sha256, ext) of a key in the documented shape, or None."""
    parts = key.split("/")
    if len(parts) != 5 or parts[:2] != ["raw", "expenses"] or not parts[3].isdigit() or "." not in parts[4]:
        return None
    digest, ext = parts[4].rsplit(".", 1)
    return parts[2], int(parts[3]), digest, ext


def retain_until(base: date, years: int) -> date:
    """The last day the receipt must be kept: `years` after the date of the cost (UK 6, DE 10: a setting, never code)."""
    try:
        return base.replace(year=base.year + years)
    except ValueError:                                                      # 29 February
        return base.replace(year=base.year + years, day=28)