"""PixorAds Offerwall 1.3 · aiogram 3 + httpx, Python 3.11+.

pip install 'aiogram>=3,<4' httpx
Set BOT_TOKEN, PIXORADS_KEY, optionally PIXORADS_DB (persistent private path).
Run: python aiogram_offerwall.py. Commands: /bonus, /bonusbalance.
Choose partner rewards and leave webhook empty: this example polls the API.
The SQLite credits table IS this example's bonus ledger. For your own currency,
replace credit_once with a transaction in YOUR database (unique conversion_id).
Never acknowledge a reward before your actual balance transaction commits.
Run one polling process per bot; protect the database file and environment.
"""

from __future__ import annotations

import asyncio
import logging
import os
import sqlite3
import time
from contextlib import suppress
from decimal import Decimal
from typing import Any, NotRequired, Protocol, TypedDict, cast
from uuid import UUID

import httpx
from aiogram import Bot, Dispatcher, F, Router
from aiogram.filters import Command
from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message

BASE = "https://pixora-ads.com/api/v1/offerwall"
log = logging.getLogger("pixorads")
router = Router()
db = sqlite3.connect(os.environ.get("PIXORADS_DB", "offerwall.sqlite3"))
db.execute("PRAGMA journal_mode=WAL")
db.execute(
    "CREATE TABLE IF NOT EXISTS attempts(id TEXT PRIMARY KEY, user_id INTEGER NOT NULL, done INTEGER NOT NULL DEFAULT 0, polled INTEGER NOT NULL DEFAULT 0)"
)
db.execute(
    "CREATE TABLE IF NOT EXISTS credits(id TEXT PRIMARY KEY, user_id INTEGER NOT NULL, units INTEGER NOT NULL, name TEXT NOT NULL)"
)
db.execute("CREATE TABLE IF NOT EXISTS outcomes(id TEXT PRIMARY KEY, status TEXT NOT NULL)")


class ButtonSpec(TypedDict):
    text: str
    url: NotRequired[str]
    callback_data: NotRequired[str]


class PollingDispatcher(Protocol):
    async def start_polling(self, bot: Bot, /) -> None: ...


async def request(
    path: str,
    body: dict[str, Any] | None = None,
    token: str | None = None,
    key: str | None = None,
) -> dict[str, Any]:
    headers = (
        {"Authorization": f"Bearer {token}"}
        if token
        else {"X-PixorAds-Key": os.environ["PIXORADS_KEY"]}
    )
    if key:
        headers["Idempotency-Key"] = key
    async with httpx.AsyncClient(timeout=20, follow_redirects=False) as client:
        response = await client.request(
            "GET" if body is None else "POST", BASE + path, json=body, headers=headers
        )
        response.raise_for_status()
        return response.json()


def keyboard(rows: list[list[ButtonSpec]]) -> InlineKeyboardMarkup:
    return InlineKeyboardMarkup(
        inline_keyboard=[[InlineKeyboardButton(**b) for b in row] for row in rows]
    )


def credit_once(result: dict[str, Any]) -> None:
    # This insert is the example's actual credit, not merely an event dedup marker.
    with db:
        db.execute(
            "INSERT OR IGNORE INTO credits VALUES(?,?,?,?)",
            (
                result["conversion_id"],
                result["telegram_id"],
                result["reward_units"],
                result["reward_name"],
            ),
        )


def finish_attempt(identity: str, status: str) -> None:
    with db:
        db.execute("INSERT OR REPLACE INTO outcomes VALUES(?,?)", (identity, status))
        db.execute("UPDATE attempts SET done=1 WHERE id=?", (identity,))


async def settle(identity: str) -> str:
    with db:
        db.execute("UPDATE attempts SET polled=? WHERE id=?", (time.time(), identity))
    result = await request(f"/publisher-api/rewards/{identity}")
    local = db.execute("SELECT user_id FROM attempts WHERE id=?", (identity,)).fetchone()
    if not local or local[0] != result["telegram_id"] or result["test"] is not False:
        raise ValueError("Reward identity mismatch")
    if result["reward_eligible"] and result["reward_mode"] == "partner":
        credit_once(result)
        await request(f"/publisher-api/rewards/{identity}/ack", {"reference": f"sqlite:{identity}"})
        if result["status"] != "rewarded":
            # Advance is the publisher's accepted risk, not a final qualification.
            # Keep polling; credit_once and the fixed ACK reference prevent repeats.
            return "Бонус выдан авансом. Итоговая проверка ещё продолжается."
        finish_attempt(identity, result["status"])
        return "Бонус начислен. Баланс: /bonusbalance"
    if (
        result["status"] in {"revoked", "cancelled", "expired"}
        or result["reward_mode"] == "pixorads"
        and result["status"] == "rewarded"
    ):
        finish_attempt(identity, result["status"])
        if (
            result["status"] != "rewarded"
            and db.execute("SELECT 1 FROM credits WHERE id=?", (identity,)).fetchone()
        ):
            # No automatic debit: advance was explicitly funded at publisher risk.
            log.warning(
                "Advance bonus %s ended as %s; review your bonus ledger, no automatic debit",
                identity,
                result["status"],
            )
    return "Статус: " + {
        "pending_retention": "ожидается повторная проверка",
        "rewarded": "подтверждено",
        "assigned": "выполните условия и нажмите Проверить",
    }.get(result["status"], result["status"])


TASK_TERMS = {
    "telegram_subscription": "Подпишитесь на канал. Прежняя подписка не засчитывается.",
    "telegram_group_join": "Вступите в группу. Прежнее участие не засчитывается.",
    "telegram_post_reaction": "Добавьте реакцию от своего аккаунта в публичной группе.",
    "telegram_poll_vote": "Проголосуйте в неанонимном опросе PixorAds.",
    "telegram_boost": "Добавьте новый Premium-буст после начала задания. Нужен доступный буст.",
    "web_visit": "Перейдите на сайт защищённой ссылкой и вернитесь для проверки.",
    "web_surf": "Посмотрите сайт с активным таймером и пройдите проверку внимания.",
    "extended_task": "Прочитайте инструкцию и отправьте отчёт в готовом экране.",
}


async def catalog(message: Message, user_id: int) -> None:
    session = await request(
        "/publisher-api/session", {"telegram_id": user_id, "placement": "bot_bonus"}
    )
    if not session["verified"]:
        await message.answer(
            "Подтвердите аккаунт в PixorAds, затем вернитесь и нажмите /bonus.",
            reply_markup=keyboard(
                [[{"text": "Подтвердить аккаунт", "url": session["verification_url"]}]]
            ),
        )
        return
    token = session["viewer_token"]
    data = await request("/viewer/catalog", token=token)
    for attempt in data["history"]:
        with db:
            db.execute(
                "INSERT OR IGNORE INTO attempts(id,user_id) VALUES(?,?)", (attempt["id"], user_id)
            )
    active = next(
        (
            a
            for a in data["history"]
            if a["status"] in {"assigned", "submitted", "verifying", "review"}
        ),
        None,
    )
    if active:
        await send_attempt(message, active, token)
        return
    if not data["items"]:
        await message.answer("Подходящих заданий пока нет. Все обычные функции бота доступны.")
        return
    task = data["items"][0]
    reward = (
        f"{task['reward_units']} {task['reward_name']}"
        if task["reward_mode"] == "partner"
        else f"{Decimal(task['reward']) / (1000000 if task['currency'] == 'USD' else 1000)} {task['currency']} на баланс PixorAds"
    )
    seconds = task["retention_seconds"]
    action = task.get("action")
    terms = (
        action["instructions"]
        if action
        else TASK_TERMS.get(task["task_type"], "Выполните действие по инструкции.")
    )
    period = "Период проверки"
    audience = "Только для новых пользователей." if action and action["new_users_only"] else ""
    await message.answer(
        f"{task['title']}\n{reward}\n{terms}\n{audience}\n{period}: {seconds // 3600} ч {seconds % 3600 // 60} мин.\nБонус — "
        + (
            "после первой проверки."
            if task.get("reward_timing") == "advance"
            else "после окончательной проверки."
        )
        + "\nУчастие добровольное.",
        reply_markup=keyboard(
            [[{"text": "Начать задание", "callback_data": "ow:s:" + task["id"]}]]
        ),
    )
    # Telegram confirms delivery, not visibility. Do not call /viewer/visible here.


async def send_attempt(message: Message, task: dict[str, Any], token: str) -> None:
    if task["task_type"] == "extended_task":
        await message.answer(
            "Инструкция, отчёт и история проверки — в готовом экране. Бонус начисляется после принятия работы.",
            reply_markup=keyboard(
                [[{"text": "Открыть задание и отчёт", "url": task["hosted_url"]}]]
            ),
        )
        return
    if task["status"] not in {"assigned", "submitted"}:
        await message.answer(
            "Статус и сроки проверки — в готовом экране.",
            reply_markup=keyboard([[{"text": "Мой прогресс", "url": task["hosted_url"]}]]),
        )
        return
    task = {**task, **await request(f"/viewer/attempts/{task['id']}/open", {}, token)}
    action = task.get("action")
    terms = (
        action["instructions"]
        if action
        else TASK_TERMS.get(task["task_type"], "Выполните действие по инструкции.")
    )
    period = "Период проверки"
    await message.answer(
        f"{task['title']}\nСрок: {task['deadline']}\n{terms}\n{period}: {task['retention_seconds'] // 60} мин.\nПосле выполнения нажмите «Проверить».",
        reply_markup=keyboard(
            [
                [{"text": "Открыть задание", "url": task["url"]}],
                [{"text": "Все условия и прогресс", "url": task["hosted_url"]}],
                [
                    {"text": "Проверить", "callback_data": "ow:c:" + task["id"]},
                    {"text": "Отменить", "callback_data": "ow:x:" + task["id"]},
                ],
            ]
        ),
    )


@router.message(Command("bonus"), F.chat.type == "private")
async def bonus(message: Message) -> None:
    if message.from_user is None:
        return
    try:
        await catalog(message, message.from_user.id)
    except (httpx.HTTPError, ValueError):
        log.warning("Offerwall temporarily unavailable")
        await message.answer(
            "Задания временно недоступны. Попробуйте позже; обычные функции работают."
        )


@router.message(Command("bonusbalance"), F.chat.type == "private")
async def balance(message: Message) -> None:
    if message.from_user is None:
        return
    rows = db.execute(
        "SELECT name,sum(units) FROM credits WHERE user_id=? GROUP BY name", (message.from_user.id,)
    ).fetchall()
    await message.answer(
        "Ваши бонусы:\n" + ("\n".join(f"{n} {name}" for name, n in rows) or "Пока 0")
    )


@router.callback_query(F.data.startswith("ow:"), F.message.chat.type == "private")
async def click(callback: CallbackQuery) -> None:
    await callback.answer()
    message, data = callback.message, callback.data
    if not isinstance(message, Message) or data is None:
        return
    try:
        _, action, raw = data.split(":", 2)
        identity, user_id = str(UUID(raw)), callback.from_user.id
        session = await request(
            "/publisher-api/session", {"telegram_id": user_id, "placement": "bot_bonus"}
        )
        if not session["verified"]:
            await catalog(message, user_id)
            return
        token = session["viewer_token"]
        if action == "s":
            # Stable across network retries and process restarts.
            task = await request(
                "/viewer/start", {"campaign_id": identity}, token, f"bot:{user_id}:{identity}"
            )
            with db:
                db.execute(
                    "INSERT OR IGNORE INTO attempts(id,user_id) VALUES(?,?)", (task["id"], user_id)
                )
            await send_attempt(message, task, token)
        elif action in {"c", "x"}:
            own = db.execute("SELECT user_id FROM attempts WHERE id=?", (identity,)).fetchone()
            if not own or own[0] != user_id:
                return
            await request(
                f"/viewer/attempts/{identity}/" + ("check" if action == "c" else "cancel"),
                {},
                token,
            )
            await message.answer(await settle(identity))
    except (httpx.HTTPError, ValueError):
        await message.answer("Проверка пока не завершена. Повторите позже или откройте /bonus.")


async def reconcile() -> None:
    while True:
        try:
            for event in (await request("/publisher-api/tests"))["events"]:
                if event.get("test") is True:
                    # Sandbox only: deliberately no credit_once call.
                    await request(f"/publisher-api/tests/{event['conversion_id']}/ack", {})
            for (identity,) in db.execute(
                "SELECT id FROM attempts WHERE done=0 ORDER BY polled LIMIT 20"
            ).fetchall():
                try:
                    await settle(identity)
                except (httpx.HTTPError, ValueError):
                    log.warning("Reward reconciliation will retry")
                await asyncio.sleep(1)
        except (httpx.HTTPError, ValueError):
            log.warning("Offerwall reconciliation unavailable; retrying")
        await asyncio.sleep(60)


async def main() -> None:
    bot = Bot(os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)
    worker = asyncio.create_task(reconcile())
    try:
        # Aiogram's sentinel default is untyped; this example uses only one Bot.
        polling = cast(PollingDispatcher, dp)
        await polling.start_polling(bot)
    finally:
        worker.cancel()
        with suppress(asyncio.CancelledError):
            await worker
        db.close()


if __name__ == "__main__":
    asyncio.run(main())