clientflow_backend/scripts/backfill_chatwoot_inbox_to_clientflow.py

#!/usr/bin/env python3
import hashlib
import hmac
import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional


def env(name: str, default: str = "") -> str:
    return os.getenv(name, default).strip()


CHATWOOT_BASE_URL = env("CHATWOOT_BASE_URL").rstrip("/")
CHATWOOT_ACCOUNT_ID = env("CHATWOOT_ACCOUNT_ID")
CHATWOOT_API_TOKEN = env("CHATWOOT_API_TOKEN")
CLIENTFLOW_WEBHOOK_SECRET = env("CLIENTFLOW_WEBHOOK_SECRET")
CLIENTFLOW_WEBHOOK_URL = env("CLIENTFLOW_WEBHOOK_URL", "http://127.0.0.1:8020/webhooks/chatwoot")

BACKFILL_DAYS = int(env("BACKFILL_DAYS", "5"))
BACKFILL_TO_CLIENTFLOW = env("BACKFILL_TO_CLIENTFLOW", "false").lower() == "true"
BACKFILL_STATUSES = [s.strip() for s in env("BACKFILL_STATUSES", "open,pending").split(",") if s.strip()]
BACKFILL_MAX_PAGES = int(env("BACKFILL_MAX_PAGES", "20"))


def request_json(method: str, url: str, body: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
    data = None
    final_headers = headers.copy() if headers else {}

    if body is not None:
        data = json.dumps(body, ensure_ascii=False).encode("utf-8")
        final_headers["Content-Type"] = "application/json"

    req = urllib.request.Request(url, data=data, headers=final_headers, method=method)

    try:
        with urllib.request.urlopen(req, timeout=45) as resp:
            raw = resp.read().decode("utf-8", errors="replace")
            return {
                "ok": 200 <= resp.status < 300,
                "status": resp.status,
                "json": json.loads(raw) if raw else {},
                "raw": raw,
            }
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        return {
            "ok": False,
            "status": e.code,
            "json": None,
            "raw": raw,
        }


def chatwoot_headers() -> Dict[str, str]:
    return {
        "api_access_token": CHATWOOT_API_TOKEN,
        "Accept": "application/json",
    }


def payload_list(data: Any) -> List[Dict[str, Any]]:
    if isinstance(data, list):
        return data

    if not isinstance(data, dict):
        return []

    candidates = [
        data.get("payload"),
        data.get("data", {}).get("payload") if isinstance(data.get("data"), dict) else None,
        data.get("data"),
        data.get("messages"),
    ]

    for item in candidates:
        if isinstance(item, list):
            return item

    return []


def ts_to_datetime(value: Any) -> Optional[datetime]:
    if value is None:
        return None

    try:
        if isinstance(value, (int, float)):
            return datetime.fromtimestamp(float(value), tz=timezone.utc)

        s = str(value).replace("Z", "+00:00")
        return datetime.fromisoformat(s).astimezone(timezone.utc)
    except Exception:
        return None


def conversation_last_activity(conversation: Dict[str, Any]) -> Optional[datetime]:
    for key in ["last_activity_at", "updated_at", "created_at"]:
        dt = ts_to_datetime(conversation.get(key))
        if dt:
            return dt
    return None


def fetch_conversations(status: str) -> List[Dict[str, Any]]:
    all_items: List[Dict[str, Any]] = []

    for page in range(1, BACKFILL_MAX_PAGES + 1):
        query = urllib.parse.urlencode({
            "status": status,
            "page": page,
        })

        url = f"{CHATWOOT_BASE_URL}/api/v1/accounts/{CHATWOOT_ACCOUNT_ID}/conversations?{query}"
        result = request_json("GET", url, headers=chatwoot_headers())

        if not result["ok"]:
            print(f"ERRO Chatwoot conversations status={status} page={page}: {result['status']} {result['raw'][:300]}")
            break

        items = payload_list(result["json"])

        if not items:
            break

        all_items.extend(items)

        if len(items) < 10:
            break

    return all_items


def fetch_messages(conversation_id: str) -> List[Dict[str, Any]]:
    url = f"{CHATWOOT_BASE_URL}/api/v1/accounts/{CHATWOOT_ACCOUNT_ID}/conversations/{conversation_id}/messages"
    result = request_json("GET", url, headers=chatwoot_headers())

    if not result["ok"]:
        print(f"ERRO Chatwoot messages conversation={conversation_id}: {result['status']} {result['raw'][:300]}")
        return []

    return payload_list(result["json"])


def is_incoming(message: Dict[str, Any]) -> bool:
    mt = message.get("message_type")
    return mt == "incoming" or mt == 0 or str(mt).lower() == "incoming"


def message_created_at(message: Dict[str, Any]) -> datetime:
    return ts_to_datetime(message.get("created_at")) or datetime.fromtimestamp(0, tz=timezone.utc)


def sender_from_conversation(conversation: Dict[str, Any], message: Dict[str, Any]) -> Dict[str, Any]:
    sender = {}

    meta = conversation.get("meta") or {}
    if isinstance(meta, dict) and isinstance(meta.get("sender"), dict):
        sender.update(meta.get("sender") or {})

    if isinstance(conversation.get("contact"), dict):
        sender.update({k: v for k, v in conversation["contact"].items() if v is not None})

    if isinstance(message.get("sender"), dict):
        sender.update({k: v for k, v in message["sender"].items() if v is not None})

    return sender


def sign_body(body_raw: str) -> Dict[str, str]:
    ts = str(int(time.time()))
    msg = ts.encode("utf-8") + b"." + body_raw.encode("utf-8")
    sig = "sha256=" + hmac.new(
        CLIENTFLOW_WEBHOOK_SECRET.encode("utf-8"),
        msg,
        hashlib.sha256,
    ).hexdigest()

    return {
        "Content-Type": "application/json",
        "X-Chatwoot-Timestamp": ts,
        "X-Chatwoot-Signature": sig,
    }


def post_to_clientflow(payload: Dict[str, Any]) -> Dict[str, Any]:
    # Importante: assinar e enviar exatamente o mesmo body_raw.
    # Se o JSON for reformatado depois da assinatura, o webhook rejeita com 401.
    body_raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
    headers = sign_body(body_raw)

    req = urllib.request.Request(
        CLIENTFLOW_WEBHOOK_URL,
        data=body_raw.encode("utf-8"),
        headers=headers,
        method="POST",
    )

    try:
        with urllib.request.urlopen(req, timeout=45) as resp:
            raw = resp.read().decode("utf-8", errors="replace")
            return {
                "ok": 200 <= resp.status < 300,
                "status": resp.status,
                "json": json.loads(raw) if raw else {},
                "raw": raw,
            }
    except urllib.error.HTTPError as e:
        raw = e.read().decode("utf-8", errors="replace")
        return {
            "ok": False,
            "status": e.code,
            "json": None,
            "raw": raw,
        }


def main() -> int:
    required = {
        "CHATWOOT_BASE_URL": CHATWOOT_BASE_URL,
        "CHATWOOT_ACCOUNT_ID": CHATWOOT_ACCOUNT_ID,
        "CHATWOOT_API_TOKEN": CHATWOOT_API_TOKEN,
        "CLIENTFLOW_WEBHOOK_SECRET": CLIENTFLOW_WEBHOOK_SECRET,
    }

    missing = [k for k, v in required.items() if not v]
    if missing:
        raise SystemExit(f"Faltam variáveis: {', '.join(missing)}")

    cutoff = datetime.now(timezone.utc) - timedelta(days=BACKFILL_DAYS)

    print(f"BACKFILL_DAYS={BACKFILL_DAYS}")
    print(f"BACKFILL_STATUSES={BACKFILL_STATUSES}")
    print(f"BACKFILL_TO_CLIENTFLOW={BACKFILL_TO_CLIENTFLOW}")
    print(f"CUTOFF={cutoff.isoformat()}")

    seen_conversations = set()
    selected = 0
    posted = 0
    failed = 0
    skipped = 0

    for status in BACKFILL_STATUSES:
        conversations = fetch_conversations(status)
        print(f"--- status={status} conversations={len(conversations)}")

        for conv in conversations:
            conv_id = str(conv.get("id") or "")
            if not conv_id or conv_id in seen_conversations:
                continue

            seen_conversations.add(conv_id)

            last_activity = conversation_last_activity(conv)
            if last_activity and last_activity < cutoff:
                skipped += 1
                continue

            messages = fetch_messages(conv_id)
            incoming_messages = [
                m for m in messages
                if is_incoming(m)
                and not m.get("private")
                and str(m.get("content") or "").strip()
                and message_created_at(m) >= cutoff
            ]

            if not incoming_messages:
                skipped += 1
                continue

            incoming_messages.sort(key=message_created_at)
            last_msg = incoming_messages[-1]

            content = str(last_msg.get("content") or "").strip()
            sender = sender_from_conversation(conv, last_msg)
            contact_id = str(sender.get("id") or conv.get("contact_id") or conv.get("contact", {}).get("id") or "")

            selected += 1

            payload = {
                "event": "message_created",
                "message": {
                    "id": str(last_msg.get("id") or f"backfill-{conv_id}"),
                    "content": content,
                    "message_type": "incoming",
                    "conversation_id": conv_id,
                    "sender": sender,
                    "created_at": last_msg.get("created_at"),
                },
                "conversation": {
                    "id": conv_id,
                    "status": conv.get("status") or status,
                    "contact": sender,
                    "meta": {
                        "sender": sender,
                    },
                },
                "backfill": {
                    "source": "chatwoot_inbox_last_days",
                    "days": BACKFILL_DAYS,
                    "status": status,
                    "last_activity_at": conv.get("last_activity_at"),
                },
            }

            print("---")
            print(f"conversation={conv_id} contact={contact_id} msg={payload['message']['id']}")
            print(f"content={content[:160].replace(chr(10), ' ')}")

            if not BACKFILL_TO_CLIENTFLOW:
                print("DRY_RUN")
                continue

            result = post_to_clientflow(payload)

            if result["ok"]:
                print(f"POSTED status={result['status']}")
                posted += 1
            else:
                print(f"FAILED status={result['status']} body={result['raw'][:500]}")
                failed += 1

    print("---")
    print(f"selected={selected}")
    print(f"posted={posted}")
    print(f"failed={failed}")
    print(f"skipped={skipped}")

    return 0 if failed == 0 else 1


if __name__ == "__main__":
    raise SystemExit(main())