Template
westfarn
f2eb4b7152
## Summary Closes #11. Mirrors ai_ml_operations/monica_site#15. - Direct-mail compose can target the postcard mailing list **or** find and select one or two USPS carrier routes by ZIP (optional radius). - Estimated cost is shown per route and for the selection (PCM design pricing, else `PCM_POSTCARD_PRICE_*` fallback). - Selected routes order through PCM `POST /order/postcard/with-list-count` — occupant addresses are not imported into Contacts. - Not official EDDM postage; PCM FirstClass/Standard purchased occupant lists on carrier routes. ## Test plan - [ ] Campaign compose: mailing-list audience still quotes list size × per-piece cost. - [ ] Switch to USPS carrier routes, search a ZIP, pick 1–2 routes, confirm household count + estimated cost. - [ ] Saving without a selection is rejected; saving with a quote creates one contact-less Message and shows the route panel on the report. - [ ] Send places `/order/postcard/with-list-count` (not per-recipient `/order/postcard`). - [ ] `uv run python manage.py test directmail.tests.CarrierRoutePostcardTests directmail.tests.PostcardDesignPickTests` Reviewed-on: #12
664 lines
21 KiB
Python
664 lines
21 KiB
Python
"""PCM Integrations (DirectMail API v3) postcard adapter."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import threading
|
||
from dataclasses import dataclass
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
|
||
import requests
|
||
from django.conf import settings
|
||
|
||
from directmail.providers.postcard import PostcardResult
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
PCM_API_BASE = "https://v3.pcmintegrations.com"
|
||
|
||
# Refresh a bit before PCM's expires timestamp.
|
||
_TOKEN_SKEW = timedelta(seconds=60)
|
||
|
||
# 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."""
|
||
|
||
|
||
_token_lock = threading.Lock()
|
||
_cached_token: str | None = None
|
||
_cached_expires: datetime | None = None
|
||
|
||
|
||
def _api_key() -> str:
|
||
return (settings.PCM_API_KEY or "").strip()
|
||
|
||
|
||
def _api_secret() -> str:
|
||
return (settings.PCM_API_SECRET or "").strip()
|
||
|
||
|
||
def _parse_expires(raw: Any) -> datetime:
|
||
"""Parse PCM expires timestamp; default to 55 minutes from now."""
|
||
if isinstance(raw, (int, float)):
|
||
# Unix seconds vs ms
|
||
ts = float(raw)
|
||
if ts > 1e12:
|
||
ts /= 1000.0
|
||
return datetime.fromtimestamp(ts, tz=timezone.utc)
|
||
if isinstance(raw, str) and raw.strip():
|
||
text = raw.strip().replace("Z", "+00:00")
|
||
try:
|
||
dt = datetime.fromisoformat(text)
|
||
except ValueError:
|
||
dt = None
|
||
if dt is not None:
|
||
if dt.tzinfo is None:
|
||
dt = dt.replace(tzinfo=timezone.utc)
|
||
return dt.astimezone(timezone.utc)
|
||
return datetime.now(timezone.utc) + timedelta(minutes=55)
|
||
|
||
|
||
def clear_token_cache() -> None:
|
||
"""Drop cached bearer token (tests / forced re-login)."""
|
||
global _cached_token, _cached_expires
|
||
with _token_lock:
|
||
_cached_token = None
|
||
_cached_expires = None
|
||
|
||
|
||
def login(*, force: bool = False) -> str:
|
||
"""POST /auth/login with apiKey + apiSecret → bearer token.
|
||
|
||
See https://docs.pcmintegrations.com/docs/directmail-api/ffef03a112bb0-logging-in
|
||
"""
|
||
global _cached_token, _cached_expires
|
||
|
||
with _token_lock:
|
||
now = datetime.now(timezone.utc)
|
||
if (
|
||
not force
|
||
and _cached_token
|
||
and _cached_expires
|
||
and now < (_cached_expires - _TOKEN_SKEW)
|
||
):
|
||
return _cached_token
|
||
|
||
key = _api_key()
|
||
secret = _api_secret()
|
||
if not key or not secret:
|
||
raise PcmApiError(
|
||
"PCM_API_KEY and PCM_API_SECRET are required "
|
||
"(POST /auth/login, then use the returned token)"
|
||
)
|
||
|
||
body: dict[str, str] = {"apiKey": key, "apiSecret": secret}
|
||
child = (getattr(settings, "PCM_CHILD_REF_NBR", None) or "").strip()
|
||
if child:
|
||
body["childRefNbr"] = child
|
||
|
||
response = requests.post(
|
||
f"{PCM_API_BASE}/auth/login",
|
||
headers={
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json",
|
||
},
|
||
json=body,
|
||
timeout=30,
|
||
)
|
||
if response.status_code >= 400:
|
||
detail = (response.text or "")[:500]
|
||
raise PcmApiError(
|
||
f"PCM POST /auth/login → {response.status_code}: {detail}"
|
||
)
|
||
try:
|
||
data = response.json()
|
||
except ValueError as exc:
|
||
raise PcmApiError("PCM login returned non-JSON") from exc
|
||
if not isinstance(data, dict):
|
||
raise PcmApiError("PCM login returned unexpected payload")
|
||
|
||
token = str(data.get("token") or "").strip()
|
||
if not token:
|
||
raise PcmApiError("PCM login response missing token")
|
||
|
||
_cached_token = token
|
||
_cached_expires = _parse_expires(data.get("expires"))
|
||
logger.info(
|
||
"PCM login ok; token expires %s",
|
||
_cached_expires.isoformat(),
|
||
)
|
||
return token
|
||
|
||
|
||
def _headers(*, force_login: bool = False) -> dict[str, str]:
|
||
token = login(force=force_login)
|
||
return {
|
||
"Accept": "application/json",
|
||
"Authorization": f"Bearer {token}",
|
||
"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,
|
||
)
|
||
# Expired/revoked session → login once and retry.
|
||
if response.status_code == 401:
|
||
clear_token_cache()
|
||
response = requests.request(
|
||
method,
|
||
url,
|
||
headers=_headers(force_login=True),
|
||
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 and/or PCM_RETURN_* vars.
|
||
|
||
PCM requires address/city/state (and zip) whenever firstName is set — an empty
|
||
street with SITE_NAME-derived firstName causes a 400 from POST /order/postcard.
|
||
"""
|
||
data: dict[str, Any] = {}
|
||
raw = (settings.PCM_RETURN_ADDRESS or "").strip()
|
||
if raw:
|
||
try:
|
||
parsed = json.loads(raw)
|
||
except json.JSONDecodeError as exc:
|
||
raise PcmApiError(
|
||
"PCM_RETURN_ADDRESS must be valid JSON "
|
||
'(e.g. {"firstName":"…","lastName":"…","address":"…",'
|
||
'"city":"…","state":"IL","zipCode":"…"})'
|
||
) from exc
|
||
if not isinstance(parsed, dict):
|
||
raise PcmApiError("PCM_RETURN_ADDRESS must be a JSON object")
|
||
data = parsed
|
||
|
||
name = (settings.SITE_NAME or "").strip()
|
||
parts = name.split(None, 1)
|
||
site_first = parts[0] if parts else ""
|
||
site_last = parts[1] if len(parts) > 1 else ""
|
||
|
||
address = {
|
||
"company": str(data.get("company") or "").strip(),
|
||
"firstName": str(
|
||
data.get("firstName") or data.get("first_name") or site_first or ""
|
||
).strip(),
|
||
"lastName": str(
|
||
data.get("lastName") or data.get("last_name") or site_last or ""
|
||
).strip(),
|
||
"address": str(
|
||
data.get("address")
|
||
or data.get("line1")
|
||
or getattr(settings, "PCM_RETURN_LINE1", "")
|
||
or ""
|
||
).strip(),
|
||
# PCM treats blank address2 oddly; space matches recipient payload.
|
||
"address2": str(
|
||
data.get("address2")
|
||
or data.get("line2")
|
||
or getattr(settings, "PCM_RETURN_LINE2", "")
|
||
or ""
|
||
).strip()
|
||
or " ",
|
||
"city": str(
|
||
data.get("city") or getattr(settings, "PCM_RETURN_CITY", "") or ""
|
||
).strip(),
|
||
"state": str(
|
||
data.get("state") or getattr(settings, "PCM_RETURN_STATE", "") or ""
|
||
).strip(),
|
||
"zipCode": str(
|
||
data.get("zipCode")
|
||
or data.get("zip")
|
||
or getattr(settings, "PCM_RETURN_ZIP", "")
|
||
or ""
|
||
).strip(),
|
||
}
|
||
|
||
missing = [
|
||
field
|
||
for field in ("firstName", "address", "city", "state", "zipCode")
|
||
if not address.get(field, "").strip()
|
||
]
|
||
if missing:
|
||
raise PcmApiError(
|
||
"PCM return address incomplete (missing "
|
||
+ ", ".join(missing)
|
||
+ "). Set PCM_RETURN_ADDRESS JSON with firstName, lastName, "
|
||
"address, city, state, zipCode "
|
||
"(or PCM_RETURN_LINE1 / CITY / STATE / ZIP)."
|
||
)
|
||
return address
|
||
|
||
|
||
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 get_design(design_id: int | str) -> dict[str, Any]:
|
||
data = pcm_request("GET", f"/design/{design_id}")
|
||
return data if isinstance(data, dict) else {}
|
||
|
||
|
||
FIRST_CLASS_ONLY_SIZES = frozenset({"46", "46S"})
|
||
MAX_SELECTED_ROUTES = 2
|
||
DEFAULT_LIST_TYPE = "IRL"
|
||
|
||
|
||
def mail_class_for_size(size: str | None) -> str:
|
||
key = str(size or "").strip()
|
||
if key in FIRST_CLASS_ONLY_SIZES:
|
||
return "FirstClass"
|
||
return "Standard"
|
||
|
||
|
||
def _env_unit_price(mail_class: str) -> float | None:
|
||
attr = (
|
||
"PCM_POSTCARD_PRICE_FIRSTCLASS"
|
||
if mail_class == "FirstClass"
|
||
else "PCM_POSTCARD_PRICE_STANDARD"
|
||
)
|
||
raw = str(getattr(settings, attr, "") or "").strip()
|
||
if not raw:
|
||
return None
|
||
try:
|
||
value = float(raw)
|
||
except ValueError:
|
||
return None
|
||
return value if value >= 0 else None
|
||
|
||
|
||
def pick_unit_price(
|
||
pricing: list[dict] | None,
|
||
*,
|
||
mail_class: str,
|
||
quantity: int,
|
||
) -> tuple[float | None, str]:
|
||
"""Return (dollars per piece, source). Source is pcm|fallback|missing."""
|
||
rows: list[dict] = []
|
||
for item in pricing or []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
if str(item.get("mailClass") or item.get("mail_class") or "") != mail_class:
|
||
continue
|
||
try:
|
||
price = float(item.get("price"))
|
||
break_qty = int(item.get("breakQty") or item.get("break_qty") or 0)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
rows.append({"price": price, "breakQty": break_qty})
|
||
qty = max(int(quantity or 0), 0)
|
||
if rows:
|
||
eligible = [row for row in rows if qty >= row["breakQty"]]
|
||
chosen = max(eligible, key=lambda row: row["breakQty"]) if eligible else min(
|
||
rows, key=lambda row: row["breakQty"]
|
||
)
|
||
return chosen["price"], "pcm"
|
||
fallback = _env_unit_price(mail_class)
|
||
if fallback is not None:
|
||
return fallback, "fallback"
|
||
return None, "missing"
|
||
|
||
|
||
def design_pricing(design_id: int | None) -> list[dict]:
|
||
"""Load PCM per-piece pricing for a design (gallery payload preferred)."""
|
||
if not design_id:
|
||
return []
|
||
try:
|
||
design = get_design(design_id)
|
||
except PcmApiError:
|
||
design = {}
|
||
pricing = design.get("pricing") if isinstance(design, dict) else None
|
||
if isinstance(pricing, list) and pricing:
|
||
return [row for row in pricing if isinstance(row, dict)]
|
||
try:
|
||
for item in list_designs(product_type="postcard"):
|
||
if not isinstance(item, dict):
|
||
continue
|
||
did = item.get("designID") or item.get("design_id") or item.get("id")
|
||
if str(did) != str(design_id):
|
||
continue
|
||
extra = item.get("pricing")
|
||
if isinstance(extra, list):
|
||
return [row for row in extra if isinstance(row, dict)]
|
||
except PcmApiError:
|
||
logger.exception("PCM list_designs failed while loading pricing")
|
||
return []
|
||
|
||
|
||
def extract_crrt_rows(payload: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Flatten ZipCRRT (or first) breakdown into {code, text, total} rows."""
|
||
breakdowns = payload.get("breakdown") if isinstance(payload, dict) else None
|
||
if not isinstance(breakdowns, list):
|
||
return []
|
||
preferred = None
|
||
fallback = None
|
||
for block in breakdowns:
|
||
if not isinstance(block, dict):
|
||
continue
|
||
kind = str(block.get("breakdownType") or "").upper().replace("_", "")
|
||
data = block.get("data") if isinstance(block.get("data"), list) else []
|
||
rows = []
|
||
for item in data:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
code = str(item.get("code") or "").strip()
|
||
if not code:
|
||
continue
|
||
try:
|
||
total = int(item.get("total") or 0)
|
||
except (TypeError, ValueError):
|
||
total = 0
|
||
rows.append(
|
||
{
|
||
"code": code,
|
||
"text": str(item.get("text") or code).strip(),
|
||
"total": total,
|
||
}
|
||
)
|
||
if not rows:
|
||
continue
|
||
if kind in {"ZIPCRRT", "CRRT", "CARRIERROUTE"}:
|
||
preferred = rows
|
||
break
|
||
if fallback is None:
|
||
fallback = rows
|
||
return preferred or fallback or []
|
||
|
||
|
||
def _list_type() -> str:
|
||
return (getattr(settings, "PCM_LIST_TYPE", None) or DEFAULT_LIST_TYPE).strip() or DEFAULT_LIST_TYPE
|
||
|
||
|
||
def list_count_zipcode(*, zip_codes: list[str], list_type: str | None = None) -> dict[str, Any]:
|
||
data = pcm_request(
|
||
"POST",
|
||
"/list/count/zipcode",
|
||
json_body={
|
||
"listType": list_type or _list_type(),
|
||
"breakdownType": "ZipCRRT",
|
||
"zipCodes": zip_codes,
|
||
"demographics": [],
|
||
},
|
||
)
|
||
return data if isinstance(data, dict) else {}
|
||
|
||
|
||
def list_count_radius(
|
||
*,
|
||
address: str,
|
||
city: str,
|
||
state: str,
|
||
zip_code: str,
|
||
miles: float,
|
||
list_type: str | None = None,
|
||
) -> dict[str, Any]:
|
||
data = pcm_request(
|
||
"POST",
|
||
"/list/count/radius",
|
||
json_body={
|
||
"listType": list_type or _list_type(),
|
||
"breakdownType": "ZipCRRT",
|
||
"radius": {
|
||
"radius": miles,
|
||
"address": address,
|
||
"city": city,
|
||
"state": state,
|
||
"zip": zip_code,
|
||
},
|
||
"demographics": [],
|
||
},
|
||
)
|
||
return data if isinstance(data, dict) else {}
|
||
|
||
|
||
def list_count_carrier_route(
|
||
*,
|
||
carrier_routes: list[str],
|
||
list_type: str | None = None,
|
||
) -> dict[str, Any]:
|
||
data = pcm_request(
|
||
"POST",
|
||
"/list/count/carrier-route",
|
||
json_body={
|
||
"listType": list_type or _list_type(),
|
||
"breakdownType": "ZipCRRT",
|
||
"carrierRoutes": carrier_routes,
|
||
"demographics": [],
|
||
},
|
||
)
|
||
return data if isinstance(data, dict) else {}
|
||
|
||
|
||
def _order_id_from_response(data: dict[str, Any]) -> str:
|
||
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 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],
|
||
}
|
||
# Product-specific path — plain POST /order is 404 ("Cannot POST /order").
|
||
# Docs: Place Postcard Order → POST /order/postcard
|
||
data = pcm_request("POST", "/order/postcard", json_body=payload)
|
||
if not isinstance(data, dict):
|
||
raise PcmApiError("Unexpected order response from PCM")
|
||
return _order_id_from_response(data)
|
||
|
||
|
||
def place_postcard_order_with_list_count(
|
||
*,
|
||
design_id: int,
|
||
list_count_id: int,
|
||
record_count: int,
|
||
ext_ref: str,
|
||
mail_class: str = "Standard",
|
||
) -> str:
|
||
"""Place a postcard order against a PCM purchased list count."""
|
||
payload = {
|
||
"designID": design_id,
|
||
"mailClass": mail_class,
|
||
"extRefNbr": ext_ref,
|
||
"listCountID": list_count_id,
|
||
"recordCount": record_count,
|
||
"returnAddress": return_address_from_settings(),
|
||
}
|
||
data = pcm_request(
|
||
"POST", "/order/postcard/with-list-count", json_body=payload
|
||
)
|
||
if not isinstance(data, dict):
|
||
raise PcmApiError("Unexpected list-count order response from PCM")
|
||
return _order_id_from_response(data)
|
||
|
||
|
||
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)"
|
||
)
|
||
|
||
targeting = {}
|
||
if message.campaign_id and isinstance(
|
||
getattr(message.campaign, "route_targeting", None), dict
|
||
):
|
||
targeting = message.campaign.route_targeting
|
||
list_count_id = targeting.get("list_count_id")
|
||
record_count = targeting.get("record_count")
|
||
if list_count_id and record_count:
|
||
mail_class = str(targeting.get("mail_class") or "Standard")
|
||
order_id = place_postcard_order_with_list_count(
|
||
design_id=design_id,
|
||
list_count_id=int(list_count_id),
|
||
record_count=int(record_count),
|
||
ext_ref=str(message.pk),
|
||
mail_class=mail_class,
|
||
)
|
||
return PostcardResult(provider_id=order_id)
|
||
|
||
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")
|