clxud
353 lines
13 KiB
Python
353 lines
13 KiB
Python
"""
|
||
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 <t:...:D>
|
||
end_discord – Discord timestamp string <t:...:D>
|
||
|
||
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"<t:{start_ts}:D>",
|
||
"end_discord": f"<t:{end_ts}:D>",
|
||
})
|
||
|
||
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"<t:{entry['end']}:D>"
|
||
|
||
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())
|