CRM_AGENTES_CARGA/backend/core/s3_keys.py

"""
Convención de claves S3/MinIO para objetos persistidos.

Todas las cargas que usen ``put_object_bytes`` deben obtener la clave mediante
funciones de este módulo (no construir ``tenants/...`` a mano en las rutas HTTP).

Árbol canónico
--------------

**Multi-tenant** (datos de clientes), siempre bajo ``tenants/{tenant_id}/``:

- ``tenants/{tid}/users/{keycloak_sub}/``
  Perfil de usuario (avatar). Ver ``tenant_user_prefix``, ``user_avatar_key``.

- ``tenants/{tid}/companies/{company_id}/``
  Recursos ligados a una empresa:

  - ``.../doda/{doda_id}/report/doda_report.pdf`` — reporte DODA en PDF. ``doda_report_pdf_key``.
  - ``.../branding/{filename}`` — logo. ``company_logo_key``.
  - ``.../certificates/{tipo}_{timestamp}.{cer|key}`` — CER/KEY FIEL, CFDI, cancelación.
    ``company_certificate_key``.
  - ``.../imports/csv/{job_type}/{job_id}.csv`` — CSV de layouts (import jobs).
    ``csv_import_key`` (usado por ``storage_s3.s3_key_for_csv_import``).
  - ``.../customs_brokers/{broker_id}/certificates/`` — CER del VU (``.cer``).
    ``customs_broker_vu_certificate_key``.
  - ``.../customs_brokers/{broker_id}/keys/`` — llave privada VU (``.key``).
    ``customs_broker_vu_private_key_key``.
  - ``.../customs_brokers/{broker_id}/cove/`` — archivos COVE (xml, zip, etc.).
    ``customs_broker_vu_cove_key``.

**Sistema global** (no por tenant):

- ``system/help/{carpeta opcional}/{archivo}`` — biblioteca de ayuda (imágenes, PDFs, vídeos).
  ``help_asset_key``, ``global_system_prefix``. Lectura HTTP mapea a este prefijo.

**Legado / migración**:

- ``imports/csv/{job_type}/{job_id}.csv`` — sin tenant/company. Solo ``legacy_csv_import_key``
  (cleanup o compatibilidad).

Constantes públicas
-------------------

``SYSTEM_HELP_PREFIX`` — prefijo literal ``system/help/`` para lecturas y utilidades
que no pasan por ``help_asset_key``.
"""
import re
from typing import Union

# Segmentos permitidos en claves (evita path traversal)
_SAFE_SEGMENT = re.compile(r"^[a-zA-Z0-9._\-]+$")

# Prefijo fijo para objetos de Help Center (debe coincidir con help_asset_key / GET /files/)
SYSTEM_HELP_PREFIX = "system/help/"


def _segment(value: Union[int, str], label: str) -> str:
    s = str(value).strip()
    if not s or "/" in s or ".." in s:
        raise ValueError(f"invalid {label} segment")
    if not _SAFE_SEGMENT.match(s):
        raise ValueError(f"invalid {label} characters")
    return s


def tenant_company_prefix(tenant_id: Union[int, str], company_id: int) -> str:
    """Prefijo `tenants/{tid}/companies/{cid}/` (termina en /)."""
    tid = _segment(tenant_id, "tenant_id")
    cid = _segment(company_id, "company_id")
    return f"tenants/{tid}/companies/{cid}/"


def tenant_user_prefix(tenant_id: Union[int, str], keycloak_user_id: str) -> str:
    """Prefijo `tenants/{tid}/users/{keycloak_sub}/` (avatar de perfil, sin company)."""
    tid = _segment(tenant_id, "tenant_id")
    kid = _segment(keycloak_user_id, "keycloak_user_id")
    return f"tenants/{tid}/users/{kid}/"


def user_avatar_key(
    tenant_id: Union[int, str],
    keycloak_user_id: str,
    ext: str,
) -> str:
    ext = ext.lower() if ext.startswith(".") else f".{ext}"
    allowed = (".jpg", ".jpeg", ".png", ".gif", ".webp")
    if ext not in allowed:
        raise ValueError("invalid avatar extension")
    return f"{tenant_user_prefix(tenant_id, keycloak_user_id)}avatar{ext}"


def public_user_avatar_api_path(tenant_id: int, keycloak_user_id: str) -> str:
    """Ruta GET pública para servir la imagen (sin host)."""
    return f"/api/v1/core/users/avatar/{tenant_id}/{keycloak_user_id}"


def global_system_prefix(subpath: str = "help") -> str:
    """Prefijo bajo `system/` para contenido global (p. ej. help). Termina en /."""
    sub = subpath.strip().strip("/")
    if not sub:
        return SYSTEM_HELP_PREFIX
    parts = sub.split("/")
    for p in parts:
        _segment(p, "system_subpath")
    return f"system/{sub}/"


def customs_broker_vu_prefix(
    tenant_id: Union[int, str],
    company_id: int,
    broker_id: int,
) -> str:
    """
    Prefijo para el bloque VU del agente aduanal.

    Forma: ``tenants/{tid}/companies/{cid}/customs_brokers/{broker_id}/``
    """
    bid = _segment(str(broker_id), "broker_id")
    return f"{tenant_company_prefix(tenant_id, company_id)}customs_brokers/{bid}/"


def customs_broker_vu_certificate_key(
    tenant_id: Union[int, str],
    company_id: int,
    broker_id: int,
    timestamp: str,
    file_ext: str,
) -> str:
    """CER del VU bajo ``.../customs_brokers/{id}/certificates/vu_cer_{timestamp}.cer``."""
    ts = _segment(timestamp, "timestamp")
    ext = file_ext.lower() if str(file_ext).startswith(".") else f".{file_ext}"
    if ext != ".cer":
        raise ValueError("VU certificate must be .cer")
    base = f"vu_cer_{ts}{ext}"
    return f"{customs_broker_vu_prefix(tenant_id, company_id, broker_id)}certificates/{base}"


def customs_broker_vu_private_key_key(
    tenant_id: Union[int, str],
    company_id: int,
    broker_id: int,
    timestamp: str,
    file_ext: str,
) -> str:
    """Llave privada del VU bajo ``.../customs_brokers/{id}/keys/vu_key_{timestamp}.key``."""
    ts = _segment(timestamp, "timestamp")
    ext = file_ext.lower() if str(file_ext).startswith(".") else f".{file_ext}"
    if ext != ".key":
        raise ValueError("VU private key must be .key")
    base = f"vu_key_{ts}{ext}"
    return f"{customs_broker_vu_prefix(tenant_id, company_id, broker_id)}keys/{base}"


def customs_broker_vu_cove_key(
    tenant_id: Union[int, str],
    company_id: int,
    broker_id: int,
    timestamp: str,
    original_filename: str,
) -> str:
    """
    Archivos COVE bajo ``.../customs_brokers/{id}/cove/cove_{timestamp}_{filename}``.
    Extensiones típicas: .xml, .zip, .txt, .pdf, .json
    """
    ts = _segment(timestamp, "timestamp")
    fn = safe_filename(original_filename)
    parts = fn.rsplit(".", 1)
    if len(parts) < 2:
        raise ValueError("COVE file must have an extension")
    ext = "." + parts[1].lower()
    allowed = (".xml", ".zip", ".txt", ".pdf", ".json")
    if ext not in allowed:
        raise ValueError(f"COVE extension not allowed: {ext}")
    base = f"cove_{ts}_{fn}"
    return f"{customs_broker_vu_prefix(tenant_id, company_id, broker_id)}cove/{base}"


def customs_broker_vu_doda_certificate_key(
    tenant_id: Union[int, str],
    company_id: int,
    broker_id: int,
    timestamp: str,
    file_ext: str,
) -> str:
    """CER DODA bajo ``.../customs_brokers/{id}/doda/certificates/doda_cer_{timestamp}.cer``."""
    ts = _segment(timestamp, "timestamp")
    ext = file_ext.lower() if str(file_ext).startswith(".") else f".{file_ext}"
    if ext != ".cer":
        raise ValueError("DODA certificate must be .cer")
    base = f"doda_cer_{ts}{ext}"
    return f"{customs_broker_vu_prefix(tenant_id, company_id, broker_id)}doda/certificates/{base}"


def customs_broker_vu_doda_private_key_key(
    tenant_id: Union[int, str],
    company_id: int,
    broker_id: int,
    timestamp: str,
    file_ext: str,
) -> str:
    """Llave DODA bajo ``.../customs_brokers/{id}/doda/keys/doda_key_{timestamp}.key``."""
    ts = _segment(timestamp, "timestamp")
    ext = file_ext.lower() if str(file_ext).startswith(".") else f".{file_ext}"
    if ext != ".key":
        raise ValueError("DODA private key must be .key")
    base = f"doda_key_{ts}{ext}"
    return f"{customs_broker_vu_prefix(tenant_id, company_id, broker_id)}doda/keys/{base}"


def customs_broker_vu_doda_cove_key(
    tenant_id: Union[int, str],
    company_id: int,
    broker_id: int,
    timestamp: str,
    original_filename: str,
) -> str:
    """
    Archivos DODA XML bajo ``.../customs_brokers/{id}/doda/cove/doda_cove_{timestamp}_{filename}``.
    Extensiones permitidas: .xml, .zip, .txt, .pdf, .json
    """
    ts = _segment(timestamp, "timestamp")
    fn = safe_filename(original_filename)
    parts = fn.rsplit(".", 1)
    if len(parts) < 2:
        raise ValueError("DODA file must have an extension")
    ext = "." + parts[1].lower()
    allowed = (".xml", ".zip", ".txt", ".pdf", ".json")
    if ext not in allowed:
        raise ValueError(f"DODA extension not allowed: {ext}")
    base = f"doda_cove_{ts}_{fn}"
    return f"{customs_broker_vu_prefix(tenant_id, company_id, broker_id)}doda/cove/{base}"


def job_type_segment(job_type: str) -> str:
    if job_type == "" or job_type == "invoice":
        return "invoice"
    return job_type


def csv_import_key(
    tenant_id: Union[int, str],
    company_id: int,
    job_type: str,
    job_id: str,
) -> str:
    _segment(job_id, "job_id")
    return (
        f"{tenant_company_prefix(tenant_id, company_id)}"
        f"imports/csv/{job_type_segment(job_type)}/{job_id}.csv"
    )


def legacy_csv_import_key(job_type: str, job_id: str) -> str:
    """Clave antigua sin tenant/company (solo migración / cleanup)."""
    _segment(job_id, "job_id")
    return f"imports/csv/{job_type_segment(job_type)}/{job_id}.csv"


def safe_filename(filename: str) -> str:
    """Nombre de archivo final sin separadores."""
    base = filename.rsplit("/", 1)[-1].rsplit("\\", 1)[-1]
    if not base or ".." in base:
        raise ValueError("invalid filename")
    return base


def company_logo_key(
    tenant_id: Union[int, str],
    company_id: int,
    filename: str,
) -> str:
    fn = safe_filename(filename)
    return f"{tenant_company_prefix(tenant_id, company_id)}branding/{fn}"


def doda_report_pdf_key(
    tenant_id: Union[int, str],
    company_id: int,
    doda_id: int,
) -> str:
    """
    Reporte DODA en PDF bajo ``.../doda/{doda_id}/report/doda_report.pdf`` (clave estable).
    """
    did = _segment(doda_id, "doda_id")
    return f"{tenant_company_prefix(tenant_id, company_id)}doda/{did}/report/doda_report.pdf"


def invoice_stamp_xml_key(
    tenant_id: Union[int, str],
    company_id: int,
    invoice_id: int,
    uuid: str,
) -> str:
    """
    XML timbrado bajo ``.../fin-invoices/{invoice_id}/stamps/{uuid}.xml``.

    La clave lleva el UUID y no un timestamp: el UUID identifica el comprobante ante el SAT y
    no cambia, así que la clave es estable y un retimbrado accidental no puede sobrescribir el
    XML de otro comprobante.
    """
    iid = _segment(invoice_id, "invoice_id")
    # El UUID va por _segment para que no pueda colar separadores de ruta.
    u = _segment(uuid, "uuid")
    return f"{tenant_company_prefix(tenant_id, company_id)}fin-invoices/{iid}/stamps/{u}.xml"


STAMP_XML_KINDS = ("request", "response")


def invoice_stamp_attempt_xml_key(
    tenant_id: Union[int, str],
    company_id: int,
    invoice_id: int,
    attempt_id: int,
    kind: str,
) -> str:
    """
    XML de un **intento** de timbrado, bajo
    ``.../fin-invoices/{invoice_id}/stamps/attempts/{attempt_id}-{kind}.xml``.

    ``kind`` es ``request`` (lo que se transmitió al PAC) o ``response`` (lo que contestó).

    La clave va por ``attempt_id`` y no por UUID porque un intento rechazado no tiene UUID, y
    es justo el rechazado el que hay que poder reconstruir: sin el par enviado/recibido, un
    error del PAC no se puede diagnosticar después de que termine la petición.
    """
    if kind not in STAMP_XML_KINDS:
        raise ValueError(f"kind inválido: {kind!r}. Sólo se admiten {STAMP_XML_KINDS}.")
    iid = _segment(invoice_id, "invoice_id")
    aid = _segment(attempt_id, "attempt_id")
    return (
        f"{tenant_company_prefix(tenant_id, company_id)}"
        f"fin-invoices/{iid}/stamps/attempts/{aid}-{kind}.xml"
    )


def company_certificate_key(
    tenant_id: Union[int, str],
    company_id: int,
    certificate_type: str,
    timestamp: str,
    file_ext: str,
) -> str:
    ct = _segment(certificate_type.replace(".", "_"), "certificate_type")
    ts = _segment(timestamp, "timestamp")
    ext = file_ext.lower() if file_ext.startswith(".") else f".{file_ext}"
    if ext not in (".cer", ".key"):
        raise ValueError("certificate file must be .cer or .key")
    base = f"{ct}_{ts}{ext}"
    return f"{tenant_company_prefix(tenant_id, company_id)}certificates/{base}"


def expediente_archivo_document_key(
    tenant_id: Union[int, str],
    company_id: int,
    expediente_id: int,
    timestamp: str,
    original_filename: str,
) -> str:
    """
    Archivo del expediente bajo ``.../expediente_archivos/{id}/documents/expediente_{timestamp}_{filename}``.
    Extensiones permitidas: .pdf, .xml, .png, .jpg, .jpeg, .json, .txt, .zip
    """
    ts = _segment(timestamp, "timestamp")
    eid = _segment(expediente_id, "expediente_id")
    fn = safe_filename(original_filename)
    parts = fn.rsplit(".", 1)
    if len(parts) < 2:
        raise ValueError("expediente file must have an extension")
    ext = "." + parts[1].lower()
    allowed = (".pdf", ".xml", ".png", ".jpg", ".jpeg", ".json", ".txt", ".zip")
    if ext not in allowed:
        raise ValueError(f"expediente file extension not allowed: {ext}")
    base = f"expediente_{ts}_{fn}"
    return f"{tenant_company_prefix(tenant_id, company_id)}expediente_archivos/{eid}/documents/{base}"


def expediente_archivo_artifact_key(
    tenant_id: Union[int, str],
    company_id: int,
    expediente_id: int,
    artifact_type: str,
    timestamp: str,
) -> str:
    """
    Artefacto de digitalización bajo ``.../expediente_archivos/{id}/artifacts/{type}_{timestamp}.{ext}``.
    artifact_type: acuse | envio_xml | respuesta_xml | consulta_envio_xml | consulta_respuesta_xml
    """
    ts = _segment(timestamp, "timestamp")
    eid = _segment(expediente_id, "expediente_id")
    at = _segment(artifact_type, "artifact_type")
    ext = ".pdf" if artifact_type == "acuse" else ".xml"
    return f"{tenant_company_prefix(tenant_id, company_id)}expediente_archivos/{eid}/artifacts/{at}_{ts}{ext}"


def cove_xml_key(
    tenant_id: Union[int, str],
    company_id: int,
    invoice_id: int,
) -> str:
    """
    XML de COVE devuelto por Ventanilla Única, bajo
    ``.../invoices/{invoice_id}/cove/cove.xml`` (clave estable por factura).
    """
    iid = _segment(invoice_id, "invoice_id")
    return f"{tenant_company_prefix(tenant_id, company_id)}invoices/{iid}/cove/cove.xml"


def cove_acuse_pdf_key(
    tenant_id: Union[int, str],
    company_id: int,
    invoice_id: int,
) -> str:
    """
    PDF de Acuse de COVE bajo
    ``.../invoices/{invoice_id}/cove/acuse_cove.pdf`` (clave estable por factura).
    """
    iid = _segment(invoice_id, "invoice_id")
    return f"{tenant_company_prefix(tenant_id, company_id)}invoices/{iid}/cove/acuse_cove.pdf"


def help_asset_key(folder: str, new_filename: str) -> str:
    """
    folder: '', 'pdfs', 'videos', 'assets' relativo a system/help/
    """
    folder = folder.strip().strip("/")
    fn = safe_filename(new_filename)
    if folder:
        for p in folder.split("/"):
            _segment(p, "help_folder")
        return f"{global_system_prefix('help')}{folder}/{fn}"
    return f"{global_system_prefix('help')}{fn}"


def help_s3_key_to_public_relative_path(key: str) -> str:
    """Parte tras `system/help/` para el path del endpoint público."""
    if not key.startswith(SYSTEM_HELP_PREFIX):
        raise ValueError("key is not under system/help/")
    return key[len(SYSTEM_HELP_PREFIX) :]


def help_public_api_path(relative_under_help: str) -> str:
    """URL de lectura pública bajo el router help-center (sin host)."""
    rel = relative_under_help.lstrip("/")
    return f"/api/v1/core/help-center/files/{rel}"


def signature_photo_key(
    tenant_id: Union[int, str],
    company_id: int,
    signature_id: int,
    timestamp: str,
    file_ext: str,
) -> str:
    """
    Foto de firma bajo ``.../signatures/{signature_id}/photo_{timestamp}.{ext}``.
    Extensiones permitidas: .jpg, .jpeg, .png, .gif, .webp
    """
    sid = _segment(signature_id, "signature_id")
    ts = _segment(timestamp, "timestamp")
    ext = file_ext.lower() if str(file_ext).startswith(".") else f".{file_ext}"
    allowed = (".jpg", ".jpeg", ".png", ".gif", ".webp")
    if ext not in allowed:
        raise ValueError(f"signature photo extension not allowed: {ext}")
    return f"{tenant_company_prefix(tenant_id, company_id)}signatures/{sid}/photo_{ts}{ext}"


def system_help_object_key(relative_path: str) -> str:
    """
    Clave S3 completa bajo ``system/help/`` para un path relativo (p. ej. GET /files/...).
    ``relative_path`` no debe empezar por / ni contener '..'.
    """
    rel = relative_path.strip().lstrip("/")
    if ".." in rel or not rel:
        raise ValueError("invalid help object path")
    return f"{SYSTEM_HELP_PREFIX}{rel}"