eve-watcher/eve_combat_watcher.py

#!/usr/bin/env python3
"""Gamelog watcher — everything EVE logs but has no API for.

Tails EVE's local Gamelogs (Documents/EVE/logs/Gamelogs) and alerts via the same
channels as the ore-hold watcher (ntfy + Windows toast + Discord). Reading the log
file is legit — it never touches the game. Runs on the PC you play on.

Catches:
  • TACKLED / EWAR'd   (notify warp-scramble/disrupt/jam/web/neut) — urgent, no cooldown
  • Incoming damage    (combat 'from') — "rats?" — cooldown'd
  • Hold full          (notify cargo/ore hold full)
  • Capacitor empty    (notify)
  • Bounty income      (bounty lines) — accumulates a session total, milestone pings

Config: reuses config.ini, optional [combat] section:
  gamelogs_dir =        ; auto-detects ~/Documents/EVE/logs/Gamelogs
  cooldown_secs = 60
  min_damage    = 1
  bounty_milestone = 25000000   ; ping every this much ratting ISK

Run:  python eve_combat_watcher.py        (--test fires one alert)
"""
import glob
import os
import re
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import eve_orehold_watcher as w   # reuse load_config() + notify()

TAG = re.compile(r"<[^>]+>")
INCOMING = re.compile(r"\(combat\)\s*([\d,]+)\s+from\s+(.+?)\s*[-\n]", re.IGNORECASE)
DANGER = re.compile(r"\(notify\).*(warp (?:scrambl|disrupt)|unable to warp|"
                    r"jam|target(?:ing)? (?:disrupt|jam)|energy neutraliz|web|stasis)",
                    re.IGNORECASE)
HOLDFULL = re.compile(r"\(notify\).*(cargo (?:hold )?is full|hold is full|"
                      r"not enough (?:cargo )?space|deactivat.*full|full.*deactivat)",
                      re.IGNORECASE)
CAPEMPTY = re.compile(r"\(notify\).*(capacitor is empty|not enough (?:capacitor|energy))",
                      re.IGNORECASE)
BOUNTY = re.compile(r"\(bounty\)\s*([\d,]+(?:\.\d+)?)\s*ISK", re.IGNORECASE)


def find_gamelogs(cp):
    d = cp.get("combat", "gamelogs_dir", fallback="").strip() if cp.has_section("combat") else ""
    if d and os.path.isdir(d):
        return d
    for base in (os.path.expanduser("~/Documents/EVE/logs/Gamelogs"),
                 os.path.expanduser("~/OneDrive/Documents/EVE/logs/Gamelogs")):
        if os.path.isdir(base):
            return base
    return None


def newest(d):
    files = glob.glob(os.path.join(d, "*.txt"))
    return max(files, key=os.path.getmtime) if files else None


def main():
    cp = w.load_config()
    g = lambda k, d: (cp.getint("combat", k, fallback=d) if cp.has_section("combat") else d)
    cooldown, min_dmg, milestone = g("cooldown_secs", 60), g("min_damage", 1), g("bounty_milestone", 25000000)

    if "--test" in sys.argv:
        w.notify(cp, "Gamelog watcher test", "Combat/tackle/hold/bounty alerts will reach you.",
                 priority="high", tags="crossed_swords")
        return

    w.heartbeat(cp, "combat")                     # announce we're alive first
    gdir = find_gamelogs(cp)
    while not gdir:                                # don't die if EVE isn't running yet
        print("[gamelog] no Gamelogs dir yet (EVE not started?) — retrying in 30s")
        time.sleep(30)
        gdir = find_gamelogs(cp)
    print(f"[gamelog] watching {gdir}")
    cur = newest(gdir)
    fh = open(cur, "r", encoding="utf-8", errors="ignore") if cur else None
    if fh:
        fh.seek(0, os.SEEK_END)
    last_dmg = 0.0
    bounty_total = 0.0
    bounty_reported = 0.0

    while True:
        nl = newest(gdir)
        if nl and nl != cur:
            cur = nl
            if fh:
                fh.close()
            fh = open(cur, "r", encoding="utf-8", errors="ignore")
            bounty_total = bounty_reported = 0.0   # new session
            print(f"[gamelog] -> {os.path.basename(cur)}")
        if not fh:
            time.sleep(3); continue
        line = fh.readline()
        if not line:
            time.sleep(1); continue
        clean = TAG.sub("", line)

        if "(notify)" in line:
            if DANGER.search(clean):
                w.notify(cp, "⚠ TACKLED / EWAR",
                         "Scrambled/jammed/webbed — you may not be able to warp. ACT NOW.",
                         priority="urgent", tags="rotating_light"); continue
            if HOLDFULL.search(clean):
                w.notify(cp, "Hold full", "Your hold is full — compress / unload / swap.",
                         priority="high", tags="package"); continue
            if CAPEMPTY.search(clean):
                w.notify(cp, "Capacitor empty", "Cap's out — modules dropping.",
                         priority="default", tags="battery"); continue
            continue
        if "(bounty)" in line:
            mb = BOUNTY.search(clean)
            if mb:
                bounty_total += float(mb.group(1).replace(",", ""))
                if bounty_total - bounty_reported >= milestone:
                    bounty_reported = bounty_total
                    w.notify(cp, "Ratting income",
                             f"~{bounty_total/1e6:.0f}M ISK in bounties this session.",
                             priority="low", tags="moneybag")
            continue
        if "(combat)" in line:
            m = INCOMING.search(clean)
            if not m:
                continue
            dmg = int(m.group(1).replace(",", ""))
            if dmg < min_dmg or time.time() - last_dmg < cooldown:
                continue
            last_dmg = time.time()
            w.notify(cp, "Taking damage — rats?",
                     f"Incoming fire from {m.group(2).strip()} ({dmg}). Engage drones, "
                     f"watch shield.", priority="urgent", tags="crossed_swords")
            print(f"[gamelog] damage {dmg} from {m.group(2).strip()}")


if __name__ == "__main__":
    main()