API EVIR

# Python 3.11+ · сохраните файл как evir_integration.py
# pip install "aiogram>=3,<4" "httpx>=0.27,<1"
import asyncio
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_qualify:"
DELIVERY_ID = re.compile(r"^[A-Za-z0-9_-]{32}quot;)

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")

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=assignment["ctaLabel"],
        url=assignment["actionUrl"],  # используйте URL без изменений
    )]]
    if assignment.get("requiresQualification"):
        rows.append([InlineKeyboardButton(
            text="Проверить",
            callback_data=f'{CALLBACK_PREFIX}{assignment["deliveryId"]}',
        )])
    return InlineKeyboardMarkup(inline_keyboard=rows)

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,
    })
    for assignment in data["assignments"]:
        remember_delivery(recipient_id, assignment)

    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:
            await message.answer(
                f'{assignment["title"]}\n\n{assignment["description"]}',
                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 qualify_handler(callback: CallbackQuery):
    delivery_id = (callback.data or "").removeprefix(CALLBACK_PREFIX)
    if not DELIVERY_ID.fullmatch(delivery_id):
        await callback.answer("Кнопка устарела.", show_alert=True)
        return
    try:
        await evir_post(
            f"/deliveries/{delivery_id}/qualify",
            {"recipientId": str(callback.from_user.id)},
        )
    except (EvirError, httpx.RequestError) as error:
        messages = {
            "OP_VISIT_REQUIRED": "Сначала откройте предложение, затем нажмите «Проверить».",
            "OP_NOT_CONFIRMED": "Подписка пока не подтверждена.",
            "OP_ALREADY_MEMBER": "Вы уже были подписаны до показа.",
        }
        await callback.answer(
            messages.get(getattr(error, "code", None), "Не удалось проверить подписку."),
            show_alert=True,
        )
        return
    await callback.answer("Подписка подтверждена.")

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