""" update_game_files.py Downloads and updates vehicle icon files from the War Thunder Datamine GitHub repository. Fetches unit atlas images used for scoreboard rendering. """ # Standard Library Imports import asyncio import json import os import platform import re import shutil import subprocess from datetime import datetime, timezone from pathlib import Path # Third-Party Library Imports import aiohttp from aiofiles import open as aioopen GITHUB_TREE_URL = "https://api.github.com/repos/gszabi99/War-Thunder-Datamine/git/trees/master?recursive=1" TARGET_PATH = "atlases.vromfs.bin_u/units/" GAMEUISKIN_PATH = "atlases.vromfs.bin_u/gameuiskin/" RAW_BASE_URL = "https://raw.githubusercontent.com/gszabi99/War-Thunder-Datamine/master/" MAX_CONCURRENT = 50 # adjust if you hit rate limits def _default_wt_dir() -> Path: """Return the default War Thunder installation directory for the current platform. Returns: Path to the War Thunder Steam installation directory. """ if platform.system() == "Windows": return Path("C:/Program Files (x86)/Steam/steamapps/common/War Thunder") return Path.home() / ".local" / "share" / "Steam" / "steamapps" / "common" / "War Thunder" WAR_THUNDER_DIR = Path(os.environ.get("WAR_THUNDER_DIR", str(_default_wt_dir()))) VROMFS_FILES = ["char.vromfs.bin", "lang.vromfs.bin"] PROJECT_DIR = Path(__file__).parent SREBOT_BOT_DIR = PROJECT_DIR.parent / "SREBOT" / "BOT" LOCAL_ICONS_DIR = PROJECT_DIR / "ICONS" / "VEHICLES" LOCAL_MINIS_DIR = PROJECT_DIR / "ICONS" / "MINIS" LOCAL_MINIS_SVG_DIR = LOCAL_MINIS_DIR / "SVGs" HEADERS = {"User-Agent": "IconUpdater/2.0"} def update_vromfs(): """Copy .vromfs.bin files from the local War Thunder installation into the project directory.""" print("Updating vromfs files from War Thunder installation...") for filename in VROMFS_FILES: src = WAR_THUNDER_DIR / filename dst = PROJECT_DIR / filename if src.exists(): shutil.copy2(src, dst) print(f"✅ Copied {filename}") else: print(f"⚠️ {filename} not found at {src}") async def fetch_file_list(session): """Fetch the list of unit icon files from the War Thunder Datamine GitHub repo. Args: session: An aiohttp.ClientSession for making HTTP requests. Returns: List of dicts with 'name' and 'download_url' keys for each icon file. Raises: RuntimeError: If the GitHub API returns a non-200 status. """ async with session.get(GITHUB_TREE_URL, headers=HEADERS) as resp: if resp.status != 200: text = await resp.text() raise RuntimeError(f"GitHub API error {resp.status}: {text[:200]}") data = await resp.json() # Filter to only files in the target directory (not subdirs) files = [] for item in data.get("tree", []): if item["type"] == "blob" and item["path"].startswith(TARGET_PATH): name = item["path"].split("/")[-1] files.append({ "name": name, "download_url": RAW_BASE_URL + item["path"] }) return files async def download_file(session, item, sem, stats): """Download a single icon file if it is new or has changed. Args: session: An aiohttp.ClientSession. item: Dict with 'name' and 'download_url' keys. sem: asyncio.Semaphore for concurrency limiting. stats: Mutable dict tracking 'new', 'updated', and 'skipped' counts. """ async with sem: name = item["name"] url = item["download_url"] local_path = LOCAL_ICONS_DIR / name try: async with session.get(url) as resp: if resp.status != 200: print(f"Failed {name}: {resp.status}") return data = await resp.read() # Check if file already exists with identical content if local_path.exists(): existing = local_path.read_bytes() if existing == data: stats["skipped"] += 1 return print(f"Updated: {name}") stats["updated"] += 1 else: print(f"New: {name}") stats["new"] += 1 async with aioopen(local_path, "wb") as f: await f.write(data) except Exception as e: print(f"⚠️ Error downloading {name}: {e}") async def fetch_gameuiskin_list(session): """Fetch the list of *_ico.svg files from the gameuiskin atlas directory. Args: session: An aiohttp.ClientSession for making HTTP requests. Returns: List of dicts with 'name' and 'download_url' keys for each _ico.svg file. """ async with session.get(GITHUB_TREE_URL, headers=HEADERS) as resp: if resp.status != 200: text = await resp.text() raise RuntimeError(f"GitHub API error {resp.status}: {text[:200]}") data = await resp.json() files = [] for item in data.get("tree", []): if item["type"] == "blob" and item["path"].startswith(GAMEUISKIN_PATH): name = item["path"].split("/")[-1] if name.endswith("_ico.svg"): files.append({ "name": name, "download_url": RAW_BASE_URL + item["path"] }) return files async def download_mini(session, item, sem): """Download a single mini icon SVG unconditionally (temporary file for PNG conversion). Args: session: An aiohttp.ClientSession. item: Dict with 'name' and 'download_url' keys. sem: asyncio.Semaphore for concurrency limiting. """ async with sem: name = item["name"] url = item["download_url"] local_path = LOCAL_MINIS_SVG_DIR / name try: async with session.get(url) as resp: if resp.status != 200: print(f"Failed {name}: {resp.status}") return async with aioopen(local_path, "wb") as f: await f.write(await resp.read()) except Exception as e: print(f"⚠️ Error downloading {name}: {e}") def convert_svgs_to_pngs(svg_dir: Path, out_dir: Path, height: int = 128): """Batch convert all SVG files in *svg_dir* to PNG in *out_dir* using rsvg-convert. Only the height is fixed; width scales to preserve aspect ratio. Reports new PNGs, updated PNGs, and unchanged PNGs. Requires ``rsvg-convert`` to be installed (librsvg2-bin). Args: svg_dir: Path to the directory containing SVG files. out_dir: Path to write the output PNG files. height: Height in pixels for the rasterised PNGs (width scales proportionally). """ out_dir.mkdir(parents=True, exist_ok=True) svgs = list(svg_dir.glob("*.svg")) if not svgs: print("No SVGs to convert.") return new = 0 updated = 0 skipped = 0 for svg_path in svgs: png_path = out_dir / svg_path.with_suffix(".png").name old_bytes = png_path.read_bytes() if png_path.exists() else None try: subprocess.run( ["rsvg-convert", "-h", str(height), str(svg_path), "-o", str(png_path)], check=True, capture_output=True, ) new_bytes = png_path.read_bytes() if old_bytes is None: print(f"New: {png_path.name}") new += 1 elif old_bytes != new_bytes: print(f"Updated: {png_path.name}") updated += 1 else: skipped += 1 except FileNotFoundError: print("⚠️ rsvg-convert not found! Install librsvg2-bin (apt) or librsvg2-tools.") return except subprocess.CalledProcessError as e: print(f"⚠️ Failed to convert {svg_path.name}: {e.stderr.decode()[:100]}") print(f"\n🎯 Mini icon update complete. {new} new, {updated} updated, {skipped} unchanged.") async def main(): """Run the full update pipeline: copy vromfs, then download all changed icons from GitHub.""" update_vromfs() print() LOCAL_ICONS_DIR.mkdir(parents=True, exist_ok=True) sem = asyncio.Semaphore(MAX_CONCURRENT) stats = {"new": 0, "updated": 0, "skipped": 0} async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=600)) as session: # --- Vehicle atlas icons (VEHICLES/) --- print("Fetching icon list from GitHub...") files = await fetch_file_list(session) print(f"Found {len(files)} files, checking for changes...") tasks = [download_file(session, item, sem, stats) for item in files] await asyncio.gather(*tasks) print(f"\n🎯 Icon update complete. {stats['new']} new, {stats['updated']} updated, {stats['skipped']} unchanged.") # --- Mini vehicle silhouette icons (MINIS/SVGs/ -> MINIS/*.png) --- LOCAL_MINIS_SVG_DIR.mkdir(parents=True, exist_ok=True) print("\nFetching mini icon (gameuiskin) list from GitHub...") mini_files = await fetch_gameuiskin_list(session) print(f"Found {len(mini_files)} _ico.svg files, downloading...") mini_tasks = [download_mini(session, item, sem) for item in mini_files] await asyncio.gather(*mini_tasks) # Convert SVGs to PNGs, report new/updated/unchanged, then clean up SVGs print("\nConverting mini icon SVGs to PNGs...") convert_svgs_to_pngs(LOCAL_MINIS_SVG_DIR, LOCAL_MINIS_DIR, height=128) shutil.rmtree(LOCAL_MINIS_SVG_DIR, ignore_errors=True) def parse_schedule_text(text, year=None, hour_offset=0): """ Parse a season schedule from pasted text into a JSON-ready list. Handles lines like: 1 week мах BR 14.3 (01.01 — 07.01) Until the end of season, мах BR 5.0 (24.02 — 28.02) Each entry contains: max_br – float, maximum battle rating for the period start – int, Unix timestamp at midnight UTC of the start date end – int, Unix timestamp when the next BR window starts start_discord – Discord timestamp string end_discord – Discord timestamp string Args: text: raw schedule text (multi-line) year: 4-digit int year (defaults to current UTC year) hour_offset: hours to shift every timestamp (e.g. -17 subtracts 17 hours) """ """ adding comment to make the bot autorestart update print("heeehehhehehehe") """ if year is None: year = datetime.now(tz=timezone.utc).year br_re = re.compile(r'BR\s+([\d.]+)') date_re = re.compile(r'\((\d{2})\.(\d{2})\s*[—–\-]+\s*(\d{2})\.(\d{2})\)') shift = hour_offset * 3600 results = [] for line in text.strip().splitlines(): line = line.strip() if not line: continue br_match = br_re.search(line) date_match = date_re.search(line) if not br_match or not date_match: continue max_br = float(br_match.group(1)) s_day, s_mon, e_day, e_mon = (int(x) for x in date_match.groups()) start_dt = datetime(year, s_mon, s_day, 0, 0, 0, tzinfo=timezone.utc) end_dt = datetime(year, e_mon, e_day, 0, 0, 0, tzinfo=timezone.utc) start_ts = int(start_dt.timestamp()) + shift end_ts = int(end_dt.timestamp()) + shift results.append({ "max_br": max_br, "start": start_ts, "end": end_ts, "start_discord": f"", "end_discord": f"", }) for i, entry in enumerate(results): if i + 1 < len(results): entry["end"] = results[i + 1]["start"] else: entry["end"] = entry["end"] + 24 * 3600 entry["end_discord"] = f"" return results _SCHEDULE_TXT = SREBOT_BOT_DIR / "SCHEDULE.txt" _SCHEDULE_JSON = SREBOT_BOT_DIR / "SCHEDULE.json" _SCHEDULE_HOUR_OFFSET = 7 # shift from midnight UTC so dates display correctly locally def update_schedule(): """Parse SCHEDULE.txt and write the resulting JSON schedule to SCHEDULE.json.""" if not _SCHEDULE_TXT.exists(): print(f"⚠️ {_SCHEDULE_TXT} not found, skipping schedule update.") return text = _SCHEDULE_TXT.read_text(encoding="utf-8") data = parse_schedule_text(text, hour_offset=_SCHEDULE_HOUR_OFFSET) _SCHEDULE_JSON.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8") print(f"✅ Schedule updated — {len(data)} entries written to {_SCHEDULE_JSON.name}") if __name__ == "__main__": update_schedule() asyncio.run(main())