254 lines
9.2 KiB
Python
254 lines
9.2 KiB
Python
#!/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()
|