DonutSMP Sniper Bot

"""bot.py - the Discord sniper bot.

Polls LZT Market on a fixed interval, enriches each candidate with DonutSMP
stats, then either auto-buys (strict gates) or fires an alert (looser gates).
Built on discord.py with a tasks.loop poller.
"""
from __future__ import annotations

import logging

import discord
from discord.ext import tasks

from config import Config, Thresholds, load_config
from lzt_donut_api import DonutStats, Listing, MarketClient

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("sniper")


def passes_auto_buy(listing: Listing, stats: DonutStats, t: Thresholds) -> bool:
    """Strict auto-buy criteria - every gate must pass."""
    if listing.price > t.max_auto_buy_price:
        return False
    if listing.discount_percent < t.min_discount_percent:
        return False
    if t.require_clean_warranty and listing.warranty.lower() not in {"eft", "12_hours"}:
        return False
    if not stats.found:
        return False
    if stats.balance < t.min_donut_balance:
        return False
    if stats.playtime_hours < t.min_donut_playtime_hours:
        return False
    return True


def passes_alert(listing: Listing, t: Thresholds) -> bool:
    """Looser alert criteria - surface for manual review, no purchase."""
    return (
        listing.price <= t.alert_max_price
        and listing.discount_percent >= t.alert_discount_percent
    )


class SniperBot(discord.Client):
    def __init__(self, config: Config) -> None:
        intents = discord.Intents.default()
        super().__init__(intents=intents)
        self._config = config
        self._seen: set[int] = set()

    async def setup_hook(self) -> None:
        self.poll.start()

    async def on_ready(self) -> None:
        log.info("logged in as %s (dry_run=%s)", self.user, self._config.dry_run)

    @tasks.loop(seconds=12.0)
    async def poll(self) -> None:
        t = self._config.thresholds
        try:
            async with MarketClient(self._config) as market:
                listings = await market.search_category()
                for listing in listings:
                    if listing.item_id in self._seen:
                        continue
                    self._seen.add(listing.item_id)

                    stats = DonutStats(listing.minecraft_username or "", 0, 0.0, 0, False)
                    if listing.minecraft_username:
                        stats = await market.donut_stats(listing.minecraft_username)

                    if passes_auto_buy(listing, stats, t):
                        bought = await market.fast_buy(listing)
                        await self._log_buy(listing, stats, bought)
                    elif passes_alert(listing, t):
                        await self._send_alert(listing, stats)
        except Exception:  # noqa: BLE001 - keep the loop alive on transient errors
            log.exception("poll iteration failed")

    @poll.before_loop
    async def _before_poll(self) -> None:
        await self.wait_until_ready()
        self.poll.change_interval(seconds=self._config.poll_interval_seconds)

    async def _send_alert(self, listing: Listing, stats: DonutStats) -> None:
        channel = self.get_channel(self._config.alert_channel_id)
        if channel is None:
            return
        embed = discord.Embed(
            title=listing.title or f"Listing #{listing.item_id}",
            description=f"{listing.discount_percent:.0f}% below median",
            color=0x3B6EA5,
        )
        embed.add_field(name="Price", value=f"{listing.price:,}")
        embed.add_field(name="Median", value=f"{listing.market_median:,}")
        embed.add_field(name="Warranty", value=listing.warranty or "unknown")
        if stats.found:
            embed.add_field(name="DonutSMP balance", value=f"{stats.balance:,}")
            embed.add_field(name="Shards", value=f"{stats.shards:,}")
            embed.add_field(name="Playtime", value=f"{stats.playtime_hours:.0f}h")
        await channel.send(embed=embed)

    async def _log_buy(
        self, listing: Listing, stats: DonutStats, bought: bool
    ) -> None:
        channel = self.get_channel(self._config.buy_log_channel_id)
        if channel is None:
            return
        status = "BOUGHT" if bought else "MISSED"
        embed = discord.Embed(
            title=f"{status}: {listing.title or listing.item_id}",
            color=0x2E8B57 if bought else 0xB04A4A,
        )
        embed.add_field(name="Price", value=f"{listing.price:,}")
        embed.add_field(name="Discount", value=f"{listing.discount_percent:.0f}%")
        if stats.found:
            embed.add_field(name="DonutSMP balance", value=f"{stats.balance:,}")
        await channel.send(embed=embed)


def main() -> None:
    config = load_config()
    bot = SniperBot(config)
    bot.run(config.discord_token)


if __name__ == "__main__":
    main()

"""lzt_donut_api.py - API integration layer.

Wraps two upstreams behind a single async client:

  * LZT Market (Zelenka) - category search + fast buy.
  * DonutSMP - player stats lookup (Bearer auth).

The client owns one aiohttp.ClientSession and one token-bucket limiter per
endpoint, so callers never have to think about rate limits or retries.
"""
from __future__ import annotations

import asyncio
import time
from dataclasses import dataclass

import aiohttp

from config import Config, RateLimit

LZT_BASE = "https://prod-api.lzt.market"
DONUT_BASE = "https://api.donutsmp.net/v1"


@dataclass
class Listing:
    item_id: int
    title: str
    price: int
    market_median: int
    warranty: str
    seller: str
    minecraft_username: str | None

    @property
    def discount_percent(self) -> float:
        if self.market_median <= 0:
            return 0.0
        return (1 - self.price / self.market_median) * 100.0


@dataclass
class DonutStats:
    username: str
    balance: int
    playtime_hours: float
    shards: int
    found: bool


class _TokenBucket:
    """Async token bucket that also enforces a minimum inter-call delay."""

    def __init__(self, policy: RateLimit) -> None:
        self._policy = policy
        self._allowance = float(policy.requests)
        self._last_check = time.monotonic()
        self._last_call = 0.0
        self._lock = asyncio.Lock()

    async def acquire(self) -> None:
        async with self._lock:
            now = time.monotonic()
            # Refill based on elapsed time.
            elapsed = now - self._last_check
            self._last_check = now
            rate = self._policy.requests / self._policy.window_seconds
            self._allowance = min(
                self._policy.requests, self._allowance + elapsed * rate
            )

            # Enforce the hard floor between calls.
            since_last = now - self._last_call
            if since_last < self._policy.min_delay_seconds:
                await asyncio.sleep(self._policy.min_delay_seconds - since_last)

            # Wait for a token if the bucket is empty.
            if self._allowance < 1.0:
                deficit = (1.0 - self._allowance) / rate
                await asyncio.sleep(deficit)
                self._allowance = 0.0
            else:
                self._allowance -= 1.0

            self._last_call = time.monotonic()


class MarketClient:
    """Async LZT Market + DonutSMP client. Use as an async context manager."""

    def __init__(self, config: Config) -> None:
        self._config = config
        self._session: aiohttp.ClientSession | None = None
        self._buckets = {
            name: _TokenBucket(policy)
            for name, policy in config.rate_limits.items()
        }

    async def __aenter__(self) -> "MarketClient":
        self._session = aiohttp.ClientSession(
            timeout=aiohttp.ClientTimeout(total=20)
        )
        return self

    async def __aexit__(self, exc_type, exc, tb) -> None:
        if self._session is not None:
            await self._session.close()
            self._session = None

    @property
    def session(self) -> aiohttp.ClientSession:
        if self._session is None:
            raise RuntimeError("MarketClient used outside its async context")
        return self._session

    # ------------------------------------------------------------------ LZT --
    async def search_category(self, page: int = 1) -> list[Listing]:
        """Search the configured LZT category.

        Respects the 20 req/min limit and the mandatory 3s delay between calls
        via the 'category_search' token bucket.
        """
        await self._buckets["category_search"].acquire()

        url = f"{LZT_BASE}/{self._config.lzt_category}"
        headers = {
            "Authorization": f"Bearer {self._config.lzt_token}",
            "Accept": "application/json",
        }
        params = {"page": page, "order_by": "price_to_up"}

        async with self.session.get(url, headers=headers, params=params) as resp:
            if resp.status == 429:
                retry_after = float(resp.headers.get("Retry-After", "3"))
                await asyncio.sleep(retry_after)
                return await self.search_category(page=page)
            resp.raise_for_status()
            payload = await resp.json()

        return [self._parse_listing(item) for item in payload.get("items", [])]

    async def fast_buy(self, listing: Listing) -> bool:
        """Attempt a fast buy with 'retry_request' semantics.

        LZT briefly reserves a listing during checkout and returns a retryable
        status. We retry up to fast_buy_max_retries times before giving up.
        """
        if self._config.dry_run:
            return True

        url = f"{LZT_BASE}/{listing.item_id}/fast-buy"
        headers = {"Authorization": f"Bearer {self._config.lzt_token}"}

        for attempt in range(self._config.fast_buy_max_retries):
            await self._buckets["fast_buy"].acquire()
            async with self.session.post(
                url, headers=headers, json={"price": listing.price}
            ) as resp:
                if resp.status in (200, 201):
                    return True
                if resp.status == 429:
                    retry_after = float(resp.headers.get("Retry-After", "1"))
                    await asyncio.sleep(retry_after)
                    continue

                body = await resp.json(content_type=None)
                if self._is_retryable(body):
                    await asyncio.sleep(self._config.fast_buy_retry_delay_seconds)
                    continue
                # Non-retryable failure (sold, insufficient funds, etc.).
                return False

        return False

    @staticmethod
    def _is_retryable(body: object) -> bool:
        if not isinstance(body, dict):
            return False
        errors = body.get("errors")
        if isinstance(errors, list):
            return any("retry_request" in str(err) for err in errors)
        return "retry_request" in str(body.get("error", ""))

    def _parse_listing(self, item: dict) -> Listing:
        return Listing(
            item_id=int(item["item_id"]),
            title=str(item.get("title", "")),
            price=int(item.get("price", 0)),
            market_median=int(item.get("median_price", item.get("price", 0))),
            warranty=str(item.get("guarantee", {}).get("type", "")),
            seller=str(item.get("seller", {}).get("username", "")),
            minecraft_username=item.get("minecraft_username"),
        )

    # -------------------------------------------------------------- DonutSMP --
    async def donut_stats(self, username: str) -> DonutStats:
        """Look up a DonutSMP player's stats using Bearer-token auth."""
        await self._buckets["donut_stats"].acquire()

        url = f"{DONUT_BASE}/stats/{username}"
        headers = {
            "Authorization": f"Bearer {self._config.donut_token}",
            "Accept": "application/json",
        }

        async with self.session.get(url, headers=headers) as resp:
            if resp.status == 404:
                return DonutStats(username, 0, 0.0, 0, found=False)
            resp.raise_for_status()
            data = await resp.json()

        result = data.get("result", data)
        return DonutStats(
            username=username,
            balance=int(result.get("money", 0)),
            playtime_hours=float(result.get("playtime", 0)) / 3600.0,
            shards=int(result.get("shards", 0)),
            found=True,
        )

"""config.py - configuration and threshold settings for the sniper.

Loads secrets from the environment, defines auto-buy / alert thresholds, and the
per-endpoint rate-limit and retry policy. Tune the THRESHOLDS to your strategy;
nothing here talks to the network.
"""
from __future__ import annotations

import os
from dataclasses import dataclass, field


def _env(name: str, default: str = "") -> str:
    value = os.environ.get(name, default)
    if not value and default == "":
        raise RuntimeError(f"missing required environment variable: {name}")
    return value


@dataclass(frozen=True)
class RateLimit:
    """Token-bucket policy for a single upstream endpoint."""

    requests: int           # allowed requests per window
    window_seconds: float   # length of the window
    min_delay_seconds: float  # hard floor between two calls


@dataclass(frozen=True)
class Thresholds:
    """Strict auto-buy and alert criteria.

    A listing is *bought* automatically only when EVERY auto-buy gate passes.
    A listing is *alerted* (pinged in Discord, no purchase) when it clears the
    looser alert gates but misses at least one auto-buy gate.
    """

    # Auto-buy gates (all must pass)
    max_auto_buy_price: int = 12_000      # never auto-buy above this (account currency)
    min_discount_percent: float = 35.0    # price must be >= this % below market median
    min_balance_ratio: float = 8.0        # account value / price must clear this
    require_clean_warranty: bool = True   # only EFT / no-chargeback warranties

    # Alert-only gates (looser - surfaced for manual review)
    alert_discount_percent: float = 18.0
    alert_max_price: int = 45_000

    # DonutSMP enrichment gates
    min_donut_balance: int = 5_000_000    # in-game $ on the linked DonutSMP profile
    min_donut_playtime_hours: float = 40.0


@dataclass(frozen=True)
class Config:
    # --- Discord ---
    discord_token: str = field(default_factory=lambda: _env("DISCORD_TOKEN"))
    alert_channel_id: int = field(
        default_factory=lambda: int(_env("ALERT_CHANNEL_ID", "0"))
    )
    buy_log_channel_id: int = field(
        default_factory=lambda: int(_env("BUY_LOG_CHANNEL_ID", "0"))
    )

    # --- LZT Market (Zelenka) ---
    lzt_token: str = field(default_factory=lambda: _env("LZT_TOKEN"))
    lzt_category: str = field(default_factory=lambda: _env("LZT_CATEGORY", "minecraft"))

    # --- DonutSMP ---
    donut_token: str = field(default_factory=lambda: _env("DONUT_TOKEN"))

    # --- Behaviour ---
    poll_interval_seconds: float = 12.0
    dry_run: bool = field(
        default_factory=lambda: _env("DRY_RUN", "false").lower() == "true"
    )

    thresholds: Thresholds = field(default_factory=Thresholds)

    # Per-endpoint policy. LZT category search is the tightest: 20 req/min with a
    # 3s floor between calls. Fast-buy is allowed to retry aggressively.
    rate_limits: dict[str, RateLimit] = field(
        default_factory=lambda: {
            "category_search": RateLimit(
                requests=20, window_seconds=60.0, min_delay_seconds=3.0
            ),
            "fast_buy": RateLimit(
                requests=120, window_seconds=60.0, min_delay_seconds=0.0
            ),
            "donut_stats": RateLimit(
                requests=60, window_seconds=60.0, min_delay_seconds=0.5
            ),
        }
    )

    # Fast-buy keeps retrying while the listing is still being reserved.
    fast_buy_max_retries: int = 100
    fast_buy_retry_delay_seconds: float = 0.4


def load_config() -> Config:
    """Build the immutable Config from the current environment."""
    return Config()