"""PCM Integrations (DirectMail API v3) postcard adapter.""" from __future__ import annotations import json import logging from dataclasses import dataclass from typing import Any import requests from django.conf import settings from messaging.providers.postcard import PostcardResult logger = logging.getLogger(__name__) PCM_API_BASE = "https://v3.pcmintegrations.com" # PCM size codes for custom designer designs. PCM_SIZE_CHOICES = ( ("46", "4.25 × 6"), ("68", "6 × 8.5"), ("69", "6 × 9"), ("611", "6 × 11"), ("811", "8.5 × 11"), ) class PcmApiError(RuntimeError): """Raised when a PCM API call fails.""" def _api_key() -> str: return (settings.PCM_API_KEY or "").strip() def _headers() -> dict[str, str]: key = _api_key() if not key: raise PcmApiError("PCM_API_KEY is not configured") return { "Accept": "application/json", "Authorization": f"Bearer {key}", "Content-Type": "application/json", } def pcm_request( method: str, path: str, *, params: dict[str, Any] | None = None, json_body: dict[str, Any] | None = None, timeout: int = 60, ) -> Any: """Call PCM v3 API. ``path`` is absolute under the API host (e.g. ``/design``).""" url = f"{PCM_API_BASE}{path}" response = requests.request( method, url, headers=_headers(), params=params, json=json_body, timeout=timeout, ) if response.status_code >= 400: detail = (response.text or "")[:500] raise PcmApiError( f"PCM {method} {path} → {response.status_code}: {detail}" ) if not response.content: return {} try: return response.json() except ValueError: return {"raw": response.text} def return_address_from_settings() -> dict[str, str]: """Build PCM returnAddress from PCM_RETURN_ADDRESS JSON or CONTACT_* vars.""" raw = (settings.PCM_RETURN_ADDRESS or "").strip() if raw: data = json.loads(raw) if not isinstance(data, dict): raise PcmApiError("PCM_RETURN_ADDRESS must be a JSON object") return { "company": str(data.get("company") or ""), "firstName": str(data.get("firstName") or data.get("first_name") or ""), "lastName": str(data.get("lastName") or data.get("last_name") or ""), "address": str(data.get("address") or data.get("line1") or ""), "address2": str(data.get("address2") or data.get("line2") or ""), "city": str(data.get("city") or ""), "state": str(data.get("state") or ""), "zipCode": str(data.get("zipCode") or data.get("zip") or ""), } name = (settings.SITE_NAME or "").strip() parts = name.split(None, 1) first = parts[0] if parts else "Monica" last = parts[1] if len(parts) > 1 else "" return { "company": "", "firstName": first, "lastName": last, "address": str(getattr(settings, "PCM_RETURN_LINE1", "") or ""), "address2": str(getattr(settings, "PCM_RETURN_LINE2", "") or ""), "city": str(getattr(settings, "PCM_RETURN_CITY", "") or ""), "state": str(getattr(settings, "PCM_RETURN_STATE", "") or ""), "zipCode": str(getattr(settings, "PCM_RETURN_ZIP", "") or ""), } def contact_to_pcm_recipient(contact, *, ext_ref: str) -> dict[str, str]: address = contact.postal_address or {} line1 = (address.get("line1") or "").strip() if not line1: raise ValueError("Contact postal_address.line1 required for postcard") first = (contact.first_name or "").strip() last = (contact.last_name or "").strip() if not first and not last: # PCM requires name or company. first = (contact.full_name or contact.email or "Resident").strip() return { "firstName": first, "lastName": last, "address": line1, "address2": (address.get("line2") or "").strip() or " ", "city": (address.get("city") or "").strip(), "state": (address.get("state") or "").strip(), "zipCode": (address.get("zip") or "").strip(), "extRefNbr": ext_ref, } def list_designs(*, product_type: str = "postcard", page: int = 1, per_page: int = 50) -> list[dict]: data = pcm_request( "GET", "/design", params={ "productType": product_type, "page": page, "perPage": per_page, }, ) if isinstance(data, dict): results = data.get("results") or data.get("designs") or [] return results if isinstance(results, list) else [] return [] def create_custom_design(*, name: str, size: str) -> dict[str, Any]: """POST /design/custom → designID + embed url.""" return pcm_request( "POST", "/design/custom", json_body={"name": name, "size": size}, ) def get_design_embed_url(design_id: int | str, *, duplicate: bool = False) -> str: """GET /design/{id}/edit?mode=embed → iframe URL.""" params: dict[str, Any] = {"mode": "embed"} if duplicate: params["duplicate"] = "true" data = pcm_request("GET", f"/design/{design_id}/edit", params=params) if not isinstance(data, dict): raise PcmApiError("Unexpected embed response from PCM") url = data.get("embed_url") or data.get("url") or "" if not url: raise PcmApiError("PCM did not return an embed URL") return str(url) def get_order(order_id: int | str) -> dict[str, Any]: data = pcm_request("GET", f"/order/{order_id}") return data if isinstance(data, dict) else {} def place_postcard_order( *, design_id: int, recipient: dict[str, str], ext_ref: str, mail_class: str = "FirstClass", ) -> str: """Place a one-recipient postcard order; return PCM orderID as string.""" payload = { "designID": design_id, "mailClass": mail_class, "extRefNbr": ext_ref, "returnAddress": return_address_from_settings(), "recipients": [recipient], } data = pcm_request("POST", "/order", json_body=payload) if not isinstance(data, dict): raise PcmApiError("Unexpected order response from PCM") order_id = data.get("orderID") or data.get("orderId") or data.get("id") if order_id is None and isinstance(data.get("results"), list) and data["results"]: order_id = data["results"][0].get("orderID") if order_id is None: raise PcmApiError(f"PCM order response missing orderID: {data!r}"[:400]) return str(order_id) def design_id_from_template(template) -> int | None: """Read design_id from MessageTemplate.postcard_front JSON.""" if not template: return None front = template.postcard_front or {} if not isinstance(front, dict): return None raw = front.get("design_id") or front.get("designID") if raw is None: return None try: return int(raw) except (TypeError, ValueError): return None @dataclass class PcmProvider: name: str = "pcm" def send_postcard(self, message) -> PostcardResult: template = message.campaign.template if message.campaign_id else None design_id = design_id_from_template(template) if not design_id: raise ValueError( "Postcard campaign template missing PCM design_id " "(save a design from the postcard designer first)" ) recipient = contact_to_pcm_recipient( message.contact, ext_ref=str(message.pk) ) mail_class = "FirstClass" if template and isinstance(template.postcard_front, dict): mail_class = ( template.postcard_front.get("mail_class") or mail_class ) order_id = place_postcard_order( design_id=design_id, recipient=recipient, ext_ref=str(message.pk), mail_class=str(mail_class), ) return PostcardResult(provider_id=order_id) def get_status(self, provider_id: str) -> str: try: data = get_order(provider_id) except PcmApiError: logger.exception("PCM get_status failed for %s", provider_id) return "unknown" return str(data.get("status") or "unknown")