c815a32bbc
Orquestra create_client/order FOSS, bridge OpenPanel com metadata, upsert res.partner Odoo, mail-bundle wizard e webhook order-activated. Co-authored-by: Cursor <cursoragent@cursor.com>
242 lines
8.4 KiB
Python
242 lines
8.4 KiB
Python
"""Mapeamento activação cliente — Spec 043 activation-field-mapping.md."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
DEFAULT_PLAN_CODE = "ligbox-mail-business"
|
|
|
|
PLAN_DEFAULTS: dict[str, dict[str, int]] = {
|
|
"ligbox-mail-starter": {"seats": 10, "mail_gb": 20, "files_gb": 100},
|
|
"ligbox-mail-business": {"seats": 25, "mail_gb": 30, "files_gb": 200},
|
|
"ligbox-mail-enterprise": {"seats": 50, "mail_gb": 50, "files_gb": 500},
|
|
"ligbox-mail-custom": {"seats": 25, "mail_gb": 30, "files_gb": 200},
|
|
"ligbox-site-cms": {"seats": 1, "mail_gb": 0, "files_gb": 10},
|
|
}
|
|
|
|
ACTIVATABLE_STATES = frozenset(
|
|
{"awaiting_billing_validation", "provisioning", "policy_pending"}
|
|
)
|
|
|
|
|
|
def _digits_only(value: str | None) -> str:
|
|
if not value:
|
|
return ""
|
|
return re.sub(r"\D", "", str(value))
|
|
|
|
|
|
def normalize_company_profile(
|
|
profile: dict[str, Any] | None,
|
|
*,
|
|
domain: str,
|
|
) -> dict[str, Any]:
|
|
"""Unifica manager_* e address vindos do webhook wizard."""
|
|
p = dict(profile or {})
|
|
addr = p.get("address")
|
|
if not isinstance(addr, dict):
|
|
addr = {}
|
|
p["address"] = {
|
|
"country": addr.get("country") or "BR",
|
|
"city": addr.get("city") or "",
|
|
"postal_code": addr.get("postal_code") or "",
|
|
"street": addr.get("street") or "",
|
|
"state": addr.get("state") or "",
|
|
}
|
|
p.setdefault("domain", domain.strip().lower())
|
|
if not p.get("manager_email"):
|
|
p["manager_email"] = f"admin@{domain.strip().lower()}"
|
|
if not p.get("manager_name"):
|
|
p["manager_name"] = p.get("trade_name") or p.get("legal_name") or ""
|
|
return p
|
|
|
|
|
|
def split_manager_name(name: str | None) -> tuple[str, str]:
|
|
parts = (name or "").strip().split(None, 1)
|
|
if not parts:
|
|
return ("", "")
|
|
if len(parts) == 1:
|
|
return (parts[0], "")
|
|
return (parts[0], parts[1])
|
|
|
|
|
|
def derive_openpanel_username(domain: str, profile: dict[str, Any] | None = None) -> str:
|
|
"""Sugestão username hub — Spec 024 (7 chars alfanum + sufixo)."""
|
|
existing = (profile or {}).get("openpanel_username")
|
|
if existing:
|
|
return str(existing).strip().lower()
|
|
label = re.sub(r"[^a-z0-9]", "", domain.split(".")[0].lower()) or "ligbox"
|
|
if len(label) <= 7:
|
|
return f"{label}x"[:16]
|
|
return f"{label[:7]}x"[:16]
|
|
|
|
|
|
def resolve_plan_config(plan_code: str | None, config: dict[str, Any] | None = None) -> tuple[str, dict[str, int]]:
|
|
code = (plan_code or DEFAULT_PLAN_CODE).strip() or DEFAULT_PLAN_CODE
|
|
defaults = dict(PLAN_DEFAULTS.get(code, PLAN_DEFAULTS[DEFAULT_PLAN_CODE]))
|
|
if config:
|
|
for key in ("seats", "mail_gb", "files_gb"):
|
|
if config.get(key) is not None:
|
|
defaults[key] = int(config[key])
|
|
return code, defaults
|
|
|
|
|
|
def map_foss_client(profile: dict[str, Any], domain: str) -> dict[str, Any]:
|
|
first, last = split_manager_name(profile.get("manager_name"))
|
|
email = (profile.get("manager_email") or profile.get("email_billing") or "").strip().lower()
|
|
addr = profile.get("address") or {}
|
|
return {
|
|
"email": email,
|
|
"first_name": first,
|
|
"last_name": last,
|
|
"company": profile.get("legal_name") or profile.get("trade_name") or domain,
|
|
"company_vat": _digits_only(profile.get("tax_id")),
|
|
"phone": profile.get("phone_mobile") or profile.get("contact_phone") or profile.get("phone_landline") or "",
|
|
"address_1": addr.get("street") or "",
|
|
"city": addr.get("city") or "",
|
|
"state": addr.get("state") or "",
|
|
"postcode": addr.get("postal_code") or "",
|
|
"country": addr.get("country") or "BR",
|
|
"currency": "BRL",
|
|
"status": "active",
|
|
"custom_ligbox_domain": domain,
|
|
}
|
|
|
|
|
|
def map_foss_order(
|
|
*,
|
|
domain: str,
|
|
profile: dict[str, Any],
|
|
plan_code: str,
|
|
config: dict[str, int],
|
|
foss_client_id: str | int | None = None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"client_id": foss_client_id,
|
|
"product_slug": plan_code,
|
|
"period": "1M",
|
|
"domain": domain,
|
|
"manager_email": profile.get("manager_email"),
|
|
"manager_name": profile.get("manager_name"),
|
|
"config": {
|
|
"seats": config.get("seats"),
|
|
"mail_gb": config.get("mail_gb"),
|
|
"files_gb": config.get("files_gb"),
|
|
},
|
|
}
|
|
|
|
|
|
def map_openpanel(
|
|
*,
|
|
domain: str,
|
|
profile: dict[str, Any],
|
|
plan_code: str,
|
|
config: dict[str, int],
|
|
) -> dict[str, Any]:
|
|
username = derive_openpanel_username(domain, profile)
|
|
return {
|
|
"username": username,
|
|
"email": profile.get("manager_email"),
|
|
"plan_name": plan_code,
|
|
"domain": domain,
|
|
"metadata": {
|
|
"bundle_type": "ligbox_mail" if plan_code.startswith("ligbox-mail") else "ligbox_hosting",
|
|
"max_seats": config.get("seats"),
|
|
"mail_gb_per_seat": config.get("mail_gb"),
|
|
"files_gb_per_seat": config.get("files_gb"),
|
|
"easydmarc": True,
|
|
"wizard_domain": domain,
|
|
"manager_name": profile.get("manager_name"),
|
|
"manager_email": profile.get("manager_email"),
|
|
},
|
|
}
|
|
|
|
|
|
def map_odoo_partner(profile: dict[str, Any], domain: str, foss_client_id: str | int | None = None) -> dict[str, Any]:
|
|
addr = profile.get("address") or {}
|
|
return {
|
|
"name": profile.get("legal_name") or profile.get("trade_name") or domain,
|
|
"display_name": profile.get("trade_name") or profile.get("legal_name") or domain,
|
|
"vat": profile.get("tax_id") or "",
|
|
"email": profile.get("email_billing") or profile.get("manager_email") or "",
|
|
"phone": profile.get("phone_mobile") or profile.get("contact_phone") or "",
|
|
"street": addr.get("street") or "",
|
|
"city": addr.get("city") or "",
|
|
"zip": addr.get("postal_code") or "",
|
|
"country_code": addr.get("country") or "BR",
|
|
"company_type": "company",
|
|
"customer_rank": 1,
|
|
"ref": domain,
|
|
"comment": {
|
|
"manager_email": profile.get("manager_email"),
|
|
"foss_client_id": foss_client_id,
|
|
"domain": domain,
|
|
},
|
|
}
|
|
|
|
|
|
def map_wizard_mail_bundle(
|
|
*,
|
|
domain: str,
|
|
profile: dict[str, Any],
|
|
config: dict[str, int],
|
|
foss_order_id: str | int | None = None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"domain": domain,
|
|
"admin_email": profile.get("manager_email"),
|
|
"admin_name": profile.get("manager_name"),
|
|
"seats": config.get("seats"),
|
|
"mail_gb_per_seat": config.get("mail_gb"),
|
|
"files_gb_per_seat": config.get("files_gb"),
|
|
"foss_order_id": foss_order_id,
|
|
}
|
|
|
|
|
|
def build_activation_preview(
|
|
account: dict[str, Any],
|
|
*,
|
|
plan_code: str | None = None,
|
|
config: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
domain = str(account.get("domain") or "").strip().lower()
|
|
profile = normalize_company_profile(account.get("company_profile"), domain=domain)
|
|
code, cfg = resolve_plan_config(plan_code or account.get("plan_code"), config)
|
|
foss_id = account.get("external_customer_id")
|
|
billing_state = account.get("billing_state") or "awaiting_billing_validation"
|
|
can_activate = billing_state in ACTIVATABLE_STATES and not account.get("recurrence_active")
|
|
|
|
return {
|
|
"billing_account_id": account.get("id"),
|
|
"domain": domain,
|
|
"billing_state": billing_state,
|
|
"can_activate": can_activate,
|
|
"plan_code": code,
|
|
"config": cfg,
|
|
"company_profile": profile,
|
|
"mapped": {
|
|
"foss_client": map_foss_client(profile, domain),
|
|
"foss_order": map_foss_order(
|
|
domain=domain,
|
|
profile=profile,
|
|
plan_code=code,
|
|
config=cfg,
|
|
foss_client_id=foss_id,
|
|
),
|
|
"openpanel": map_openpanel(domain=domain, profile=profile, plan_code=code, config=cfg),
|
|
"odoo_partner": map_odoo_partner(profile, domain, foss_client_id=foss_id),
|
|
"wizard_mail_bundle": map_wizard_mail_bundle(
|
|
domain=domain,
|
|
profile=profile,
|
|
config=cfg,
|
|
),
|
|
},
|
|
"existing_ids": {
|
|
"foss_client_id": foss_id,
|
|
"external_subscription_id": account.get("external_subscription_id"),
|
|
"openpanel_username": profile.get("openpanel_username"),
|
|
},
|
|
"notes": (
|
|
"Preview read-only — confirmar dados antes de Activar conta."
|
|
),
|
|
}
|