Integrating Instagram, Lead Ads and Telegram with amoCRM and Bitrix24

import json, os
from aiohttp import web
from socialift.webhooks import verify_request, parse_events

APP_SECRET = os.environ["META_APP_SECRET"]

async def meta_webhook(request: web.Request) -> web.Response:
    body = await request.read()  # raw bytes — the signature covers these
    if not verify_request(APP_SECRET, body, request.headers):
        return web.Response(status=403)
    for event in parse_events(json.loads(body)):
        if event.field == "leadgen" and await mark_seen(event.key):
            await queue.put({"leadgen_id": event.leadgen_id, "page_id": event.page_id})
    return web.Response(text="ok")

import re
from socialift import Facebook
from socialift.leads import normalize

def to_e164_uz(raw: str | None) -> str | None:
    digits = re.sub(r"\D", "", raw or "")
    if digits.startswith("00"):
        digits = digits[2:]
    if len(digits) == 12 and digits.startswith("998"):
        return "+" + digits
    if len(digits) == 9:  # 90 123 45 67
        return "+998" + digits
    return None  # route to manual review

async def load_lead(fb: Facebook, leadgen_id: str, page_token: str) -> dict:
    lead = await fb.lead(leadgen_id, page_token)
    person = normalize(lead.field_data)
    return {"lead_id": lead.id, "name": person.full_name, "email": person.email,
            "phone": to_e164_uz(person.phone), "phone_raw": person.phone,
            "form_id": lead.form_id, "ad_id": lead.ad_id,
            "campaign": lead.campaign_name, "organic": lead.is_organic}

import os, aiohttp

AMO = f"https://{os.environ['AMO_SUBDOMAIN']}.amocrm.ru/api/v4"
HEADERS = {"Authorization": f"Bearer {os.environ['AMO_ACCESS_TOKEN']}"}

async def amo_create(session: aiohttp.ClientSession, lead: dict) -> dict:
    payload = [{
        "name": f"Lead Ads: {lead['campaign'] or lead['form_id']}",
        "request_id": lead["lead_id"],
        "_embedded": {
            "tags": [{"name": "instagram-lead-ads"}],
            "contacts": [{
                "first_name": lead["name"] or "Lead",
                "custom_fields_values": [{"field_code": "PHONE",
                    "values": [{"enum_code": "MOB", "value": lead["phone"]}]}],
            }],
        },
    }]
    async with session.post(f"{AMO}/leads/complex", json=payload, headers=HEADERS) as r:
        if r.status == 429:
            raise RetryLater(r.status)
        r.raise_for_status()
        return (await r.json())[0]  # id, contact_id, merged

import os, aiohttp

B24 = os.environ["B24_WEBHOOK_URL"]  # https://<portal>/rest/<user_id>/<secret>/

async def b24(session: aiohttp.ClientSession, method: str, params: dict):
    async with session.post(f"{B24}{method}.json", json=params) as r:
        data = await r.json()
    if "error" in data:  # QUERY_LIMIT_EXCEEDED -> retry later
        raise B24Error(data["error"], data.get("error_description"))
    return data["result"]

async def b24_create(session: aiohttp.ClientSession, lead: dict) -> int:
    dup = await b24(session, "crm.duplicate.findbycomm",
                    {"entity_type": "LEAD", "type": "PHONE", "values": [lead["phone"]]})
    if isinstance(dup, dict) and dup.get("LEAD"):
        return dup["LEAD"][0]
    res = await b24(session, "crm.item.add", {"entityTypeId": 1, "fields": {
        "title": f"Lead Ads: {lead['campaign'] or lead['form_id']}",
        "name": lead["name"],
        "utmCampaign": lead["campaign"],
        "sourceDescription": f"form={lead['form_id']} ad={lead['ad_id']}",
        "fm": [{"typeId": "PHONE", "valueType": "MOBILE", "value": lead["phone"]}],
    }})
    return res["item"]["id"]