#!/usr/bin/env python3 """ ALMACÉN DEL ESCÁNER — app híbrida (una cuenta, un cliente). El módulo escáner del agente emite los mismos "kinds" que el escáner de producción (batch/config/wcfg). Aquí se acumulan y se sirven en el MISMO formato que devolvía `/api/data` del :8773, para que todo el bot (presets, auto-farm, rally, auto-join) siga funcionando sin tocar su lógica: sólo cambia de dónde vienen los datos. Origen del código de familias: evony-scout-pixel-v5/iscout_web.py (líneas 2670-2774). ⚠️ FAMILIES/_id2fam guardan el grupo en MINÚSCULA y group_of() devuelve Mayúscula. Ya hubo un bug por mezclarlos: comparar siempre en el mismo caso. """ import threading, time, json # ── Ventanas de frescura (las de producción) ───────────────────────────────── ONMAP_FARM = 300 # 5 min: monstruos roaming, se mueven ONMAP_RALLY = 2700 # 45 min: boss/event son estacionarios RALLY_GROUPS_ONMAP = ("boss", "event") FAMILIES = [ # ── 1) SHADOW OF DAWN (primero: captura "(Shadow of Dawn) X" antes que la criatura base) ── ("Shadow of Dawn", ["shadow of dawn"], "shadow"), ("Dawn Sanctuary", ["dawn sanctuary"], "shadow"), ("Dawn Temple", ["dawn temple"], "shadow"), ("Eclipse Stronghold", ["eclipse stronghold"], "shadow"), # ── 2) Excepciones de event que contienen keyword de boss (antes que el boss) ── ("Flame Cerberus", ["flame cerberus"], "event"), # event, NO boss (pese a "cerberus") # ── 3) BOSS — monstruos permanentes del mapa ── ("Cerberus", ["cerberus"], "boss"), ("Knight Bayard", ["bayard", "bayar knight"], "boss"), ("Harpy", ["harpy"], "boss"), ("Peryton", ["peryton"], "boss"), ("Minotaur", ["minotaur"], "boss"), ("Griffin", ["griffin"], "boss"), ("Ifrit", ["ifrit"], "boss"), ("Kamaitachi", ["kamaitachi"], "boss"), ("Fafnir", ["fafnir"], "boss"), ("Behemoth", ["behemoth"], "boss"), ("Phoenix", ["phoenix"], "boss"), ("Jormungandr", ["jormungandr"], "boss"), ("Typhon", ["typhon"], "boss"), ("Ammit", ["ammit"], "boss"), ("Centaur", ["centaur"], "boss"), ("Werewolf", ["werewolf"], "boss"), ("Yasha", ["yasha"], "boss"), ("Redcap", ["redcap"], "boss"), ("Skeleton Dragon", ["skeleton dragon"], "boss"), ("Manticore", ["manticore"], "boss"), ("Zombie", ["zombie"], "boss"), ("Airavata", ["airavata"], "boss"), ("Ghidorah", ["ghidorah"], "boss"), ("Nine-tails", ["nine-tails", "nine tails"], "boss"), ("Stymphalian Bird",["stymphalian"], "boss"), ("Kraken", ["kraken"], "boss"), ("Azazel", ["azazel"], "boss"), ("Leviathan", ["leviathan"], "boss"), ("Garmr", ["garmr"], "boss"), # ── 4) EVENT — monstruos de eventos temporales ── # Barbary Pirate: dos monstruos distintos -> dos pills (Normal 24.6M / Elite 130M) ("Normal Barbary Pirate", ["normal barbary pirate"], "event"), ("Elite Barbary Pirate", ["elite barbary pirate"], "event"), ("Hydra", ["hydra"], "event"), ("Warlord", ["warlord"], "event"), ("Ymir", ["ymir"], "event"), ("Pan", ["pan "], "event"), ("Lava Turtle", ["lava turtle"], "event"), ("Sphinx", ["sphinx"], "event"), # cubre Sphinx y Sphinx Castle ("Witch", ["witch"], "event"), # ojo: Mysterious Witch va a Other (antes) ("Golem", ["golem"], "event"), ("Mire Squid", ["mire squid"], "event"), ("Pumpkin", ["pumpkin"], "event"), ("Viking", ["viking"], "event"), ("Golden Goblin", ["golden goblin"], "event"), ("Arctic Barbarian",["arctic barbarian"], "event"), ("Aglaope", ["aglaope"], "event"), ("Nasu", ["nasu"], "event"), ("Gugler Knight", ["gugler"], "event"), ("Carcinus", ["carcinus"], "event"), ("Serpopard", ["serpopard"], "event"), ("Savage Tiger Gladiator", ["savage tiger"], "event"), ("Taotie", ["taotie"], "event"), ("Nian", ["nian"], "event"), ("Cranky Snowman", ["cranky snowman", "snowman"], "event"), ("Surtr", ["surtr"], "event"), ("Angry Turkey", ["angry turkey", "turkey"], "event"), ("Fortress Pyramid",["fortress pyramid"], "event"), ("Arachne", ["arachne"], "event"), # L1/L2, 120M/361M ("Garuda", ["garuda"], "event"), # Junior/Senior/Excellent, L1-3 ("Lord of Lava", ["lord of lava"], "event"), # L25 ("Thunder Scorpion",["thunder scorpion"], "event"), # L25 ("Bird of Hurricane",["bird of hurricane"], "event"), # L25 ("Silver Lionheart Knight", ["silver lionheart"], "event"), # Obsidian/Mica/Cinnabar, L1-3 # ── 5) OTHER — festivos / decorativos ── ("Easter Bunny", ["easter bunny", "bunny"], "other"), ("Santa", ["santa"], "other"), ("Mysterious Witch",["mysterious witch"], "other"), # antes que Witch→event (más específico) ("Ferris Wheel", ["ferris wheel"], "other"), ("Celebration Squad",["celebration squad"], "other"), ("Ares Statue", ["ares statue"], "other"), ("Fire Spirit", ["fire spirit"], "other"), # ── 6) wildcard genérico de bosses prefijados "(Boss)" (último) ── ("Bosses (Boss)", ["(boss)"], "boss"), ] # Orden de precedencia: "Mysterious Witch" debe ir ANTES de "Witch" (event) para no # clasificarse como event. Lo reordenamos: las entradas Other con substring que colisiona # con event se evalúan por su posición — movemos Mysterious Witch arriba programáticamente. _FAM_OTHER_FIRST = [f for f in FAMILIES if f[2] == "other" and "witch" in f[1][0]] FAMILIES = _FAM_OTHER_FIRST + [f for f in FAMILIES if f not in _FAM_OTHER_FIRST] def fam_match(name, kws): n = (name or "").lower() return any(k in n for k in kws) _GROUP_LABEL = {"boss": "Boss", "event": "Event", "shadow": "Shadow of Dawn", "other": "Other"} _GROUP_CACHE = {} # name -> label (memo: los nombres se repiten miles de veces) def group_of(name): """Clasifica un monstruo: Boss | Event | Shadow of Dawn | Other | Normal.""" cached = _GROUP_CACHE.get(name) if cached is not None: return cached n = (name or "").lower() res = "Normal" for disp, kws, grp in FAMILIES: if any(k in n for k in kws): res = _GROUP_LABEL.get(grp, "Normal"); break else: if "(boss)" in n: res = "Boss" _GROUP_CACHE[name] = res return res # ============================================================================ # ALMACÉN DE OBJETOS DEL MAPA # ============================================================================ _LOCK = threading.RLock() OBJS = {} # (x,y) -> {"t","id","lv","x","y","ts"} ts = última vez visto CFG = {} # id -> {"name","level","type","power"} (MonsterConfig del juego) WCFG = {} # id -> {"name","level","type"} (WorldResourceConfig: farms/minas) STATE = { "objs": 0, "batches": 0, "cfg_n": 0, "wcfg_n": 0, "last_batch": 0, "server": 0, "sweep": {}, "mt": {}, # telemetría del agente (barrido y salud del cliente) "scan_on": True, } MONSTER_T = 2 # tipo de mapinfo que son monstruos (verificado en vivo 2026-08-13) # Etiquetas del resto de tipos de tile. El bot sólo roba objetivos cuyo "tgroup" está en # {Boss, Event, Normal, Shadow, Other}, así que estas etiquetas los excluyen por sí solas. 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_tile"} RESOURCE_T = 3 # tiles de recurso (farms/minas) def ingest_batch(items, server=0): """Lote de objetos del mapa que manda el agente cada 1,5 s.""" now = int(time.time()) with _LOCK: if server: STATE["server"] = int(server) for it in items or []: try: x = int(it.get("wx", 0) or 0); y = int(it.get("wy", 0) or 0) if not x or not y: continue OBJS[(x, y)] = {"t": int(it.get("t", 0) or 0), "id": int(it.get("id", 0) or 0), "lv": int(it.get("lv", 0) or 0), "x": x, "y": y, "ts": now} except Exception: continue STATE["objs"] = len(OBJS) STATE["batches"] += 1 STATE["last_batch"] = now def ingest_config(cfg): with _LOCK: for k, v in (cfg or {}).items(): CFG[str(k)] = v STATE["cfg_n"] = len(CFG) _save_cfg() def ingest_wcfg(wcfg): with _LOCK: for k, v in (wcfg or {}).items(): WCFG[str(k)] = v STATE["wcfg_n"] = len(WCFG) def ingest_metrics(p): with _LOCK: STATE["sweep"] = p.get("sweep", {}) or {} STATE["mt"] = p.get("mt", {}) or {} STATE["scan_on"] = bool(p.get("on", True)) if p.get("server"): STATE["server"] = int(p["server"]) def name_of(oid, t=MONSTER_T): """Nombre legible del objeto. Sin config del juego devolvemos 'id', que es exactamente lo que el bot ya sabe descartar (no es un monstruo nombrable).""" c = CFG.get(str(oid)) if t != RESOURCE_T else WCFG.get(str(oid)) if c and c.get("name") and c["name"] != "?": return c["name"] return "id%d" % int(oid or 0) def map_rows(monsters_only=True): """Filas en el MISMO formato que devolvía /api/data del escáner externo: {x, y, name, level, group, id, power, seen_age}. seen_age = segundos desde la última vez que el barrido vio el tile. El bot lo usa para no marchar a fantasmas: filtra con ventanas distintas según el grupo (los boss/event son estacionarios y toleran 45 min; los roaming, 5 min). Devolvemos TODO el histórico on-map y que filtre el bot, igual que hacía el V4. """ now = int(time.time()) out = [] with _LOCK: for (x, y), o in OBJS.items(): t = o["t"] if monsters_only and t != MONSTER_T: continue age = now - o["ts"] if age > ONMAP_RALLY: # más viejo que la ventana más laxa: ya no sirve continue nm = name_of(o["id"], t) c = CFG.get(str(o["id"])) or {} lv = o["lv"] or int(c.get("level", 0) or 0) # nivel del CFG si el mapa no lo trae out.append({ "x": x, "y": y, "name": nm, "level": lv, "group": group_of(nm), "id": o["id"], "power": float(c.get("power", 0) or 0), "seen_age": age, }) return out # ============================================================================ # SHORTCUTS — familias × nivel con conteo live (tab de accesos rápidos, estilo V5) # ============================================================================ QUICK_HIDE = {"Viking"} # familias ocultas del menú de Shortcuts (no del escáner) _QUICK_CACHE = {"ts": 0.0, "data": None} _ID2FAM = {"map": None, "cfg_n": -1} def _id2fam(): """id(str) -> (familia, grupo_lower, nivel_cfg). Memoizado; se rehace si crece CFG.""" with _LOCK: n = len(CFG) if _ID2FAM["map"] is not None and _ID2FAM["cfg_n"] == n: return _ID2FAM["map"] m = {} for cid, cf in CFG.items(): nm = cf.get("name", "") if not nm or nm == "?": continue for disp, kws, grp in FAMILIES: if fam_match(nm, kws): m[str(cid)] = (disp, grp, int(cf.get("level", 0) or 0)) break _ID2FAM["map"] = m; _ID2FAM["cfg_n"] = n return m def _fam_onmap_win(grp_lower): """Ventana on-map por grupo (boss/event/shadow estacionarios 45min; roaming 5min).""" return ONMAP_RALLY if grp_lower in ("boss", "event", "shadow") else ONMAP_FARM def quick_families(): """[{name, group, levels:[{lv,count}], live}] por familia — niveles de config + conteo live on-map por nivel (para colorear los botones Lvx). Solo lectura; cache 15s.""" now = time.time(); cc = _QUICK_CACHE if cc["data"] is not None and (now - cc["ts"]) < 15: return cc["data"] id2fam = _id2fam() cfg_levels = {} # familia -> set(niveles de config) for disp, grp, lv in id2fam.values(): if lv > 0: cfg_levels.setdefault(disp, set()).add(lv) with _LOCK: # snapshot mínimo bajo LOCK snap = [(o["id"], o["lv"], o["ts"]) for o in OBJS.values() if o.get("t") == MONSTER_T] live = {} # (familia, nivel) -> conteo on-map for oid, olv, ts in snap: f = id2fam.get(str(oid)) if not f: continue disp, grp, clv = f if (now - (ts or 0)) > _fam_onmap_win(grp): continue lv = int(olv or 0) or clv live[(disp, lv)] = live.get((disp, lv), 0) + 1 out = [] for disp, kws, grp in FAMILIES: if disp in QUICK_HIDE: continue lvls = sorted(l for l in cfg_levels.get(disp, ()) if l > 0) if not lvls: continue levels = [{"lv": lv, "count": live.get((disp, lv), 0)} for lv in lvls] out.append({"name": disp, "group": grp, "levels": levels, "live": sum(l["count"] for l in levels)}) cc["ts"] = now; cc["data"] = out return out def quick_list(fams, lv=0, limit=40, city=None, min_power_m=0, keys=None): """Monstruos on-map de las familias dadas (y nivel lv si >0), ordenados por distancia a `city`=(x,y). Devuelve {total, shown:[...]} para el panel de lista de Shortcuts. `keys`=lista de "Familia|Nivel" exactos (para MyPresets): si se da, ignora fams/lv.""" id2fam = _id2fam() fams = set(fams or []) keyset = set(keys or []) # {"Familia|Nivel"} para MyPresets now = time.time() cx, cy = (city or (0, 0)) minp = float(min_power_m or 0) * 1_000_000.0 rows = [] with _LOCK: for (x, y), o in OBJS.items(): if o.get("t") != MONSTER_T: continue f = id2fam.get(str(o["id"])) if not f: continue disp, grp, clv = f olv = int(o["lv"] or 0) or clv if keyset: # MyPresets: solo los pares exactos "Familia|Nivel" if (disp + "|" + str(olv)) not in keyset: continue else: if fams and disp not in fams: continue if lv and olv != lv: continue if (now - o["ts"]) > _fam_onmap_win(grp): continue c = CFG.get(str(o["id"])) or {} pw = float(c.get("power", 0) or 0) if minp and pw < minp: continue dist = round(((x - cx) ** 2 + (y - cy) ** 2) ** 0.5, 1) if (cx or cy) else 0 rows.append({"x": x, "y": y, "name": name_of(o["id"]), "level": olv, "group": _GROUP_LABEL.get(grp, "Normal"), "power": pw, "dist": dist, "seen_age": int(now - o["ts"])}) rows.sort(key=lambda r: r["dist"]) return {"total": len(rows), "shown": rows[:max(1, int(limit or 40))]} def prune(max_age=None): """Descarta lo que lleva demasiado tiempo sin verse (evita crecer sin límite).""" cutoff = int(time.time()) - int(max_age or ONMAP_RALLY * 2) with _LOCK: dead = [k for k, o in OBJS.items() if o["ts"] < cutoff] for k in dead: del OBJS[k] for xy in [k for k, ts in _UNKNOWN.items() if ts and time.time() - ts > UNKNOWN_RETRY * 4]: _UNKNOWN.pop(xy, None) # ya resuelto o irrelevante STATE["objs"] = len(OBJS) return len(dead) def status(): with _LOCK: return { "objs": len(OBJS), "batches": STATE["batches"], "cfg": STATE["cfg_n"], "wcfg": STATE["wcfg_n"], "server": STATE["server"], "last_batch": STATE["last_batch"], "scan_on": STATE["scan_on"], "sweep": dict(STATE["sweep"]), "mt": dict(STATE["mt"]), } # ============================================================================ # MARCHAS EN VUELO -> ATAQUES ACTIVOS # ============================================================================ # El reply de worldmap trae las marchas (map_target_info). Agrupadas por objetivo y # cruzadas con los objetos del mapa dan la vista de "ataques activos", que además es # lo que alimenta el motor de STEAL del bot (robar el kill a otro jugador). MARCHES = {} # troop_id -> marcha + ts_recv PLAYERS = {} # uid -> {name, tag, gid} PLAYER_POWER = {} # uid -> {power, ts} (poder exacto del power_rank_reply; el broadcast del map manda 0) GUILD_MEMBERS = {} # uid -> {name, gid, tag, ts} (roster completo de la alianza, member-list PASIVA) def ingest_powers(items): """Poder exacto por uid, capturado por el hook PASIVO del power_rank_reply (cuando el juego recibe un ranking de poder). El broadcast del mapa manda power=0, así que ésta es la fuente real.""" now = int(time.time()) with _LOCK: for it in items or []: try: 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, "ts": now} except Exception: continue if PLAYER_POWER: _pp_meta["ts"] = now _save_power() ATTACK_MTYS = {2, 10, 19, 20, 21, 43} # 2=monstruo, 10=ataque a jugador, 19/20/21=fases de rally, 43=scout ALLY_MTYS = {19, 20, 21} PHASE_NAME = {2: "march", 10: "way", 19: "wait", 20: "way", 21: "combat", 43: "scout"} # Una marcha activa debería refrescarse cada vuelta del barrido. Si lleva demasiado sin # actualizarse, lo más probable es que el rally se cancelara o ya aterrizara y el # servidor dejó de anunciarla. MARCH_STALE = 120 # Seguimos mostrando una marcha unos segundos DESPUÉS de aterrizar: una marcha de 2 s # aterriza entre dos lecturas y, sin esto, no se vería nunca. LANDED_GRACE = 12 # Tiles a los que alguien marcha pero que el barrido aún no ha visto. Se piden con un # scan DIRIGIDO: sin el nombre del monstruo la fila no pasa el filtro de grupo y Rally # Steals se los pierde (visto con el Ymir @867,627 que DaSea sí tenía y nosotros no). _UNKNOWN = {} # (x,y) -> ts de la última petición UNKNOWN_RETRY = 120 # no volver a pedir el mismo tile antes de esto def take_unknown_tiles(max_n=3): """Devuelve hasta max_n tiles pendientes de resolver y los marca como pedidos. El tope es a propósito: los scans dirigidos viajan por el MISMO canal que las marchas, así que pedir muchos de golpe reproduce la saturación que nos costó que no saliera ninguna marcha.""" now = time.time() out = [] with _LOCK: for xy, ts in sorted(_UNKNOWN.items(), key=lambda kv: kv[1]): if ts and now - ts < UNKNOWN_RETRY: continue _UNKNOWN[xy] = now out.append(xy) if len(out) >= max_n: break return out def ingest_marches(items): now = int(time.time()) with _LOCK: for m in items or []: try: trp = int(m.get("trp", 0) or 0) if not trp: continue m = dict(m); m["ts_recv"] = now MARCHES[trp] = m # destino desconocido -> apuntarlo para pedir un scan dirigido tx, ty = int(m.get("tx", 0) or 0), int(m.get("ty", 0) or 0) if tx and ty and (tx, ty) not in OBJS and (tx, ty) not in _UNKNOWN: _UNKNOWN[(tx, ty)] = 0.0 # 0 = nunca pedido -> sale en la próxima tanda except Exception: continue # purga: marchas aterrizadas hace rato o sin refresco dead = [k for k, v in MARCHES.items() if int(v.get("te", 0) or 0) <= now - LANDED_GRACE and (now - v["ts_recv"]) > MARCH_STALE] for k in dead: del MARCHES[k] def ingest_members(items): """Roster de la alianza (uid+name) capturado por el hook PASIVO GuildMembers.UpdateMembers. Da TODOS los miembros (no solo los vistos en el mapa) para el selector de whisper.""" now = int(time.time()) with _LOCK: for it in items or []: try: uid = int(it.get("uid", 0) or 0) if not uid or not it.get("name"): continue GUILD_MEMBERS[uid] = {"name": it.get("name"), "gid": int(it.get("gid", 0) or 0), "tag": it.get("tag", "") or "", "ts": now} except Exception: continue def ingest_players(items): with _LOCK: for p in items or []: try: uid = int(p.get("uid", 0) or 0) if not uid: continue prev = PLAYERS.get(uid) or {} # no pisar un nombre bueno con uno vacío: el reply no siempre trae todo PLAYERS[uid] = {"uid": uid, "name": p.get("name") or prev.get("name") or "", "tag": p.get("tag") or prev.get("tag") or "", "gid": int(p.get("gid", 0) or 0) or prev.get("gid", 0), "wx": int(p.get("wx", 0) or 0) or prev.get("wx", 0), "wy": int(p.get("wy", 0) or 0) or prev.get("wy", 0), "power": int(p.get("power", 0) or 0) or prev.get("power", 0), # __power (Bubble) "shield": int(p.get("shield", 0) or 0), # __peace_shield: >0 = burbuja "ts": int(time.time())} # última vez visto (Bubble: recientes) except Exception: continue def bubble_list(city=None, limit=40, min_power_m=0, max_age=None): """Jugadores SIN burbuja (shield<=0) vistos recientemente, ordenados por distancia a `city`. Devuelve {total, shown:[...]} para el panel Bubble de Shortcuts (estilo escáner V5).""" now = time.time() cx, cy = (city or (0, 0)) minp = float(min_power_m or 0) * 1_000_000.0 maxage = int(max_age or ONMAP_FARM) # 5 min: los jugadores se mueven rows = [] with _LOCK: for uid, p in PLAYERS.items(): if int(p.get("shield", 0) or 0) > 0: # tiene burbuja/escudo -> NO atacable, fuera continue x = int(p.get("wx", 0) or 0); y = int(p.get("wy", 0) or 0) if not x or not y: continue if (now - (p.get("ts", 0) or 0)) > maxage: continue pw = float(p.get("power", 0) or 0) ppi = PLAYER_POWER.get(uid) # poder exacto del power_rank (el broadcast manda 0) if ppi and ppi.get("power", 0) > 0: pw = float(ppi["power"]) if minp and pw < minp: continue dist = round(((x - cx) ** 2 + (y - cy) ** 2) ** 0.5, 1) if (cx or cy) else 0 rows.append({"uid": uid, "name": p.get("name", ""), "tag": p.get("tag", ""), "x": x, "y": y, "power": pw, "dist": dist, "seen_age": int(now - (p.get("ts", 0) or 0))}) rows.sort(key=lambda r: r["dist"]) return {"total": len(rows), "shown": rows[:max(1, int(limit or 40))]} def player_state(uid): """DEAD MAN: estado actual de un jugador por uid (coords/power/shield/edad). None si el escáner no lo ha visto. Lock-safe, solo lectura. `power` usa el exacto de PLAYER_POWER si existe (el broadcast manda 0).""" uid = int(uid or 0) if not uid: return None now = time.time() with _LOCK: p = PLAYERS.get(uid) if not p: return None pw = float(p.get("power", 0) or 0) ppi = PLAYER_POWER.get(uid) if ppi and ppi.get("power", 0) > 0: pw = float(ppi["power"]) return {"uid": uid, "name": p.get("name", ""), "tag": p.get("tag", ""), "x": int(p.get("wx", 0) or 0), "y": int(p.get("wy", 0) or 0), "power": pw, "shield": int(p.get("shield", 0) or 0), "age": int(now - (p.get("ts", 0) or 0))} def players_search(q, limit=20): """DEAD MAN: busca jugadores por nombre (substring, case-insensitive) o uid, para el selector de targets. Lock-safe, solo lectura. Ordena por poder desc.""" q = str(q or "").strip().lower() if not q: return [] out = [] with _LOCK: for uid, p in PLAYERS.items(): nm = (p.get("name") or "").lower() if q in nm or q in str(uid): pw = float(p.get("power", 0) or 0) ppi = PLAYER_POWER.get(uid) if ppi and ppi.get("power", 0) > 0: pw = float(ppi["power"]) out.append({"uid": uid, "name": p.get("name", ""), "tag": p.get("tag", ""), "x": int(p.get("wx", 0) or 0), "y": int(p.get("wy", 0) or 0), "power": pw}) out.sort(key=lambda r: -(r["power"] or 0)) return out[:max(1, int(limit or 20))] def attacks(own_uid=0, own_gid=0, include_helps=False, skip_relocations=True): """Ataques activos en el formato que consume el bot. Cada fila: mty, tx, ty, uid, tag, name, tg_id, tlevel, tname, tgroup, eta, phase, kind, count. Es una versión directa de lo que servía `/api/attacks` del escáner, sin sus filtros de relocation/SVS (dependían de su censo completo de jugadores). """ now = int(time.time()) out = [] with _LOCK: # índice tile -> jugador: identifica de quién es el castillo atacado tile_owner = {} for p in PLAYERS.values(): wx, wy = int(p.get("wx", 0) or 0), int(p.get("wy", 0) or 0) if wx and wy: tile_owner[(wx, wy)] = p # agrupar por objetivo: varias marchas al mismo tile son UN ataque con count>1 by_tgt = {} for m in MARCHES.values(): mty = int(m.get("mty", 0) or 0) if mty not in ATTACK_MTYS: continue te = int(m.get("te", 0) or 0) if te <= now - LANDED_GRACE: # ya aterrizó hace rato continue if (now - m["ts_recv"]) > MARCH_STALE: # sin refrescar: probablemente cancelada continue tx, ty = int(m.get("tx", 0) or 0), int(m.get("ty", 0) or 0) if not tx or not ty: continue by_tgt.setdefault((tx, ty), []).append(m) for (tx, ty), ms in by_tgt.items(): # la marcha que ANTES llega manda: es la que define el eta a batir lead = min(ms, key=lambda z: int(z.get("te", 0) or 0)) uid = int(lead.get("ow", 0) or 0) pl = PLAYERS.get(uid) or {} gid = int(pl.get("gid", 0) or 0) if own_uid and uid == own_uid: continue # mis propias marchas no son ataques ajenos owner_here = tile_owner.get((tx, ty)) if skip_relocations and owner_here and uid and int(owner_here.get("uid", 0) or 0) == uid: continue # RELOCATION: el jugador vuelve/refuerza su propia ciudad, no ataca o = OBJS.get((tx, ty)) or {} ot = int(o.get("t", 0) or 0) oid = int(o.get("id", 0) or 0) c = CFG.get(str(oid)) or {} tlevel = int(o.get("lv", 0) or 0) or int(c.get("level", 0) or 0) if ot == MONSTER_T and oid: tname = name_of(oid, ot) tgroup = group_of(tname) if not tname.startswith("id") else "" else: # No es un monstruo: casi siempre un castillo de jugador (ataque PvP). # Mostramos el nombre del DUEÑO, que es lo útil, en vez de un id críptico. owner = tile_owner.get((tx, ty)) tname = (owner or {}).get("name", "") if owner else "" tgroup = T_LABEL.get(ot, "") if not tname and ot: tname = T_LABEL.get(ot, "") or "" mty = int(lead.get("mty", 0) or 0) if (not include_helps) and own_gid and gid and gid == own_gid and ot != MONSTER_T: continue # refuerzo a un castillo de la propia alianza: no es un ataque out.append({ "mty": mty, "tx": tx, "ty": ty, "uid": uid, "tag": pl.get("tag", ""), "name": pl.get("name", ""), "tg_id": oid or int(lead.get("tsub", 0) or 0), "tlevel": tlevel, "tname": tname, "tgroup": tgroup, "eta": max(0, int(lead.get("te", 0) or 0) - now), "phase": PHASE_NAME.get(mty, ""), "kind": "alliance" if mty in ALLY_MTYS else "solo", "count": len(ms), "landed": int(lead.get("te", 0) or 0) <= now, "war_id": int(lead.get("war", 0) or 0), "sx": int(lead.get("sx", 0) or 0), "sy": int(lead.get("sy", 0) or 0), }) out.sort(key=lambda a: a["eta"]) return out def marches_status(): now = int(time.time()) with _LOCK: return {"marches": len(MARCHES), "players": len(PLAYERS), "active": sum(1 for m in MARCHES.values() if int(m.get("te", 0) or 0) > now)} def uid_at(wx, wy): """uid del jugador cuyo castillo está en ese tile. Sirve para deducir el uid propio a partir de las coordenadas de la ciudad, que es lo único que expone la cuenta.""" wx, wy = int(wx or 0), int(wy or 0) if not wx or not wy: return 0, 0 with _LOCK: for p in PLAYERS.values(): if int(p.get("wx", 0) or 0) == wx and int(p.get("wy", 0) or 0) == wy: return int(p.get("uid", 0) or 0), int(p.get("gid", 0) or 0) return 0, 0 # ============================================================================ # CONTINUIDAD DEL BARRIDO (persistir el índice entre freezes/reinicios) # ============================================================================ # El sweep del agente vive en memoria: al reengancharse tras un freeze/reinicio, su # índice volvía a 0 y el barrido re-empezaba por el núcleo una y otra vez, sin llegar # nunca a la periferia profunda. Aquí guardamos el índice en disco y se lo devolvemos al # agente en su arranque (mensaje scan_boot -> resume_idx), de modo que las vueltas al # mapa se completan aunque el agente reinicie muchas veces. import os as _os _PROG_PATH = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "scan_progress.json") _prog = {"idx": 0, "order": 0, "pass": 0, "ts": 0} _prog_last_save = 0.0 try: with open(_PROG_PATH) as _f: _prog.update(json.load(_f)) except Exception: pass def save_progress(idx, order, passes, min_interval=15): """Guarda el índice del barrido (throttled). Escritura atómica: no dejar un JSON a medias si el proceso muere durante el volcado.""" global _prog_last_save now = time.time() idx = int(idx or 0); order = int(order or 0) if idx <= 0: return _prog["idx"] = idx; _prog["order"] = order; _prog["pass"] = int(passes or 0); _prog["ts"] = int(now) if now - _prog_last_save < min_interval: return _prog_last_save = now try: tmp = _PROG_PATH + ".tmp" with open(tmp, "w") as f: json.dump(_prog, f) _os.replace(tmp, _PROG_PATH) except Exception: pass def resume_index(order=0): """Índice desde el que el agente debe retomar el barrido. Se clampa al tamaño de la vuelta actual por si los parámetros del sweep cambiaron (order distinto).""" idx = int(_prog.get("idx", 0) or 0) order = int(order or _prog.get("order", 0) or 0) if order and idx >= order: return 0 return max(0, idx) # ============================================================================ # PERSISTENCIA de OBJS (monstruos) + CFG (nombres) + PLAYER_POWER (poder) # ============================================================================ # Tras un reinicio del backend, sin esto: la lista de monstruos queda VACÍA hasta que el # barrido re-cubre el mapa (minutos) y el poder desaparece (el poll es cada 48h). Con esto, # ambos están al instante (último estado conocido); los filtros de seen_age descartan lo viejo. _CFG_PATH = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "scan_cfg.json") _cfg_last_save = 0.0 try: with open(_CFG_PATH) as _f: CFG.update(json.load(_f)) STATE["cfg_n"] = len(CFG) except Exception: pass def _save_cfg(min_interval=60): global _cfg_last_save now = time.time() if now - _cfg_last_save < min_interval or not CFG: return _cfg_last_save = now try: with _LOCK: data = dict(CFG) tmp = _CFG_PATH + ".tmp" with open(tmp, "w") as f: json.dump(data, f) _os.replace(tmp, _CFG_PATH) except Exception: pass _OBJS_PATH = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "scan_objs.json") _objs_last_save = 0.0 try: with open(_OBJS_PATH) as _f: for _o in json.load(_f): _x = int(_o.get("x", 0) or 0); _y = int(_o.get("y", 0) or 0) if _x and _y: OBJS[(_x, _y)] = {"t": int(_o.get("t", 0) or 0), "id": int(_o.get("id", 0) or 0), "lv": int(_o.get("lv", 0) or 0), "x": _x, "y": _y, "ts": int(_o.get("ts", 0) or 0)} STATE["objs"] = len(OBJS) except Exception: pass _PLAYERS_PATH = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "scan_players.json") try: with open(_PLAYERS_PATH) as _f: for _p in json.load(_f): _u = int(_p.get("uid", 0) or 0) if _u: PLAYERS[_u] = _p except Exception: pass def save_objs(min_interval=45): """Guarda OBJS a disco (throttled, atómico). Lo llama el backend en el ciclo de métricas.""" global _objs_last_save now = time.time() if now - _objs_last_save < min_interval: return _objs_last_save = now try: with _LOCK: data = list(OBJS.values()) tmp = _OBJS_PATH + ".tmp" with open(tmp, "w") as f: json.dump(data, f) _os.replace(tmp, _OBJS_PATH) except Exception: pass try: with _LOCK: pdata = list(PLAYERS.values()) tmp = _PLAYERS_PATH + ".tmp" with open(tmp, "w") as f: json.dump(pdata, f) _os.replace(tmp, _PLAYERS_PATH) except Exception: pass _PP_PATH = _os.path.join(_os.path.dirname(_os.path.abspath(__file__)), "player_power.json") _pp_meta = {"ts": 0} # epoch de la última captura de ranking con éxito _pp_last_save = 0.0 try: with open(_PP_PATH) as _f: _d = json.load(_f) for _k, _v in (_d.get("power") or {}).items(): try: PLAYER_POWER[int(_k)] = _v except Exception: pass _pp_meta["ts"] = int(_d.get("ts", 0) or 0) except Exception: pass def _save_power(min_interval=5): global _pp_last_save now = time.time() if now - _pp_last_save < min_interval: return _pp_last_save = now try: with _LOCK: data = {"ts": int(_pp_meta.get("ts", 0) or 0), "power": {str(k): v for k, v in PLAYER_POWER.items()}} tmp = _PP_PATH + ".tmp" with open(tmp, "w") as f: json.dump(data, f) _os.replace(tmp, _PP_PATH) except Exception: pass def power_last_ts(): """Epoch de la última captura de ranking (para decidir si toca re-pollear cada ~48h).""" return int(_pp_meta.get("ts", 0) or 0)