feat(crm): PDF de cotización (formato maestro) + marca por tenant + envío por correo

- Modelo crm.quote_settings (emisor, logo, color, prefijo, términos) por compañía
  + columna quotes.pdf_file_key + migración con down().
- Generador PDF (formato maestro: emisor+logo, cliente, carga/ruta, costos,
  resumen, condiciones); logo incrustado como JPEG (Pillow).
- Endpoints: GET /quotes/{id}/pdf-url, POST /quotes/{id}/send-email (adjunta PDF),
  GET/PUT /quote-settings, POST /quote-settings/logo.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/alembic/versions/a0b1c2d3e4f5_quote_settings_and_pdf.py

@@ -0,0 +1,60 @@
+"""crm quote_settings (marca por tenant) + quotes.pdf_file_key
+
+Revision ID: a0b1c2d3e4f5
+Revises: f8a9b0c1d2e3
+Create Date: 2026-07-29 00:00:00.000000
+
+PDF de cotización con formato maestro + branding por tenant + envío por correo.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "a0b1c2d3e4f5"
+down_revision: Union[str, None] = "f8a9b0c1d2e3"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+SCHEMA = "crm"
+
+
+def upgrade() -> None:
+    op.add_column("quotes", sa.Column("pdf_file_key", sa.String(length=512), nullable=True), schema=SCHEMA)
+
+    op.create_table(
+        "quote_settings",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("emitter_name", sa.String(length=255), nullable=True),
+        sa.Column("emitter_rfc", sa.String(length=13), nullable=True),
+        sa.Column("emitter_address", sa.Text(), nullable=True),
+        sa.Column("emitter_phone", sa.String(length=60), nullable=True),
+        sa.Column("emitter_email", sa.String(length=255), nullable=True),
+        sa.Column("emitter_website", sa.String(length=255), nullable=True),
+        sa.Column("logo_file_key", sa.String(length=512), nullable=True),
+        sa.Column("accent_color", sa.String(length=9), nullable=True, server_default=sa.text("'#2f6bf0'")),
+        sa.Column("quote_prefix", sa.String(length=12), nullable=True, server_default=sa.text("'COT'")),
+        sa.Column("default_terms", sa.Text(), nullable=True),
+        sa.Column("footer_note", sa.Text(), nullable=True),
+        sa.Column("tenant_id", sa.Integer(), nullable=False),
+        sa.Column("company_id", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
+        sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.text("now()")),
+        sa.Column("deleted_at", sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        sa.ForeignKeyConstraint(["tenant_id"], ["core.tenants.id"]),
+        schema=SCHEMA,
+    )
+    op.create_index("ix_crm_quote_settings_id", "quote_settings", ["id"], schema=SCHEMA)
+    op.create_index("ix_crm_quote_settings_tenant_id", "quote_settings", ["tenant_id"], schema=SCHEMA)
+    op.create_index("ix_crm_quote_settings_company_id", "quote_settings", ["company_id"], schema=SCHEMA)
+    # Una configuración por compañía
+    op.create_index(
+        "uq_crm_quote_settings_company", "quote_settings", ["tenant_id", "company_id"],
+        unique=True, schema=SCHEMA, postgresql_where=sa.text("deleted_at IS NULL"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("quote_settings", schema=SCHEMA)
+    op.drop_column("quotes", "pdf_file_key", schema=SCHEMA)

backend/api/v1/modules/crm/quotes/dto.py

@@ -86,6 +86,7 @@ class QuoteResponse(QuoteBase):
     status: str
     total_cost: Decimal
     total_sale: Decimal
+    pdf_file_key: str | None = None
     sent_at: datetime | None = None
     accepted_at: datetime | None = None
     rejected_at: datetime | None = None
@@ -100,3 +101,30 @@ class QuoteResponse(QuoteBase):
     @property
     def margin(self) -> Decimal:
         return (self.total_sale or Decimal(0)) - (self.total_cost or Decimal(0))
+
+
+# ----- Configuración de marca del formato de cotización -----
+
+class QuoteSettingsInput(BaseModel):
+    emitter_name: str | None = Field(None, max_length=255)
+    emitter_rfc: str | None = Field(None, max_length=13)
+    emitter_address: str | None = None
+    emitter_phone: str | None = Field(None, max_length=60)
+    emitter_email: str | None = Field(None, max_length=255)
+    emitter_website: str | None = Field(None, max_length=255)
+    accent_color: str | None = Field(None, max_length=9)
+    quote_prefix: str | None = Field(None, max_length=12)
+    default_terms: str | None = None
+    footer_note: str | None = None
+
+
+class QuoteSettingsResponse(QuoteSettingsInput):
+    model_config = ConfigDict(from_attributes=True)
+    id: int | None = None
+    logo_file_key: str | None = None
+
+
+class SendQuoteEmailRequest(BaseModel):
+    to: str | None = None
+    subject: str | None = None
+    message: str | None = None

backend/api/v1/modules/crm/quotes/models.py

@@ -34,10 +34,35 @@ class Quote(Base, TenantScopedMixin, TimestampMixin):
     notes: Mapped[str | None] = mapped_column(Text, nullable=True)
     terms: Mapped[str | None] = mapped_column(Text, nullable=True)
     owner_user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    # Clave del PDF generado en MinIO (para regenerar/enviar)
+    pdf_file_key: Mapped[str | None] = mapped_column(String(512), nullable=True)
     created_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
     updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
 
 
+class QuoteSettings(Base, TenantScopedMixin, TimestampMixin):
+    """Configuración de marca del formato de cotización, por compañía (tenant).
+
+    Encabezado del emisor, logo y textos por defecto que se imprimen en el PDF.
+    """
+
+    __tablename__ = "quote_settings"
+    __table_args__ = {"schema": "crm"}
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
+    emitter_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    emitter_rfc: Mapped[str | None] = mapped_column(String(13), nullable=True)
+    emitter_address: Mapped[str | None] = mapped_column(Text, nullable=True)
+    emitter_phone: Mapped[str | None] = mapped_column(String(60), nullable=True)
+    emitter_email: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    emitter_website: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    logo_file_key: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    accent_color: Mapped[str | None] = mapped_column(String(9), nullable=True, server_default=text("'#2f6bf0'"))
+    quote_prefix: Mapped[str | None] = mapped_column(String(12), nullable=True, server_default=text("'COT'"))
+    default_terms: Mapped[str | None] = mapped_column(Text, nullable=True)
+    footer_note: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+
 class QuoteItem(Base, TenantScopedMixin, TimestampMixin):
     """Concepto de una cotización (flete, transporte terrestre, despacho, gastos destino, otros)."""
 

backend/api/v1/modules/crm/quotes/pdf.py

@@ -0,0 +1,282 @@
+"""Generador del PDF de Cotización (formato maestro) sin dependencias de sistema.
+
+Compone un PDF 1.4 válido byte a byte (fuente Helvetica) e incrusta el logo como
+imagen JPEG (XObject /DCTDecode) usando Pillow para normalizarlo. El branding del
+emisor (nombre, RFC, dirección, contacto, color) viene de la configuración por
+tenant.
+"""
+
+from __future__ import annotations
+
+import io
+from decimal import Decimal
+
+_PAGE_W = 612
+_PAGE_H = 792
+_MARGIN = 50
+
+CONCEPT_LABELS = {
+    "flete_internacional": "Flete internacional",
+    "transporte_terrestre": "Transporte terrestre",
+    "despacho_aduanal": "Despacho aduanal",
+    "gastos_destino": "Gastos en destino",
+    "otros": "Otros cargos",
+}
+
+
+def _esc(text: str) -> str:
+    out = (str(text) if text is not None else "").encode("latin-1", "replace").decode("latin-1")
+    return out.replace("\\", r"\\").replace("(", r"\(").replace(")", r"\)")
+
+
+def _money(value, currency: str = "") -> str:
+    d = Decimal(str(value or 0)).quantize(Decimal("0.01"))
+    return (f"{currency} " if currency else "") + f"{d:,.2f}"
+
+
+def _wrap(text: str, width: int) -> list[str]:
+    words = (text or "").split()
+    if not words:
+        return []
+    lines, cur = [], ""
+    for w in words:
+        cand = f"{cur} {w}".strip()
+        if len(cand) > width and cur:
+            lines.append(cur)
+            cur = w
+        else:
+            cur = cand
+    if cur:
+        lines.append(cur)
+    return lines
+
+
+def _hex_rgb(hexs: str | None) -> tuple[float, float, float]:
+    try:
+        h = (hexs or "#2f6bf0").lstrip("#")
+        return tuple(int(h[i : i + 2], 16) / 255 for i in (0, 2, 4))  # type: ignore[return-value]
+    except Exception:
+        return (0.184, 0.42, 0.94)
+
+
+def _prep_logo(logo_bytes: bytes | None):
+    """Normaliza el logo a JPEG RGB. Devuelve (jpeg_bytes, w, h) o None."""
+    if not logo_bytes:
+        return None
+    try:
+        from PIL import Image
+
+        im = Image.open(io.BytesIO(logo_bytes)).convert("RGB")
+        im.thumbnail((600, 300))
+        buf = io.BytesIO()
+        im.save(buf, format="JPEG", quality=85)
+        return buf.getvalue(), im.width, im.height
+    except Exception:
+        return None
+
+
+def _kv_lines(pairs: list[tuple[str, str]]) -> list[tuple[str, int]]:
+    out: list[tuple[str, int]] = []
+    for k, v in pairs:
+        if v not in (None, "", "None"):
+            out.append((f"{k}: {v}", 10))
+    return out
+
+
+def build_quote_pdf(
+    *,
+    emitter: dict,
+    head: dict,
+    client: dict,
+    cargo: list[tuple[str, str]],
+    route: list[tuple[str, str]],
+    items: list[dict],
+    currency: str,
+    subtotal,
+    terms: str | None,
+    footer: str | None,
+    logo_bytes: bytes | None = None,
+    accent: str | None = "#2f6bf0",
+) -> bytes:
+    accent_rgb = _hex_rgb(accent)
+    logo = _prep_logo(logo_bytes)
+
+    # ---- Cuerpo (debajo del encabezado) ----
+    body: list[tuple[str, int]] = []
+
+    def section(title: str):
+        body.append(("", 6))
+        body.append((title.upper(), 11))
+        body.append(("_" * 92, 8))
+
+    # Cliente
+    section("Cliente")
+    for line in _kv_lines([
+        ("Cliente", client.get("name")), ("RFC", client.get("rfc")),
+        ("Correo", client.get("email")), ("Teléfono", client.get("phone")),
+    ]):
+        body.append(line)
+
+    # Carga / Ruta
+    if cargo:
+        section("Información de la carga")
+        for line in _kv_lines(cargo):
+            body.append(line)
+    if route:
+        section("Ruta logística")
+        for line in _kv_lines(route):
+            body.append(line)
+
+    # Costos
+    section("Costos cotizados")
+    body.append(("Concepto                                  Cant.        Tarifa           Importe", 9))
+    body.append(("-" * 92, 8))
+    for it in items:
+        code = str(it.get("concept") or "")
+        label = CONCEPT_LABELS.get(code, code)
+        desc = str(it.get("description") or "")
+        if desc:
+            label = f"{label} — {desc}"
+        qty = Decimal(str(it.get("quantity") or 0))
+        unit = Decimal(str(it.get("unit_sale") or 0))
+        amount = (qty * unit).quantize(Decimal("0.01"))
+        row = f"{label[:40].ljust(40)}  {qty:>6.2f}  {unit:>14,.2f}  {amount:>14,.2f}"
+        body.append((row, 9))
+    body.append(("-" * 92, 8))
+    body.append((f"Subtotal {currency}: {_money(subtotal)}", 11))
+    body.append(("IVA: según aplique", 9))
+    body.append((f"Total {currency}: {_money(subtotal)} + IVA", 12))
+
+    # Condiciones
+    if terms:
+        section("Condiciones comerciales")
+        for para in terms.splitlines():
+            for line in _wrap(para, 105) or [""]:
+                body.append((line, 9))
+
+    # ---- Paginación (página 1 con encabezado; siguientes solo cuerpo) ----
+    p1_top = _PAGE_H - 150   # y donde inicia el cuerpo en la página 1
+    pN_top = _PAGE_H - _MARGIN
+    line_h = 14
+    pages: list[list[tuple[float, tuple[str, int]]]] = []
+    cur: list[tuple[float, tuple[str, int]]] = []
+    y = p1_top
+    for item in body:
+        if y < _MARGIN + 40:
+            pages.append(cur)
+            cur = []
+            y = pN_top
+        cur.append((y, item))
+        y -= line_h
+    pages.append(cur)
+
+    # ---- Content streams ----
+    streams: list[bytes] = []
+    for pi, page in enumerate(pages):
+        parts: list[str] = []
+        if pi == 0:
+            # barra de acento arriba
+            r, g, b = accent_rgb
+            parts.append(f"{r:.3f} {g:.3f} {b:.3f} rg")
+            parts.append(f"0 {_PAGE_H - 8} {_PAGE_W} 8 re f")
+            # logo
+            logo_y = _PAGE_H - 30
+            if logo:
+                _, lw, lh = logo
+                dw = 150.0
+                dh = dw * lh / lw
+                if dh > 60:
+                    dh = 60.0
+                    dw = dh * lw / lh
+                parts.append(f"q {dw:.2f} 0 0 {dh:.2f} {_MARGIN} {logo_y - dh:.2f} cm /Im0 Do Q")
+            # emisor (columna derecha)
+            ex = 330
+            ey = logo_y - 6
+            parts.append("BT /F1 12 Tf 0.09 0.14 0.24 rg")
+            parts.append(f"1 0 0 1 {ex} {ey} Tm ({_esc(emitter.get('name') or 'Emisor')}) Tj")
+            parts.append("/F1 9 Tf 0.35 0.41 0.5 rg 13 TL")
+            em_lines = []
+            if emitter.get("rfc"):
+                em_lines.append(f"RFC: {emitter['rfc']}")
+            for a in (emitter.get("address") or "").splitlines():
+                if a.strip():
+                    em_lines.append(a.strip())
+            contact = "  ".join([x for x in [emitter.get("phone"), emitter.get("email"), emitter.get("website")] if x])
+            if contact:
+                em_lines.append(contact)
+            for ln in em_lines[:5]:
+                parts.append(f"T* ({_esc(ln)}) Tj")
+            parts.append("ET")
+            # título
+            parts.append("BT /F1 22 Tf 0.09 0.14 0.24 rg")
+            parts.append(f"1 0 0 1 {_MARGIN} {_PAGE_H - 120} Tm (COTIZACION) Tj ET")
+            # datos de cabecera (derecha del título)
+            parts.append("BT /F1 9 Tf 0.2 0.25 0.35 rg 12 TL")
+            parts.append(f"1 0 0 1 330 {_PAGE_H - 100} Tm ({_esc('No.: ' + str(head.get('reference') or '-'))}) Tj")
+            for hl in [
+                f"Fecha: {head.get('issue_date') or '-'}",
+                f"Vigencia: {head.get('valid_until') or '-'}",
+                f"Ejecutivo: {head.get('owner') or '-'}    Estatus: {head.get('status') or '-'}",
+            ]:
+                parts.append(f"T* ({_esc(hl)}) Tj")
+            parts.append("ET")
+        # cuerpo
+        for yy, (text, size) in page:
+            parts.append(f"BT /F1 {size} Tf 0 0 0 rg 1 0 0 1 {_MARGIN} {yy:.2f} Tm ({_esc(text)}) Tj ET")
+        # pie
+        if footer:
+            parts.append(f"BT /F1 8 Tf 0.5 0.5 0.5 rg 1 0 0 1 {_MARGIN} {_MARGIN - 20} Tm ({_esc(footer[:110])}) Tj ET")
+        streams.append("\n".join(parts).encode("latin-1", "replace"))
+
+    # ---- Ensamblado de objetos ----
+    objects: list[bytes] = []
+
+    def add(obj: bytes) -> int:
+        objects.append(obj)
+        return len(objects)
+
+    n_pages = len(pages)
+    has_img = 1 if logo else 0
+    font_num = 3
+    img_num = 4 if has_img else None
+    base = 5 if has_img else 4
+    page_nums = list(range(base, base + n_pages))
+    content_nums = list(range(base + n_pages, base + 2 * n_pages))
+
+    kids = " ".join(f"{n} 0 R" for n in page_nums)
+    add(b"<< /Type /Catalog /Pages 2 0 R >>")
+    add(f"<< /Type /Pages /Kids [{kids}] /Count {n_pages} >>".encode("latin-1"))
+    add(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica /Encoding /WinAnsiEncoding >>")
+    if logo:
+        jpeg, lw, lh = logo
+        img_obj = (
+            f"<< /Type /XObject /Subtype /Image /Width {lw} /Height {lh} "
+            f"/ColorSpace /DeviceRGB /BitsPerComponent 8 /Filter /DCTDecode /Length {len(jpeg)} >>\n"
+        ).encode("latin-1") + b"stream\n" + jpeg + b"\nendstream"
+        add(img_obj)
+    for i in range(n_pages):
+        res = f"/Font << /F1 {font_num} 0 R >>"
+        if has_img and i == 0:
+            res += f" /XObject << /Im0 {img_num} 0 R >>"
+        page_dict = (
+            f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 {_PAGE_W} {_PAGE_H}] "
+            f"/Resources << {res} >> /Contents {content_nums[i]} 0 R >>"
+        )
+        add(page_dict.encode("latin-1"))
+    for stream in streams:
+        add(b"<< /Length " + str(len(stream)).encode() + b" >>\nstream\n" + stream + b"\nendstream")
+
+    out = bytearray()
+    out += b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n"
+    offsets: list[int] = []
+    for i, obj in enumerate(objects, start=1):
+        offsets.append(len(out))
+        out += f"{i} 0 obj\n".encode("latin-1") + obj + b"\nendobj\n"
+    xref_pos = len(out)
+    total = len(objects) + 1
+    out += f"xref\n0 {total}\n".encode("latin-1")
+    out += b"0000000000 65535 f \n"
+    for off in offsets:
+        out += f"{off:010d} 00000 n \n".encode("latin-1")
+    out += f"trailer\n<< /Size {total} /Root 1 0 R >>\nstartxref\n{xref_pos}\n%%EOF".encode("latin-1")
+    return bytes(out)

backend/api/v1/modules/crm/quotes/pdf_service.py

@@ -0,0 +1,235 @@
+"""PDF de cotización, configuración de marca por tenant y envío por correo."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+
+from fastapi import HTTPException, status
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from ..accounts.models import Account
+from ..service_requests.models import ServiceRequest
+from .models import Quote, QuoteItem, QuoteSettings
+from .pdf import build_quote_pdf
+from .service import get_quote
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_TERMS = (
+    "Tarifas sujetas a disponibilidad de espacio.\n"
+    "Cualquier variación en peso o volumen generará ajuste tarifario.\n"
+    "No incluye cargos extraordinarios, maniobras especiales o servicios no especificados.\n"
+    "Tarifas sujetas a revisión por parte de la línea transportista y autoridades correspondientes."
+)
+
+
+# ---------------- Configuración de marca ----------------
+def get_settings(db: Session, tenant_id: int, company_id: int) -> QuoteSettings | None:
+    return (
+        db.query(QuoteSettings)
+        .filter(QuoteSettings.tenant_id == tenant_id, QuoteSettings.company_id == company_id,
+                QuoteSettings.deleted_at.is_(None))
+        .first()
+    )
+
+
+def upsert_settings(db: Session, tenant_id: int, company_id: int, data: dict) -> QuoteSettings:
+    obj = get_settings(db, tenant_id, company_id)
+    if obj is None:
+        obj = QuoteSettings(tenant_id=tenant_id, company_id=company_id)
+        db.add(obj)
+    for field, value in data.items():
+        if value is not None:
+            setattr(obj, field, value)
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def set_logo_key(db: Session, tenant_id: int, company_id: int, file_key: str) -> QuoteSettings:
+    obj = get_settings(db, tenant_id, company_id)
+    if obj is None:
+        obj = QuoteSettings(tenant_id=tenant_id, company_id=company_id)
+        db.add(obj)
+    obj.logo_file_key = file_key
+    db.commit()
+    db.refresh(obj)
+    return obj
+
+
+def _company_row(db: Session, company_id: int) -> dict:
+    try:
+        row = db.execute(
+            text("SELECT name, rfc, logo FROM a76.company WHERE id = :c"), {"c": company_id}
+        ).first()
+        if row:
+            return {"name": row[0], "rfc": row[1], "logo": row[2]}
+    except Exception:
+        pass
+    return {}
+
+
+# ---------------- Construcción del PDF ----------------
+def build_pdf_bytes(db: Session, quote: Quote, tenant_id: int, company_id: int) -> bytes:
+    items = (
+        db.query(QuoteItem)
+        .filter(QuoteItem.quote_id == quote.id, QuoteItem.deleted_at.is_(None))
+        .order_by(QuoteItem.id.asc())
+        .all()
+    )
+    account = (
+        db.query(Account).filter(Account.id == quote.account_id).first() if quote.account_id else None
+    )
+    sr = (
+        db.query(ServiceRequest).filter(ServiceRequest.id == quote.service_request_id).first()
+        if quote.service_request_id else None
+    )
+    settings = get_settings(db, tenant_id, company_id)
+    company = _company_row(db, company_id)
+
+    # Emisor: config del tenant con respaldo en a76.company
+    emitter = {
+        "name": (settings.emitter_name if settings else None) or company.get("name") or "Emisor",
+        "rfc": (settings.emitter_rfc if settings else None) or company.get("rfc"),
+        "address": settings.emitter_address if settings else None,
+        "phone": settings.emitter_phone if settings else None,
+        "email": settings.emitter_email if settings else None,
+        "website": settings.emitter_website if settings else None,
+    }
+    accent = (settings.accent_color if settings else None) or "#2f6bf0"
+    prefix = (settings.quote_prefix if settings else None) or "COT"
+    terms = quote.terms or (settings.default_terms if settings else None) or DEFAULT_TERMS
+    footer = settings.footer_note if settings else None
+
+    # Logo (MinIO)
+    logo_bytes = None
+    logo_key = settings.logo_file_key if settings else None
+    if logo_key:
+        try:
+            from core.storage_s3 import get_object_bytes
+            logo_bytes = get_object_bytes(logo_key)
+        except Exception as exc:
+            logger.warning("No se pudo leer el logo del tarifario: %s", exc)
+
+    reference = quote.reference or f"{prefix}-{datetime.now().strftime('%Y%m%d')}-{quote.id:03d}"
+    head = {
+        "reference": reference,
+        "issue_date": quote.issue_date.isoformat() if quote.issue_date else None,
+        "valid_until": quote.valid_until.isoformat() if quote.valid_until else None,
+        "owner": quote.owner_user_id or "-",
+        "status": quote.status,
+    }
+    client = {
+        "name": account.name if account else None,
+        "rfc": account.rfc if account else None,
+        "email": account.email if account else None,
+        "phone": account.phone if account else None,
+    }
+    cargo = []
+    route = []
+    if sr:
+        cargo = [
+            ("Tipo de mercancía", sr.cargo_type), ("Descripción", sr.commodity),
+            ("Peso", str(sr.weight) if sr.weight is not None else None),
+            ("Volumen", str(sr.volume) if sr.volume is not None else None),
+            ("Tipo de carga", sr.load_type), ("Equipo", sr.container_equipment),
+        ]
+        route = [
+            ("Operación", sr.operation_type), ("Modo", sr.transport_mode),
+            ("Servicio", sr.service_type), ("Incoterm", sr.incoterm),
+            ("Origen", sr.origin), ("Destino", sr.destination),
+            ("Fecha requerida", sr.required_date.isoformat() if sr.required_date else None),
+        ]
+
+    return build_quote_pdf(
+        emitter=emitter, head=head, client=client, cargo=cargo, route=route,
+        items=[{"concept": i.concept, "description": i.description, "quantity": i.quantity, "unit_sale": i.unit_sale} for i in items],
+        currency=quote.currency, subtotal=quote.total_sale, terms=terms, footer=footer,
+        logo_bytes=logo_bytes, accent=accent,
+    )
+
+
+def _store_pdf(db: Session, quote: Quote, tenant_id: int, company_id: int, pdf_bytes: bytes) -> str:
+    from core.storage_s3 import put_object_bytes
+    ref = (quote.reference or f"cot-{quote.id}").replace("/", "-")
+    key = f"tenants/{tenant_id}/companies/{company_id}/crm-quotes/{quote.id}/cotizacion-{ref}.pdf"
+    put_object_bytes(key, pdf_bytes, content_type="application/pdf")
+    quote.pdf_file_key = key
+    db.commit()
+    return key
+
+
+def get_pdf_url(db: Session, quote_id: int, tenant_id: int, company_id: int) -> str:
+    from core.storage_s3 import presigned_get_url
+    quote = get_quote(db, quote_id, tenant_id, company_id)
+    pdf_bytes = build_pdf_bytes(db, quote, tenant_id, company_id)
+    key = _store_pdf(db, quote, tenant_id, company_id, pdf_bytes)
+    return presigned_get_url(key)
+
+
+# ---------------- Envío por correo ----------------
+async def send_quote_email(
+    db: Session, quote_id: int, tenant_id: int, company_id: int,
+    to: str | None, subject: str | None, message: str | None,
+) -> dict:
+    import ssl
+    from email import encoders
+    from email.mime.base import MIMEBase
+    from email.mime.multipart import MIMEMultipart
+    from email.mime.text import MIMEText
+
+    import aiosmtplib
+
+    from core.config import settings as cfg
+
+    quote = get_quote(db, quote_id, tenant_id, company_id)
+    account = db.query(Account).filter(Account.id == quote.account_id).first() if quote.account_id else None
+    recipient = to or (account.email if account else None)
+    if not recipient:
+        raise HTTPException(status_code=400, detail="No hay correo destino (captura uno o pon el correo del cliente).")
+
+    pdf_bytes = build_pdf_bytes(db, quote, tenant_id, company_id)
+    _store_pdf(db, quote, tenant_id, company_id, pdf_bytes)
+    ref = quote.reference or f"COT-{quote.id}"
+
+    msg = MIMEMultipart()
+    msg["From"] = f"{cfg.SMTP_FROM_NAME} <{cfg.SMTP_USER}>"
+    msg["To"] = recipient
+    msg["Subject"] = subject or f"Cotización {ref}"
+    html = (
+        "<div style='font-family:Arial,sans-serif;color:#333;max-width:600px'>"
+        f"<p>{(message or 'Adjunto la cotización solicitada. Quedamos atentos.').replace(chr(10), '<br>')}</p>"
+        f"<p style='color:#6b7280;font-size:12px'>Cotización {ref}</p></div>"
+    )
+    msg.attach(MIMEText(html, "html"))
+    part = MIMEBase("application", "pdf")
+    part.set_payload(pdf_bytes)
+    encoders.encode_base64(part)
+    part.add_header("Content-Disposition", f'attachment; filename="cotizacion-{ref}.pdf"')
+    msg.attach(part)
+
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    try:
+        if cfg.SMTP_PORT == 465:
+            async with aiosmtplib.SMTP(hostname=cfg.SMTP_HOST, port=cfg.SMTP_PORT, use_tls=True, tls_context=ctx) as smtp:
+                await smtp.login(cfg.SMTP_USER, cfg.SMTP_PASSWORD)
+                await smtp.send_message(msg)
+        else:
+            async with aiosmtplib.SMTP(hostname=cfg.SMTP_HOST, port=cfg.SMTP_PORT, tls_context=ctx) as smtp:
+                await smtp.starttls(tls_context=ctx)
+                await smtp.login(cfg.SMTP_USER, cfg.SMTP_PASSWORD)
+                await smtp.send_message(msg)
+    except Exception as exc:
+        logger.error("Error enviando cotización %s: %s", quote_id, exc)
+        raise HTTPException(status_code=502, detail=f"No se pudo enviar el correo: {exc}")
+
+    # Marca como enviada
+    if quote.status == "borrador":
+        quote.status = "enviada"
+    quote.sent_at = datetime.now(timezone.utc)
+    db.commit()
+    return {"sent_to": recipient, "reference": ref}

backend/api/v1/modules/crm/quotes/routes.py

@@ -1,22 +1,76 @@
-from fastapi import APIRouter, Depends, Query, status
+from fastapi import APIRouter, Depends, File, Query, UploadFile, status
 from sqlalchemy.orm import Session
 
 from core.database import get_core_db
 from core.security import get_current_user
 
-from . import service
+from . import pdf_service, service
 from .dto import (
     QuoteCreate,
     QuoteItemCreate,
     QuoteItemResponse,
     QuoteItemUpdate,
     QuoteResponse,
+    QuoteSettingsInput,
+    QuoteSettingsResponse,
     QuoteUpdate,
+    SendQuoteEmailRequest,
 )
 
 router = APIRouter()
 
 
+# ----- Configuración de marca del formato de cotización -----
+
+@router.get("/quote-settings", response_model=QuoteSettingsResponse)
+def get_quote_settings(
+    company_id: int = Query(...),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    obj = pdf_service.get_settings(db, current_user["tenant_id"], company_id)
+    return obj or QuoteSettingsResponse()
+
+
+@router.put("/quote-settings", response_model=QuoteSettingsResponse)
+def save_quote_settings(
+    payload: QuoteSettingsInput,
+    company_id: int = Query(...),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    return pdf_service.upsert_settings(db, current_user["tenant_id"], company_id, payload.model_dump(exclude_unset=True))
+
+
+@router.post("/quote-settings/logo", response_model=QuoteSettingsResponse)
+async def upload_quote_logo(
+    company_id: int = Query(...),
+    file: UploadFile = File(...),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    from core.storage_s3 import put_object_bytes
+    tenant_id = current_user["tenant_id"]
+    content = await file.read()
+    safe = (file.filename or "logo").replace("/", "-")
+    key = f"tenants/{tenant_id}/companies/{company_id}/crm-quote-logo/{safe}"
+    put_object_bytes(key, content, content_type=file.content_type or "image/png")
+    return pdf_service.set_logo_key(db, tenant_id, company_id, key)
+
+
+@router.get("/quote-settings/logo-url")
+def get_logo_url(
+    company_id: int = Query(...),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    from core.storage_s3 import presigned_get_url
+    obj = pdf_service.get_settings(db, current_user["tenant_id"], company_id)
+    if not obj or not obj.logo_file_key:
+        return {"url": None}
+    return {"url": presigned_get_url(obj.logo_file_key)}
+
+
 def _user_id(current_user: dict) -> str | None:
     return current_user.get("sub") or current_user.get("id")
 
@@ -98,6 +152,32 @@ def reject_quote(
     return service.reject_quote(db, quote_id, current_user["tenant_id"], company_id)
 
 
+@router.get("/quotes/{quote_id}/pdf-url")
+def quote_pdf_url(
+    quote_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Genera el PDF de la cotización (formato maestro + marca) y devuelve una URL."""
+    url = pdf_service.get_pdf_url(db, quote_id, current_user["tenant_id"], company_id)
+    return {"url": url}
+
+
+@router.post("/quotes/{quote_id}/send-email")
+async def quote_send_email(
+    quote_id: int,
+    payload: SendQuoteEmailRequest,
+    company_id: int = Query(..., description="Company ID"),
+    current_user: dict = Depends(get_current_user),
+    db: Session = Depends(get_core_db),
+):
+    """Genera el PDF y lo envía por correo (al cliente o al destinatario indicado)."""
+    return await pdf_service.send_quote_email(
+        db, quote_id, current_user["tenant_id"], company_id, payload.to, payload.subject, payload.message
+    )
+
+
 @router.post("/quotes/{quote_id}/clone", response_model=QuoteResponse, status_code=status.HTTP_201_CREATED)
 def clone_quote(
     quote_id: int,