API EVIR

# Python 3.11+ · сохраните файл как evir_integration.py
# pip install "aiogram>=3,<4" "httpx>=0.27,<1"
import asyncio
import hashlib
import json
import logging
import os
import re
import sqlite3

import httpx
from aiogram import F, Router
from aiogram.types import (
    CallbackQuery,
    InlineKeyboardButton,
    InlineKeyboardMarkup,
    Message,
)

EVIR_KEY = os.environ["EVIR_API_KEY"]
EVIR_URL = "https://evir.me/api/v1/integrations"
CALLBACK_PREFIX = "evir_check:"
DELIVERY_ID = re.compile(r"^[A-Za-z0-9_-]{32}$")

evir_router = Router(name="evir")
http = httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0))
db = sqlite3.connect(
    os.getenv("EVIR_STATE_DB", "evir-deliveries.sqlite3"),
    isolation_level=None,
)
db.row_factory = sqlite3.Row
db.execute("PRAGMA journal_mode=WAL")
db.execute("PRAGMA busy_timeout=5000")
db.executescript("""
CREATE TABLE IF NOT EXISTS evir_deliveries (
    delivery_id TEXT PRIMARY KEY,
    recipient_id TEXT NOT NULL,
    payload_json TEXT NOT NULL,
    state TEXT NOT NULL,
    requires_served_ack INTEGER NOT NULL,
    expires_at TEXT NOT NULL,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS evir_delivery_recipient_state
ON evir_deliveries(recipient_id, state);
""")
delivery_columns = {
    row["name"] for row in db.execute("PRAGMA table_info(evir_deliveries)").fetchall()
}
if "expires_at" not in delivery_columns:
    db.execute("ALTER TABLE evir_deliveries ADD COLUMN expires_at TEXT")

db.execute("""CREATE TABLE IF NOT EXISTS evir_check_batches (
    batch_id TEXT PRIMARY KEY, recipient_id TEXT NOT NULL, delivery_ids_json TEXT NOT NULL
)""")

class EvirError(RuntimeError):
    def __init__(self, message, status=None, code=None):
        super().__init__(message)
        self.status = status
        self.code = code

async def evir_post(path, payload):
    for attempt in range(3):
        try:
            response = await http.post(
                f"{EVIR_URL}{path}",
                headers={
                    "Authorization": f"Bearer {EVIR_KEY}",
                    "Content-Type": "application/json",
                },
                json=payload,
            )
        except httpx.RequestError:
            if attempt == 2:
                raise
            await asyncio.sleep(0.5 * (2 ** attempt))
            continue

        if response.status_code == 429 and attempt < 2:
            try:
                delay = max(1.0, float(response.headers.get("Retry-After", "1")))
            except ValueError:
                delay = 1.0
            await asyncio.sleep(delay)
            continue
        if response.status_code >= 500 and attempt < 2:
            await asyncio.sleep(0.5 * (2 ** attempt))
            continue

        try:
            body = response.json()
        except ValueError:
            body = {}
        if response.is_error:
            error = body.get("error", {})
            raise EvirError(
                error.get("message", f"EVIR: HTTP {response.status_code}"),
                status=response.status_code,
                code=error.get("code"),
            )
        return body["data"]
    raise RuntimeError("EVIR request failed")

def telegram_keyboard(assignment):
    rows = [[InlineKeyboardButton(
        text=button["label"], url=button["actionUrl"],
        icon_custom_emoji_id=button.get("iconCustomEmojiId"),
        style=button.get("style"),
    ) for button in row] for row in assignment["buttonRows"]] if assignment.get("buttonRows") else [[InlineKeyboardButton(
        text=assignment["ctaLabel"],
        url=assignment["actionUrl"],  # используйте URL без изменений
        icon_custom_emoji_id=assignment.get("ctaIconCustomEmojiId"),
    )]]
    if assignment.get("productType") == "op" or assignment.get("billingMode") == "qualified_action":
        rows.append([InlineKeyboardButton(
            text="Проверить все",
            callback_data=f'{CALLBACK_PREFIX}{assignment["evirCheckBatch"]}',
        )])
    return InlineKeyboardMarkup(inline_keyboard=rows)

def telegram_content(assignment):
    prefix = "" if assignment.get("preserveContent") else assignment["title"] + "\n\n"
    prefix_units = len(prefix.encode("utf-16-le")) // 2
    entities = []
    for entity in assignment.get("descriptionEntities", []):
        item = {"type": entity["type"], "offset": prefix_units + entity["offset"], "length": entity["length"]}
        if entity.get("customEmojiId"):
            item["custom_emoji_id"] = entity["customEmojiId"]
        if entity.get("language"):
            item["language"] = entity["language"]
        if entity.get("url"):
            item["url"] = entity["url"]
        entities.append(item)
    return prefix + assignment["description"], entities

def remember_delivery(recipient_id, assignment):
    db.execute(
        """INSERT OR IGNORE INTO evir_deliveries
        (delivery_id, recipient_id, payload_json, state, requires_served_ack, expires_at)
        VALUES (?, ?, ?, 'allocated', ?, ?)""",
        (
            assignment["deliveryId"],
            recipient_id,
            json.dumps(assignment, ensure_ascii=False),
            1 if assignment.get("requiresServedAck") else 0,
            assignment["expiresAt"],
        ),
    )

async def confirm_served(row):
    await evir_post(
        f'/deliveries/{row["delivery_id"]}/served',
        {},
    )
    db.execute(
        "UPDATE evir_deliveries SET state = 'acked', updated_at = CURRENT_TIMESTAMP "
        "WHERE delivery_id = ?",
        (row["delivery_id"],),
    )

async def retry_served_acks():
    rows = db.execute(
        "SELECT * FROM evir_deliveries WHERE state = 'sent' "
        "AND requires_served_ack = 1 ORDER BY rowid LIMIT 10",
    ).fetchall()
    for row in rows:
        try:
            await confirm_served(row)
        except EvirError as error:
            if error.status in {404, 410}:
                db.execute(
                    "UPDATE evir_deliveries SET state = 'expired', updated_at = CURRENT_TIMESTAMP "
                    "WHERE delivery_id = ?",
                    (row["delivery_id"],),
                )
                continue
            logging.exception("EVIR delivery acknowledgment retry failed: %s", row["delivery_id"])
        except httpx.RequestError:
            logging.exception("EVIR delivery acknowledgment retry failed: %s", row["delivery_id"])

async def evir_ack_worker():
    while True:
        try:
            await retry_served_acks()
        except asyncio.CancelledError:
            raise
        except Exception:
            logging.exception("EVIR acknowledgment worker failed; it will retry")
        await asyncio.sleep(15)

async def start_evir_ack_worker(**_):
    global evir_ack_task
    if evir_ack_task is None or evir_ack_task.done():
        evir_ack_task = asyncio.create_task(evir_ack_worker(), name="evir-ack-worker")

async def stop_evir_ack_worker(**_):
    global evir_ack_task
    if evir_ack_task is None:
        return
    evir_ack_task.cancel()
    try:
        await evir_ack_task
    except asyncio.CancelledError:
        pass
    evir_ack_task = None

evir_ack_task = None
evir_router.startup.register(start_evir_ack_worker)
evir_router.shutdown.register(stop_evir_ack_worker)

async def send_evir_cards(message: Message):
    user = message.from_user
    if user is None:
        return 0
    recipient_id = str(user.id)
    db.execute(
        "UPDATE evir_deliveries SET state = 'expired', updated_at = CURRENT_TIMESTAMP "
        "WHERE recipient_id = ? AND state = 'allocated' "
        "AND (expires_at IS NULL OR julianday(expires_at) <= julianday('now'))",
        (recipient_id,),
    )

    data = await evir_post("/next", {
        "recipientId": recipient_id,
        "limit": 3,
        "productTypes": ["op","transition","impression"],
        "supportsRichCards": True,
        "supportsButtonRows": True,
        "supportsQualifiedImpressions": True,
        "supportsPremiumEmoji": False,  # Enable only when this sender can send custom emoji.
    })
    ids = [card["deliveryId"] for card in data["assignments"] if card["productType"] == "op" or card.get("billingMode") == "qualified_action"]
    batch_id = hashlib.sha256((recipient_id + ":" + json.dumps(ids)).encode()).hexdigest()[:32]
    if ids:
        db.execute("INSERT OR IGNORE INTO evir_check_batches VALUES (?, ?, ?)",
                   (batch_id, recipient_id, json.dumps(ids)))
    for original in data["assignments"]:
        assignment = {**original, "evirCheckBatch": batch_id}
        remember_delivery(recipient_id, assignment)
        db.execute("UPDATE evir_deliveries SET payload_json = ? WHERE delivery_id = ? "
                   "AND recipient_id = ? AND state = 'allocated'",
                   (json.dumps(assignment), assignment["deliveryId"], recipient_id))

    rows = db.execute(
        "SELECT * FROM evir_deliveries WHERE recipient_id = ? "
        "AND state = 'allocated' AND julianday(expires_at) > julianday('now') ORDER BY rowid",
        (recipient_id,),
    ).fetchall()
    sent_count = 0
    for row in rows:
        claimed = db.execute(
            "UPDATE evir_deliveries SET state = 'sending', updated_at = CURRENT_TIMESTAMP "
            "WHERE delivery_id = ? AND state = 'allocated' "
            "AND julianday(expires_at) > julianday('now')",
            (row["delivery_id"],),
        )
        if claimed.rowcount != 1:
            continue
        assignment = json.loads(row["payload_json"])

        try:
            text, entities = telegram_content(assignment)
            await message.answer(
                text,
                entities=entities,
                reply_markup=telegram_keyboard(assignment),
                parse_mode=None,
            )
        except Exception:
            logging.exception(
                "Эту карточку нельзя отправлять повторно: проверьте запись sending в SQLite. %s",
                row["delivery_id"],
            )
            continue

        db.execute(
            "UPDATE evir_deliveries SET state = 'sent', updated_at = CURRENT_TIMESTAMP "
            "WHERE delivery_id = ? AND state = 'sending'",
            (row["delivery_id"],),
        )
        sent_count += 1

        if row["requires_served_ack"] == 1:
            try:
                await confirm_served(row)
            except (EvirError, httpx.RequestError):
                logging.exception(
                    "EVIR delivery acknowledgment failed; it will be retried without resending: %s",
                    row["delivery_id"],
                )
        else:
            db.execute(
                "UPDATE evir_deliveries SET state = 'acked', updated_at = CURRENT_TIMESTAMP "
                "WHERE delivery_id = ?",
                (row["delivery_id"],),
            )
    return sent_count

@evir_router.callback_query(F.data.startswith(CALLBACK_PREFIX))
async def check_handler(callback: CallbackQuery):
    batch_id = (callback.data or "").removeprefix(CALLBACK_PREFIX)
    recipient_id = str(callback.from_user.id)
    batch = db.execute("SELECT * FROM evir_check_batches WHERE batch_id = ? AND recipient_id = ?",
                       (batch_id, recipient_id)).fetchone() if DELIVERY_ID.fullmatch(batch_id) else None
    if batch is None:
        await callback.answer("Кнопка устарела.", show_alert=True)
        return
    await callback.answer()
    try:
        result = await evir_post("/deliveries/check", {
            "recipientId": recipient_id,
            "deliveryIds": json.loads(batch["delivery_ids_json"]),
        })
    except (EvirError, httpx.RequestError):
        if callback.message:
            await callback.message.answer("Не удалось проверить. Попробуйте ещё раз.")
        return
    if result["allCompleted"]:
        await on_evir_completed(callback)
    elif callback.message:
        expired = any(item["status"] == "expired" for item in result["results"])
        await callback.message.answer(
            "Срок предложений истёк. Запросите новые."
            if expired else "Осталось заданий: " + str(result["remaining"])
        )

async def on_evir_completed(callback: CallbackQuery):
    if callback.message:
        await callback.message.answer("Все задания выполнены.")

# В основном файле: from evir_integration import evir_router, send_evir_cards
# Один раз при настройке: dispatcher.include_router(evir_router)
# В нужном существующем обработчике: await send_evir_cards(message)