108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Chat-intel watcher — closest thing to a Local early-warning, from the logs.
|
|
|
|
Tails EVE's Chatlogs (Documents/EVE/logs/Chatlogs) and pings (ntfy + toast + Discord)
|
|
when a configured keyword or hostile name appears in the watched channels. Chatlogs
|
|
are UTF-16. Reading them is legit; it never touches the game.
|
|
|
|
LIMIT: chatlogs record *messages*, not live enters/leaves — so this catches people
|
|
*talking* (intel channels, someone in Local), not a silent new entrant. Watch the
|
|
in-game Local window for that.
|
|
|
|
Config: reuses config.ini, optional [chat] section:
|
|
chatlogs_dir = ; auto-detects ~/Documents/EVE/logs/Chatlogs
|
|
channels = Local,Intel ; filename prefixes to watch
|
|
keywords = ; comma list; if empty, alerts on EVERY msg in these channels
|
|
cooldown_secs = 20
|
|
|
|
Run: python eve_chat_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
|
|
|
|
LINE = re.compile(r"^\[\s*[\d.]+\s+[\d:]+\s*\]\s*(.+?)\s*>\s*(.*)$")
|
|
|
|
|
|
def find_dir(cp):
|
|
d = cp.get("chat", "chatlogs_dir", fallback="").strip() if cp.has_section("chat") else ""
|
|
if d and os.path.isdir(d):
|
|
return d
|
|
for base in (os.path.expanduser("~/Documents/EVE/logs/Chatlogs"),
|
|
os.path.expanduser("~/OneDrive/Documents/EVE/logs/Chatlogs")):
|
|
if os.path.isdir(base):
|
|
return base
|
|
return None
|
|
|
|
|
|
def newest_for(d, prefix):
|
|
files = glob.glob(os.path.join(d, f"{prefix}_*.txt"))
|
|
return max(files, key=os.path.getmtime) if files else None
|
|
|
|
|
|
def main():
|
|
cp = w.load_config()
|
|
sec = cp.has_section("chat")
|
|
channels = [c.strip() for c in (cp.get("chat", "channels", fallback="Local,Intel")
|
|
if sec else "Local,Intel").split(",") if c.strip()]
|
|
keywords = [k.strip().lower() for k in (cp.get("chat", "keywords", fallback="")
|
|
if sec else "").split(",") if k.strip()]
|
|
cooldown = cp.getint("chat", "cooldown_secs", fallback=20) if sec else 20
|
|
|
|
if "--test" in sys.argv:
|
|
w.notify(cp, "Chat watcher test", "Chat-intel alerts will reach you.",
|
|
priority="high", tags="speech_balloon")
|
|
return
|
|
|
|
w.heartbeat(cp, "chat") # announce we're alive first
|
|
d = find_dir(cp)
|
|
while not d: # wait for EVE to create the logs
|
|
print("[chat] no Chatlogs dir yet (EVE not started?) — retrying in 30s")
|
|
time.sleep(30)
|
|
d = find_dir(cp)
|
|
print(f"[chat] watching {channels} in {d}; keywords={keywords or 'ALL'}")
|
|
|
|
handles = {} # channel -> (path, fh)
|
|
last = {} # channel -> last alert time
|
|
while True:
|
|
for ch in channels:
|
|
nl = newest_for(d, ch)
|
|
if not nl:
|
|
continue
|
|
path, fh = handles.get(ch, (None, None))
|
|
if nl != path:
|
|
if fh:
|
|
fh.close()
|
|
fh = open(nl, "r", encoding="utf-16", errors="ignore")
|
|
fh.seek(0, os.SEEK_END)
|
|
handles[ch] = (nl, fh)
|
|
continue
|
|
line = fh.readline()
|
|
if not line:
|
|
continue
|
|
m = LINE.match(line.strip())
|
|
if not m:
|
|
continue
|
|
speaker, msg = m.group(1), m.group(2)
|
|
if speaker in ("EVE System",):
|
|
continue
|
|
if keywords and not any(k in msg.lower() or k in speaker.lower()
|
|
for k in keywords):
|
|
continue
|
|
if time.time() - last.get(ch, 0) < cooldown:
|
|
continue
|
|
last[ch] = time.time()
|
|
w.notify(cp, f"{ch}: {speaker}", msg[:180],
|
|
priority="high", tags="speech_balloon,eyes")
|
|
print(f"[chat] {ch} {speaker}: {msg[:80]}")
|
|
time.sleep(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|