#!/usr/bin/env python3 """OGame Data Updater — snapshot pubblici dell'universo + classifiche. Endpoint usati (API pubblica ufficiale OGame, JSON via ?toJson=1): serverData.xml -> impostazioni universo (aggiornato ~1 volta al giorno) players.xml -> giocatori + stato (aggiornato ~1 volta al giorno) alliances.xml -> alleanze (aggiornato ~1 volta al giorno) universe.xml -> pianeti e lune (aggiornato ~1 volta a settimana) localization.xml -> nomi localizzati (raramente) highscore.xml -> classifiche giocatori (aggiornato ~1 volta all'ora) type=0 -> Totale (generale) type=3 -> Militare (armamenti) Uso: python3 update_data.py aggiornamento completo (consigliato 1x/giorno) python3 update_data.py --highscore-only solo classifiche (per un job orario) """ import json import os import sys import time import urllib.request from datetime import datetime, timezone from pathlib import Path BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" LATEST = DATA_DIR / "latest" SNAPSHOTS = DATA_DIR / "snapshots" CONFIG_PATH = BASE_DIR / "config.json" ENDPOINTS = ["serverData", "universe", "players", "alliances", "localization"] # (nome_file, category, type) — category 1 = giocatori HIGHSCORES = [("highscore_total", 1, 0), ("highscore_military", 1, 3)] UA = "OGameGalaxyViewer/1.0 (+personal tool; https://ogapi.faw-kes.de/ documented flow)" def load_config(): with open(CONFIG_PATH, encoding="utf-8") as f: return json.load(f) def http_get(url, timeout=120): req = urllib.request.Request(url, headers={"User-Agent": UA}) with urllib.request.urlopen(req, timeout=timeout) as r: return r.read().decode("utf-8") def fetch_xml_json(cfg, name, query=""): api = f"https://s{cfg['server_number']}-{cfg['community']}.ogame.gameforge.com/api" url = f"{api}/{name}.xml?toJson=1{query}" data = json.loads(http_get(url)) data["_fetched_at"] = int(time.time()) return data def fetch_endpoint(cfg, name): return fetch_xml_json(cfg, name) def fetch_highscore(cfg, name, category, htype): data = fetch_xml_json(cfg, name, f"&category={category}&type={htype}") data["_meta"] = {"category": category, "type": htype} return data def save_latest(name, data): with open(LATEST / f"{name}.json", "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False) def save_snapshot(name, data): today = datetime.now(timezone.utc).strftime("%Y-%m-%d") SNAPSHOTS.mkdir(exist_ok=True) dest = SNAPSHOTS / f"{name}_{today}.json" if not dest.exists(): with open(dest, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False) keep = load_config().get("keep_snapshots", 30) files = sorted(SNAPSHOTS.glob(f"{name}_*.json")) for old in files[:-keep]: old.unlink() def _as_dict(payload, key, attr_key="id"): """Rende {'id': {...attributi}} da {'key': [{'@attributes': {...}}]}.""" out = {} for item in payload.get(key, []): att = item.get("@attributes", {}) out[att[attr_key]] = {k: v for k, v in att.items()} return out def build_world(cfg): """Costruisce world.json: indici per giocatori, pianeti, alleanze e classifiche.""" server = json.load(open(LATEST / "serverData.json", encoding="utf-8")) players = json.load(open(LATEST / "players.json", encoding="utf-8")) alliances = json.load(open(LATEST / "alliances.json", encoding="utf-8")) universe = json.load(open(LATEST / "universe.json", encoding="utf-8")) sd = server.get("serverData", server) if "@attributes" in sd: sd = {k: v for k, v in sd.items() if k != "@attributes"} ally_map = {} for a in alliances.get("alliance", []): att = a.get("@attributes", {}) ally_map[att["id"]] = { "id": att["id"], "name": att.get("name", ""), "tag": att.get("tag", ""), "founder": att.get("founder", ""), } player_map = {} for p in players.get("player", []): att = p.get("@attributes", {}) pid = att["id"] player_map[pid] = { "id": pid, "name": att.get("name", ""), "status": att.get("status", ""), "alliance": att.get("alliance", ""), } if att.get("alliance") and att["alliance"] in ally_map: player_map[pid]["alliance_tag"] = ally_map[att["alliance"]]["tag"] player_map[pid]["alliance_name"] = ally_map[att["alliance"]]["name"] planets = {} avatar = {} for p in universe.get("planet", []): att = p.get("@attributes", {}) coords = att["coords"] entry = {"id": att["id"], "player": att["player"], "name": att.get("name", ""), "coords": coords} if "moon" in p: ma = p["moon"].get("@attributes", {}) entry["moon"] = {"id": ma.get("id", ""), "name": ma.get("name", "Luna"), "size": int(ma.get("size", 0) or 0)} planets[coords] = entry avatar.setdefault(att["player"], coords) # classifiche: mappa playerId -> posizione e punteggio def load_ranks(fname): try: hs = json.load(open(LATEST / f"{fname}.json", encoding="utf-8")) d = _as_dict(hs, "player") return {pid: (int(a.get("position", 0)), float(a.get("score", 0))) for pid, a in d.items()} except FileNotFoundError: return {} ranks_total = load_ranks("highscore_total") ranks_military = load_ranks("highscore_military") ranks = {} for pid in player_map: t = ranks_total.get(pid) m = ranks_military.get(pid) ranks[pid] = { "total": t[0] if t else None, "military": m[0] if m else None, "total_score": int(t[1]) if t else None, "military_score": int(m[1]) if m else None, } world = { "server": { "number": cfg["server_number"], "community": cfg["community"], "domain": cfg.get("domain"), "name": sd.get("name", ""), "galaxies": int(sd.get("galaxies", 9)), "systems": int(sd.get("systems", 499)), "speed": sd.get("speed", ""), "version": sd.get("version", ""), "language": sd.get("language", ""), "timezone": sd.get("timezone", ""), "acs": sd.get("acs", ""), }, "fetched": { "serverData": server.get("_fetched_at"), "universe": universe.get("_fetched_at"), "players": players.get("_fetched_at"), "alliances": alliances.get("_fetched_at"), "highscore_total": _file_time("highscore_total"), "highscore_military": _file_time("highscore_military"), }, "players": player_map, "alliances": ally_map, "planets": planets, "avatar": avatar, "ranks": ranks, } with open(LATEST / "world.json", "w", encoding="utf-8") as f: json.dump(world, f, ensure_ascii=False) return world def _file_time(name): try: d = json.load(open(LATEST / f"{name}.json", encoding="utf-8")) return d.get("_fetched_at") except (FileNotFoundError, json.JSONDecodeError): return None def update_highscores(cfg, with_snapshot=True): for name, cat, typ in HIGHSCORES: data = fetch_xml_json(cfg, "highscore", f"&category={cat}&type={typ}") data["_meta"] = {"category": cat, "type": typ} save_latest(name, data) if with_snapshot: save_snapshot(name, data) ts = datetime.fromtimestamp(data["_fetched_at"], tz=timezone.utc) print(f" [ok] {name}.json (dati server del {ts:%Y-%m-%d %H:%M} UTC)") def main(): highscore_only = "--highscore-only" in sys.argv cfg = load_config() if highscore_only: print(f"Aggiornamento classifiche s{cfg['server_number']}-{cfg['community']}") try: update_highscores(cfg, with_snapshot=False) w = build_world(cfg) print(f"world.json aggiornato (classifiche incluse, {len(w['players'])} giocatori)") return except Exception as e: print(f"ERRORE: {e}", file=sys.stderr) sys.exit(1) print(f"Aggiornamento dati universo s{cfg['server_number']}-{cfg['community']} " f"({cfg['domain']})") ok = [] for name in ENDPOINTS: try: data = fetch_endpoint(cfg, name) save_latest(name, data) save_snapshot(name, data) ts = datetime.fromtimestamp(data["_fetched_at"], tz=timezone.utc) print(f" [ok] {name}.json (dati server del {ts:%Y-%m-%d %H:%M} UTC)") ok.append(name) except Exception as e: print(f" [ERRORE] {name}: {e}", file=sys.stderr) try: update_highscores(cfg, with_snapshot=True) ok.append("highscore") except Exception as e: print(f" [ERRORE] highscore: {e}", file=sys.stderr) if ok: world = build_world(cfg) print(f"world.json costruito: {len(world['planets'])} pianeti, " f"{len(world['players'])} giocatori, {len(world['alliances'])} alleanze") else: print("Nessun dato scaricato.", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()