#!/usr/bin/env python3
"""
iScout Web - UI local sobre el backend Frida.
Adjunta a Evony en el emulador rooteado, barre el mapa automaticamente,
y sirve una interfaz web para filtrar por boss / radio / nivel.
Uso:
/Applications/Xcode.app/Contents/Developer/usr/bin/python3 iscout_web.py
Abre http://127.0.0.1:8770 en el navegador.
"""
import os, sys, json, math, threading, time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import frida
HERE = os.path.dirname(os.path.abspath(__file__))
AGENT = os.path.join(HERE, "agent.js")
PORT = 8771 # v2 dual: puerto distinto a v1 (8770)
# Persistencia del cache entre restarts del backend. Guarda OBJS, PLAYERS, etc.
# en JSON cada PERSIST_INTERVAL segundos y al cerrar. Filtra al cargar entradas
# con ts más viejas que PERSIST_TTL (24h) para no acumular basura indefinidamente.
CACHE_FILE = os.path.join(HERE, "cache_persist.json")
PERSIST_INTERVAL = 30 # guardar cada 30s
PERSIST_TTL = 86400 # 24h: entries más viejas se descartan al cargar
# WorldResourceConfig: tipo enum farm_type (1=food, 2=wood, 3=stone, 4=iron, 5+=gems/pumpkin/rose/etc.)
WCFG_TYPE_LABEL = {1: "Farm", 2: "Sawmill", 3: "Quarry", 4: "Iron Mine"}
SUBCITY_QUALITY = {0: "white", 1: "green", 2: "blue", 3: "purple", 4: "gold", 5: "red"}
# Mapping de Msg.sub_city_summary.__evony (Int32) -> nombre de cultura.
# 1-7 = las 7 culturas clasicas (NPC sub-cities normales).
# 10010+ = Famous Cities (sub-cities especiales con buffs unicos, p.ej. Vietnam).
SUBCITY_CULTURE = {
1: "European", 2: "American", 3: "Chinese", 4: "Russian",
5: "Korean", 6: "Arabia", 7: "Japan",
10010: "Vietnam", # Famous City (visto en este server)
}
# mapinfo_type enum: solo los que usamos para las pestañas
MAPINFO_TYPE = {
0: "field", 1: "player", 2: "npc", 3: "farm", 4: "garrison",
5: "ruins", 6: "king_city", 7: "subcity", 8: "snowberg",
9: "guild_city", 10: "guild_store", 11: "guild_farm", 12: "boss",
34: "altar", 36: "barbarian_castle", 57: "pyramid",
}
# v2: dos escaneres, cada uno una mitad del mapa (corte vertical en X=650).
# W = emulator-5554 (cuenta #1, Oeste X<=650)
# E = emulator-5556 (cuenta #2, Este X>=650)
SCANNERS = [("emulator-5554", "W"), ("emulator-5556", "E")]
LOCK = threading.Lock()
FRIDA_SETUP_LOCK = threading.Lock() # serializa attach/create_script/load de los 2 escaneres
CFG = {} # id -> {name, level, type, power} (MonsterConfig)
WCFG = {} # id -> {name, level, type} (WorldResourceConfig: farms/mines/etc)
OBJS = {} # (wx,wy) -> {t,id,lv,wx,wy,ts} (pool fusionado de ambos)
PLAYERS = {} # uid -> {...} (fusionado de ambos)
MARCHES = {} # troop_id (int) -> {mty,ow,sx,sy,tx,ty,ts,te,tg,ts_recv}
MARCH_BY_TGT = {} # (tx,ty) -> {troop_id: True} (indice rapido para enrich)
SUBCITIES = {} # subcity_id -> {wx,wy,owner_uid,name,shield_end,...}
SHIELD_ETA = {} # uid -> {end_time, src, confidence, activation_ts, ts}
# confidence: 'exact' (scout/mail/subcity con uid) |
# 'inferred_default' (0->1 transition, asume DEFAULT) |
# 'inferred_history' (0->1 con histórico previo, usa mediana) |
# 'newbie_inferred' (0->2, asume 7d max)
SHIELD_HISTORY = {} # uid -> list[{activation_ts, expired_ts, duration_s, tier, src}]
# archivo cuando shield cae 1->0 o 2->0. Sobrevive reinicios via cache_persist.
SHIELD_HISTORY_MAX = 5 # max muestras por uid (las mas recientes ganan)
SHIELD_LAST_TRANSITION = {} # uid -> ts del ultimo cambio de tier (para anti-flicker cooldown)
SHIELD_FLICKER_COOLDOWN_S = 60 # ignorar transitions opuestas <60s del ultimo cambio
SHIELD_MIN_DURATION_S = 300 # 5min: duraciones menores se asumen flicker, no se archivan
SHIELD_DEFAULT_DURATION_S = 8 * 3600 # 8h: default mas comun de item-shields en Evony
SHIELD_NEWBIE_MAX_S = 7 * 86400 # 7d: maximo del newbie shield (tier=2)
SHIELD_RECENT = [] # ultimos N shield captures (mas reciente primero) p/ UI toast
SHIELD_RECENT_MAX = 20
WATCHLIST = set() # set[int uid] de uids marcados como prioritarios (alertas + scout queue)
WATCHLIST_NOTES = {} # uid -> {note: str, added_ts: int, source: 'manual'|'auto'}
WATCHLIST_BLACKLIST = set() # uids descartados por el usuario (no se re-auto-anaden)
# Reglas para auto-watchlist. Worker auto_watchlist_thread() lee PLAYERS cada 60s
# y anade uids que cumplan reglas (no estan ya en WATCHLIST ni en BLACKLIST).
WATCHLIST_RULES = {
"enabled": True,
"enemy_tags": ["NBB"], # tags que cuentan como enemigo (auto-add si coincide)
"exclude_tags": ["MVP"], # tu alianza + aliados (nunca auto-add)
"min_castle": 0, # castle level minimo del player
"min_power_M": 0, # power minimo en millones
"require_shield_seen": True, # solo si hemos visto tier>=1 alguna vez (SHIELD_HISTORY o SHIELD_ETA)
}
PLAYER_POWER = {} # uid -> {power, rank, ts} (de power_rank_reply)
SELF = {"W": {"uid": 0, "guild_id": 0}, "E": {"uid": 0, "guild_id": 0}} # uid+guild_id por scanner
MONSTERS_BY_TILE = {} # (wx,wy) -> {id, lv, name, ts} cache LARGO de ultimo monstruo visto
# en cada tile. Sobrevive al prune de OBJS. Permite resolver el
# target de attacks aunque el monstruo ya no se broadcasta (combate).
MONSTERS_TTL = 3600 # 1h: si en 1h no se ve nada en el tile, se descarta
# ── Boss Respawn Predictor ──────────────────────────────────────────────────
# Por cada tile guardamos hasta SPAWN_HISTORY_MAX timestamps de "primera vez visto"
# (un nuevo spawn). El intervalo entre spawns consecutivos = lifetime + respawn_gap;
# para predecir el próximo basta con: last_first_ts + median(intervalos).
SPAWN_HISTORY = {} # (wx,wy) -> [{ts, id, lv}, ...] últimas N apariciones
SPAWN_HISTORY_MAX = 8 # suficiente para mediana estable sin inflar memoria
# ── Player Relocation Tracker ───────────────────────────────────────────────
# Por cada uid guardamos hasta RELOCATIONS_MAX eventos de cambio de coords.
# Se detecta comparando wx/wy de PLAYERS[uid] con el batch nuevo.
RELOCATIONS = {} # uid -> [{ts, from:(x,y), to:(x,y)}, ...]
RELOCATIONS_MAX = 20
SHIELD_ALERT_THRESHOLD = 600 # 10 minutos: ETAs < esto se marcan como "expiring soon" para alertas
SHIELD_ALERTED = set() # uids ya alertados (para no spamear toasts)
SCRIPTS = {"W": None, "E": None} # refs a los scripts cargados (para pause/resume via .post)
SCANNER_PAUSED = False # estado global pausa/resume
LAST_HEARTBEAT = {"W": 0.0, "E": 0.0} # ts del último mensaje del agent (cualquier kind)
HEARTBEAT_TIMEOUT = 90 # si no llega nada en 90s, el agent se considera muerto
REATTACH_FAILS = {"W": 0, "E": 0} # contador de reattaches fallidos consecutivos
HARD_RESET_THRESHOLD = 3 # tras N fallos seguidos, force-stop Evony del half
NUCLEAR_RESET_THRESHOLD = 6 # tras N fallos seguidos, kill qemu entero y relanzar
# AVD names + emulator binary para nuclear reset
NUCLEAR_AVD = {"W": ("evony_root", 5554), "E": ("evony_root_2", 5556)}
EMULATOR_BIN = "/opt/homebrew/share/android-commandlinetools/emulator/emulator"
STATE = {"cfg_n": 0,
"W": {"sweep": "?", "num": 0, "last_secs": None},
"E": {"sweep": "?", "num": 0, "last_secs": None}}
FIRST_PASS_TS = None # ts en que AMBOS terminaron su 1a vuelta (mapa entero visto)
_FP_DONE = {"W": False, "E": False}
LAST_PASS_TS = {"W": None, "E": None} # ts de la ultima vuelta por escaner
FRESH_WINDOW = 120 # 2 min: tras esto el "nuevo" deja de estar resaltado
TTL_SECONDS = 86400 # 24h: persistencia larga para acumular cobertura entre sesiones
# (se guarda en disco también via persist_thread)
STALE_SECONDS = 90 # una vuelta del barrido es ~24s. Si un objeto no se re-ve
# en ~3-4 vueltas se considera muerto -> se quita de la lista
PRUNE_EVERY = 20 # cada N seg el hilo de limpieza purga muertos
RESPAWN_GAP_S = 250 # si un tile no se ve >= esto y reaparece => spawn NUEVO
# (el boss murio, el tile estuvo vacio entre pasadas y
# ahora hay otro). Igualado a v1/iScout para descartar que
# la diferencia de detection count entre v1 y v2 venga de aqui.
# 250s = ~8-10 vueltas de v2: mucho margen para no falsear
# respawns por gaps puntuales de 1-2 pasadas.
# ---------------- Shield inference helpers ----------------
def _shield_median_duration(uid):
"""Devuelve la mediana de duraciones (s) del historico del uid, o None si <2 muestras."""
h = SHIELD_HISTORY.get(uid)
if not h or len(h) < 2: return None
durs = sorted(int(x.get("duration_s", 0) or 0) for x in h if x.get("duration_s"))
if not durs: return None
n = len(durs)
return durs[n // 2] if n % 2 == 1 else (durs[n // 2 - 1] + durs[n // 2]) // 2
def _on_shield_tier_transition(uid, prev_tier, new_tier, now):
"""Logica de inferencia cuando un player cambia su tier de shield en el broadcast.
LLAMAR DENTRO DEL LOCK.
REGLAS:
- El broadcast tier es GROUND TRUTH. Si llega tier=0, SIEMPRE limpiamos SHIELD_ETA.
- El COOLDOWN solo protege ARCHIVING (no envenenar historico con flickers <5min).
- Una transition 0->1 dentro del cooldown del ultimo evento NO crea nueva SHIELD_ETA
(era flicker: el shield real nunca se fue, el undershot_check lo recreara con +50%).
- No-op si existe SHIELD_ETA[uid] con confidence='exact' y end_time>now (gana lo exacto).
"""
cur = SHIELD_ETA.get(uid)
has_exact_live = (cur and cur.get("confidence") == "exact" and int(cur.get("end_time", 0) or 0) > now)
last_trans = SHIELD_LAST_TRANSITION.get(uid, 0)
in_cooldown = bool(last_trans and (now - last_trans) < SHIELD_FLICKER_COOLDOWN_S)
SHIELD_LAST_TRANSITION[uid] = int(now)
# ---- ACTIVACION: 0 -> 1 (item) o 0 -> 2 (newbie)
if prev_tier == 0 and new_tier in (1, 2):
if has_exact_live:
return # respeto la ETA exacta vigente
if in_cooldown:
# 0->1 dentro del cooldown de un 1->0 reciente: era FLICKER, no nueva activacion.
# El undershot_check posterior recreara SHIELD_ETA con +50% si tier sigue >0.
return
if new_tier == 2:
# newbie: asume el maximo (7d desde ahora). Es upper-bound.
SHIELD_ETA[uid] = {
"end_time": int(now) + SHIELD_NEWBIE_MAX_S,
"src": "tier_transition",
"confidence": "newbie_inferred",
"activation_ts": int(now),
"ts": now,
}
else:
# item shield (tier=1): usa mediana del historico si hay, sino default 8h
med = _shield_median_duration(uid)
# SANIDAD: si la mediana es absurdamente corta (<10min), probablemente esta envenenada
# por flicker pasado -> usa default. Eso da auto-recovery sin necesidad de wipe manual.
if med and med < 600:
med = None
dur = med if med else SHIELD_DEFAULT_DURATION_S
SHIELD_ETA[uid] = {
"end_time": int(now) + int(dur),
"src": "tier_transition",
"confidence": "inferred_history" if med else "inferred_default",
"activation_ts": int(now),
"duration_assumed_s": int(dur),
"ts": now,
}
return
# ---- EXPIRACION/ROTURA: 1 -> 0 o 2 -> 0
# SIEMPRE limpiamos SHIELD_ETA: broadcast es ground truth.
if prev_tier in (1, 2) and new_tier == 0:
# Archivar duracion real en historico (si tenemos activation_ts)
# Solo archivar fuera del cooldown — el cooldown indica que probablemente fue flicker.
if cur and not in_cooldown:
act_ts = int(cur.get("activation_ts", 0) or 0)
end_t = int(cur.get("end_time", 0) or 0)
if act_ts > 0:
duration_real = int(now) - act_ts
# Solo guardar si parece razonable. Threshold SHIELD_MIN_DURATION_S=300s (5min).
if SHIELD_MIN_DURATION_S <= duration_real <= 14 * 86400:
h = SHIELD_HISTORY.setdefault(uid, [])
h.append({
"activation_ts": act_ts,
"expired_ts": int(now),
"duration_s": duration_real,
"tier": prev_tier,
"src": cur.get("src", "?"),
"confidence_at_activation": cur.get("confidence", "?"),
})
if len(h) > SHIELD_HISTORY_MAX:
del h[:-SHIELD_HISTORY_MAX]
# Si tenia ETA exact y end_time aun futuro, fue rotura por ataque
if cur.get("confidence") == "exact" and end_t > now:
print(f"[shield-break] uid={uid} ETA exact era {end_t} ({end_t-int(now)}s left), rota ahora (probable ataque)", flush=True)
SHIELD_ETA.pop(uid, None)
return
# ---- SWITCH 1<->2: raro, conservar lo que tengamos
# no hacemos nada especial
def _shield_check_undershot(uid, current_tier, now):
"""Llamar en cada broadcast players. Si el cliente tiene tier>0 (shield real activo)
pero NO tenemos SHIELD_ETA para uid, significa que la inferencia previa expiro
pero el shield real sigue activo (undershot).
En ese caso, extender SHIELD_ETA con duracion previa * 1.5 + estado 'inferred_undershot'.
LLAMAR DENTRO DEL LOCK.
"""
if current_tier not in (1, 2):
return
if uid in SHIELD_ETA:
return # ya tenemos algo (exact o inferred vivo)
# NO crear undershot por defecto en el primer broadcast (puede ser primera vez que lo vemos).
# Solo si tenemos historico previo: significa que ya hicimos ciclos.
h = SHIELD_HISTORY.get(uid)
if not h:
return
# Tomar la duracion mas reciente del historico como base
last_dur = int(h[-1].get("duration_s", SHIELD_DEFAULT_DURATION_S) or SHIELD_DEFAULT_DURATION_S)
extended_dur = max(int(last_dur * 1.5), 3600) # min 1h
SHIELD_ETA[uid] = {
"end_time": int(now) + extended_dur,
"src": "tier_undershot",
"confidence": "inferred_undershot",
"activation_ts": int(now) - last_dur, # asumimos activacion al final del ultimo ciclo
"duration_assumed_s": extended_dur,
"ts": now,
}
print(f"[undershot] uid={uid} tier={current_tier} inferencia previa expiro pero shield sigue. Extiendo +{extended_dur}s", flush=True)
# ---------------- Frida backend ----------------
def on_message(msg, data, half):
if msg.get("type") == "error":
print(f"[{half} agent error]", msg.get("stack") or msg.get("description"), flush=True)
return
if msg.get("type") != "send":
return
p = msg["payload"]
if not isinstance(p, dict):
return
LAST_HEARTBEAT[half] = time.time() # watchdog: cualquier mensaje cuenta como vida
if p.get("kind") == "heartbeat":
return # solo sirve como ping; no necesita processing
if p.get("kind") == "config":
with LOCK:
for k, v in p["cfg"].items():
CFG[str(k)] = v
STATE["cfg_n"] = len(CFG)
print(f"[{half} cfg] {p['n']} configs", flush=True)
elif p.get("kind") == "wcfg":
with LOCK:
for k, v in p["wcfg"].items():
WCFG[str(k)] = v
print(f"[{half} wcfg] {p['n']} world resource configs", flush=True)
elif p.get("kind") == "batch":
now = time.time()
with LOCK:
for it in p["items"]:
k = (it["wx"], it["wy"])
it["half"] = half # de qué scanner (W o E) llegó este item
# cache LARGO de monstruos por tile: si es ttype=2 (monstruo NPC) y
# tiene id, lo guardamos para que active attacks pueda resolverlo
# incluso si OBJS lo purga durante combate.
if int(it.get("t", 0)) == 2 and int(it.get("id", 0)) > 0:
cfg = CFG.get(str(it["id"])) or {}
MONSTERS_BY_TILE[k] = {
"id": it["id"], "lv": it.get("lv", 0),
"name": cfg.get("name", "") or f"id{it['id']}",
"ts": now,
}
old = OBJS.get(k)
if old is None:
# tile nunca visto -> descubrimiento nuevo
it["ts"] = now
it["first_ts"] = now
OBJS[k] = it
# respawn predictor: registrar PRIMERA aparicion (solo si es monstruo NPC)
if int(it.get("t", 0)) == 2 and int(it.get("id", 0)) > 0:
h = SPAWN_HISTORY.setdefault(k, [])
h.append({"ts": now, "id": it["id"], "lv": it.get("lv", 0)})
if len(h) > SPAWN_HISTORY_MAX: del h[:-SPAWN_HISTORY_MAX]
else:
# spawn NUEVO en un tile recurrente si: cambio el tipo/nivel,
# o el tile estuvo sin verse >= RESPAWN_GAP_S (vacio entre
# pasadas = el boss anterior murio y aparecio otro).
respawn = (old.get("id") != it["id"]
or old.get("lv") != it["lv"]
or (now - old.get("ts", now)) >= RESPAWN_GAP_S)
if respawn:
it["ts"] = now
it["first_ts"] = now # re-marca como recien descubierto
OBJS[k] = it
# respawn predictor: registrar nuevo spawn (intervalo desde el anterior)
if int(it.get("t", 0)) == 2 and int(it.get("id", 0)) > 0:
h = SPAWN_HISTORY.setdefault(k, [])
h.append({"ts": now, "id": it["id"], "lv": it.get("lv", 0)})
if len(h) > SPAWN_HISTORY_MAX: del h[:-SPAWN_HISTORY_MAX]
else:
old.update(it)
old["ts"] = now
old["half"] = half # actualiza scanner que lo re-detectó
old.setdefault("first_ts", old.get("ts", now))
elif p.get("kind") == "players":
now = time.time()
with LOCK:
for it in p["items"]:
uid = it["uid"]
old_p = PLAYERS.get(uid)
# Player Relocation Tracker: detectar cambio de coords del castillo
new_wx, new_wy = int(it.get("wx", 0)), int(it.get("wy", 0))
if (new_wx != 0 or new_wy != 0) and old_p is not None:
old_wx, old_wy = int(old_p.get("wx", 0)), int(old_p.get("wy", 0))
if (old_wx != 0 or old_wy != 0) and (old_wx != new_wx or old_wy != new_wy):
# ¡Relocalización detectada!
r = RELOCATIONS.setdefault(uid, [])
r.append({
"ts": now,
"from_x": old_wx, "from_y": old_wy,
"to_x": new_wx, "to_y": new_wy,
"name": it.get("name", old_p.get("name", "")),
"tag": it.get("tag", old_p.get("tag", "")),
"gid": int(it.get("gid", old_p.get("gid", 0)) or 0),
})
if len(r) > RELOCATIONS_MAX: del r[:-RELOCATIONS_MAX]
# Shield tier transition detection (inferencia de activacion/expiracion)
new_tier = int(it.get("shield", 0) or 0)
prev_tier = int((old_p or {}).get("shield", 0) or 0)
if old_p is not None and new_tier != prev_tier:
_on_shield_tier_transition(uid, prev_tier, new_tier, now)
# Anti-undershot: si tier sigue siendo 1/2 pero no tenemos SHIELD_ETA,
# extender con duracion del ultimo ciclo + 50%
if new_tier in (1, 2):
_shield_check_undershot(uid, new_tier, now)
it["ts"] = now
PLAYERS[uid] = it
elif p.get("kind") == "marches":
now = time.time()
with LOCK:
for it in p["items"]:
trp = int(it.get("trp", 0))
if trp == 0:
continue
it["ts_recv"] = now
old = MARCHES.get(trp)
if old is not None:
okey = (int(old.get("tx", 0)), int(old.get("ty", 0)))
sub = MARCH_BY_TGT.get(okey)
if sub is not None:
sub.pop(trp, None)
if not sub: MARCH_BY_TGT.pop(okey, None)
MARCHES[trp] = it
key = (int(it["tx"]), int(it["ty"]))
MARCH_BY_TGT.setdefault(key, {})[trp] = True
elif p.get("kind") == "subcities":
# Sub-cities con peace_shield_end_time + power_calc (GetPower del cliente).
# NPCs: power_calc > 0 (config-based, exacto). Player-owned: power_calc = 0
# porque el broadcast pasivo no incluye troops/buildings; sólo un scout report
# del sub-city (CLV 19+) traería el dato real.
now = time.time()
with LOCK:
for it in p["items"]:
sid = int(it.get("id", 0))
if sid == 0:
continue
it["ts"] = now
pc = int(it.get("power_calc", 0) or 0)
ouid = int(it.get("owner_uid", 0) or 0)
# preserva power_calc previo si llega 0 ahora
old = SUBCITIES.get(sid)
if old:
old_pc = int(old.get("power_calc", 0) or 0)
if pc == 0 and old_pc > 0:
it["power_calc"] = old_pc
SUBCITIES[sid] = it
end_t = int(it.get("shield_end", 0) or 0)
if end_t > 0 and ouid > 0:
SHIELD_ETA[ouid] = {"end_time": end_t, "src": "subcity",
"confidence": "exact",
"activation_ts": int(SHIELD_ETA.get(ouid, {}).get("activation_ts", 0) or 0),
"ts": now}
elif p.get("kind") == "shields":
# Ground truth de shield ETA. Fuentes:
# - addSubCity (owner_uid + end_time) -> ya viene con uid
# - UIMailScoutReport.ShowScoutReport -> ya viene con uid (cuando user abre mail)
# - sys_mail.set__peace_shield_report -> viene con wx/wy/name, NO con uid: hay que
# matchear contra PLAYERS por (wx,wy) o por nombre para obtener el uid.
now = time.time()
with LOCK:
for it in p["items"]:
et = int(it.get("end_time", 0) or 0)
if et <= 0:
continue
src = it.get("src", "?")
uid = int(it.get("uid", 0) or 0)
wx = int(it.get("wx", 0) or 0); wy = int(it.get("wy", 0) or 0)
nm = (it.get("name", "") or "").strip()
matched_uid = 0
# caso A: el agent ya nos da uid directamente (scout / subcity / self / sys_mail)
if uid > 0:
SHIELD_ETA[uid] = {"end_time": et, "src": src,
"confidence": "exact",
"activation_ts": int(SHIELD_ETA.get(uid, {}).get("activation_ts", 0) or 0),
"ts": now}
matched_uid = uid
else:
# caso B: viene por sys_mail (wx, wy, name). Match a un uid del cache PLAYERS.
# 1) match exacto por coordenadas (lo mas fiable)
if wx and wy:
for puid, p2 in PLAYERS.items():
if int(p2.get("wx", 0) or 0) == wx and int(p2.get("wy", 0) or 0) == wy:
matched_uid = puid; break
# 2) fallback: match por nombre (case-insensitive)
if matched_uid == 0 and nm:
lo = nm.lower()
for puid, p2 in PLAYERS.items():
if (p2.get("name", "") or "").lower() == lo:
matched_uid = puid; break
if matched_uid > 0:
SHIELD_ETA[matched_uid] = {"end_time": et, "src": src,
"confidence": "exact",
"activation_ts": int(SHIELD_ETA.get(matched_uid, {}).get("activation_ts", 0) or 0),
"ts": now}
print(f"[psr->uid] matched {nm or '?'} @({wx},{wy}) -> uid={matched_uid} end_time={et} left={et-int(now)}s", flush=True)
else:
# no encontrado en PLAYERS: guarda por coords para futuro match cuando
# el sweep capture al jugador. query_players hara la migracion.
SHIELD_ETA[f"coord:{wx},{wy}"] = {"end_time": et, "src": src,
"confidence": "exact",
"ts": now,
"wx": wx, "wy": wy, "name": nm}
print(f"[psr] not in PLAYERS yet: {nm or '?'} @({wx},{wy}) end_time={et} (cached by coord)", flush=True)
# registra en SHIELD_RECENT para el toast de la UI (independientemente del match)
SHIELD_RECENT.insert(0, {
"ts": now, "name": nm, "wx": wx, "wy": wy,
"end_time": et, "uid": matched_uid, "src": src,
})
del SHIELD_RECENT[SHIELD_RECENT_MAX:]
elif p.get("kind") == "self":
# Identidad del scanner: uid + guild_id. La usamos para clasificar tiles
# ocupados como "in MY alliance" vs "external alliance".
SELF[half] = {"uid": int(p.get("uid", 0) or 0),
"guild_id": int(p.get("guild_id", 0) or 0)}
print(f"[{half} self] uid={SELF[half]['uid']} guild_id={SELF[half]['guild_id']}", flush=True)
elif p.get("kind") == "powers":
# Power exacto per uid (de MsgDown.power_rank_reply). El usuario tiene que abrir
# Rankings -> Power Ranking en el juego para que el cliente solicite la lista.
now = time.time()
with LOCK:
for it in p["items"]:
uid = int(it.get("uid", 0) or 0)
pw = int(it.get("power", 0) or 0)
if uid > 0 and pw > 0:
PLAYER_POWER[uid] = {"power": pw, "rank": int(it.get("rank", 0) or 0), "ts": now}
elif p.get("kind") == "sweep_done":
STATE[half]["sweep"] = "done"
elif p.get("kind") == "sweep_pass":
global FIRST_PASS_TS
now = time.time()
_FP_DONE[half] = True
if FIRST_PASS_TS is None and all(_FP_DONE.values()):
FIRST_PASS_TS = now # AMBOS dieron su 1a vuelta: mapa entero visto
print("[sweep] 1a vuelta completa de AMBOS; nuevos a partir de ahora se resaltan", flush=True)
if LAST_PASS_TS[half] is not None:
STATE[half]["last_secs"] = round(now - LAST_PASS_TS[half], 1)
LAST_PASS_TS[half] = now
STATE[half]["num"] = p["pass"]
STATE[half]["sweep"] = f"pasada {p['pass']}"
class _NeedsEscalation(Exception):
"""Señal interna: forzar caída al bloque de threshold checks (HARD/NUCLEAR).
NO incrementa REATTACH_FAILS automáticamente — el que la lanza debe hacerlo."""
pass
# Reset coordinado UI <-> frida_thread:
# - RESET_LOCK por half: evita 2 resets simultáneos en la misma mitad.
# - EXTERNAL_RESET[half]: cuando la UI dispara un reset, lo pone en True ANTES
# de empezar y a False al terminar. El frida_thread, al ver True, se abstiene
# de hacer su propia escalación y solo reintenta el attach (la UI ya está
# haciendo el trabajo pesado de force-stop/relanzar).
RESET_LOCK = {"W": threading.Lock(), "E": threading.Lock()}
EXTERNAL_RESET = {"W": False, "E": False}
def _hard_reset_impl(serial, half):
"""Force-stop Evony + relanzar frida-server. ~40s. Idempotente."""
import subprocess
print(f"[{half}] HARD-RESET (UI): force-stop Evony + frida-server", flush=True)
STATE[half]["sweep"] = "hard-reset"
try:
pids = subprocess.run(["adb", "-s", serial, "shell", "pidof", "frida-server"],
capture_output=True, text=True, timeout=5).stdout.strip().split()
for pid in pids:
if pid: subprocess.run(["adb", "-s", serial, "shell", "kill", "-9", pid], timeout=5)
subprocess.run(["adb", "-s", serial, "shell", "am", "force-stop", "com.topgamesinc.evony"], timeout=10)
time.sleep(3)
subprocess.run(["adb", "-s", serial, "root"], timeout=10)
time.sleep(3)
subprocess.run(["adb", "-s", serial, "shell",
"nohup", "/data/local/tmp/frida-server", ">/dev/null", "2>&1", "&"],
timeout=5)
time.sleep(3)
subprocess.run(["adb", "-s", serial, "shell", "am", "start", "-n",
"com.topgamesinc.evony/com.topgamesinc.androidplugin.UnityActivity"],
timeout=10)
print(f"[{half}] HARD-RESET (UI) aplicado, esperando 35s a Evony", flush=True)
time.sleep(35)
print(f"[{half}] HARD-RESET (UI) completado, esperando reattach del watchdog", flush=True)
except Exception as e:
print(f"[{half}] HARD-RESET (UI) fail: {e}", flush=True)
def _nuclear_reset_impl(serial, half):
"""Kill qemu + relanzar AVD entero. ~3min. Idempotente."""
import subprocess
avd, port = NUCLEAR_AVD[half]
print(f"[{half}] ☠ NUCLEAR-RESET (UI): kill qemu pid + relanzar AVD {avd}", flush=True)
STATE[half]["sweep"] = "nuclear-reset"
try:
r = subprocess.run(["sh", "-c",
f"ps -A -o pid,command | grep 'qemu-system' | grep -- '-port {port}' | grep -v grep | awk '{{print $1}}'"],
capture_output=True, text=True, timeout=5)
qpid = (r.stdout or "").strip().split("\n")[0].strip()
if qpid:
subprocess.run(["kill", "-9", qpid], timeout=5)
print(f"[{half}] qemu pid {qpid} killed", flush=True)
time.sleep(5)
# NO hacemos adb kill-server aquí: afectaría a la OTRA mitad. El emulador
# nuevo se re-registrará solo cuando arranque.
subprocess.Popen([EMULATOR_BIN, "-avd", avd, "-port", str(port),
"-no-snapshot", "-writable-system", "-no-boot-anim",
"-gpu", "auto", "-memory", "4096",
"-partition-size", "8192", "-no-audio"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
print(f"[{half}] esperando 180s a boot del emulador...", flush=True)
time.sleep(180)
subprocess.run(["adb", "-s", serial, "root"], timeout=15)
time.sleep(5)
for _ in range(10):
try:
r2 = subprocess.run(["adb", "-s", serial, "shell", "echo", "ready"],
capture_output=True, text=True, timeout=5)
if "ready" in (r2.stdout or ""): break
except Exception: pass
time.sleep(2)
subprocess.run(["adb", "-s", serial, "shell",
"nohup", "/data/local/tmp/frida-server", ">/dev/null", "2>&1", "&"],
timeout=5)
time.sleep(4)
subprocess.run(["adb", "-s", serial, "shell", "am", "start", "-n",
"com.topgamesinc.evony/com.topgamesinc.androidplugin.UnityActivity"],
timeout=10)
print(f"[{half}] esperando 35s a Evony tras nuclear...", flush=True)
time.sleep(35)
print(f"[{half}] ☠ NUCLEAR-RESET (UI) completado", flush=True)
except Exception as e:
print(f"[{half}] ☠ NUCLEAR-RESET (UI) fail: {e}", flush=True)
def _run_reset(half, kind):
"""Ejecuta hard o nuclear reset en un thread separado, coordinado con frida_thread.
Si ya hay un reset corriendo en esta mitad, no hace nada (no se acumulan)."""
if not RESET_LOCK[half].acquire(blocking=False):
print(f"[{half}] reset ya en curso, ignorando duplicado ({kind})", flush=True)
return False
def worker():
try:
EXTERNAL_RESET[half] = True
REATTACH_FAILS[half] = 0 # limpia contador para que frida_thread no escale
serial = next(s for s, h in SCANNERS if h == half)
if kind == "nuclear":
_nuclear_reset_impl(serial, half)
else:
_hard_reset_impl(serial, half)
# tras el reset, fuerza al watchdog del frida_thread a reattachar inmediato
LAST_HEARTBEAT[half] = 0.0
finally:
EXTERNAL_RESET[half] = False
RESET_LOCK[half].release()
threading.Thread(target=worker, daemon=True, name=f"reset-{half}-{kind}").start()
return True
def frida_thread(serial, half):
# agentes PREcompilados por mitad (no se puede editar el bundle de
# frida-compile en runtime: tiene manifiesto de offsets -> "malformed package").
agent_path = os.path.join(HERE, f"agent_{half}.js")
with open(agent_path) as f:
src = f.read()
attempt = 0
while True:
attempt += 1
try:
# setup serializado: dos hilos haciendo attach/load a la vez en el
# core de frida corrompe el transporte ("malformed package").
with FRIDA_SETUP_LOCK:
dev = frida.get_device(serial, timeout=30)
# attach by name evita el enumerate_processes() que estaba timeando.
try:
session = dev.attach("Evony")
except frida.ProcessNotFoundError:
# Evony no corre: contar como fallo para que la escalación HARD/NUCLEAR
# (que sí lanza Evony) se dispare en lugar de quedar en bucle eterno.
REATTACH_FAILS[half] += 1
print(f"[{half}] Evony no corre en {serial}; fails={REATTACH_FAILS[half]}", flush=True)
time.sleep(5)
# FALL THROUGH al bloque de thresholds al final del while
raise _NeedsEscalation()
# create_script falla a veces con TransportError porque transferir 167KB
# via adb forward es lento. Retry 3 veces; si todas fallan -> escalar
# (probablemente frida-server degradado, sólo HARD/NUCLEAR lo arregla).
script = None
for cs_try in range(3):
try:
script = session.create_script(src)
break
except frida.TransportError as e:
wait = 3 + cs_try * 2
print(f"[{half}] create_script try {cs_try+1} fail ({e}); retry {wait}s", flush=True)
time.sleep(wait)
if script is None:
# 3 fallos seguidos de create_script == frida-server degradado.
# Marcar como fallo grave (suma HARD_RESET_THRESHOLD) para forzar
# HARD-RESET inmediato en lugar de seguir reintentando attach inútilmente.
REATTACH_FAILS[half] += HARD_RESET_THRESHOLD
print(f"[{half}] create_script falló 3 veces -> escalando a HARD-RESET (fails={REATTACH_FAILS[half]})", flush=True)
try: session.detach()
except Exception: pass
raise _NeedsEscalation()
script.on("message", lambda m, d, h=half: on_message(m, d, h))
try:
script.load(timeout=120)
except TypeError:
script.load()
SCRIPTS[half] = script # ref para pause/resume via .post
# si la UI ya pidio pausa antes de que arrancara este escaner, propaga
if SCANNER_PAUSED:
try: script.post({"type": "ctl", "cmd": "pause"})
except Exception: pass
LAST_PASS_TS[half] = time.time() # base para medir la 1a vuelta
STATE[half]["sweep"] = "running"
STATE[half]["num"] = 0
STATE[half]["last_secs"] = None
LAST_HEARTBEAT[half] = time.time()
REATTACH_FAILS[half] = 0 # éxito: resetea contador de fallos
print(f"[{half}] frida activo en {serial} (intento {attempt})", flush=True)
# lock liberado: el otro escaner puede hacer su setup; los mensajes
# llegan por el hilo interno de frida, no por este.
while True:
time.sleep(5)
hb = LAST_HEARTBEAT.get(half, 0)
age = time.time() - hb
if age > HEARTBEAT_TIMEOUT:
# hb=0 es señal de "reset forzado externamente": no mostramos
# el delta absurdo (~1.7e9s = epoch). Solo decimos "forzado".
age_str = "(forzado)" if hb == 0 else f"hace {int(age)}s"
print(f"[{half}] watchdog: sin heartbeat {age_str}, forzando reattach", flush=True)
# NO llamar unload/detach: pueden colgarse si la sesión está corrupta.
# Frida limpiará la sesión zombi cuando hagamos attach nuevo.
SCRIPTS[half] = None
STATE[half]["sweep"] = "reattach"
REATTACH_FAILS[half] += 1
break
except _NeedsEscalation:
# Señal explícita desde dentro del bloque para caer al threshold check
# (REATTACH_FAILS ya fue incrementado por quien lanzó la excepción).
STATE[half]["sweep"] = "reconectando"
except frida.TransportError as e:
REATTACH_FAILS[half] += 1
print(f"[{half}] TransportError ({e}); fails={REATTACH_FAILS[half]}", flush=True)
STATE[half]["sweep"] = "reconectando"
time.sleep(4)
except Exception as e:
REATTACH_FAILS[half] += 1
print(f"[{half}] error: {e}; fails={REATTACH_FAILS[half]}", flush=True)
STATE[half]["sweep"] = "reconectando"
time.sleep(5)
# Si la UI disparó un reset externo, NO hagas escalación propia:
# _run_reset() ya está haciendo el trabajo desde otro thread.
if EXTERNAL_RESET[half]:
print(f"[{half}] reset externo en curso, frida_thread espera...", flush=True)
while EXTERNAL_RESET[half]:
time.sleep(2)
time.sleep(3) # un margen extra para que Evony termine de arrancar
continue
# NUCLEAR RESET AUTOMÁTICO: si N≥NUCLEAR_RESET_THRESHOLD fallos seguidos,
# ni los hard-resets funcionan (qemu colgado a nivel proceso). Solución:
# kill -9 al proceso qemu + relanzar el emulador entero. ~3 min total.
if REATTACH_FAILS[half] >= NUCLEAR_RESET_THRESHOLD:
avd, port = NUCLEAR_AVD[half]
print(f"[{half}] ☠ NUCLEAR-RESET: {REATTACH_FAILS[half]} fallos -> kill qemu pid + relanzar AVD {avd}", flush=True)
STATE[half]["sweep"] = "nuclear-reset"
try:
import subprocess
# 1) encontrar el qemu del puerto
r = subprocess.run(["sh", "-c",
f"ps -A -o pid,command | grep 'qemu-system' | grep -- '-port {port}' | grep -v grep | awk '{{print $1}}'"],
capture_output=True, text=True, timeout=5)
qpid = (r.stdout or "").strip().split("\n")[0].strip()
if qpid:
subprocess.run(["kill", "-9", qpid], timeout=5)
print(f"[{half}] qemu pid {qpid} killed", flush=True)
time.sleep(5)
# 2) limpiar adb (registrará device offline)
subprocess.run(["adb", "kill-server"], timeout=10)
time.sleep(3)
subprocess.run(["adb", "start-server"], timeout=10)
time.sleep(3)
# 3) relanzar emulador
subprocess.Popen([EMULATOR_BIN, "-avd", avd, "-port", str(port),
"-no-snapshot", "-writable-system", "-no-boot-anim",
"-gpu", "auto", "-memory", "4096",
"-partition-size", "8192", "-no-audio"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
print(f"[{half}] esperando 180s a boot del emulador...", flush=True)
time.sleep(180)
# 4) adb root (CRÍTICO: frida-server necesita root para acceder al proceso Evony)
subprocess.run(["adb", "-s", serial, "root"], timeout=15)
time.sleep(5) # tiempo para que adb se reinicie como root
# esperar a que el dispositivo vuelva tras 'adb root' (puede tardar)
for _ in range(10):
try:
r2 = subprocess.run(["adb", "-s", serial, "shell", "echo", "ready"],
capture_output=True, text=True, timeout=5)
if "ready" in (r2.stdout or ""): break
except Exception: pass
time.sleep(2)
subprocess.run(["adb", "-s", serial, "shell",
"nohup", "/data/local/tmp/frida-server", ">/dev/null", "2>&1", "&"],
timeout=5)
time.sleep(4)
# 5) lanzar Evony
subprocess.run(["adb", "-s", serial, "shell", "am", "start", "-n",
"com.topgamesinc.evony/com.topgamesinc.androidplugin.UnityActivity"],
timeout=10)
print(f"[{half}] esperando 35s a Evony...", flush=True)
time.sleep(35)
REATTACH_FAILS[half] = 0
except Exception as e:
print(f"[{half}] ☠ NUCLEAR-RESET fail: {e}", flush=True)
time.sleep(30)
continue # vuelve al while True externo para reintentar attach
# HARD RESET AUTOMÁTICO: si ya hemos fallado N veces seguidas, algo más
# profundo está mal. Force-stop Evony + relaunch + kill frida-server.
# Esto soluciona el caso "Evony en estado degradado tras horas".
if REATTACH_FAILS[half] >= HARD_RESET_THRESHOLD:
print(f"[{half}] HARD-RESET: {REATTACH_FAILS[half]} fallos seguidos -> force-stop Evony + frida-server", flush=True)
STATE[half]["sweep"] = "hard-reset"
try:
# kill frida-server local (solo del emulador relevante)
import subprocess
pids = subprocess.run(["adb", "-s", serial, "shell", "pidof", "frida-server"],
capture_output=True, text=True, timeout=5).stdout.strip().split()
for pid in pids:
if pid: subprocess.run(["adb", "-s", serial, "shell", "kill", "-9", pid], timeout=5)
# force-stop Evony
subprocess.run(["adb", "-s", serial, "shell", "am", "force-stop", "com.topgamesinc.evony"], timeout=10)
time.sleep(3)
# CRÍTICO: garantizar adb root antes de lanzar frida-server
subprocess.run(["adb", "-s", serial, "root"], timeout=10)
time.sleep(3)
# relaunch frida-server
subprocess.run(["adb", "-s", serial, "shell",
"nohup", "/data/local/tmp/frida-server", ">/dev/null", "2>&1", "&"],
timeout=5)
time.sleep(3)
# relaunch Evony
subprocess.run(["adb", "-s", serial, "shell", "am", "start", "-n",
"com.topgamesinc.evony/com.topgamesinc.androidplugin.UnityActivity"],
timeout=10)
print(f"[{half}] HARD-RESET aplicado, esperando 35s a Evony", flush=True)
time.sleep(35)
# NO reset el contador aquí: si el reattach posterior falla, el
# contador sigue subiendo y eventualmente escala a NUCLEAR (>=6).
# Solo se resetea al lograr `frida activo` exitosamente.
except Exception as e:
print(f"[{half}] HARD-RESET fail: {e}", flush=True)
time.sleep(10)
# ---------------- API helpers ----------------
def enrich(o):
c = CFG.get(str(o["id"])) or {}
now = time.time()
ft = o.get("first_ts", o.get("ts", now))
# resaltado SOLO si: ya termino la 1a vuelta, el objeto se descubrio DESPUES
# de esa 1a vuelta, y lleva <=5 min desde su primera deteccion.
fresh = (FIRST_PASS_TS is not None
and ft > FIRST_PASS_TS
and (now - ft) <= FRESH_WINDOW)
# estado de ataque/rally REAL: viene del hook ServerMapLayer.addTargetObj que
# captura MsgDown.map_target_info (march en curso). Si existe march activa
# hacia (wx,wy) con end_time > now, hay ataque/rally de verdad.
# ataque/rally REAL via hook ServerMapLayer.addTargetObj. Solo cuentan
# marches cuyo `mty` (Msg.march_type) sea de ATAQUE a monstruo:
# 2 = monster (solo de monarca)
# 19 = boss_war_wait (alianza: miembros uniendose)
# 20 = boss_war_way (alianza: rally marchando)
# 21 = boss_war (alianza: en combate)
# 43 = scout_monster (exploracion del monstruo)
# Otros mty (reap_resource=7, cargo=49, etc.) son ruido y se filtran.
atk = ""; atk_uid = 0; atk_guild = 0; atk_eta = 0; atk_mty = 0
atk_name = ""; atk_tag = ""; atk_count = 0
ATTACK_MTYS = {2, 19, 20, 21, 43}
ALLY_MTYS = {19, 20, 21}
key = (int(o["wx"]), int(o["wy"]))
subs = MARCH_BY_TGT.get(key)
if subs:
rel = []
for trp in list(subs.keys()):
m = MARCHES.get(trp)
if not m: continue
if int(m.get("mty", 0)) not in ATTACK_MTYS: continue
if int(m.get("te", 0)) <= now: continue
rel.append(m)
if rel:
ally = [m for m in rel if int(m.get("mty", 0)) in ALLY_MTYS]
if ally:
# rally de alianza: elegimos representante priorizando mty=21
# (en combate) > 20 (marchando) > 19 (formandose).
by = {}
for m in ally: by.setdefault(int(m["mty"]), []).append(m)
if 21 in by: rep = min(by[21], key=lambda m: int(m["te"]))
elif 20 in by: rep = min(by[20], key=lambda m: int(m["te"]))
else: rep = min(by[19], key=lambda m: int(m["te"]))
atk = "rally"
atk_count = len(ally)
else:
rep = max(rel, key=lambda m: int(m["te"]))
atk = "rally"
atk_count = 1
atk_uid = int(rep.get("ow", 0))
atk_mty = int(rep.get("mty", 0))
atk_eta = max(0, int(rep.get("te", 0) - now))
pl = PLAYERS.get(str(atk_uid)) or PLAYERS.get(atk_uid)
if pl:
atk_guild = int(pl.get("gid", 0) or 0)
atk_name = pl.get("name", "") or ""
atk_tag = pl.get("tag", "") or ""
# legado: campos antiguos del agente (st/hp/oid) — los seguimos exponiendo
# para que el tooltip los muestre, pero NO se usan para derivar `atk` ya.
oid = int(o.get("oid", 0) or 0)
ogd = int(o.get("ogd", 0) or 0)
st = int(o.get("st", 0) or 0)
hp = int(o.get("hp", 0) or 0)
hpt = int(o.get("hpt", 0) or 0)
if not atk_uid and oid > 0 and o["t"] in (3, 4, 5, 7):
# mapinfo_type: farm/garrison/ruins/subcity -> ocupacion de territorio
atk = "owned"; atk_uid = oid; atk_guild = ogd
pl = PLAYERS.get(str(oid)) or PLAYERS.get(oid)
if pl:
atk_name = pl.get("name", "") or ""
atk_tag = pl.get("tag", "") or ""
return {
"name": c.get("name", f"id{o['id']}"),
"level": o["lv"],
"type": o["t"],
"power": c.get("power", 0),
"x": o["wx"], "y": o["wy"], "id": o["id"],
"group": group_of(c.get("name", "")),
"age": int(now - ft),
"seen_age": int(now - o.get("ts", now)),
"fresh": fresh,
"half": o.get("half", ""), # scanner que lo capturó (W o E)
"atk": atk, "atk_uid": atk_uid, "atk_guild": atk_guild,
"atk_eta": atk_eta, "atk_mty": atk_mty,
"atk_name": atk_name, "atk_tag": atk_tag, "atk_count": atk_count,
"atk_hp": hp, "atk_hpt": hpt, "atk_st": st,
}
# Familias de rally-boss de Evony (match por substring, case-insensitive)
BOSS_KEYWORDS = [
"cerberus", "sphinx", "bayard", "bayar knight", "lava turtle", "hydra",
"ymir", "pan ", "harpy", "warlord", "witch", "golem", "mire squid",
"pumpkin", "ghidorah", "nine-tails", "nine tails", "centaur", "minotaur",
"peryton", "redcap", "skeleton dragon", "fafnir", "kamaitachi", "ifrit",
"griffin", "behemoth", "manticore", "werewolf", "yasha", "typhon",
"jormungandr", "phoenix", "gluttonous", "viking", "fire spirit",
"golden goblin", "arctic barbarian",
]
def is_boss(name):
n = (name or "").lower()
if "(boss)" in n:
return True
return any(k in n for k in BOSS_KEYWORDS)
# Familias rapidas (display -> substrings que matchean el nombre)
# (display, keywords, grupo) grupo: "boss" | "event"
FAMILIES = [
("Cerberus", ["cerberus"], "boss"),
("Knight Bayard", ["bayard", "bayar knight"], "boss"),
("Harpy", ["harpy"], "boss"),
("Hydra", ["hydra"], "boss"),
("Warlord", ["warlord"], "boss"),
("Ghidorah", ["ghidorah"], "boss"),
("Nine-tails", ["nine-tails", "nine tails"], "boss"),
("Bosses (Boss)", ["(boss)"], "boss"),
("Ymir", ["ymir"], "event"),
("Pan", ["pan "], "event"),
("Lava Turtle", ["lava turtle"], "event"),
("Sphinx", ["sphinx"], "event"),
("Witch", ["witch"], "event"),
("Golem", ["golem"], "event"),
("Mire Squid", ["mire squid"], "event"),
("Pumpkin", ["pumpkin"], "event"),
("Viking", ["viking"], "event"),
("Golden Goblin", ["golden goblin"], "event"),
("Arctic Barbarian",["arctic barbarian"], "event"),
("Fire Spirit", ["fire spirit"], "event"),
]
def fam_match(name, kws):
n = (name or "").lower()
return any(k in n for k in kws)
def group_of(name):
"""Clasifica un monstruo por su nombre de config: Boss / Event / Normal."""
n = (name or "").lower()
for disp, kws, grp in FAMILIES:
if any(k in n for k in kws):
return "Boss" if grp == "boss" else "Event"
if "(boss)" in n:
return "Boss"
return "Normal"
def respawn_predictions(min_samples=2, only_empty=True,
ids=None, fams=None, allow_normal=True):
"""Devuelve [{x, y, id, name, lv, last_spawn_ts, median_cycle_s,
samples, predicted_in_s, confidence, group}] ordenado por predicted_in_s asc.
Filtros:
- min_samples: ≥2 intervalos (≥3 spawns) para tener mediana fiable.
- only_empty: solo tiles actualmente vacíos (los vivos aún no han muerto).
- ids: set de monster_ids; si no vacío, solo esos.
- fams: set de nombres de familia (display name de FAMILIES); si no vacío,
solo monstruos cuyo nombre matchee alguna keyword de esas familias.
- allow_normal: si False, descarta el grupo "Normal" (monstruos no-Boss no-Event).
ids y fams se combinan en OR (igual que en query()).
"""
ids = set(ids) if ids else set()
fams = set(fams) if fams else set()
# expand fams -> ids via fam_ids (reusa la función ya existente)
if fams:
ids |= fam_ids(fams)
filter_by_id = bool(ids) or bool(fams)
now = time.time()
out = []
with LOCK:
for (wx, wy), hist in SPAWN_HISTORY.items():
if len(hist) < 2:
continue
# spawn más reciente
last = hist[-1]
id_, lv = int(last.get("id", 0)), int(last.get("lv", 0))
if id_ <= 0:
continue
# intervalos consecutivos
deltas = [hist[i+1]["ts"] - hist[i]["ts"] for i in range(len(hist)-1)]
deltas = [d for d in deltas if 60 <= d <= 86400] # filtra outliers absurdos
if not deltas:
continue
deltas_sorted = sorted(deltas)
median = deltas_sorted[len(deltas_sorted)//2]
# reliability: 'high' >=3 intervals & std<30%; 'medium' >=3; 'low' rest
mean = sum(deltas)/len(deltas)
var = sum((d-mean)**2 for d in deltas)/len(deltas)
std = var**0.5
cv = std/mean if mean > 0 else 1.0
if len(deltas) >= 3 and cv < 0.30: conf = "high"
elif len(deltas) >= 3: conf = "medium"
else: conf = "low"
# tile alive?
o = OBJS.get((wx, wy))
alive = o is not None and (now - o.get("ts", 0)) <= STALE_SECONDS \
and int(o.get("id", 0)) == id_ and int(o.get("lv", 0)) == lv
if only_empty and alive:
continue
next_spawn_ts = last["ts"] + median
predicted_in_s = next_spawn_ts - now
cfg = CFG.get(str(id_)) or {}
name = cfg.get("name", "") or f"id{id_}"
grp = group_of(name)
# filtros server-side
if filter_by_id and id_ not in ids:
continue
if not allow_normal and grp == "Normal":
continue
out.append({
"x": wx, "y": wy, "id": id_, "lv": lv, "name": name,
"group": grp, # "Boss" | "Event" | "Normal" — para badge UI
"last_spawn_ts": int(last["ts"]),
"last_spawn_age_s": int(now - last["ts"]),
"median_cycle_s": int(median),
"samples": len(deltas),
"predicted_in_s": int(predicted_in_s),
"predicted_ts": int(next_spawn_ts),
"confidence": conf,
"alive": alive,
})
# ordenar: los que ya deberían haber respawneado primero, luego los más cercanos
out.sort(key=lambda r: r["predicted_in_s"])
return out
def relocations_recent(max_age_h=24, near=None):
"""Devuelve [{uid, name, tag, gid, ts, age_s, from:[x,y], to:[x,y],
distance_km, hops_total}] ordenado por ts desc.
- max_age_h: descarta eventos más viejos que esto.
- near: tupla (cx, cy, radius_tiles) opcional; si se da, solo eventos
cuyo destino (to) caiga dentro del radio.
"""
import math
now = time.time()
cutoff = now - max_age_h * 3600
out = []
with LOCK:
for uid, hist in RELOCATIONS.items():
for ev in hist:
if ev.get("ts", 0) < cutoff:
continue
tx, ty = int(ev.get("to_x", 0)), int(ev.get("to_y", 0))
if near is not None:
cx, cy, rad = near
if math.hypot(tx - cx, ty - cy) > rad:
continue
# distancia del salto (en km como en el resto de la UI)
fx, fy = int(ev.get("from_x", 0)), int(ev.get("from_y", 0))
hop_dist = int(math.hypot(tx - fx, ty - fy))
out.append({
"uid": uid,
"name": ev.get("name", ""),
"tag": ev.get("tag", ""),
"gid": int(ev.get("gid", 0) or 0),
"ts": int(ev.get("ts", 0)),
"age_s": int(now - ev.get("ts", 0)),
"from": [fx, fy], "to": [tx, ty],
"hop_dist": hop_dist,
"hops_total": len(hist),
})
out.sort(key=lambda r: r["ts"], reverse=True)
return out
def families(max_seen=None):
"""[{name, variants, live}] — variants=tiers en config, live=cuantos en el mapa
según el mismo filtro de max_seen que /api/data. Sin esto, el pill cuenta el
cache acumulado de 24h (persistencia) mientras la tabla solo muestra los del
último ciclo (default STALE_SECONDS=90s), causando discrepancia visual."""
cat = catalog()
if max_seen is None: max_seen = STALE_SECONDS
now = time.time()
with LOCK:
objs = [o for o in OBJS.values() if (now - o.get("ts", 0)) <= max_seen]
live = {}
for o in objs:
c = CFG.get(str(o["id"])) or {}
nm = c.get("name", "")
if not nm:
continue
for disp, kws, grp in FAMILIES:
if fam_match(nm, kws):
live[disp] = live.get(disp, 0) + 1
out = []
for disp, kws, grp in FAMILIES:
v = sum(1 for x in cat if fam_match(x["name"], kws))
if v:
out.append({"name": disp, "variants": v, "live": live.get(disp, 0), "group": grp})
return out
def fam_ids(selected_fams):
"""ids de config cuyos nombres caen en alguna familia seleccionada."""
allkw = []
for disp, kws, grp in FAMILIES:
if disp in selected_fams:
allkw += kws
if not allkw:
return set()
ids = set()
with LOCK:
for cid, c in CFG.items():
if fam_match(c.get("name", ""), allkw):
ids.add(int(cid))
return ids
def catalog():
"""Lista de configs de rally-boss: [{id,name,level,type,power,label}] ordenada."""
with LOCK:
items = []
for cid, c in CFG.items():
nm = c.get("name", "")
if not nm or not is_boss(nm):
continue
lv = c.get("level", 0)
items.append({
"id": int(cid), "name": nm, "level": lv,
"type": c.get("type", 0), "power": c.get("power", 0),
"label": f"{nm} {lv}" if lv else nm,
})
items.sort(key=lambda r: (r["name"], r["level"]))
return items
def query(params):
ids = set()
for s in params.get("ids", "").split(","):
s = s.strip()
if s.isdigit():
ids.add(int(s))
fams = [f for f in params.get("fams", "").split("||") if f]
if fams:
ids |= fam_ids(set(fams))
try: cx = float(params.get("cx", "")) if params.get("cx") else None
except: cx = None
try: cy = float(params.get("cy", "")) if params.get("cy") else None
except: cy = None
try: rad = float(params.get("radius", "")) if params.get("radius") else None
except: rad = None
only_type = params.get("type", "") # "" o "2" etc
sort = params.get("sort", "newest")
# NOTA: query() devuelve la lista COMPLETA ordenada/filtrada. El paginado
# (limit + page) lo hace el handler HTTP /api/data, no esta funcion.
# max_seen: objetos vistos por ultima vez hace <= N seg.
# Default = STALE_SECONDS: solo lo que sigue existiendo en el mapa ahora mismo
# (lo que ya no se re-detecta en ~3-4 vueltas desaparece de la lista).
try: max_seen = float(params["max_seen"]) if params.get("max_seen") else STALE_SECONDS
except: max_seen = STALE_SECONDS
with LOCK:
rows = [enrich(o) for o in OBJS.values()]
out = []
for r in rows:
if only_type and str(r["type"]) != only_type:
continue
if ids and r["id"] not in ids:
continue
if max_seen is not None and r["seen_age"] > max_seen:
continue
if cx is not None and cy is not None and rad is not None:
d = math.hypot(r["x"] - cx, r["y"] - cy)
if d > rad:
continue
r["dist"] = round(d)
else:
r["dist"] = None
out.append(r)
# order: "desc" (default natural — level/power grande primero, newest reciente
# primero) o "asc" (invertido). El frontend usa esto para los headers clicables.
order = (params.get("order", "desc") or "desc").lower()
rev = (order == "asc") # invertimos los defaults
if sort == "level":
out.sort(key=lambda r: (r["level"], r["name"]) if rev else (-r["level"], r["name"]))
elif sort == "power":
out.sort(key=lambda r: r["power"] if rev else -r["power"])
elif sort == "dist" and cx is not None:
out.sort(key=lambda r: (r["dist"] if r["dist"] is not None else 9e9), reverse=rev)
elif sort == "newest":
# default: age asc (más reciente primero). asc invertido = más viejos primero
out.sort(key=lambda r: r["age"], reverse=rev)
return out
def names_list():
with LOCK:
seen = {}
for o in OBJS.values():
c = CFG.get(str(o["id"])) or {}
nm = c.get("name", f"id{o['id']}")
seen[nm] = seen.get(nm, 0) + 1
return sorted(seen.items(), key=lambda x: (-x[1], x[0]))
def query_players(params):
now = time.time()
def num(k):
v = params.get(k, "")
try: return float(v) if v not in ("", None) else None
except: return None
lv_min = num("lv_min"); lv_max = num("lv_max")
pw_min = num("pw_min"); pw_max = num("pw_max") # en M (millones)
shield = params.get("shield", "") # "", "yes", "no"
tag = params.get("tag", "").strip().lower()
# Buscador libre: match parcial case-insensitive contra name/uid/tag
# Si es un numero, tambien intenta match exacto contra uid.
q = params.get("q", "").strip().lower()
q_int = None
if q:
try: q_int = int(q)
except: pass
cx, cy, rad = num("cx"), num("cy"), num("radius")
sort = params.get("sort", "power")
limit = int(params.get("limit", "300"))
max_seen = num("max_seen") or TTL_SECONDS
with LOCK:
ps = list(PLAYERS.values())
out = []
for p in ps:
if now - p.get("ts", 0) > max_seen:
continue
lv = p.get("lv", 0)
# power: el broadcast lleva 0; preferimos PLAYER_POWER[uid] (de power_rank_reply).
pw_src = ""
pw = int(p.get("power", 0) or 0)
pp_info = PLAYER_POWER.get(p["uid"])
if pp_info and pp_info.get("power", 0) > 0:
pw = int(pp_info["power"])
pw_src = "power_rank"
elif pw > 0:
pw_src = "broadcast"
pwM = pw / 1e6
has_shield = bool(p.get("shield", 0))
if lv_min is not None and lv < lv_min: continue
if lv_max is not None and lv > lv_max: continue
if pw_min is not None and pwM < pw_min: continue
if pw_max is not None and pwM > pw_max: continue
if shield == "yes" and not has_shield: continue
if shield == "no" and has_shield: continue
if tag and tag not in (p.get("tag", "") or "").lower(): continue
# Search libre: match si q aparece en name (parcial), tag (parcial), o uid (exacto si q es entero)
if q:
nm_lo = (p.get("name", "") or "").lower()
tg_lo = (p.get("tag", "") or "").lower()
uid_match = (q_int is not None and int(p.get("uid", 0)) == q_int)
if not (q in nm_lo or q in tg_lo or uid_match):
continue
d = None
if cx is not None and cy is not None and rad is not None:
d = math.hypot(p["wx"] - cx, p["wy"] - cy)
if d > rad: continue
d = round(d)
# ETA exacta de burbuja. Tres caches posibles, por prioridad:
# 1) SHIELD_ETA[uid] -> match directo (subcity / scout / psr ya matcheado)
# 2) SHIELD_ETA["coord:wx,wy"] -> psr capturado antes de tener uid en PLAYERS
# (al encontrar uid ahora, migramos el cache de coord a uid)
sh_eta = 0; sh_src = ""
eta_info = SHIELD_ETA.get(p["uid"])
if eta_info is None:
ckey = f"coord:{p['wx']},{p['wy']}"
eta_info = SHIELD_ETA.get(ckey)
if eta_info is not None:
# migra coord -> uid: ahora que conocemos el uid, lo movemos
SHIELD_ETA[p["uid"]] = eta_info
SHIELD_ETA.pop(ckey, None)
sh_confidence = ""; sh_activation_ts = 0
# Sanity check: broadcast tier es ground truth. Si tier=0 + confidence!=exact, descartamos eta.
# Esto cubre el caso de cache stale donde no hubo transition que invalidara la entry.
if eta_info and int(p.get("shield", 0) or 0) == 0 and eta_info.get("confidence") != "exact":
SHIELD_ETA.pop(p["uid"], None)
eta_info = None
if eta_info:
et = int(eta_info.get("end_time", 0) or 0)
if et > int(now):
sh_eta = et - int(now)
sh_src = eta_info.get("src", "")
sh_confidence = eta_info.get("confidence", "")
sh_activation_ts = int(eta_info.get("activation_ts", 0) or 0)
else:
# caducado -> limpia lazy
SHIELD_ETA.pop(p["uid"], None)
out.append({
"name": p.get("name", ""), "level": lv, "castle": p.get("clv", 0),
"power": pw, "power_src": pw_src, "tag": p.get("tag", ""),
"shield": int(p.get("shield", 0) or 0), # tier code (0=ninguno, 1=item, 2=newbie)
"shield_eta": sh_eta, # >0 = segundos restantes; 0 desconocida
"shield_src": sh_src, # "subcity" | "scout" | "tier_transition" | ...
"shield_confidence": sh_confidence, # "exact" | "inferred_default" | "inferred_history" | "newbie_inferred"
"shield_activation_ts": sh_activation_ts, # unix ts de cuando se activo (si conocido)
"shield_history_n": len(SHIELD_HISTORY.get(p["uid"], [])), # cuantas muestras de historico hay
"x": p["wx"], "y": p["wy"], "uid": p["uid"], "dist": d,
"age": int(now - p.get("ts", now)),
})
if sort == "power": out.sort(key=lambda r: -r["power"])
elif sort == "level": out.sort(key=lambda r: (-r["level"], -r["power"]))
elif sort == "dist": out.sort(key=lambda r: (r["dist"] if r["dist"] is not None else 9e9))
elif sort == "newest":out.sort(key=lambda r: r["age"])
elif sort == "eta":
# ascendente por ETA: los que tienen burbuja exacta (eta>0) primero (los que
# expiran antes a la cabeza); el resto al final. Util para anticipar ataques.
out.sort(key=lambda r: (r.get("shield_eta", 0) <= 0, r.get("shield_eta", 0) or 9e9))
return out[:limit]
# ---------------- HTTP ----------------
HTML = """
Evony Scout · Dual
Auto-add applies rules every 60s on PLAYERS. Manual = clicked ⭐. AUTO = added by rules.
Top-right banner triggers when a watchlist shield expires in <10min.
Player
Alliance
Castle
Coords
Shield
Confidence
Src
Note
Action
"""
class H(BaseHTTPRequestHandler):
def log_message(self, *a): pass
def _send(self, code, body, ctype="application/json"):
b = body.encode() if isinstance(body, str) else body
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(b)))
self.send_header("Cache-Control", "no-store, no-cache, must-revalidate")
self.send_header("Pragma", "no-cache")
self.end_headers()
self.wfile.write(b)
def do_GET(self):
from urllib.parse import urlparse, parse_qs
u = urlparse(self.path)
if u.path == "/" or u.path == "/index.html":
return self._send(200, HTML, "text/html; charset=utf-8")
if u.path == "/api/names":
return self._send(200, json.dumps(names_list()))
if u.path == "/api/catalog":
return self._send(200, json.dumps(catalog()))
if u.path == "/api/families":
params = {k: v[0] for k, v in parse_qs(u.query).items()}
try: ms = float(params["max_seen"]) if params.get("max_seen") else None
except: ms = None
return self._send(200, json.dumps(families(max_seen=ms)))
if u.path == "/api/data":
params = {k: v[0] for k, v in parse_qs(u.query).items()}
all_rows = query(params)
try: limit = int(params.get("limit", "500"))
except: limit = 500
try: page = max(1, int(params.get("page", "1")))
except: page = 1
total = len(all_rows)
if limit <= 0:
page, pages, rows = 1, 1, all_rows
else:
pages = max(1, (total + limit - 1) // limit)
page = min(page, pages)
rows = all_rows[(page-1)*limit : page*limit]
with LOCK:
alln = len(OBJS)
return self._send(200, json.dumps({
"rows": rows, "total": total, "all": alln,
"page": page, "pages": pages, "limit": limit,
"cfg_n": STATE["cfg_n"],
"sw_W": STATE["W"], "sw_E": STATE["E"]}))
if u.path == "/api/players":
params = {k: v[0] for k, v in parse_qs(u.query).items()}
rows = query_players(params)
with LOCK:
alln = len(PLAYERS)
return self._send(200, json.dumps({
"rows": rows, "total": len(rows), "all": alln,
"sw_W": STATE["W"], "sw_E": STATE["E"]}))
if u.path == "/api/watchlist":
# Devuelve uids del watchlist con su data live (player + shield ETA).
now = time.time(); now_i = int(now)
out = []
with LOCK:
for uid in sorted(WATCHLIST):
pl = PLAYERS.get(uid) or {}
eta_info = SHIELD_ETA.get(uid) or {}
et = int(eta_info.get("end_time", 0) or 0)
note = (WATCHLIST_NOTES.get(uid) or {}).get("note", "")
note_info = WATCHLIST_NOTES.get(uid) or {}
out.append({
"uid": uid,
"name": pl.get("name", "") or "?",
"level": pl.get("lv", 0), "castle": pl.get("clv", 0),
"tag": pl.get("tag", "") or "",
"x": pl.get("wx", 0), "y": pl.get("wy", 0),
"shield_tier": int(pl.get("shield", 0) or 0),
"shield_eta": max(0, et - now_i) if et > now_i else 0,
"shield_end_time": et,
"shield_src": eta_info.get("src", ""),
"shield_confidence": eta_info.get("confidence", ""),
"shield_activation_ts": int(eta_info.get("activation_ts", 0) or 0),
"shield_history_n": len(SHIELD_HISTORY.get(uid, [])),
"note": note,
"source": note_info.get("source", "manual"),
"in_players_cache": uid in PLAYERS,
})
return self._send(200, json.dumps({"rows": out, "total": len(out),
"rules": dict(WATCHLIST_RULES),
"blacklist_size": len(WATCHLIST_BLACKLIST)}))
if u.path == "/api/watchlist/rules":
return self._send(200, json.dumps(WATCHLIST_RULES))
if u.path == "/api/scanner":
# GET = estado actual; POST = cambiar (action=pause|resume)
return self._send(200, json.dumps({"paused": SCANNER_PAUSED}))
if u.path == "/api/shields":
# diagnostico/UI: contadores + recientes + alerts de proxima expiracion.
# alerts: ETAs activos cuyo tiempo restante <= SHIELD_ALERT_THRESHOLD (10min)
# y que aun no hayamos alertado en esta session (gestionado por SHIELD_ALERTED).
now = time.time()
now_i = int(now)
new_alerts = []
active_list = [] # todos los SHIELD_ETA activos, expanded con info de PLAYERS
with LOCK:
active_uid = 0
pending_coord = 0
for k, e in SHIELD_ETA.items():
et = int(e.get("end_time", 0) or 0)
if et <= now_i: continue
if isinstance(k, int):
active_uid += 1
left = et - now_i
pl = PLAYERS.get(k) or {}
active_list.append({
"uid": k, "left": left, "end_time": et,
"name": pl.get("name", "") or "?",
"tag": pl.get("tag", "") or "",
"wx": int(pl.get("wx", 0) or 0),
"wy": int(pl.get("wy", 0) or 0),
"castle": int(pl.get("clv", 0) or 0),
"src": e.get("src", "?"),
"confidence": e.get("confidence", ""),
})
# alerta solo si esta en el threshold y no se ha alertado aun
if 0 < left <= SHIELD_ALERT_THRESHOLD and k not in SHIELD_ALERTED:
new_alerts.append({
"uid": k, "left": left, "end_time": et,
"name": pl.get("name", "") or e.get("name", "") or "?",
"tag": pl.get("tag", "") or "",
"src": e.get("src", "?"),
})
SHIELD_ALERTED.add(k)
elif isinstance(k, str) and k.startswith("coord:"):
pending_coord += 1
# tambien expone los pending coord (sin uid match aun)
active_list.append({
"uid": 0, "left": int(e.get("end_time", 0)) - now_i,
"end_time": int(e.get("end_time", 0)),
"name": e.get("name", "") or "?",
"tag": "",
"wx": int(e.get("wx", 0) or 0),
"wy": int(e.get("wy", 0) or 0),
"castle": 0,
"src": e.get("src", "?"),
"confidence": "pending_coord_match",
})
# sort ascending por tiempo restante (los que expiran antes primero)
active_list.sort(key=lambda x: x["left"])
recent = []
for r in SHIELD_RECENT[:10]:
et = int(r.get("end_time", 0) or 0)
if et <= now_i: continue # ocultar caducados del feed
recent.append({
"name": r.get("name", "") or "?",
"wx": r.get("wx", 0), "wy": r.get("wy", 0),
"uid": r.get("uid", 0),
"left": et - now_i,
"src": r.get("src", "?"),
"age": int(now - r.get("ts", now)),
})
return self._send(200, json.dumps({
"active_uid": active_uid,
"pending_coord": pending_coord,
"total": active_uid + pending_coord,
"active_list": active_list,
"recent": recent,
"alerts": new_alerts,
"alert_threshold": SHIELD_ALERT_THRESHOLD,
}))
if u.path == "/api/resources":
# Resource tiles del mapa: t=3 (mapinfo_farm). El __id de mapinfo indexa
# WorldResourceConfig (no MonsterConfig). farm_type: 1=food, 2=wood, 3=stone, 4=iron,
# 5+=gems/pumpkin/rose/especiales.
# Campos extra (capturados ahora):
# __resource_num (Int64) -> cantidad disponible
# __occupy_id (UInt32) -> uid del ocupante (0 si Free)
# __occupy_guild_id -> guild_id del ocupante
# __occupy_time -> instante de ocupacion
params = {k: v[0] for k, v in parse_qs(u.query).items()}
try: cx = float(params.get("cx", "")) if params.get("cx") else None
except: cx = None
try: cy = float(params.get("cy", "")) if params.get("cy") else None
except: cy = None
try: rad = float(params.get("radius", "")) if params.get("radius") else None
except: rad = None
try: lv_min = int(params["lv_min"]) if params.get("lv_min") else None
except: lv_min = None
try: lv_max = int(params["lv_max"]) if params.get("lv_max") else None
except: lv_max = None
rtype = params.get("rtype", "") # "" = todos | "1" food | "2" wood | "3" stone | "4" iron
# occupied: "" all | "free" | "ext" (occupied no-alliance) | "mine" (occupied IN my alliance)
occ = params.get("occupied", "")
alliance_tag = (params.get("alliance", "") or "").strip().lower()
try: limit = int(params.get("limit", "500"))
except: limit = 500
try: page = max(1, int(params.get("page", "1")))
except: page = 1
try: max_seen = float(params["max_seen"]) if params.get("max_seen") else STALE_SECONDS
except: max_seen = STALE_SECONDS
sort = params.get("sort", "newest")
now = time.time()
# set de guild_ids del propio jugador (uno por scanner W/E). Tiles cuyo
# __occupy_guild_id este en este set -> "in MY alliance".
my_guilds = {g for g in (SELF["W"]["guild_id"], SELF["E"]["guild_id"]) if g > 0}
out = []
with LOCK:
for o in OBJS.values():
if int(o.get("t", 0)) != 3: continue
if (now - o.get("ts", 0)) > max_seen: continue
wcfg = WCFG.get(str(o["id"])) or {}
rtp = int(wcfg.get("type", 0) or 0)
if rtype and rtp != int(rtype): continue
lv = int(o.get("lv", 0) or wcfg.get("level", 0) or 0)
if lv_min is not None and lv < lv_min: continue
if lv_max is not None and lv > lv_max: continue
oid = int(o.get("oid", 0) or 0)
ogid = int(o.get("ogd", 0) or 0)
is_free = (oid == 0)
is_mine = (not is_free) and (ogid in my_guilds) if my_guilds else False
is_ext = (not is_free) and (not is_mine)
if occ == "free" and not is_free: continue
if occ == "mine" and not is_mine: continue
if occ == "ext" and not is_ext: continue
# info del ocupante para mostrar y filtrar por alliance tag
occ_name = ""; occ_tag = ""
if oid > 0:
pl = PLAYERS.get(oid) or {}
occ_name = pl.get("name", "") or ""
occ_tag = (pl.get("tag", "") or "")
if alliance_tag and alliance_tag not in occ_tag.lower():
continue
d = None
if cx is not None and cy is not None and rad is not None:
d = math.hypot(o["wx"] - cx, o["wy"] - cy)
if d > rad: continue
d = round(d)
out.append({
"id": o["id"],
"name": wcfg.get("name", "") or f"id{o['id']}",
"type_id": rtp,
"type_label": WCFG_TYPE_LABEL.get(rtp, f"type{rtp}" if rtp else "?"),
"level": lv,
"available": int(o.get("rn", 0) or 0),
"occupy_uid": oid,
"occupy_guild_id": ogid,
"occupy_name": occ_name,
"occupy_tag": occ_tag,
"occupy_status": "free" if is_free else ("mine" if is_mine else "ext"),
"x": o["wx"], "y": o["wy"], "dist": d,
"age": int(now - o.get("ts", now)),
})
if sort == "level": out.sort(key=lambda r: (-r["level"], r["name"]))
elif sort == "type": out.sort(key=lambda r: (r["type_id"], -r["level"]))
elif sort == "dist": out.sort(key=lambda r: (r["dist"] if r["dist"] is not None else 9e9))
elif sort == "available": out.sort(key=lambda r: -r["available"])
elif sort == "newest": out.sort(key=lambda r: r["age"])
total = len(out)
if limit <= 0:
page, pages, rows = 1, 1, out
else:
pages = max(1, (total + limit - 1) // limit)
page = min(page, pages)
rows = out[(page-1)*limit : page*limit]
return self._send(200, json.dumps({"rows": rows, "total": total,
"page": page, "pages": pages, "limit": limit}))
if u.path == "/api/relics":
# Relics/Pyramids: tiles especiales del mapa. Combina:
# t=5 mapinfo_ruins
# t=34 mapinfo_altar
# t=57 pyramid_castle
# t=12 mapinfo_boss (bosses rallyeables)
# Filtros adicionales: occupied (free/ext/mine) + alliance tag
params = {k: v[0] for k, v in parse_qs(u.query).items()}
try: cx = float(params.get("cx", "")) if params.get("cx") else None
except: cx = None
try: cy = float(params.get("cy", "")) if params.get("cy") else None
except: cy = None
try: rad = float(params.get("radius", "")) if params.get("radius") else None
except: rad = None
try: lv_min = int(params["lv_min"]) if params.get("lv_min") else None
except: lv_min = None
try: lv_max = int(params["lv_max"]) if params.get("lv_max") else None
except: lv_max = None
kind_filter = params.get("kind", "") # "" / "ruins" / "altar" / "pyramid" / "boss"
occ = params.get("occupied", "") # "" / "free" / "ext" / "mine"
alliance_tag = (params.get("alliance", "") or "").strip().lower()
try: limit = int(params.get("limit", "500"))
except: limit = 500
try: page = max(1, int(params.get("page", "1")))
except: page = 1
try: max_seen = float(params["max_seen"]) if params.get("max_seen") else STALE_SECONDS
except: max_seen = STALE_SECONDS
sort = params.get("sort", "level")
now = time.time()
my_guilds = {g for g in (SELF["W"]["guild_id"], SELF["E"]["guild_id"]) if g > 0}
RELIC_TYPES = {5: "ruins", 34: "altar", 57: "pyramid", 12: "boss"}
out = []
with LOCK:
for o in OBJS.values():
tval = int(o.get("t", 0))
if tval not in RELIC_TYPES: continue
if (now - o.get("ts", 0)) > max_seen: continue
kind_name = RELIC_TYPES[tval]
if kind_filter and kind_filter != kind_name: continue
cfg = CFG.get(str(o["id"])) or {}
nm = cfg.get("name", "") or f"{kind_name} id{o['id']}"
lv = int(o.get("lv", 0) or 0)
if lv_min is not None and lv < lv_min: continue
if lv_max is not None and lv > lv_max: continue
oid = int(o.get("oid", 0) or 0)
ogid = int(o.get("ogd", 0) or 0)
is_free = (oid == 0)
is_mine = (not is_free) and (ogid in my_guilds) if my_guilds else False
is_ext = (not is_free) and (not is_mine)
if occ == "free" and not is_free: continue
if occ == "mine" and not is_mine: continue
if occ == "ext" and not is_ext: continue
occ_name = ""; occ_tag = ""
if oid > 0:
pl = PLAYERS.get(oid) or {}
occ_name = pl.get("name", "") or ""
occ_tag = (pl.get("tag", "") or "")
if alliance_tag and alliance_tag not in occ_tag.lower():
continue
d = None
if cx is not None and cy is not None and rad is not None:
d = math.hypot(o["wx"] - cx, o["wy"] - cy)
if d > rad: continue
d = round(d)
out.append({
"id": o["id"], "name": nm, "kind": kind_name,
"level": lv, "power": int(cfg.get("power", 0) or 0),
"occupy_uid": oid,
"occupy_guild_id": ogid,
"occupy_name": occ_name,
"occupy_tag": occ_tag,
"occupy_status": "free" if is_free else ("mine" if is_mine else "ext"),
"x": o["wx"], "y": o["wy"], "dist": d,
"age": int(now - o.get("ts", now)),
})
if sort == "level": out.sort(key=lambda r: (-r["level"], r["name"]))
elif sort == "kind": out.sort(key=lambda r: (r["kind"], -r["level"]))
elif sort == "power": out.sort(key=lambda r: -r["power"])
elif sort == "dist": out.sort(key=lambda r: (r["dist"] if r["dist"] is not None else 9e9))
elif sort == "newest": out.sort(key=lambda r: r["age"])
total = len(out)
if limit <= 0:
page, pages, rows = 1, 1, out
else:
pages = max(1, (total + limit - 1) // limit)
page = min(page, pages)
rows = out[(page-1)*limit : page*limit]
return self._send(200, json.dumps({"rows": rows, "total": total,
"page": page, "pages": pages, "limit": limit}))
if u.path == "/api/arctic":
# Arctic Barbarians: dos fuentes
# 1) t=36 (barbarian_castle)
# 2) monstruos t=2/12 cuyo nombre contenga "arctic" (familia FAMILIES)
# Filtros adicionales: occupied + alliance (igual que resources/relics)
params = {k: v[0] for k, v in parse_qs(u.query).items()}
try: cx = float(params.get("cx", "")) if params.get("cx") else None
except: cx = None
try: cy = float(params.get("cy", "")) if params.get("cy") else None
except: cy = None
try: rad = float(params.get("radius", "")) if params.get("radius") else None
except: rad = None
try: lv_min = int(params["lv_min"]) if params.get("lv_min") else None
except: lv_min = None
try: lv_max = int(params["lv_max"]) if params.get("lv_max") else None
except: lv_max = None
occ = params.get("occupied", "")
alliance_tag = (params.get("alliance", "") or "").strip().lower()
try: limit = int(params.get("limit", "500"))
except: limit = 500
try: page = max(1, int(params.get("page", "1")))
except: page = 1
try: max_seen = float(params["max_seen"]) if params.get("max_seen") else STALE_SECONDS
except: max_seen = STALE_SECONDS
sort = params.get("sort", "level")
now = time.time()
my_guilds = {g for g in (SELF["W"]["guild_id"], SELF["E"]["guild_id"]) if g > 0}
arctic_kws = ["arctic barbarian", "arctic"]
out = []
with LOCK:
for o in OBJS.values():
if (now - o.get("ts", 0)) > max_seen: continue
tval = int(o.get("t", 0))
cfg = CFG.get(str(o["id"])) or {}
nm = cfg.get("name", "") or ""
is_arctic = tval == 36 or fam_match(nm, arctic_kws)
if not is_arctic: continue
kind_name = "barbarian_castle" if tval == 36 else "barbarian"
if not nm:
nm = f"Barbarian Castle id{o['id']}" if tval == 36 else f"id{o['id']}"
lv = int(o.get("lv", 0) or 0)
if lv_min is not None and lv < lv_min: continue
if lv_max is not None and lv > lv_max: continue
oid = int(o.get("oid", 0) or 0)
ogid = int(o.get("ogd", 0) or 0)
is_free = (oid == 0)
is_mine = (not is_free) and (ogid in my_guilds) if my_guilds else False
is_ext = (not is_free) and (not is_mine)
if occ == "free" and not is_free: continue
if occ == "mine" and not is_mine: continue
if occ == "ext" and not is_ext: continue
occ_name = ""; occ_tag = ""
if oid > 0:
pl = PLAYERS.get(oid) or {}
occ_name = pl.get("name", "") or ""
occ_tag = (pl.get("tag", "") or "")
if alliance_tag and alliance_tag not in occ_tag.lower():
continue
d = None
if cx is not None and cy is not None and rad is not None:
d = math.hypot(o["wx"] - cx, o["wy"] - cy)
if d > rad: continue
d = round(d)
out.append({
"id": o["id"], "name": nm, "kind": kind_name,
"level": lv, "power": int(cfg.get("power", 0) or 0),
"occupy_uid": oid, "occupy_guild_id": ogid,
"occupy_name": occ_name, "occupy_tag": occ_tag,
"occupy_status": "free" if is_free else ("mine" if is_mine else "ext"),
"x": o["wx"], "y": o["wy"], "dist": d,
"age": int(now - o.get("ts", now)),
})
if sort == "level": out.sort(key=lambda r: (-r["level"], r["name"]))
elif sort == "power": out.sort(key=lambda r: -r["power"])
elif sort == "dist": out.sort(key=lambda r: (r["dist"] if r["dist"] is not None else 9e9))
elif sort == "newest": out.sort(key=lambda r: r["age"])
total = len(out)
if limit <= 0:
page, pages, rows = 1, 1, out
else:
pages = max(1, (total + limit - 1) // limit)
page = min(page, pages)
rows = out[(page-1)*limit : page*limit]
return self._send(200, json.dumps({"rows": rows, "total": total,
"page": page, "pages": pages, "limit": limit}))
if u.path == "/api/subcities":
# Sub-cities capturadas via PlayersManager.addSubCity. Enriquecidas con:
# - shield_eta del owner (SHIELD_ETA)
# - quality (white/green/blue/purple/gold/red)
# - famous_id (>0 = Famous City con cultura, p.ej. Roma)
# - occupied: free (npc), ext (jugador otra alianza), mine (jugador mi alianza)
params = {k: v[0] for k, v in parse_qs(u.query).items()}
try: cx = float(params.get("cx", "")) if params.get("cx") else None
except: cx = None
try: cy = float(params.get("cy", "")) if params.get("cy") else None
except: cy = None
try: rad = float(params.get("radius", "")) if params.get("radius") else None
except: rad = None
try: lv_min = int(params["lv_min"]) if params.get("lv_min") else None
except: lv_min = None
try: lv_max = int(params["lv_max"]) if params.get("lv_max") else None
except: lv_max = None
quality_filter = params.get("quality", "") # "" / "white"/"green"/"blue"/"purple"/"gold"/"red"
culture_filter = params.get("culture", "") # "" / "1".."7" (id de cultura)
famous_filter = params.get("famous", "") # "" / "yes" / "no" (no usado en UI ahora pero queda)
occ = params.get("occupied", "") # "" / "free" / "ext" / "mine"
alliance_tag = (params.get("alliance", "") or "").strip().lower()
try: limit = int(params.get("limit", "500"))
except: limit = 500
try: page = max(1, int(params.get("page", "1")))
except: page = 1
try: max_seen = float(params["max_seen"]) if params.get("max_seen") else TTL_SECONDS
except: max_seen = TTL_SECONDS
sort = params.get("sort", "level")
now = time.time()
now_i = int(now)
my_guilds = {g for g in (SELF["W"]["guild_id"], SELF["E"]["guild_id"]) if g > 0}
out = []
with LOCK:
for sid, s in SUBCITIES.items():
if (now - s.get("ts", 0)) > max_seen: continue
lv = int(s.get("lv", 0) or 0)
if lv_min is not None and lv < lv_min: continue
if lv_max is not None and lv > lv_max: continue
quality_id = int(s.get("quality", 0) or 0)
quality_label = SUBCITY_QUALITY.get(quality_id, f"q{quality_id}")
if quality_filter and quality_filter != quality_label: continue
culture_id = int(s.get("evony", 0) or 0)
culture_label = SUBCITY_CULTURE.get(culture_id, f"Culture {culture_id}" if culture_id else "")
if culture_filter and str(culture_id) != culture_filter: continue
famous_id = int(s.get("famous_id", 0) or 0)
if famous_filter == "yes" and famous_id <= 0: continue
if famous_filter == "no" and famous_id > 0: continue
ouid = int(s.get("owner_uid", 0) or 0)
# ogd: guild_id capturado del owner en agent, o fallback a PLAYERS[ouid].gid
ogid = int(s.get("owner_guild_id", 0) or 0)
if ouid > 0 and ogid == 0:
pl0 = PLAYERS.get(ouid) or {}
ogid = int(pl0.get("gid", 0) or 0)
is_free = (ouid == 0)
is_mine = (not is_free) and (ogid in my_guilds) if my_guilds else False
is_ext = (not is_free) and (not is_mine)
if occ == "free" and not is_free: continue
if occ == "mine" and not is_mine: continue
if occ == "ext" and not is_ext: continue
occ_name = ""; occ_tag = ""
if ouid > 0:
pl = PLAYERS.get(ouid) or {}
occ_name = pl.get("name", "") or s.get("owner_name", "") or ""
occ_tag = (pl.get("tag", "") or "")
if alliance_tag and alliance_tag not in occ_tag.lower():
continue
d = None
if cx is not None and cy is not None and rad is not None:
d = math.hypot(s["wx"] - cx, s["wy"] - cy)
if d > rad: continue
d = round(d)
# Bubble:
# - shield_tier viene del user_summary del OWNER en cache PLAYERS
# (0=ninguno, 1=item, 2=newbie). Las sub-cities heredan la burbuja
# del owner, asi que el tier del owner indica el estado de bubble.
# - shield_eta: si tenemos exact ETA en SHIELD_ETA[owner_uid].
sh_eta = 0; sh_tier = 0
if ouid > 0:
pl_o = PLAYERS.get(ouid) or {}
sh_tier = int(pl_o.get("shield", 0) or 0)
e = SHIELD_ETA.get(ouid)
if e:
et = int(e.get("end_time", 0) or 0)
if et > now_i: sh_eta = et - now_i
# power final: power_calc (GetPower) si > 0, sino raw, sino 0.
# Player-owned suele dar 0 porque el broadcast pasivo no expone
# troops/buildings — sólo un scout report del sub-city lo traería.
pw_raw = int(s.get("power", 0) or 0)
pw_calc = int(s.get("power_calc", 0) or 0)
pw_final = pw_calc if pw_calc > 0 else pw_raw
pw_src = "calc" if pw_calc > 0 else ("broadcast" if pw_raw > 0 else "")
out.append({
"id": sid, "name": s.get("name", "") or "?",
"level": lv, "power": pw_final, "power_src": pw_src,
"quality": quality_id, "quality_label": quality_label,
"culture": culture_id, "culture_label": culture_label,
"famous_id": famous_id, "is_famous": famous_id > 0,
"shield_tier": sh_tier,
"x": s["wx"], "y": s["wy"], "dist": d,
"owner_uid": ouid, "occupy_uid": ouid,
"occupy_name": occ_name, "occupy_tag": occ_tag,
"occupy_status": "free" if is_free else ("mine" if is_mine else "ext"),
"shield_eta": sh_eta,
"shield_end": int(s.get("shield_end", 0) or 0),
"age": int(now - s.get("ts", now)),
})
if sort == "level": out.sort(key=lambda r: (-r["level"], r["name"]))
elif sort == "quality": out.sort(key=lambda r: (-r["quality"], -r["level"]))
elif sort == "power": out.sort(key=lambda r: -r["power"])
elif sort == "dist": out.sort(key=lambda r: (r["dist"] if r["dist"] is not None else 9e9))
elif sort == "newest": out.sort(key=lambda r: r["age"])
total = len(out)
if limit <= 0:
page, pages, rows = 1, 1, out
else:
pages = max(1, (total + limit - 1) // limit)
page = min(page, pages)
rows = out[(page-1)*limit : page*limit]
return self._send(200, json.dumps({"rows": rows, "total": total,
"page": page, "pages": pages, "limit": limit}))
if u.path == "/api/marches":
now = time.time()
with LOCK:
arr = [dict(m) for m in MARCHES.values() if int(m.get("te", 0)) > now]
arr.sort(key=lambda m: int(m.get("te", 0)))
return self._send(200, json.dumps({"now": int(now), "n": len(arr), "marches": arr}))
if u.path == "/api/respawns":
qs = parse_qs(u.query or "")
try: ms = int(qs.get("min_samples", ["2"])[0])
except: ms = 2
only_empty = qs.get("only_empty", ["1"])[0] != "0"
allow_normal = qs.get("allow_normal", ["1"])[0] != "0"
try: limit = int(qs.get("limit", ["100"])[0])
except: limit = 100
# ids: lista separada por comas (igual formato que /api/data)
ids = set()
for s in qs.get("ids", [""])[0].split(","):
s = s.strip()
if s.isdigit(): ids.add(int(s))
# fams: separadas por "||" (igual formato que /api/data)
fams = [f for f in qs.get("fams", [""])[0].split("||") if f]
arr = respawn_predictions(min_samples=ms, only_empty=only_empty,
ids=ids, fams=set(fams), allow_normal=allow_normal)
if limit > 0: arr = arr[:limit]
return self._send(200, json.dumps({"now": int(time.time()), "n": len(arr), "respawns": arr}))
if u.path == "/api/relocations":
qs = parse_qs(u.query or "")
try: max_h = float(qs.get("max_age_h", ["24"])[0])
except: max_h = 24.0
near = None
cx = qs.get("cx", [""])[0]; cy = qs.get("cy", [""])[0]; rad = qs.get("radius", [""])[0]
if cx and cy and rad:
try: near = (float(cx), float(cy), float(rad))
except: near = None
try: limit = int(qs.get("limit", ["200"])[0])
except: limit = 200
arr = relocations_recent(max_age_h=max_h, near=near)
if limit > 0: arr = arr[:limit]
return self._send(200, json.dumps({"now": int(time.time()), "n": len(arr), "relocations": arr}))
if u.path == "/api/attacks":
ATTACK_MTYS = {2, 19, 20, 21, 43}
ALLY_MTYS = {19, 20, 21}
PHASE_NAME = {19: "wait", 20: "way", 21: "combat", 2: "march", 43: "scout"}
now = time.time()
out = []
with LOCK:
# indice rapido tile -> player (para identificar castillos de jugador)
tile_pl = {}
for pl in PLAYERS.values():
wx = int(pl.get("wx", 0) or 0); wy = int(pl.get("wy", 0) or 0)
if wx and wy: tile_pl[(wx, wy)] = pl
# labels por mapinfo_type para targets no-monstruo
T_LABEL = {1:"castle", 3:"farm", 4:"garrison", 5:"ruins",
6:"king_city", 7:"sub_city", 8:"snowberg",
9:"guild_city", 10:"guild_store", 11:"guild_farm",
12:"boss"}
for (tx, ty), sub in MARCH_BY_TGT.items():
ms = []
for trp in sub.keys():
m = MARCHES.get(trp)
if not m: continue
if int(m.get("mty", 0)) not in ATTACK_MTYS: continue
if int(m.get("te", 0)) <= now: continue
ms.append(m)
if not ms: continue
ally_ms = [m for m in ms if int(m.get("mty", 0)) in ALLY_MTYS]
if ally_ms:
by = {}
for m in ally_ms: by.setdefault(int(m["mty"]), []).append(m)
if 21 in by: rep = min(by[21], key=lambda m: int(m["te"]))
elif 20 in by: rep = min(by[20], key=lambda m: int(m["te"]))
else: rep = min(by[19], key=lambda m: int(m["te"]))
kind = "alliance"; count = len(ally_ms)
else:
rep = max(ms, key=lambda m: int(m["te"]))
mty = int(rep.get("mty", 0))
kind = "scout" if mty == 43 else "solo"
count = 1
mty = int(rep.get("mty", 0))
uid = int(rep.get("ow", 0))
pl = PLAYERS.get(str(uid)) or PLAYERS.get(uid) or {}
obj = OBJS.get((tx, ty)) or {}
tg_id = int(rep.get("tg", 0) or 0) # target_id del map_target_info
tname = ""; tlevel = 0; tgroup = ""; ttype = 0
if obj:
ttype = int(obj.get("t", 0))
tlevel = obj.get("lv", 0)
if ttype == 2:
# monstruo NPC -> resolver via CFG
c = CFG.get(str(obj.get("id"))) or {}
nm = c.get("name", "") or ""
tname = nm or f"id{obj.get('id')}"
tgroup = group_of(nm)
elif ttype in (1, 7):
# castillo de jugador (1) o sub-city (7): buscar dueno por coords
tpl = tile_pl.get((tx, ty))
if tpl:
nm = tpl.get("name", "") or f"u{tpl.get('uid', '?')}"
tg = tpl.get("tag", "") or ""
tname = f"[{tg}] {nm}" if tg else nm
tlevel = tpl.get("clv", 0) or tlevel
tgroup = "Player" if ttype == 1 else "SubCity"
else:
tname = T_LABEL.get(ttype, "?") + f" id{obj.get('id', 0)}"
tgroup = "Player" if ttype == 1 else "SubCity"
else:
lbl = T_LABEL.get(ttype, f"t{ttype}")
tname = lbl
tgroup = lbl.capitalize()
else:
# OBJS no tiene el tile. El server normalmente NO popula __target_id
# en los broadcasts de march -> tg_id casi siempre llega 0.
# Estrategia de fallbacks:
# 1) MONSTERS_BY_TILE (cache LARGO de ultimo monstruo visto en el tile).
# 2) tg_id + CFG (raro pero posible).
# 3) PLAYERS por coords (castillo de jugador).
# 4) Label semantico segun mty (no se queda como "tile(X,Y)").
mhist = MONSTERS_BY_TILE.get((tx, ty))
if mhist and (now - mhist.get("ts", 0)) <= MONSTERS_TTL:
tname = mhist["name"]
tlevel = mhist.get("lv", 0)
tgroup = group_of(mhist["name"])
ttype = 2
elif tg_id > 0 and mty in ATTACK_MTYS:
c = CFG.get(str(tg_id)) or {}
nm = c.get("name", "") or ""
if nm:
tname = nm
tlevel = c.get("level", 0) or 0
tgroup = group_of(nm)
ttype = 2
if not tname:
tpl = tile_pl.get((tx, ty))
if tpl:
nm = tpl.get("name", "") or f"u{tpl.get('uid', '?')}"
tg = tpl.get("tag", "") or ""
tname = f"[{tg}] {nm}" if tg else nm
tlevel = tpl.get("clv", 0)
tgroup = "Player"
ttype = 1
# Si tras todos los fallbacks aun no resolvimos, dar un label
# semantico segun el tipo de marcha en vez de "tile(X,Y)":
if not tname:
label_by_mty = {
2: "Monster (unknown)",
19: "Boss rally (unknown)",
20: "Boss rally (marching)",
21: "Boss rally (combat)",
43: "Scout monster (unknown)",
}
tname = label_by_mty.get(mty, f"Target tile")
tgroup = "Unknown"
out.append({
"tx": tx, "ty": ty,
"tname": tname, "tlevel": tlevel, "tgroup": tgroup, "ttype": ttype,
"tg_id": tg_id, # debug: target_id del march, para diagnosticar resolucion
"kind": kind, "phase": PHASE_NAME.get(mty, "march"), "mty": mty,
"count": count,
"uid": uid,
"name": pl.get("name", "") or "",
"tag": pl.get("tag", "") or "",
"guild": int(pl.get("gid", 0) or 0),
"eta": max(0, int(rep.get("te", 0) - now)),
"te": int(rep.get("te", 0)),
"sx": int(rep.get("sx", 0)),
"sy": int(rep.get("sy", 0)),
})
out.sort(key=lambda a: a["eta"])
return self._send(200, json.dumps({"now": int(now), "n": len(out), "attacks": out}))
return self._send(404, "{}")
def do_POST(self):
from urllib.parse import urlparse, parse_qs
u = urlparse(self.path)
length = int(self.headers.get("Content-Length", "0") or "0")
raw = self.rfile.read(length) if length > 0 else b""
try: body = json.loads(raw) if raw else {}
except: body = {}
if u.path == "/api/scanner":
global SCANNER_PAUSED
action = (body.get("action") or "").lower()
target_half = (body.get("half") or "").upper() # opcional: "W", "E" o ""
# Acción 'restart': fuerza reattach del agent en uno o ambos halves.
# El watchdog lo detecta vía LAST_HEARTBEAT y reattachea solo si pasan
# 60s sin recibir nada; aquí lo aceleramos seteando heartbeat a 0.
if action == "restart":
halves = [target_half] if target_half in ("W", "E") else ["W", "E"]
forced = []
for h in halves:
LAST_HEARTBEAT[h] = 0.0 # fuerza al watchdog a reattachar inmediatamente
forced.append(h)
print(f"[scanner ctl] restart forzado en {forced}", flush=True)
return self._send(200, json.dumps({"ok": True, "restarting": forced}))
if action == "hard_reset":
# Hard reset: ejecuta en thread propio, INDEPENDIENTE de frida_thread.
# Antes la UI solo seteaba flags que frida_thread consultaba — si
# ese hilo estaba colgado (dev.attach/script.load), el reset no
# llegaba a ejecutarse. Ahora corre desde el handler HTTP.
halves = [target_half] if target_half in ("W", "E") else ["W", "E"]
forced = []
for h in halves:
if _run_reset(h, "hard"):
forced.append(h)
print(f"[scanner ctl] HARD-RESET (UI) lanzado en {forced}", flush=True)
return self._send(200, json.dumps({"ok": True, "hard_resetting": forced}))
if action == "nuclear_reset":
# Nuclear reset: kill qemu entero + relanzar emulador. ~3-4 min.
# Ejecuta en thread propio (igual que hard_reset).
# SEGURIDAD: requiere half explícito ("W" o "E"), o "BOTH" para ambos.
# No se acepta string vacío para evitar disparar ambos por error.
if target_half == "BOTH":
halves = ["W", "E"]
elif target_half in ("W", "E"):
halves = [target_half]
else:
return self._send(400, json.dumps({
"error": "nuclear_reset requires half='W' | 'E' | 'BOTH' (no empty)"
}))
forced = []
for h in halves:
if _run_reset(h, "nuclear"):
forced.append(h)
print(f"[scanner ctl] ☠ NUCLEAR-RESET (UI) lanzado en {forced}", flush=True)
return self._send(200, json.dumps({"ok": True, "nuclear_resetting": forced}))
if action not in ("pause", "resume"):
return self._send(400, json.dumps({"error": "action must be 'pause'|'resume'|'restart'"}))
SCANNER_PAUSED = (action == "pause")
sent = []
for half, sc in SCRIPTS.items():
if sc is None: continue
try:
sc.post({"type": "ctl", "cmd": action})
sent.append(half)
except Exception as e:
print(f"[scanner ctl] {half} post fail: {e}", flush=True)
print(f"[scanner ctl] {action} -> {sent}", flush=True)
return self._send(200, json.dumps({"ok": True, "paused": SCANNER_PAUSED, "halves": sent}))
if u.path == "/api/watchlist":
# Anade uid al watchlist. body = {"uid": int, "note": "...opcional..."}
try:
uid = int(body.get("uid", 0) or 0)
except: uid = 0
if uid <= 0:
return self._send(400, json.dumps({"error": "uid required (int > 0)"}))
with LOCK:
WATCHLIST.add(uid)
WATCHLIST_BLACKLIST.discard(uid) # si lo anades manual, quita de blacklist
note = (body.get("note") or "").strip()[:200]
WATCHLIST_NOTES[uid] = {"note": note, "added_ts": int(time.time()), "source": "manual"}
print(f"[watchlist] +{uid} manual ({note or 'no note'})", flush=True)
return self._send(200, json.dumps({"ok": True, "uid": uid, "watchlist_size": len(WATCHLIST)}))
if u.path == "/api/watchlist/rules":
# Body: subconjunto de WATCHLIST_RULES. Solo claves conocidas se aceptan.
allowed = {"enabled", "enemy_tags", "exclude_tags", "min_castle", "min_power_M", "require_shield_seen"}
with LOCK:
for k, v in (body or {}).items():
if k not in allowed: continue
if k in ("enemy_tags", "exclude_tags"):
if not isinstance(v, list): continue
WATCHLIST_RULES[k] = [str(x).strip().upper() for x in v if str(x).strip()]
elif k == "enabled":
WATCHLIST_RULES[k] = bool(v)
elif k == "require_shield_seen":
WATCHLIST_RULES[k] = bool(v)
else:
try: WATCHLIST_RULES[k] = float(v) if k == "min_power_M" else int(v)
except: pass
print(f"[watchlist] rules updated: {WATCHLIST_RULES}", flush=True)
return self._send(200, json.dumps(WATCHLIST_RULES))
if u.path == "/api/watchlist/blacklist":
# Descarta uid: lo quita del watchlist + impide que el worker lo re-anada.
try: uid = int(body.get("uid", 0) or 0)
except: uid = 0
if uid <= 0:
return self._send(400, json.dumps({"error": "uid required"}))
with LOCK:
WATCHLIST.discard(uid)
WATCHLIST_NOTES.pop(uid, None)
WATCHLIST_BLACKLIST.add(uid)
print(f"[watchlist] BLACKLIST +{uid}", flush=True)
return self._send(200, json.dumps({"ok": True, "uid": uid,
"watchlist_size": len(WATCHLIST),
"blacklist_size": len(WATCHLIST_BLACKLIST)}))
if u.path == "/api/shield_history/clear":
# Limpia SHIELD_HISTORY[uid] + SHIELD_ETA[uid] + SHIELD_LAST_TRANSITION[uid].
# Util si el histo se envenena por flicker (mediana corrupta).
try: uid = int(body.get("uid", 0) or 0)
except: uid = 0
if uid <= 0:
return self._send(400, json.dumps({"error": "uid required"}))
with LOCK:
had_hist = uid in SHIELD_HISTORY
SHIELD_HISTORY.pop(uid, None)
SHIELD_ETA.pop(uid, None)
SHIELD_LAST_TRANSITION.pop(uid, None)
print(f"[shield-history] cleared uid={uid} (had_history={had_hist})", flush=True)
return self._send(200, json.dumps({"ok": True, "uid": uid, "cleared_history": had_hist}))
if u.path == "/api/watchlist/reset_auto":
# Borra todos los uids con source='auto'. Conserva los manuales y NO los anade a blacklist
# (asi se pueden re-anadir si reglas cambian para incluirlos).
removed = []
with LOCK:
for uid in list(WATCHLIST):
src = (WATCHLIST_NOTES.get(uid) or {}).get("source", "manual")
if src == "auto":
WATCHLIST.discard(uid)
WATCHLIST_NOTES.pop(uid, None)
removed.append(uid)
print(f"[watchlist] reset_auto: removed {len(removed)} (kept {len(WATCHLIST)} manuals)", flush=True)
return self._send(200, json.dumps({"ok": True, "removed": len(removed), "watchlist_size": len(WATCHLIST)}))
return self._send(404, "{}")
def do_DELETE(self):
from urllib.parse import urlparse, parse_qs
u = urlparse(self.path)
if u.path == "/api/watchlist":
params = {k: v[0] for k, v in parse_qs(u.query).items()}
try: uid = int(params.get("uid", 0) or 0)
except: uid = 0
if uid <= 0:
return self._send(400, json.dumps({"error": "uid query param required"}))
with LOCK:
WATCHLIST.discard(uid)
WATCHLIST_NOTES.pop(uid, None)
print(f"[watchlist] -{uid}", flush=True)
return self._send(200, json.dumps({"ok": True, "uid": uid, "watchlist_size": len(WATCHLIST)}))
return self._send(404, "{}")
def prune_thread():
"""Purga objetos no re-vistos en TTL_SECONDS (auto-expira muertos -> mejor que iScout)."""
while True:
time.sleep(PRUNE_EVERY)
now = time.time()
cutoff = now - TTL_SECONDS
with LOCK:
dead = [k for k, o in OBJS.items() if o.get("ts", 0) < cutoff]
for k in dead:
del OBJS[k]
pdead = [k for k, p in PLAYERS.items() if p.get("ts", 0) < cutoff]
for k in pdead:
del PLAYERS[k]
# marchas expiradas (end_time <= now-30s): se eliminan
mdead = [trp for trp, m in MARCHES.items() if int(m.get("te", 0)) <= now - 30]
for trp in mdead:
m = MARCHES.pop(trp, None)
if m:
key = (int(m.get("tx", 0)), int(m.get("ty", 0)))
sub = MARCH_BY_TGT.get(key)
if sub is not None:
sub.pop(trp, None)
if not sub: MARCH_BY_TGT.pop(key, None)
# sub-cities/shield-cache: purga sub-cities no re-vistas, y ETAs vencidas.
scdead = [k for k, s in SUBCITIES.items() if s.get("ts", 0) < cutoff]
for k in scdead:
del SUBCITIES[k]
shdead = [k for k, e in SHIELD_ETA.items() if int(e.get("end_time", 0)) <= int(now)]
for k in shdead:
del SHIELD_ETA[k]
if isinstance(k, int):
SHIELD_ALERTED.discard(k) # caducado: rearmar alerta si vuelve a aparecer
# power cache: purga entradas no actualizadas en >2h (uids ya rankeados que ya no estan)
pwdead = [uid for uid, p in PLAYER_POWER.items() if p.get("ts", 0) < now - 7200]
for uid in pwdead:
del PLAYER_POWER[uid]
# MONSTERS_BY_TILE: cache LARGO (1h) para resolver targets de attacks
mbtdead = [k for k, v in MONSTERS_BY_TILE.items() if v.get("ts", 0) < now - MONSTERS_TTL]
for k in mbtdead:
del MONSTERS_BY_TILE[k]
if dead or pdead or mdead or scdead or shdead:
print(f"[prune] -{len(dead)} objs / -{len(pdead)} players / -{len(mdead)} marchas / "
f"-{len(scdead)} subcity / -{len(shdead)} shieldETA, "
f"quedan {len(OBJS)}/{len(PLAYERS)}/{len(MARCHES)}/{len(SUBCITIES)}/{len(SHIELD_ETA)}", flush=True)
def save_cache():
"""Volcado atómico (write tmp + rename) de los caches en memoria a JSON.
Las keys con tuplas (wx,wy) se convierten a "wx,wy" para serialización."""
try:
with LOCK:
payload = {
"version": 1,
"saved_at": time.time(),
"OBJS": {f"{k[0]},{k[1]}": v for k, v in OBJS.items()},
"PLAYERS": {str(k): v for k, v in PLAYERS.items()},
"MARCHES": {str(k): v for k, v in MARCHES.items()},
"SUBCITIES": {str(k): v for k, v in SUBCITIES.items()},
"MONSTERS_BY_TILE": {f"{k[0]},{k[1]}": v for k, v in MONSTERS_BY_TILE.items()},
"PLAYER_POWER": {str(k): v for k, v in PLAYER_POWER.items()},
"SHIELD_ETA": {(f"int:{k}" if isinstance(k, int) else k): v
for k, v in SHIELD_ETA.items()},
"SHIELD_HISTORY": {str(k): v for k, v in SHIELD_HISTORY.items()},
"WATCHLIST": sorted(WATCHLIST),
"WATCHLIST_NOTES": {str(k): v for k, v in WATCHLIST_NOTES.items()},
"WATCHLIST_BLACKLIST": sorted(WATCHLIST_BLACKLIST),
"WATCHLIST_RULES": dict(WATCHLIST_RULES),
"CFG": CFG, "WCFG": WCFG, "STATE": {"cfg_n": STATE["cfg_n"]},
"SPAWN_HISTORY": {f"{k[0]},{k[1]}": v for k, v in SPAWN_HISTORY.items()},
"RELOCATIONS": {str(k): v for k, v in RELOCATIONS.items()},
}
tmp = CACHE_FILE + ".tmp"
with open(tmp, "w") as f:
json.dump(payload, f)
os.replace(tmp, CACHE_FILE)
except Exception as e:
print(f"[persist] save fail: {e}", flush=True)
def load_cache():
"""Carga el JSON al arrancar. Descarta entries con ts más viejas que PERSIST_TTL."""
if not os.path.exists(CACHE_FILE):
print("[persist] no cache previo, arranque limpio", flush=True)
return
try:
with open(CACHE_FILE) as f:
d = json.load(f)
now = time.time()
cutoff = now - PERSIST_TTL
with LOCK:
n_obj = n_pl = n_mch = n_sc = n_mbt = n_pwr = n_sh = 0
for k_str, v in d.get("OBJS", {}).items():
if v.get("ts", 0) < cutoff: continue
wx, wy = k_str.split(",")
OBJS[(int(wx), int(wy))] = v
n_obj += 1
for k_str, v in d.get("PLAYERS", {}).items():
if v.get("ts", 0) < cutoff: continue
PLAYERS[int(k_str)] = v
n_pl += 1
for k_str, v in d.get("MARCHES", {}).items():
if int(v.get("te", 0)) <= now: continue # marcha caducada
trp = int(k_str); MARCHES[trp] = v
key = (int(v.get("tx", 0)), int(v.get("ty", 0)))
MARCH_BY_TGT.setdefault(key, {})[trp] = True
n_mch += 1
for k_str, v in d.get("SUBCITIES", {}).items():
if v.get("ts", 0) < cutoff: continue
SUBCITIES[int(k_str)] = v
n_sc += 1
for k_str, v in d.get("MONSTERS_BY_TILE", {}).items():
if v.get("ts", 0) < cutoff: continue
wx, wy = k_str.split(",")
MONSTERS_BY_TILE[(int(wx), int(wy))] = v
n_mbt += 1
for k_str, v in d.get("PLAYER_POWER", {}).items():
if v.get("ts", 0) < cutoff: continue
PLAYER_POWER[int(k_str)] = v
n_pwr += 1
for k_str, v in d.get("SHIELD_ETA", {}).items():
if int(v.get("end_time", 0)) <= int(now): continue
k = int(k_str[4:]) if k_str.startswith("int:") else k_str
SHIELD_ETA[k] = v
n_sh += 1
# SHIELD_HISTORY: cargar todo (lo usamos para mediana de duracion)
n_sh_hist_pl = 0
for k_str, v in d.get("SHIELD_HISTORY", {}).items():
if not isinstance(v, list) or not v: continue
SHIELD_HISTORY[int(k_str)] = v[-SHIELD_HISTORY_MAX:]
n_sh_hist_pl += 1
# WATCHLIST + notes + blacklist + rules
for uid in d.get("WATCHLIST", []):
try: WATCHLIST.add(int(uid))
except: pass
for k_str, v in (d.get("WATCHLIST_NOTES", {}) or {}).items():
try: WATCHLIST_NOTES[int(k_str)] = v
except: pass
for uid in (d.get("WATCHLIST_BLACKLIST", []) or []):
try: WATCHLIST_BLACKLIST.add(int(uid))
except: pass
saved_rules = d.get("WATCHLIST_RULES")
if isinstance(saved_rules, dict):
# merge: claves nuevas/desconocidas no se aceptan, valores preservan tipo
for k in list(WATCHLIST_RULES.keys()):
if k in saved_rules: WATCHLIST_RULES[k] = saved_rules[k]
# CLEANUP post-load: descartar SHIELD_ETA stale (tier=0 + no exact).
# Necesario tras cambios de logica: el cache previo puede tener entries
# huerfanas que no se invalidan via transition (porque prev_tier ya era 0
# en PLAYERS cargado). Mantiene solo 'exact' (scout/mail) que tienen ground truth propio.
stale = []
for uid_k, info in list(SHIELD_ETA.items()):
if not isinstance(uid_k, int): continue
if info.get("confidence") == "exact": continue
pl = PLAYERS.get(uid_k)
tier = int((pl or {}).get("shield", 0) or 0) if pl else 0
if tier == 0:
stale.append(uid_k)
for uid_k in stale: SHIELD_ETA.pop(uid_k, None)
if stale: print(f"[persist] cleaned {len(stale)} stale SHIELD_ETA entries (tier=0 + non-exact)", flush=True)
# CFG y WCFG: cargar siempre (no caducan)
for k, v in d.get("CFG", {}).items():
CFG[str(k)] = v
for k, v in d.get("WCFG", {}).items():
WCFG[str(k)] = v
STATE["cfg_n"] = len(CFG)
# SPAWN_HISTORY: filtrar entradas demasiado viejas (>PERSIST_TTL)
n_sh_hist = 0
for k_str, v in d.get("SPAWN_HISTORY", {}).items():
if not isinstance(v, list) or not v: continue
if max(e.get("ts", 0) for e in v) < cutoff: continue
wx, wy = k_str.split(",")
SPAWN_HISTORY[(int(wx), int(wy))] = v
n_sh_hist += 1
# RELOCATIONS
n_reloc = 0
for k_str, v in d.get("RELOCATIONS", {}).items():
if not isinstance(v, list) or not v: continue
# filtrar eventos individuales viejos
kept = [e for e in v if e.get("ts", 0) >= cutoff]
if not kept: continue
RELOCATIONS[int(k_str)] = kept
n_reloc += 1
age_h = (now - d.get("saved_at", now)) / 3600
print(f"[persist] cache cargado (snapshot {age_h:.1f}h): "
f"{n_obj} objs / {n_pl} players / {n_mch} marchas / "
f"{n_sc} subcity / {n_mbt} monsters / {n_pwr} power / "
f"{n_sh} shield / {n_sh_hist} spawn-hist / {n_reloc} reloc-uids / "
f"{len(CFG)} CFG / {len(WCFG)} WCFG", flush=True)
except Exception as e:
print(f"[persist] load fail: {e}", flush=True)
def persist_thread():
while True:
time.sleep(PERSIST_INTERVAL)
save_cache()
def auto_watchlist_thread():
"""Cada 60s aplica WATCHLIST_RULES a PLAYERS y auto-anade los que cumplan.
No re-anade los que estan en BLACKLIST. No pisa entries 'manual' existentes.
"""
while True:
time.sleep(60)
if not WATCHLIST_RULES.get("enabled", False):
continue
try:
enemy = set(t.upper() for t in WATCHLIST_RULES.get("enemy_tags", []))
exclude = set(t.upper() for t in WATCHLIST_RULES.get("exclude_tags", []))
min_castle = int(WATCHLIST_RULES.get("min_castle", 0) or 0)
min_power = float(WATCHLIST_RULES.get("min_power_M", 0) or 0) # en M
require_shield = bool(WATCHLIST_RULES.get("require_shield_seen", False))
added = 0
now = int(time.time())
with LOCK:
# snapshot para no holdear el lock durante toda la iteracion
ps_snapshot = list(PLAYERS.items())
bl_snapshot = set(WATCHLIST_BLACKLIST)
wl_snapshot = set(WATCHLIST)
pw_snapshot = dict(PLAYER_POWER)
sh_eta_snapshot = dict(SHIELD_ETA)
sh_hist_keys = set(SHIELD_HISTORY.keys())
to_add = []
for uid, p in ps_snapshot:
if uid in wl_snapshot or uid in bl_snapshot:
continue
tag = (p.get("tag", "") or "").upper()
if enemy and tag not in enemy:
continue
if exclude and tag in exclude:
continue
clv = int(p.get("clv", 0) or 0)
if clv < min_castle:
continue
# power: preferir PLAYER_POWER (exacto). Fallback a broadcast (suele ser 0).
pw_info = pw_snapshot.get(uid) or {}
pw_val = int(pw_info.get("power", 0) or p.get("power", 0) or 0)
if (pw_val / 1e6) < min_power:
continue
if require_shield:
has_shield_ever = (uid in sh_hist_keys) or (uid in sh_eta_snapshot) or (int(p.get("shield", 0) or 0) > 0)
if not has_shield_ever:
continue
to_add.append(uid)
if to_add:
with LOCK:
for uid in to_add:
if uid in WATCHLIST or uid in WATCHLIST_BLACKLIST:
continue # double-check tras re-adquirir lock
WATCHLIST.add(uid)
p = PLAYERS.get(uid) or {}
tag = (p.get("tag", "") or "").upper()
WATCHLIST_NOTES[uid] = {
"note": f"auto: tag={tag} C{p.get('clv', 0)}",
"added_ts": now,
"source": "auto",
}
added += 1
if added:
print(f"[watchlist-auto] +{added} uids (total={len(WATCHLIST)}, rules={WATCHLIST_RULES})", flush=True)
except Exception as e:
print(f"[watchlist-auto] error: {e}", flush=True)
def main():
load_cache() # restaurar histórico antes de arrancar nada
for i, (serial, half) in enumerate(SCANNERS):
threading.Thread(target=frida_thread, args=(serial, half), daemon=True).start()
time.sleep(6) # escalonado: W termina su setup antes de que arranque E
threading.Thread(target=prune_thread, daemon=True).start()
threading.Thread(target=persist_thread, daemon=True).start()
threading.Thread(target=auto_watchlist_thread, daemon=True).start()
srv = ThreadingHTTPServer(("127.0.0.1", PORT), H)
print(f"[web] http://127.0.0.1:{PORT} v2 DUAL (W=emulator-5554 / E=emulator-5556) (Ctrl+C para parar)", flush=True)
try:
srv.serve_forever()
except KeyboardInterrupt:
print("\n[web] guardando cache antes de salir...")
save_cache()
print("[web] parado")
if __name__ == "__main__":
main()