From b1ff45794a0397a32b9dc43419445ae9b6395680 Mon Sep 17 00:00:00 2001 From: esefxdz Date: Wed, 23 Sep 2026 14:47:52 +0300 Subject: [PATCH] gacha patch 2 --- blue_archive/constants.py | 2 + blue_archive/data.py | 19 ++- blue_archive/gacha.py | 52 +++++++- blue_archive/health.py | 253 ++++++++++++++++++++++++++++++++++++++ blue_archive/inventory.py | 7 ++ 5 files changed, 331 insertions(+), 2 deletions(-) create mode 100644 blue_archive/health.py diff --git a/blue_archive/constants.py b/blue_archive/constants.py index f75e16a..183ec0b 100644 --- a/blue_archive/constants.py +++ b/blue_archive/constants.py @@ -89,4 +89,6 @@ FILTER_LABELS: dict[str, int | None] = {"All": None, "3★": 3, "2★": 2, "1★ FONT_PATHS = [ "/usr/share/fonts/TTF/DejaVuSans.ttf", "/usr/share/fonts/TTF/DejaVuSans-Bold.ttf", + "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", # Debian (fonts-dejavu-core) + "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", ] diff --git a/blue_archive/data.py b/blue_archive/data.py index 7dedb7f..826d9b9 100644 --- a/blue_archive/data.py +++ b/blue_archive/data.py @@ -412,6 +412,22 @@ async def fetch_banners(session: Optional[aiohttp.ClientSession] = None) -> Opti await session.close() +def banner_unavailable_reason(banner: dict) -> Optional[str]: + """Return why a banner can't be pulled right now, or None if it's live. + + Banners without timestamps are treated as permanent. + """ + now_ms = datetime.now(timezone.utc).timestamp() * 1000 + started = banner.get("startedAt", 0) + ended = banner.get("endedAt", 0) + if started and now_ms < started: + start_dt = datetime.fromtimestamp(started / 1000, tz=timezone.utc) + return f"This banner hasn't started yet (starts {start_dt.strftime('%b %d')})." + if ended and now_ms >= ended: + return "This banner has ended." + return None + + def format_banner_embed(banner: dict, index: int) -> str: """Format a single banner as an embed field value string.""" gtype = banner.get("gachaType", "PickupGacha") @@ -427,8 +443,9 @@ def format_banner_embed(banner: dict, index: int) -> str: dates = "Permanent" is_fes = " [FES - 6% 3*!]" if gtype == "FesGacha" else "" + upcoming = " [UPCOMING]" if started and started > datetime.now(timezone.utc).timestamp() * 1000 else "" return ( - f"**#{index + 1} — {gtype}{is_fes}**\n" + f"**#{index + 1} — {gtype}{is_fes}{upcoming}**\n" f"Rate-up: {rateups}\n" f"{dates}" ) diff --git a/blue_archive/gacha.py b/blue_archive/gacha.py index 0b45c5f..9ef7db1 100644 --- a/blue_archive/gacha.py +++ b/blue_archive/gacha.py @@ -4,6 +4,7 @@ import asyncio import json import logging +import time from pathlib import Path from typing import Optional @@ -15,10 +16,12 @@ from .data import ( db, fetch_banners, format_banner_embed, + banner_unavailable_reason, roll_rarity, get_rates_for_banner, ) from .gacha_renderer import render_pull +from .health import collect_health, OK, INFO from .constants import SPARK_TARGET, BANNER_FILE, DEFAULT_RATES, PULL10_RATES, GACHA_ANIM_PATH, GACHA_ANIM_DURATION log = logging.getLogger(__name__) @@ -69,6 +72,10 @@ class BlueArchiveGacha(commands.Cog): self.bot = bot self._banner_cache: dict = {"current": [], "upcoming": [], "ended": []} self._cache_ready = asyncio.Event() + # Refresh bookkeeping, reported by !gacha health + self._last_refresh_at: Optional[float] = None + self._last_refresh_ok: Optional[bool] = None + self._last_good_at: Optional[float] = None # ── Lifecycle ─────────────────────────────────────────────────────── @@ -87,14 +94,17 @@ class BlueArchiveGacha(commands.Cog): """Refresh banners, then merge any new students they might feature.""" try: banners = await fetch_banners() + self._last_refresh_ok = banners is not None if banners is None: log.warning("Banner refresh failed; keeping previous banner data") else: self._banner_cache = banners + self._last_good_at = time.time() await db.merge_ennead(self._banner_cache) except Exception: log.exception("Banner refresh failed") finally: + self._last_refresh_at = time.time() self._cache_ready.set() async def _periodic_refresh(self) -> None: @@ -167,7 +177,8 @@ class BlueArchiveGacha(commands.Cog): "`!gacha` — list current banners\n" "`!gacha pick ` — select banner #n\n" "`!gacha pick regular` — permanent pool\n" - "`!gacha info` — your status & spark" + "`!gacha info` — your status & spark\n" + "`!gacha health` — check everything is working" ), inline=False, ) @@ -229,6 +240,10 @@ class BlueArchiveGacha(commands.Cog): return banner = banners[idx] + reason = banner_unavailable_reason(banner) + if reason: + await ctx.reply(reason) + return rateups = ", ".join(banner.get("rateups", [])) or "Standard Pool" gtype = banner.get("gachaType", "PickupGacha") @@ -280,6 +295,13 @@ class BlueArchiveGacha(commands.Cog): "Pick another with `!gacha pick ` (see `!gacha`) or `!gacha pick regular`." ) return + reason = banner_unavailable_reason(banner) + if reason: + await ctx.reply( + f"{reason}\n" + "Pick another with `!gacha pick ` (see `!gacha`) or `!gacha pick regular`." + ) + return rateups = ", ".join(banner.get("rateups", [])) or "Standard Pool" gtype = banner.get("gachaType", "PickupGacha") banner_name = f"{gtype} — {rateups}" @@ -420,6 +442,10 @@ class BlueArchiveGacha(commands.Cog): "Your selected banner has ended or is unavailable right now, so it can't be sparked." ) return + reason = banner_unavailable_reason(banner) + if reason: + await ctx.reply(reason) + return rateups = [n.lower() for n in banner.get("rateups", [])] if student["Name"].lower() not in db.resolve_rateups(rateups): rateup_str = ", ".join(banner.get("rateups", [])) @@ -445,6 +471,30 @@ class BlueArchiveGacha(commands.Cog): f"Remaining Recruitment Points: {remaining}" ) + @commands.cooldown(1, 30, commands.BucketType.user) + @gacha.command(name="health") + async def gacha_health(self, ctx: commands.Context) -> None: + """Run live checks on everything a pull depends on.""" + async with ctx.typing(): + rows = await collect_health(self, ctx.author.id, await _get_banner_id(ctx.author.id)) + + bad = [r for r in rows if r[0] not in (OK, INFO)] + embed = discord.Embed( + title="Gacha Health — all checks passed" if not bad else f"Gacha Health — {len(bad)} issue(s)", + color=0x57F287 if not bad else 0xED4245, + timestamp=discord.utils.utcnow(), + ) + sections: dict[str, list[str]] = {} + for status, section, text in rows: + sections.setdefault(section, []).append(f"{status} {text}") + for section, lines in sections.items(): + value = "\n".join(lines) + if len(value) > 1024: + value = value[:1020] + " …" + embed.add_field(name=section, value=value, inline=False) + embed.set_footer(text="Every line was checked just now") + await ctx.reply(embed=embed) + @gacha.command(name="info") async def gacha_info(self, ctx: commands.Context) -> None: """Show your active banner, spark count, and collection summary.""" diff --git a/blue_archive/health.py b/blue_archive/health.py new file mode 100644 index 0000000..627e771 --- /dev/null +++ b/blue_archive/health.py @@ -0,0 +1,253 @@ +#this runs live health checks for the gacha system, used by !gacha health## +"""Gacha health checks — every row is a check actually run just now. + +Nothing here changes bot state: live data is fetched and compared, never +written into the banner cache or the roster. +""" +###################################################################### +import asyncio +import io +import time +from pathlib import Path +from typing import Optional + +import aiohttp +from PIL import Image + +from . import db as gacha_db +from .data import db, fetch_banners, banner_unavailable_reason +from .gacha_renderer import render_pull +from .constants import CHARACTER_API, BG_PATH, GACHA_ANIM_PATH, FONT_PATHS + +OK, WARN, FAIL, INFO = "✅", "⚠️", "❌", "ℹ️" +REFRESH_INTERVAL = 6 * 3600 # must match BlueArchiveGacha._periodic_refresh +MAX_PORTRAIT_PROBES = 6 +BANNER_TIMEOUT = 20 + +Row = tuple[str, str, str] # (status, section, text) + + +def _ago(ts: Optional[float]) -> str: + return f"" if ts else "never" + + +async def _probe_image(session: aiohttp.ClientSession, url: str) -> Optional[str]: + """Fetch and decode an image exactly like the renderer. None = success.""" + try: + async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp: + if resp.status != 200: + return f"HTTP {resp.status}" + data = await resp.read() + Image.open(io.BytesIO(data)).convert("RGBA") + return None + except asyncio.TimeoutError: + return "timed out after 5s" + except Exception as e: + return type(e).__name__ + + +async def _check_banners(cog) -> tuple[list[Row], list[dict]]: + """Cached banner state (what pulls use) + a live fetch compared against it.""" + rows: list[Row] = [] + s = "Banners" + cache = cog._banner_cache + cached = cache.get("current", []) + cache.get("upcoming", []) + + if cog._last_good_at is None: + rows.append((FAIL, s, "Banner data has never been fetched successfully — every pull falls back to no rate-up")) + elif cog._last_refresh_ok is False: + rows.append((WARN, s, f"Last refresh {_ago(cog._last_refresh_at)} FAILED; pulls use data from {_ago(cog._last_good_at)}")) + else: + rows.append((OK, s, f"Pull data refreshed {_ago(cog._last_good_at)} ({len(cached)} current/upcoming)")) + + try: + live = await asyncio.wait_for(fetch_banners(), timeout=BANNER_TIMEOUT) + except asyncio.TimeoutError: + live = None + rows.append((FAIL, s, f"Banner API check now: timed out after {BANNER_TIMEOUT}s")) + else: + if live is None: + rows.append((FAIL, s, "Banner API check now: request failed (see bot log)")) + if live is not None: + live_list = live.get("current", []) + live.get("upcoming", []) + if {str(b.get("id")) for b in live_list} == {str(b.get("id")) for b in cached}: + rows.append((OK, s, "Banner API check now: reachable, matches pull data")) + else: + rows.append((WARN, s, f"Banner API check now: reachable, but has {len(live_list)} banners vs {len(cached)} " + "in pull data — pulls use stale data until the next refresh")) + + now_ms = time.time() * 1000 + live_banners = [b for b in cached if banner_unavailable_reason(b) is None] + upcoming = [b for b in cached if b.get("startedAt", 0) > now_ms] + ended = len(cached) - len(live_banners) - len(upcoming) + rows.append((INFO, s, f"{len(live_banners)} live now, {len(upcoming)} upcoming" + + (f", {ended} ended but still listed (blocked from pulls)" if ended else ""))) + return rows, live_banners + + +def _check_rateups(live_banners: list[dict]) -> tuple[list[Row], list[dict]]: + """Run each live banner through the exact pool builder pulls use.""" + rows: list[Row] = [] + s = "Rate-ups (live banners)" + students: list[dict] = [] + if not live_banners: + rows.append((INFO, s, "No live banners to check")) + return rows, students + + for b in live_banners[:6]: + pool = db.build_pool(b) + pooled = {st["Name"].lower() for r in pool.values() for st in r} + label = f"#{b.get('id')} {b.get('gachaType', '?')}" + problems: list[tuple[str, str]] = [] # (status, text) + for name in b.get("rateups", []): + st = db.get_by_name(name, strict=True) + if not st: + problems.append((FAIL, f"{name}: not in roster, can't be pulled")) + elif st["Name"].lower() not in pooled: + problems.append((FAIL, f"{name}: marked welfare, excluded from pulls")) + else: + students.append(st) + if st["StarGrade"] < 3: + problems.append((WARN, f"{name}: {st['StarGrade']}★, rate-up boost only applies to 3★")) + if problems: + status = FAIL if any(p[0] == FAIL for p in problems) else WARN + rows.append((status, s, f"{label}: " + "; ".join(p[1] for p in problems))) + else: + names = ", ".join(b.get("rateups", [])) or "no rate-ups" + rows.append((OK, s, f"{label}: all pullable ({names})")) + if len(live_banners) > 6: + rows.append((INFO, s, f"{len(live_banners) - 6} more live banners not checked")) + return rows, students + + +async def _check_portraits(students: list[dict]) -> list[Row]: + """Probe wiki art and the sprite fallback separately for rate-up students.""" + s = "Portraits" + if not students: + return [(INFO, s, "No live rate-up students to probe")] + seen, sample = set(), [] + for st in students: + if st["Name"] not in seen: + seen.add(st["Name"]) + sample.append(st) + sample = sample[:MAX_PORTRAIT_PROBES] + + rows: list[Row] = [] + timeout = aiohttp.ClientTimeout(total=30, connect=5) # same as render_pull + async with aiohttp.ClientSession(timeout=timeout) as session: + results = await asyncio.gather(*( + asyncio.gather(_probe_image(session, db.cdn_portrait(st)), _probe_image(session, db.cdn_sprite(st))) + for st in sample + )) + for st, (wiki, sprite) in zip(sample, results): + if wiki is None: + rows.append((OK, s, f"{st['Name']}: wiki art OK")) + elif sprite is None: + rows.append((WARN, s, f"{st['Name']}: wiki art failed ({wiki}), sprite fallback OK")) + else: + rows.append((FAIL, s, f"{st['Name']}: no art (wiki {wiki}, sprite {sprite}) — card renders blank")) + return rows + + +async def _check_roster() -> list[Row]: + s = "Roster" + std = db.build_pool(None) # exactly what Regular Recruitment pulls from + welfare = sum(1 for st in db.students if (st.get("IsLimited") or 0) >= 2) + limited = sum(1 for st in db.students if (st.get("IsLimited") or 0) == 1) + rows: list[Row] = [(OK if db.students else FAIL, s, + f"{len(db.students)} students loaded — standard pool 3★ {len(std[3])}, " + f"2★ {len(std[2])}, 1★ {len(std[1])}; {limited} limited (Fes/own banner only); " + f"{welfare} welfare (never pulled)")] + try: + async with aiohttp.ClientSession() as session: + async with session.get(CHARACTER_API, timeout=aiohttp.ClientTimeout(total=15)) as resp: + if resp.status != 200: + rows.append((FAIL, s, f"Character API check now: HTTP {resp.status}")) + return rows + chars = await resp.json() + except asyncio.TimeoutError: + rows.append((FAIL, s, "Character API check now: timed out after 15s")) + return rows + except Exception as e: + rows.append((FAIL, s, f"Character API check now: {type(e).__name__}")) + return rows + missing = [c.get("name", "?") for c in chars if c.get("id") not in db.by_id] + if missing: + shown = ", ".join(missing[:5]) + (f" +{len(missing) - 5} more" if len(missing) > 5 else "") + rows.append((WARN, s, f"Character API check now: {len(missing)} not loaded yet ({shown}); " + "added on next refresh")) + else: + rows.append((OK, s, f"Character API check now: reachable, all {len(chars)} loaded")) + return rows + + +def _check_refresh_task(cog) -> list[Row]: + s = "Background refresh" + t = getattr(cog, "_refresh_task", None) + if t is None: + return [(FAIL, s, "Refresh task was never started — banners will never update")] + if t.cancelled(): + return [(FAIL, s, "Refresh task was cancelled — banners will not update")] + if t.done(): + return [(FAIL, s, f"Refresh task crashed: {t.exception()!r} — banners will not update")] + due = (cog._last_refresh_at or 0) + REFRESH_INTERVAL + return [(OK, s, f"Running, next refresh ")] + + +async def _check_storage(user_id: int) -> list[Row]: + s = "Storage & render" + rows: list[Row] = [] + try: + await gacha_db.get_inventory_stats(user_id) + rows.append((OK, s, "Inventory database: read OK (writes not tested)")) + except Exception as e: + rows.append((FAIL, s, f"Inventory database read failed: {type(e).__name__}")) + + if any(Path(fp).exists() for fp in FONT_PATHS): + rows.append((OK, s, "Font found")) + else: + rows.append((WARN, s, "No font found — render uses tiny default font")) + rows.append((OK, s, "Background image found") if BG_PATH.exists() + else (WARN, s, "Background image missing — plain colour used")) + rows.append((OK, s, "Pull animation found") if GACHA_ANIM_PATH.exists() + else (WARN, s, "Pull animation missing — text teaser used")) + + sample = next(iter(db.by_rarity.get(3) or db.students), None) + if sample is None: + rows.append((FAIL, s, "Render test skipped: no students loaded")) + return rows + try: + start = time.monotonic() + await asyncio.wait_for(render_pull([sample], "Health check", 0), timeout=30) + rows.append((OK, s, f"Render test OK in {time.monotonic() - start:.1f}s (art checked under Portraits)")) + except asyncio.TimeoutError: + rows.append((FAIL, s, "Render test timed out after 30s")) + except Exception as e: + rows.append((FAIL, s, f"Render test failed: {type(e).__name__}: {e}")) + return rows + + +def _check_user_banner(cog, banner_id: Optional[str]) -> list[Row]: + s = "Your banner" + if not banner_id or banner_id == "regular": + return [(OK, s, "Regular Recruitment")] + for b in cog._all_banners(): + if str(b.get("id")) == banner_id: + reason = banner_unavailable_reason(b) + if reason: + return [(WARN, s, f"#{banner_id}: {reason} Pick another with `!gacha pick`")] + return [(OK, s, f"#{banner_id} {b.get('gachaType', '?')} is live")] + return [(WARN, s, f"#{banner_id} is no longer listed — `!pull` will ask you to pick another")] + + +async def collect_health(cog, user_id: int, banner_id: Optional[str]) -> list[Row]: + """Run every check. Network checks run concurrently.""" + (banner_rows, live_banners), roster_rows, storage_rows = await asyncio.gather( + _check_banners(cog), _check_roster(), _check_storage(user_id), + ) + rateup_rows, students = _check_rateups(live_banners) + portrait_rows = await _check_portraits(students) + return ( + banner_rows + rateup_rows + portrait_rows + roster_rows + + _check_refresh_task(cog) + storage_rows + _check_user_banner(cog, banner_id) + ) diff --git a/blue_archive/inventory.py b/blue_archive/inventory.py index 3533f3e..043c794 100644 --- a/blue_archive/inventory.py +++ b/blue_archive/inventory.py @@ -251,6 +251,13 @@ class Inventory(commands.Cog): view = InventoryView(uid, ctx.author.id) await ctx.reply(embed=await view._build_dashboard(), view=view) + @commands.command(name="eligma") + async def eligma(self, ctx: commands.Context) -> None: + """Show your Eligma balance (advertised in !gacha help).""" + view = InventoryView(ctx.author.id, ctx.author.id) + view.stop() # only used to build the embed, no buttons attached + await ctx.reply(embed=await view._build_eligma_page()) + async def setup(bot: commands.Bot) -> None: await bot.add_cog(Inventory(bot))