A small program you leave running while you play. It keeps this site honest about your world.
It opens Player.log and the ChatLogs folder, which the
game writes as you play, and the Books folder, where the game puts any
book you save. From those it sends only the specific reports listed below. It never uploads a file, and
it never sends your chat, your inventory or anything about your account.
The only things it writes are two small notes files beside itself, so it
remembers what it already reported, and a copy of every book you open, saved into
the game’s own Books folder exactly as the in-game save button
would. Those stay on your machine: a saved book is never uploaded. Every line of it is printed on this
page, so you do not have to take this box’s word for it.
Project Gorgon keeps a few plain text files on your own hard drive. As you play, it logs the NPCs you talk to, what dropped off the thing you just killed, and the chat around you. When you save a book in game, such as Today’s Weather or the Gardening Almanac, it writes that book out as a file too. This program reads those files, picks out the handful of things the whole community benefits from knowing, and posts those here: weather, gardens, the blood season, the daily missions, what creatures actually drop.
None of that can come from the game's data files. It has to be seen by somebody who was there. A dozen people running this while they play is what keeps the Clock, Weather, Garden and Bestiary pages worth opening.
It is one file, it installs nothing, and it needs no key or account. Stop it any time by closing its window.
All of it is public world knowledge, the sort of thing players shout in global chat. Each line below feeds the page next to it.
Your character name travels with the mob-drop and arena reports, so that nonsense reports can be spotted and thrown out. That is the only thing about you that is sent.
Not one byte of any of this leaves your PC:
Every report is built from one specific pattern the program goes looking for. Everything else in the log is read past and forgotten. And it only ever reads the game's files, so it cannot damage a character or do anything the game would object to. It is not a mod, an add-on or an overlay, and the game never knows it is there.
pip install, no extra libraries.Windows: py gorgon_monitor.py
Mac/Linux: python3 gorgon_monitor.pyCtrl-C in that window to stop
it, and nothing is left running afterwards.Which world you are on is read out of the game's own log, so you do not have to say. A world this site does not track, the public beta for instance, is recognised and reporting simply switches itself off.
py gorgon_monitor.py --help every option
py gorgon_monitor.py --version which build you have
py gorgon_monitor.py --game-dir "C:\path" name your game folder yourself
py gorgon_monitor.py --game-server dreva skip world detection
py gorgon_monitor.py --save-books off do not save any books for me--game-dir "path/to/Project Gorgon". It deliberately refuses to run on
without a folder, because a monitor that reports nothing looks exactly like a quiet
evening in game.The whole program, exactly as you would download it. Nothing is minified, obfuscated or hidden, and the comments are written to be read by somebody who does not write code. If a claim above matters to you, this is where you check it.
PART 4 is the part to read if you are suspicious. Every pattern there
quotes a real line from a real log, so you can open your own Player.log in
Notepad and see precisely what this program is allowed to notice.
#!/usr/bin/env python3 """ Gorgon Monitor | the Gorgonzola community reporter for Project Gorgon ======================================================================= WHAT THIS IS A small program you leave running while you play. It reads the log files the game already writes on your own PC, notices a handful of things the whole community benefits from knowing (today's weather, the blood season, what a creature dropped) and posts those to the community site: https://gorgonzola.spellboundtavern.com It never touches the game itself. It cannot press keys, move your character, read your screen or talk to the game server. It only reads files, the same way Notepad would. That is the whole trick: Project Gorgon writes plain text logs of what you saw, plus a file for any book you save, and this reads them. You do not need to understand any of the code below to use it. Scroll to "HOW TO RUN" and you are done. The comments are there for the curious, and so that anybody can check for themselves that it does what this header claims. WHAT IT SENDS TO THE SITE All of this is public world knowledge, the sort of thing players shout in global chat. Never anything private about you or your account. - Weather and Garden The in-game "Today's Weather" and "Gardening Almanac" books, so the site's live boards match the world you are actually playing on. - Warden alerts The "Active Warden Alerts" book. - Aurest mission Lady Alethina's daily mission. - Agrashab's bounty The Fae Realm god's daily group bounty. - Blood target The day's vampire blood season. - Casino tips and fights Kuzavek's hot tips and Kyuroma's current matchup, for the community gambling page. - Arena bet results Whether the fighter you backed won or lost. The WAGER AMOUNT IS NEVER SENT. Only who fought and who won, which is what fighter win/loss ratios are made of. - Mob drops What creatures dropped, for the community bestiary. This is the big one: nobody can map a drop table alone, and a few dozen players playing normally will map the whole game. - Death XP How much XP each cause of death awards, and the whole ladder of repeat awards it pays after that, filling in the blanks on the Dying page. Nobody dies of everything, which is exactly why this is pooled. Your character's name rides along with the mob-drop and arena-bet reports, so duplicate or nonsense reports can be spotted and dropped. That is the only thing about you that is sent. WHAT IT NEVER SENDS Not your chat, not your whispers, not your guild chat, not your inventory, not your councils, not your account, not your password, not your IP beyond the ordinary fact of making a web request, and no file from your computer. Nothing is uploaded wholesale: every report above is built from a specific pattern this program looks for, and everything else in the logs is ignored and forgotten. It is also read-only against the game. It opens the game's log files for reading and nothing else, so it cannot corrupt a character or trip anything the game would object to. WHAT IT DOES FOR YOU IN RETURN - Saves work-order books to your Books folder by itself, so you can skip the in-game "save this book" click. Gorgonzola's Shop P&L page reads those files to book what a work order cost you. - Keeps a local research file (deaths_observed.json) of the death XP you have personally seen, next to this script. - Keeps the site's live boards correct for YOUR world, which is what makes the clock, weather, garden and daily pages worth opening. WHAT YOU NEED - Python 3.9 or newer. Nothing else: no "pip install", no accounts, no API key, no configuration. One file, standard library only. - Project Gorgon installed. Its folder is found for you on Windows, WSL, macOS, Linux and Steam Proton. If yours is somewhere unusual, name it once with --game-dir and nothing else needs changing. - Nothing else is required. If the files quests.json / deaths.json happen to be sitting next to this script it will use them for nicer names, and it works perfectly well without them. HOW TO RUN Save this file somewhere, open a terminal in that folder, and run: python3 gorgon_monitor.py On Windows that is usually: py gorgon_monitor.py It opens with a summary: the version, the world you are on, the game folder it found, and the list of what it will report. Read that, then play. Leave the window open in the background; it prints a line whenever it reports something. Press Ctrl-C in that window to stop it, and nothing is left running afterwards. Useful extras: python3 gorgon_monitor.py --game-dir "path" Name your game folder python3 gorgon_monitor.py --game-server dreva Skip world auto-detection python3 gorgon_monitor.py --save-books off Do not auto-save books python3 gorgon_monitor.py --version Print the version python3 gorgon_monitor.py --help List every option Which world you are on is detected from the game's own log, so you normally do not have to say. A world the site does not track (the public beta, for instance) is detected and world reporting simply switches itself off. IF SOMETHING LOOKS WRONG - "Could not find your Project Gorgon folder": it prints every place it looked. Point it at yours with --game-dir "path/to/Project Gorgon". It will not run on regardless, because a monitor that reports nothing looks exactly like a quiet evening in game. - Nothing is ever printed: it only reports what happens FROM NOW ON, and it only reports things that are worth reporting. Read a weather book in game and you should see a line within about ten seconds. - "not a world this site tracks": you are on beta or a private world. Expected; everything else keeps working. - A "429" or "rate limited" message: somebody already reported that, seconds ago. Harmless, and it backs off on its own. OPTIONAL EXTRAS (off unless you set the matching environment variable) - redwing.bet mirror. Set GORGON_KEY (hot tips) and/or LIVE_REPORT_SECRET (live fights) to forward casino data to redwing.bet as well. Left unset, as it is for everyone by default, that code never runs. HOW THIS FILE IS LAID OUT If you do want to read the code, it goes in this order: PART 1 Talking to the website (one small POST helper) PART 2 Reading the game's books (weather / almanac text -> data) PART 3 Settings (finding your game, and tuning knobs) PART 4 Patterns (what a log line looks like) PART 5 The monitor (the loop, and one method per thing it watches for) PART 6 Command line (options, and the starting point) Every pattern in PART 4 is a real line seen in a real log, quoted in the comment above it, so you can check the claim against your own Player.log. VERSION 1.5.0 (2026-09-17) Mob drops now say which world they came from, and when they were seen. Both are for one new thing: vampire loot names the day's blood target, so a creature that pays out Lifeblood during a season carries that blood, even where the site's bestiary guessed otherwise. Nothing new is read and nothing new is sent about you. 1.4.0 (2026-09-15) Saves every book you open, not a curated three. Opening a book is the only signal there is that you wanted it, and the weather, garden and warden reports are all built from a saved book file, so the old default let you read the almanac in game and report nothing. Saved books stay on your disk, as they always have. Narrow it with --save-books if you want the old behaviour. 1.3.1 (2026-09-15) Wording only. The header called the game's files a "diary", which blurred two different things: logs the game writes as you play, and a book file that exists only because you saved that book. Nothing it reads or sends has changed. 1.3.0 (2026-09-08) Death XP actually uploads. The site was refusing every copy but the author's, so a shared monitor recorded its findings to disk and nowhere else. The full set of award values now goes up too, not only the highest. 1.2.0 (2026-09-08) Runs on anybody's machine: finds your Project Gorgon folder on Windows, WSL, macOS, Linux and Steam Proton, and says which one it found. Reads chat as utf-8 (an accented name used to end the run on Windows), prints a summary of what it is about to do, and stops cleanly on Ctrl-C. 1.1.0 (2026-09-08) The arena roster comes from the site rather than four built-in lists, so a new fighter needs no new download. 1.0.0 (2026-09-07) First public release. In the Gorgonzola source tree this file is gorgon_monitor.py, the self-contained, shareable twin of the maintainer's own chat_monitor.py. """ __version__ = "1.5.0" import os import re import sys import json import time import hashlib import datetime import collections import urllib.request import urllib.error import urllib.parse from pathlib import Path # ============================================================================== # PART 1 | TALKING TO THE WEBSITE # # One function does every report in this program: hand it a web address and a bundle # of facts, and it posts them as JSON. It never raises an error upwards: a website # that is down, slow or unhappy prints a line, and the monitor carries on playing. # # (These two helpers are copied in from the Gorgonzola tools rather than imported, so # that this stays one file you can read, save anywhere and run with no installation.) # ============================================================================== # How many reports the site has accepted this run. Counted here because every push goes # through post_json, and printed on the way out, so stopping the program tells you what # it actually did rather than leaving you to scroll back. REPORTS_MADE = 0 def post_json(server_url, endpoint, payload, headers=None): """POST `payload` as JSON to server_url+endpoint; return the parsed response dict — the server's error body (plus `status`) on an HTTP error, None only when the connection itself failed. Logged, never raised.""" url = server_url.rstrip('/') + endpoint data = json.dumps(payload).encode('utf-8') hdrs = {'Content-Type': 'application/json', 'User-Agent': f'GorgonMonitor/{__version__}'} if headers: hdrs.update(headers) req = urllib.request.Request(url, data=data, method='POST', headers=hdrs) try: with urllib.request.urlopen(req, timeout=15) as resp: body = json.loads(resp.read().decode('utf-8')) if isinstance(body, dict) and body.get('ok'): global REPORTS_MADE REPORTS_MADE += 1 return body except urllib.error.HTTPError as e: # Return the error BODY (never None) so a caller can tell a 429 "come back in # N seconds" from a 400 "this payload is wrong" — every caller gates success # on result.get('ok'), so an error dict still reads as failure. body = {} try: body = json.loads(e.read().decode('utf-8')) print(f" ERROR {e.code}: {body.get('error', e.reason)}", file=sys.stderr) except Exception: print(f" HTTP {e.code}: {e.reason}", file=sys.stderr) if not isinstance(body, dict): body = {'error': str(body)} body.setdefault('error', str(e.reason)) body['status'] = e.code if 'retry_after' not in body: try: body['retry_after'] = int(e.headers.get('Retry-After')) except (TypeError, ValueError): pass return body except urllib.error.URLError as e: print(f" Connection failed: {e.reason}", file=sys.stderr) print(f" Is the server running at {server_url}?", file=sys.stderr) return None def book_timestamp(path): """Parse the _YYMMDD_HHMMSS stamp out of a book filename -> datetime, or None.""" m = re.search(r'_(\d{6})_(\d{6})', path.name) if not m: return None try: return datetime.datetime.strptime(m.group(1) + m.group(2), '%y%m%d%H%M%S') except Exception: return None # ============================================================================== # PART 2 | READING THE GAME'S BOOKS # # Two of the game's in-game books are worth sharing: Today's Weather and the # Gardening Almanac. Both are written as marked-up text, roughly like a web page, so # these two functions pick the facts back out of them. # # The almanac's times are all relative ("ends in 2 hours"), never a clock time, which # is why the moment the book was read has to be carried along with it. # ============================================================================== def parse_weather_text(text): """TodaysWeather book -> [(zone, weather)]. Each line is '<b>Zone</b>: Weather'.""" zones = [] for m in re.finditer(r'<b>([^<]+)</b>\s*:\s*([^\n<]+)', text): zones.append((m.group(1).strip(), m.group(2).strip())) return zones def _parse_duration_secs(s): """'1 day 3 hours 55 minutes' -> total seconds (int).""" s = s.lower() total = 0 for val, unit in re.findall(r'(\d+)\s*(day|hour|min|sec)', s): v = int(val) if unit.startswith('day'): total += v * 86400 elif unit.startswith('hour'): total += v * 3600 elif unit.startswith('min'): total += v * 60 elif unit.startswith('sec'): total += v return total def parse_almanac_text(text, file_time): """GardeningAlmanac book -> [event dicts], with file_time as the timestamp reference the book's relative times ('ends in 2 hours') resolve against.""" clean = re.sub(r'</?indent[^>]*>', '', text) # strip <indent=N> tags events = [] for section_m in re.finditer(r'<h1>(.*?)</h1>(.*?)(?=<h1>|$)', clean, re.DOTALL): header = section_m.group(1).strip() body = section_m.group(2).strip() is_current = 'current' in header.lower() for block in re.split(r'(?=<h2>)', body): title_m = re.search(r'<h2>([^<]+)</h2>', block) if not title_m: continue title = title_m.group(1).strip() desc_m = re.search(r'</h2>\s*([^<\n]+)', block) desc = desc_m.group(1).strip() if desc_m else '' time_m = re.search(r'<i>([^<]+)</i>', block) if not time_m: continue dur_m = re.search(r'(?:ends|starts)\s+in\s+(.+)', time_m.group(1).strip(), re.IGNORECASE) if not dur_m: continue dur_secs = _parse_duration_secs(dur_m.group(1)) abs_ts_ms = int((file_time.timestamp() + dur_secs) * 1000) zone_m = re.search(r'\s+[Ii]n\s+(.+)$', title) zone = zone_m.group(1).strip() if zone_m else '' crop_m = re.match(r'(.+?)\s+Season', title) crop = crop_m.group(1) if crop_m else title.split()[0] events.append({ 'active': is_current, 'title': title, 'desc': desc, 'crop': crop, 'zone': zone, 'abs_ts_ms': abs_ts_ms, # No 'emoji': the server computes it from garden_crops.json on save # (GET /api/garden-options), so the monitor stays plant-agnostic. }) return events # ============================================================================== # PART 3 | SETTINGS # # Where your game keeps its logs, and a handful of timings and safety limits. # # You should not have to change anything here. The game folder is found for you on # Windows, WSL, macOS, Linux and Steam Proton, and if the search comes up empty the # program says exactly where it looked and stops, rather than running on doing # nothing. Point it straight at the folder with --game-dir if your install is # somewhere unusual. # # Anything reading os.environ below is an optional extra that stays switched off # unless you set it. # ============================================================================== DEFAULT_SERVER = "https://gorgonzola.spellboundtavern.com/" # Project Gorgon is a Unity game, and Unity puts a game's files in a different place on # every operating system. These are those places, in the order they are tried. The # first one that actually holds ChatLogs, Books or Player.log wins. # # A `*` is a wildcard: under WSL the Windows drive letter and your Windows username are # both unknown, and under Steam Proton the game sits inside a numbered prefix, so those # entries are searched rather than named. GAME_DIR_CANDIDATES = [ # Windows (the ordinary case). LOCALAPPDATA is ...\AppData\Local; the sibling # LocalLow is where Unity writes, and it has no environment variable of its own. ('Windows', '{USERPROFILE}/AppData/LocalLow/Elder Game/Project Gorgon'), # WSL: the same folder, reached through the mounted Windows drive. ('WSL', '/mnt/*/Users/*/AppData/LocalLow/Elder Game/Project Gorgon'), # macOS. Note Unity splits this one: the data is here, Player.log is not (below). ('macOS', '{HOME}/Library/Application Support/Elder Game/Project Gorgon'), # Linux, playing natively. ('Linux', '{HOME}/.config/unity3d/Elder Game/Project Gorgon'), # Linux, playing the Windows build through Steam Proton: a Windows filesystem # inside a numbered compatibility prefix. ('Steam Proton', '{HOME}/.steam/steam/steamapps/compatdata/*/pfx/drive_c/users/' 'steamuser/AppData/LocalLow/Elder Game/Project Gorgon'), ('Steam Proton', '{HOME}/.local/share/Steam/steamapps/compatdata/*/pfx/drive_c/' 'users/steamuser/AppData/LocalLow/Elder Game/Project Gorgon'), ] # macOS is the one place Unity does NOT keep Player.log beside the game's data, so a # found macOS folder gets its log looked for here instead. Everywhere else the two are # the same folder, which is why find_game_dir returns a PAIR. MAC_LOG_DIR = '{HOME}/Library/Logs/Elder Game/Project Gorgon' # What makes a folder the right one. Any single hit is enough: a player who has never # exported a book has no Books folder, and a fresh install has no ChatLogs yet. GAME_DIR_MARKERS = ('ChatLogs', 'Books', 'Player.log') def _expand(pattern): """Fill {HOME} / {USERPROFILE} into a candidate path and expand any wildcard. Returns the list of real folders it matches, which is empty for every candidate belonging to an operating system you are not on. """ home = os.path.expanduser('~') path = pattern.format(HOME=home, USERPROFILE=os.environ.get('USERPROFILE', home)) if '*' not in path: p = Path(path) return [p] if p.is_dir() else [] # Path.glob needs a fixed root to search from, so split the pattern at its first # wildcard and glob the remainder relative to what comes before it. head, _, tail = path.partition('*') root = Path(head).parent if not head.endswith('/') else Path(head) rest = path[len(str(root)):].lstrip('/') try: return sorted(p for p in root.glob(rest) if p.is_dir()) except (OSError, ValueError): return [] def find_game_dir(explicit=None): """Locate Project Gorgon's folder: (data_dir, log_dir), or None if not found. `explicit` is --game-dir or GORGON_GAME_DIR, and it is taken at its word: if you name a folder, that is the folder, even if it looks empty, because you may be pointing at a copy of somebody else's logs. """ if explicit: d = Path(explicit).expanduser() return (d, d) for label, pattern in GAME_DIR_CANDIDATES: for d in _expand(pattern): if any((d / m).exists() for m in GAME_DIR_MARKERS): log_dir = d if label == 'macOS': for alt in _expand(MAC_LOG_DIR): log_dir = alt break return (d, log_dir) return None def use_game_dir(data_dir, log_dir=None): """Point the four path constants below at a game folder. They stay module-level constants because two dozen places read them; this is the one function that ever moves them, called once at startup. """ global GAME_DIR, LOG_FOLDER, BOOKS_DIR, PLAYER_LOG, PLAYER_PREV_LOG data_dir = Path(data_dir) log_dir = Path(log_dir) if log_dir else data_dir GAME_DIR = data_dir LOG_FOLDER = data_dir / 'ChatLogs' BOOKS_DIR = data_dir / 'Books' PLAYER_LOG = log_dir / 'Player.log' PLAYER_PREV_LOG = log_dir / 'Player-prev.log' def game_dir_help(explicit=None): """The message shown when no game folder could be found: where it looked, and how to name the right one. A silent do-nothing run is the worst outcome available here, so this is deliberately loud. `explicit` is the folder the reader named themselves, if they named one. Listing the default search paths at somebody who overrode them would answer a question they did not ask, so their folder is named first and the defaults are labelled as defaults. """ if explicit: lines = [f'No ChatLogs, Books or Player.log in: {explicit}', '', 'That is the folder you named. It should be the one holding your', 'ChatLogs and Books folders. Without --game-dir, these are looked at:'] else: lines = ["Could not find your Project Gorgon folder.", "", "Looked in:"] for label, pattern in GAME_DIR_CANDIDATES: lines.append(f" {label:<13} {pattern.format(HOME='~', USERPROFILE='%USERPROFILE%')}") lines += [ "", "If the game is installed somewhere else, point at the folder that holds your", "ChatLogs and Books folders:", "", " python3 gorgon_monitor.py --game-dir \"path/to/Project Gorgon\"", "", "(or set the GORGON_GAME_DIR environment variable to the same path).", ] return '\n'.join(lines) # Found at import so the constants exist for the two dozen places that read them; the # command line can move them afterwards with use_game_dir(). A blank default is fine: # nothing reads a path until the monitor starts, and __main__ refuses to start without # a real folder. GAME_DIR = Path('.') LOG_FOLDER = Path('.') BOOKS_DIR = Path('.') PLAYER_LOG = Path('.') PLAYER_PREV_LOG = Path('.') _found = find_game_dir(os.environ.get('GORGON_GAME_DIR', '').strip() or None) if _found: use_game_dir(*_found) # Mob-drop push (POST /api/loot → the community bestiary observed-drop DB). Buffered # and flushed in batches, never per log line. The endpoint is OPEN for now; LOOT_TOKEN # is a placeholder sent as a Bearer header if set, so this same code can be community- # gated later (poisoning defense) without a client change. See _scan_loot / _flush_loot. LOOT_TOKEN = os.environ.get('GORGON_LOOT_TOKEN', '').strip() LOOT_FLUSH_SECS = 15 # max seconds a buffered loot event waits before a flush LOOT_MAX_BUFFER = 40 # or flush early once this many events are queued LOOT_BUFFER_CAP = 1000 # hard cap so a persistent outage can't grow it unbounded # Arena-bet push (POST /api/bets → community win/loss ratios). Resolved bets are rare # (a few a day), so each is pushed immediately on resolution. Open + per-ip cooldown; # BETS_TOKEN is a placeholder Bearer header for later community gating, like LOOT_TOKEN. BETS_TOKEN = os.environ.get('GORGON_BETS_TOKEN', '').strip() # Casino hot-tip mirror to redwing.bet (POST /api/v1/tips, {fighter,matchup,adjustment}). # OFF unless GORGON_KEY is set; sent as a Bearer token. Every casino tip we report to # gorgonzola is also forwarded here. Isolated: a redwing failure never touches the # gorgonzola report or the monitor loop. See _mirror_redwing_tip / _redwing_payload. REDWING_URL = os.environ.get('REDWING_URL', 'https://redwing.bet').strip().rstrip('/') REDWING_TOKEN = os.environ.get('GORGON_KEY', '').strip() # redwing sits behind Cloudflare, which 403s the default 'Python-urllib/x.y' User-Agent. # Send a real UA (the same one the server's GET uses, which Cloudflare already lets through). REDWING_UA = 'GorgonChatMonitor/1.0' # Live current-fight report to redwing.bet (POST /api/live/report). Separate endpoint, # shape (numeric fighter IDs + an epoch-ms start time) AND secret from the tip mirror — # /api/live/report 401s the tips token, so this needs its own LIVE_REPORT_SECRET and is # OFF until it's set. REDWING_REPORTER identifies this source (falls back to the game # server). See _mirror_redwing_live_fight. LIVE_REPORT_SECRET = os.environ.get('LIVE_REPORT_SECRET', '').strip() REDWING_REPORTER = os.environ.get('REDWING_REPORTER', '').strip() REDWING_FIGHTER_IDS = {'Corrrak': 1, 'Dura': 2, 'Gloz': 3, 'Leo': 4, 'Otis': 5, 'Ushug': 6, 'Viz': 7} # Viz = redwing's Vizlark # Books/ exports carry RELATIVE times ("ends in 2 hours") that the server resolves # against the moment it receives them, so a stale export doesn't just add nothing — # it OVERWRITES fresher world state with already-expired seasons/alerts. An export # older than this is baselined and never uploaded (a book read in-game arrives via # Player.log within seconds, so this only ever rejects genuinely old files). BOOK_MAX_AGE_SECS = 2 * 3600 # How long to wait before re-trying a book push the server rejected, when it didn't # tell us (no `retry_after`). Without this the 10-second book poll retried a # rate-limited garden book ~360 times an hour. BOOK_RETRY_SECS = 300 # Worlds the site keeps per-server data for. The live server is the authority # (GET /api/reporting-schema, read once at startup); this is the offline fallback. VALID_GAME_SERVERS = {'strekios', 'miraverre', 'laeth', 'dreva', 'arisetsu'} # Which world to assume when Player.log doesn't say (an old log, a fresh install). # Pass --game-server to override. Auto-detection is right in practically every case, # so this only decides where a first report goes before you have logged in once. FALLBACK_GAME_SERVER = 'dreva' # ============================================================================== # PART 4 | PATTERNS: WHAT A LOG LINE LOOKS LIKE # # This is the part that does the actual reading, and the part worth checking if you # are suspicious of what this program looks at. # # Project Gorgon's client writes a running log to Player.log: every book you open, # every NPC screen you read, every item you loot, every death. It is plain text and # you can open it yourself. Each pattern below describes ONE kind of line, with a real # example quoted above it, and nothing else in the log is looked at at all. # # If you want to see for yourself, open Player.log in a text editor and search for one # of the quoted examples. That is exactly, and only, what this program sees. # ============================================================================== # Player.log login marker (verified against a live log): "Logged in as character X." # names the active character, used to attribute mob-drop observations to a toon. LOGIN_RE = re.compile(r'Logged in as character (\w+)\.') # Lady Alethina's daily Aurest mission comes from HER CONVERSATION, not the quest: # when she gives it, the talk screen body is a localization key naming the quest, e.g. # ProcessTalkScreen(11370, "", "<<<quest_25523_Preface>>>", ...) # The quest id resolves to the mission name via quests.json (works for every quest / # any future mission); the map below is a fallback if quests.json isn't present. An # Aurest mission is any quest named "Aurest Mission: X"; X → /nb/aurest/set. AUREST_PREFACE_RE = re.compile(r'<<<quest_(\d+)_Preface>>>') AUREST_PREFIX = 'Aurest Mission: ' AUREST_MISSIONS = { # fallback (quest id -> mission), used only if quests.json is absent 25521: 'Labyrinth Chests', 25522: "Claudia's Hoard", 25523: 'Gazluk Assassination', 25524: "Gazluk's Dark Basement", 25525: 'Rahu Sewers', 25526: 'Wintertide Assault', 25527: 'Wintertide Sabotage', } # Agrashab's daily group bounty works exactly like the Aurest mission: the Fae Realm # god's talk screen names the day's quest via a preface localization key, e.g. # ProcessTalkScreen(15054, "", "<<<quest_25104_Preface>>>", ...) # The quest resolves to "Agrashab's Bounty: X"; X (the target) → /nb/agrashab/set. # Same generic <<<quest_N_Preface>>> matcher (AUREST_PREFACE_RE) drives both. AGRASHAB_PREFIX = "Agrashab's Bounty: " AGRASHAB_TARGETS = { # fallback (quest id -> target), used only if quests.json is absent 25104: 'Phoenices', 25105: 'Fire Droaches', 25106: 'Shocking Droaches', 25107: 'Basilisk Controllers', 25108: 'Winter War-Trolls', } # Vampire "Current Blood Season" is announced in a talk screen, e.g. # ProcessTalkScreen(5783, "", "<h1>Current Blood Season:</h1>...<h2>Giant Blood</h2>...") # The <h2>X Blood</h2> names the day's blood target (global → /api/blood-target). The # capture handles multi-word targets ("Big Cat", "Fish and Snail"). BLOOD_SEASON_RE = re.compile(r'Current Blood Season:.*?<h2>\s*([^<]+?)\s+Blood\s*</h2>', re.IGNORECASE | re.DOTALL) TS_RE = re.compile(r'^\d{2}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\t') BLOOD_TARGET_RE = re.compile(r'blood_target_update:\s*(.*)', re.IGNORECASE) BULK_TIP_START_RE = re.compile(r'^"\w[\w\s]*[Tt]ips\w*\s*\(\d+/\d+\):', re.IGNORECASE) # Player.log death award, e.g. # ProcessDeathMessage("<<<cause_of_death_EyeOfDeath>>>", 50, True, 5, False) # Groups: cause key, death XP, firstTime (first death from this cause). The XP is # the canonical "death XP" ONLY when firstTime=True (50); repeat deaths award a # small diminishing amount (0/1/5) — the useful "repeat XP" is the HIGHEST of those, # since it decays toward 0 over successive deaths. Two trailing values follow that we # don't use. Recorded locally AND pushed to the site's death-XP overlay # (POST /api/death-xp) so it can fill deaths.json's null xp/rxp on the Dying page — # see _record/_init_deaths_state + _report_deaths_xp. DEATH_RE = re.compile( r'ProcessDeathMessage\("<<<cause_of_death_([A-Za-z0-9_]+)>>>",\s*(\d+),\s*(True|False),' ) # Player.log world-report books (weather / gardening almanac / warden alerts). # OPENING one of these in-game always emits a ProcessBook line; EXPORTING it to # Books/*.txt is a separate manual step the player usually skips. The two carry # byte-identical bodies (same <h1>/<b>/<indent=20> markup), the only difference # being that the log escapes newlines as literal backslash-n. So the log is the # strictly better source — same content, no export required — and _scan_books # feeds the very same parsers the Books/ watcher uses. # The call is ProcessBook(title, body, exportName, "", "", False×5, ""). Title and # body are matched as proper quoted strings — `(?:[^"\\]|\\.)*` stops at the first # UNESCAPED quote — rather than by anchoring on the arguments that follow. An # earlier version anchored on a literal `", "", ""` and silently over-captured: # weather and warden leave exportName empty so it appeared to work, but the # almanac sets it ("GardeningAlmanac", which is where its export filename comes # from), so the body ran on past its closing quote. Never anchor on a trailing # argument that only SOME books populate. # No DOTALL: one ProcessBook is always ONE physical line and `.` must not run past it. # Group 3 is the book's EXPORT NAME, the filename base the game itself uses when you # click "save this book" ("Today's Shop Logs" exports as PlayerShopLog_*.txt, and # shop_pnl.html only recognises that name). Most books leave it empty and are named # from the title instead, so the group is optional. See _book_filename. BOOK_LOG_RE = re.compile(r'ProcessBook\("((?:[^"\\]|\\.)*)",\s*"((?:[^"\\]|\\.)*)"' r'(?:,\s*"((?:[^"\\]|\\.)*)")?') # In-log book titles -> the kind we report. All three VERIFIED against a live log. # The almanac's in-game title ("Gardening Almanac") differs from its export # filename (GardeningAlmanac_*.txt), and it has NEVER been exported here — the # log is the only source that has ever carried a gardening almanac at all. BOOK_KINDS = [ ('weather', re.compile(r"^Today's Weather$", re.I)), ('warden', re.compile(r'^Active Warden Alerts$', re.I)), ('garden', re.compile(r'^Gardening Almanac$', re.I)), ] _BOOK_SIG_HISTORY = 500 # warden sigs remembered, so a stale re-export can't regress # a fresher push; 500 distinct snapshots far outlasts any # realistic delayed-export window, at ~40 bytes each. # ── Book auto-save ──────────────────────────────────────────────────────────── # Books the monitor WRITES to disk for you, so the in-game "save this book" click can # be skipped. Same reasoning as BOOK_KINDS above: opening a book emits a ProcessBook # line carrying the very body the save would write, so we can produce the file the # player would have produced — same name shape, same body-only content, same CRLF # (see _write_book_file). Order books are the default because shop_pnl.html books a # filled work order as a material cost from exactly this file. # EVERY book you open is saved by default. Opening one is the only signal there is # that you wanted it, and a book you have to remember to save is a book you lose: the # weather, garden and warden reports all come from a saved book file, so a curated list # meant a player could read the almanac in game and still report nothing. # Configure with GORGON_SAVE_BOOKS / --save-books: 'off' disables, 'all' is the # default, or pass a comma-separated list of title substrings to narrow it (a piece of # a title is enough, so 'Shop Logs' catches both today's and yesterday's, which the # game exports as PlayerShopLog_*.txt, the file shop_pnl.html reads for revenue). # Saved books stay on YOUR disk: one is never uploaded anywhere, and a shop log names # your customers, so it is one of the things this program deliberately does not report. SAVE_BOOKS_DEFAULT = 'all' # Written filenames match the game's own convention exactly: the book's export name # when it has one, else its title with every non-alphanumeric character dropped, then # _YYMMDD_HHMMSS (LOCAL time) and .txt. "Order Fulfilled" -> OrderFulfilled_260902_200023.txt, # "Today's Shop Logs" -> PlayerShopLog_260907_210009.txt, matching TodaysWeather_*.txt # and ActiveWardenAlerts_*.txt on disk. Consumers date a book from this name, since a # saved book's own text carries no timestamp. _BOOKS_WRITTEN_MAX = 2000 # remembered book signatures, oldest dropped # Player.log mob-loot markers (verified against a live log). Looting a killed creature # emits, in order across several lines: # ProcessTalkScreen(<eid>, "Search Corpse of <Mob>", "<body>", "", [], ..., 0, Corpse) # ProcessAddItem(<InternalName>(<instanceId>), -1, True) ← item enters inventory # ProcessRemoveLoot(<instanceId>) ← SAME id confirms loot origin # CORPSE_RE names the mob (the trailing ", 0, Corpse)" distinguishes a corpse loot # window from an NPC/dialog talk screen). LOOT_ADD_RE's trailing True is the real- # acquisition flag (False = inventory sync on login — ignored); an add is only counted # as a drop once a matching ProcessRemoveLoot with the SAME instanceId is seen, which # filters out crafting/quest/vendor adds that never produce a RemoveLoot. Item is an # InternalName (stable id), not the display name. See _scan_loot. # The trailing int differs by screen: the opening/refresh corpse screen ends ", 0, Corpse)"; # a harvest RESULT summary ends ", 1, Corpse)". Match either (was hardcoded 0, which missed # every harvest summary) and let _harvest_action read the verb to tell them apart. CORPSE_RE = re.compile(r'ProcessTalkScreen\((\d+),\s*"Search Corpse of (.+?)",.*,\s*-?\d+,\s*Corpse\)') LOOT_ADD_RE = re.compile(r'ProcessAddItem\((\w+)\((\d+)\),\s*-1,\s*True\)') LOOT_REMOVE_RE = re.compile(r'ProcessRemoveLoot\((\d+)\)') # A corpse's harvest RESULT summary names the action, and it is the RELIABLE signal for # skin/butcher/extract: those produce items straight to inventory (AddItem + UpdateItemCode) # and never emit ProcessRemoveLoot, so the base-loot AddItem+RemoveLoot pairing below misses # them. The opening/refresh corpse screen carries no verb. See _scan_loot / _harvest_action. HARVEST_VERBS = (('butchered the corpse', 'butcher'), ('skinned the corpse', 'skin')) # The player's current zone is named on each area load, e.g. "Loading Level AreaSerbule". # Player.log has no finer position and no mob levels, so zone is as specific as sourcing gets. ZONE_RE = re.compile(r'Loading Level (Area\w+)') # ── Kuzavek arena fighter tip parsing ──────────────────────────────────────── _FIGHTERS = { "otis": "Otis", "otiz": "Otis", "gloz": "Gloz", "goz": "Gloz", "dura": "Dura", "ushug": "Ushug", "ush": "Ushug", "viz": "Viz", "vizlark": "Viz", "vozlark": "Viz", "leo": "Leo", "corrrak": "Corrrak", "corrak": "Corrrak", "crrrak": "Corrrak", "corr": "Corrrak", "cor": "Corrrak", "na'a": "Na'a", "naa": "Na'a", "naaa": "Na'a", "lamp": "Lamp", } # Canonical names for arena outcome tracking (separate from tip-parsing _FIGHTERS) _ARENA_FIGHTERS = { 'cor': 'Corrrak', # matches corr/corrr/cooo variants after dedup 'dura': 'Dura', 'gloz': 'Gloz', 'leo': 'Leo', 'otis': 'Otis', 'ushug': 'Ushug', 'viz': 'Vizlark', 'vizlark': 'Vizlark', } FIGHT_TIMEOUT = 240 # seconds — discard fight state older than this _DEDUP_RE = re.compile(r'(.)\1+') def _dedup(s): """Collapse runs of repeated characters: 'corrrrrrr' → 'cor'.""" return _DEDUP_RE.sub(r'\1', s) def _canon_fighter(text): """Return canonical arena fighter name from raw text (handles all-caps stretches).""" for word in reversed(re.findall(r'[A-Za-z]+', text)): s = _dedup(word.lower()) for key, name in _ARENA_FIGHTERS.items(): kd = _dedup(key) if s.startswith(kd) or (kd.startswith(s) and len(s) >= 3): return name return None def _parse_line_ts(line): """Parse the log-line timestamp prefix into a datetime, or None.""" m = re.match(r'^(\d{2}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})', line) return datetime.datetime.strptime(m.group(1), "%y-%m-%d %H:%M:%S") if m else None # The fighter maps above are the OFFLINE FALLBACK. The live site is the authority: # _refresh_fighters() below replaces them (and rebuilds the two patterns that are built # from them) with GET /api/reporting-schema's `fighters` at startup, so a roster change # on the site does not need a new copy of this file. They had drifted before this existed # - _FIGHTERS called Viz "Viz" while _ARENA_FIGHTERS called the same fighter "Vizlark". _F = "|".join(re.escape(k) for k in _FIGHTERS) _TIP_DIR_RE = re.compile( rf"\b({_F})\s+(?:over|>)\s+({_F})(?:\s+([+-]?\d+)%?)?(?=\s|$|[^\w])", re.IGNORECASE ) _TIP_SOLO_RE = re.compile(rf"\b({_F})\s+([+-]\d+)%?(?=\s|$|[^\w])", re.IGNORECASE) def _apply_fighter_schema(data): """Replace the fighter maps with the site's, and rebuild what is derived from them. `data` is /api/reporting-schema's `fighters` block. Anything missing or malformed leaves the built-in fallback untouched, because a half-applied roster is worse than a slightly old one. The two tip patterns are alternations built FROM the alias list, so they must be recompiled here or they would still only match yesterday's names. """ global _FIGHTERS, _ARENA_FIGHTERS, REDWING_FIGHTER_IDS, REDWING_FIGHTER_NAMES global _F, _TIP_DIR_RE, _TIP_SOLO_RE if not isinstance(data, dict): return False aliases = data.get('aliases') if not isinstance(aliases, dict) or not aliases: return False _FIGHTERS = {str(k).lower(): str(v) for k, v in aliases.items()} # Outcome tracking matches on prefixes of these keys, so it takes the same map. _ARENA_FIGHTERS = dict(_FIGHTERS) ids = data.get('redwing_ids') if isinstance(ids, dict): REDWING_FIGHTER_IDS = {str(k): int(v) for k, v in ids.items()} names = data.get('redwing_names') if isinstance(names, dict): REDWING_FIGHTER_NAMES = {str(k): str(v) for k, v in names.items()} _F = "|".join(re.escape(k) for k in _FIGHTERS) _TIP_DIR_RE = re.compile( rf"\b({_F})\s+(?:over|>)\s+({_F})(?:\s+([+-]?\d+)%?)?(?=\s|$|[^\w])", re.IGNORECASE) _TIP_SOLO_RE = re.compile(rf"\b({_F})\s+([+-]\d+)%?(?=\s|$|[^\w])", re.IGNORECASE) return True def _parse_fight_tip(text): """Turn one line of casino tip chatter into a tidy one-line summary, or None. Two shapes are understood: a matchup ("Otis over Gloz +10%") and a fighter on their own ("Gloz +5%"). Printed to your console only. The tips actually reported to the site are the ones read from the NPC in _scan_hot_tips. """ m = _TIP_DIR_RE.search(text) if m: fa = _FIGHTERS[m.group(1).lower()] fb = _FIGHTERS[m.group(2).lower()] if m.group(3) is not None: pct = int(m.group(3)) sign = "+" if pct >= 0 else "" return f"{fa} over {fb} {sign}{pct}%" return f"{fa} over {fb}" m = _TIP_SOLO_RE.search(text) if m: fa = _FIGHTERS[m.group(1).lower()] pct = int(m.group(2)) sign = "+" if pct >= 0 else "" return f"{fa} {sign}{pct}%" return None # redwing.bet expects full fighter names; our system (and gorgonzola) uses the short # canonical forms. Only Viz differs — remap our name → redwing's on the way out. REDWING_FIGHTER_NAMES = {'Viz': 'Vizlark'} def _redwing_payload(key): """Translate a gorgonzola gambling-tip key into the redwing.bet /api/v1/tips body, or None for a key shape redwing can't take. Edge tips carry the matchup ('edge:winner:loser' → fighter=winner, matchup=loser, adjustment +10); the fighter modifiers ('plus5:X'/'minus5:X' → adjustment ±5) have no opponent, so `matchup` is omitted. Fighter names are remapped to redwing's full-name vocabulary. Kept module-level and side-effect-free so it's unit-testable.""" def rw(name): return REDWING_FIGHTER_NAMES.get(name, name) parts = key.split(':') if parts[0] == 'edge' and len(parts) == 3 and parts[1] and parts[2]: return {'fighter': rw(parts[1]), 'matchup': rw(parts[2]), 'adjustment': 10} if parts[0] in ('plus5', 'minus5') and len(parts) == 2 and parts[1]: return {'fighter': rw(parts[1]), 'adjustment': 5 if parts[0] == 'plus5' else -5} return None # Casino "Here's a hot tip:" / "Gotta hot tip for ya:" NPC reads in Player.log (UTC). # Two shapes seen in a live log: # edge : "If Dura and Corrrak fight today, ... that Dura will have the upper hand. (+10%)" # solo : "Gloz ... She's feeling great! (+5%)" / "... in great form today. (-5%)" _HOT_TIP_LINE_RE = re.compile(r"\bhot tip\b", re.IGNORECASE) _HOT_TIP_EDGE_RE = re.compile( rf"\bIf\s+({_F})\s+and\s+({_F})\s+fight today\b.*?\bthat\s+({_F})\s+will have the upper hand\b" rf".*?\(([+-]?\d+)%\)", re.IGNORECASE) _HOT_TIP_SOLO_RE = re.compile(rf"\b({_F})\b.*?\(([+-]?\d+)%\)", re.IGNORECASE) _PLAYERLOG_TS_RE = re.compile(r"^\[(\d{2}:\d{2}:\d{2})\]") # Kyuroma the bookie's live-matchup talk screen, e.g. # ProcessTalkScreen(13767, "", "The next fight will be between Otis and Dura. Would # you like to wager...?\n\nThe fight begins in <em>4 minutes 55 seconds</em>.", ...) # The two fighters + a relative countdown; reported to /api/current-fight (the server # resolves the countdown to a start time at receipt, like weather/garden books). _NEXT_FIGHT_RE = re.compile( rf"next fight will be between\s+({_F})\s+and\s+({_F})\b", re.IGNORECASE) def _parse_fight_countdown(text): """Seconds until the fight begins, from 'begins in <em>4 minutes 55 seconds</em>' (either unit may be absent). Scans minutes and seconds independently in the segment after 'begins in'; 0 when no countdown is present (fight imminent).""" m = re.search(r"begins in\b(.*)", text, re.IGNORECASE) seg = m.group(1) if m else text mins = re.search(r"(\d+)\s*minute", seg, re.IGNORECASE) secs = re.search(r"(\d+)\s*second", seg, re.IGNORECASE) return (int(mins.group(1)) * 60 if mins else 0) + (int(secs.group(1)) if secs else 0) # Kyuroma arena BETTING (Player.log). A placed-and-resolved bet spans three talk screens: # Confirm Bet: "You are betting <em>N</em> Councils that A defeats B in the next arena # match. ..." — names the matchup (who bet on vs opponent). # Placement: "Success! You have placed your bet for A. ..." — only on a REAL placement, # never on a cancelled preview, so this is what proves the bet went through. # Outcome: "Your previous bet was correct, and your fighter won!" OR # "Your previous bet was incorrect. Your fighter lost. Sorry." # The outcome for a battle arrives at the NEXT Kyuroma visit, and it's one bet per battle, # so the state machine (pending -> open -> resolved) never overlaps. Only the matchup + # win/loss matters (wager amounts are not tracked). Fighter names are captured freely # (multi-word / apostrophes like Na'a), not gated on the known set. See _scan_bets. BET_CONFIRM_RE = re.compile( r'betting <em>\d+</em> Councils that (.+?) defeats (.+?) in the next arena match', re.DOTALL) BET_PLACED_RE = re.compile(r'You have placed your bet for (.+?)\.\s') BET_WIN_RE = re.compile(r'previous bet was correct') BET_LOSS_RE = re.compile(r'previous bet was incorrect') # Intro: capture everything after the position phrase; _canon_fighter does the name resolution INTRO_RE = re.compile( r"(?:far side is|farthest from me is|near side is|side closer to me is|nearest me is)\s+(.*)", re.IGNORECASE ) def _extract_winner(line): """The winner named in a line of Kuzavek's arena commentary, or None. Three phrasings, all seen in real chat: "X wins", "the battle goes to X" and "victory for X". The name is then canonicalised, because the announcer stretches it out when excited (CORRRRRAK). """ m = re.search(r"([A-Z][A-Za-z]+)\s+(?:wins?|has won|cleans up)\b", line, re.IGNORECASE) if m: return _canon_fighter(m.group(1)) m = re.search(r"battle goes to ([A-Z][A-Za-z]+)", line, re.IGNORECASE) if m: return _canon_fighter(m.group(1)) m = re.search(r"victory for ([A-Z]+)", line, re.IGNORECASE) if m: return _canon_fighter(m.group(1)) return None def _extract_loser(line): """The loser named in a line of arena commentary, or None. Sometimes the commentary only names who went down ("X has fallen", "X is out"), so the winner is whoever else was in the ring.""" m = re.search( r"([A-Z][A-Za-z]+)\s+(?:loses?|has lost|has fallen|has collapsed)\b", line, re.IGNORECASE ) if m: return _canon_fighter(m.group(1)) m = re.search(r"([A-Z][A-Za-z]+)\s+is out\b", line, re.IGNORECASE) if m: return _canon_fighter(m.group(1)) return None # ============================================================================== # PART 5 | THE MONITOR # # Everything from here on lives on one object. It is long, but it is repetitive: a # short block per thing the program watches for, in the order they appear in the file. # The class docstring immediately below explains the shape they all share. # ============================================================================== class ChatMonitor: """The monitor itself: one object that watches your logs and reports what it finds. It keeps three files under watch, all of them written by the game: ChatLogs/Chat-YY-MM-DD.log what was said around you (arena commentary) Player.log the client's own log: books you opened, NPCs you talked to, what you looted, how you died Books/*.txt books you exported by hand in game Everything else here is one method per thing worth noticing. They all follow the same three-step shape, and once you have read one you have read them all: _init_<thing> run once at startup: look back through the log so today's answer is picked up even if you saw it before starting this _scan_<thing> run on each new slice of log: notice the thing _report_<thing> send it to the site, if it is new and this world is tracked Two habits run through the whole class and are worth knowing about: FORWARD-ONLY. On startup it notes how big each log already is and reads only what is written after that, so restarting it never re-reports things you did yesterday. NOTHING IS ALLOWED TO CRASH IT. Every scanner is wrapped in its own try/except. A game patch that changes the wording of one line should cost you that one feature for the session, printing a complaint, while everything else carries on. """ def __init__(self, server_url=DEFAULT_SERVER, game_server=None, data_dir=None, save_books=None): """Set everything up, then look back over the logs so today's answers are already known before the first new line arrives. Nothing here is required of you: which world you are on is read out of Player.log, and every _init_ call below is wrapped so that a missing or unreadable file costs that one feature and never the whole program. """ self.server_url = server_url self._schema = None # /api/reporting-schema, fetched once (see _reporting_schema) self._refresh_fighters() # before _scan_existing reads a single fight line self.game_server = game_server self._detected_world = False if not self.game_server: detected = self._detect_server() self.game_server = detected or FALLBACK_GAME_SERVER self._detected_world = bool(detected) if not detected: print(f"[server] World not detected, assuming {FALLBACK_GAME_SERVER}. " f"Use --game-server if that is wrong.") self.world_reports = self._world_is_tracked() self._current_fighters = [] self._fight_start = None self._fight_last = None # (a,b) of the last current-fight we reported self._fight_last_at = 0.0 # monotonic time of that report (re-report gate) self._last_weather_path = None self._last_garden_path = None self._last_warden_path = None self._last_book_check = 0.0 # kind -> sha1 of the LAST body reported, so a book seen BOTH in Player.log # and as a Books/ export is pushed once. Weather/garden legitimately repeat # (a weather pattern recurs on a later day), so last-sig is right for them — # a set would wrongly swallow the real repeat. See _report_book. self._book_sigs = {} # kind -> epoch before which a FAILED push is not re-attempted. A rejection # (rate limit, server hiccup) must not turn the 10s book poll into a retry # storm; the wait is the server's `retry_after` when it sends one, else # BOOK_RETRY_SECS. See _report_book. self._book_retry_at = {} self._retry_hint = 0 # seconds the last rejected push asked us to wait # Warden alone needs MORE than last-sig: its countdown strictly decreases, # so every genuine read differs and no legit snapshot ever repeats — but a # delayed Books/ export of an OLDER snapshot arriving after a fresher push # would (with last-sig only) sail through and overwrite the server's fresher # state. Remembering every warden sig already sent blocks that stale re-send # while still forwarding any moved-countdown snapshot (a new sig). Bounded. self._warden_sigs = collections.deque(maxlen=_BOOK_SIG_HISTORY) self._tip_block = None # accumulated multi-line HotTips block text self._last_player_log_check = 0.0 self._player_log_pos = 0 self._last_aurest = None # last Aurest mission reported (dedup) self._last_agrashab = None # last Agrashab bounty reported (dedup) self._last_blood = None # last blood target reported (dedup) self._tips_seen = set() # casino tip keys reported this casino day (dedup) self._tips_day = None # eastern casino-day the tip dedup set belongs to self._active_char = None # current character (LOGIN_RE), tags mob drops self._quests_cat = None # lazy quests.json (quest_<id> -> {...}) data_root = Path(data_dir) if data_dir else Path(__file__).parent # Local death-XP research log (kept locally AND pushed to the site overlay). # Maps each cause_of_death key -> observed death-XP. See _init_deaths_state. self._deaths_file = data_root / 'deaths_observed.json' self._deaths_map = {} self._deaths_cat = None # lazy deaths.json catalogue (key -> {name, xp}) # Books written to disk on the player's behalf (see _save_books): sha1 of # title+body -> {title, file, seen}. Persisted because re-OPENING a book emits # another identical ProcessBook line, and a second file for one work order # would book its councils twice in shop_pnl.html. self._books_written_file = data_root / 'books_written.json' self._books_written = {} # Mob-drop capture state (pushed to /api/loot; see _scan_loot). Loot spans # several log lines and chunks, so this state lives across polls; it is reset # on log rotation (new session = new entity ids). Forward-only: the Player.log # cursor baselines at EOF, so restarts never reprocess old loot. self._loot_corpse = None # (eid, mob) currently open loot window self._loot_pending = {} # instanceId -> InternalName awaiting RemoveLoot self._loot_window_adds = {} # instanceId -> InternalName: AddItems since the last # summary/open, candidates for the next harvest action self._loot_seen_corpses = set() # corpse eids already counted this session self._loot_seen_harvest = set() # (eid, action) already emitted (result screen repeats) self._loot_zone = None # current area (from ZONE_RE), tags each drop self._loot_buffer = [] # loot/corpse events awaiting a batch flush self._last_loot_flush = 0.0 # Arena-bet state (see _scan_bets): a Confirm Bet screen fills _bet_pending, the # "Success! placed your bet" line promotes it to _bet_open, and the next outcome # screen resolves + reports it. Reset on log rotation. self._bet_pending = None # {fighter, opponent} previewed on a Confirm Bet self._bet_open = None # the placed bet awaiting its win/loss outcome self._init_book_saving(save_books) self._print_summary() self._scan_existing(print_fights=True) self._init_book_state() self._init_player_log_state() self._init_aurest_state() self._init_agrashab_state() self._init_blood_state() self._init_hot_tips_state() self._init_deaths_state() # ── Startup ────────────────────────────────────────────────────────────── def _print_summary(self): """Say what this is, where it is looking and what it will report, before it starts. Somebody who has just downloaded a program off the internet should be able to answer "what did I just run" from its first few lines, without having to take a website's word for it.""" titles = getattr(self, '_save_titles', None) if not titles: saving = 'nothing (--save-books is off)' else: what = 'every book you open' if titles == ['*'] else ', '.join(titles) saving = (f"{what}\n to your own Books folder, never uploaded" f" ({len(self._books_written)} saved so far)") how = 'detected from Player.log' if self._detected_world else 'you told me' print(f""" Gorgon Monitor {__version__} reporting to {self.server_url.rstrip('/')} world {self.game_server} ({how}) game folder {GAME_DIR} saving books {saving} reporting weather, garden, warden alerts, Aurest, Agrashab, blood season, casino tips and fights, arena results, mob drops, death XP not sending chat, inventory, councils, account, or any file on this PC Every report is printed below as it is made. Ctrl-C to stop. """) def _scan_existing(self, print_fights=False): """Read today's chat log from the top once at startup, to print the arena fights that already happened today. Display only: nothing here is reported, it just means the window is not blank when you start it mid-session.""" log_file = LOG_FOLDER / ( "Chat-" + datetime.datetime.now().strftime("%y-%m-%d") + ".log" ) if not log_file.exists(): return 0 fighters_buf = [] fight_start_b = None # utf-8 with errors='replace', like every other read in this file: chat is full # of accented names and emoji, and on Windows the default is the local code page, # where one such character would end the run. with open(log_file, "r", encoding="utf-8", errors="replace") as f: lines = f.readlines() for line in lines: line = line.strip() if "[NPC Chatter] Kuzavek:" not in line: continue line_dt = _parse_line_ts(line) # Timeout: stale fight state if fight_start_b and line_dt and (line_dt - fight_start_b).total_seconds() > FIGHT_TIMEOUT: fighters_buf = [] fight_start_b = None fm = INTRO_RE.search(line) if fm: name = _canon_fighter(fm.group(1)) if name and name not in fighters_buf: if not fighters_buf: fight_start_b = line_dt fighters_buf.append(name) continue winner = _extract_winner(line) if winner: loser = next((f for f in fighters_buf if f != winner), "?") if print_fights: print(f"[fight] {winner} beat {loser}") fighters_buf = [] fight_start_b = None continue loser = _extract_loser(line) if loser and fighters_buf: winner = next((f for f in fighters_buf if f != loser), None) if winner: if print_fights: print(f"[fight] {winner} beat {loser}") fighters_buf = [] fight_start_b = None return len(lines) def _init_book_state(self): """Record current latest books so we only upload truly new arrivals.""" self._last_weather_path = self._find_latest_book("TodaysWeather_*.txt") self._last_garden_path = self._find_latest_book("GardeningAlmanac_*.txt") if self._last_weather_path: print(f"[books] Weather baseline: {self._last_weather_path.name}") if self._last_garden_path: print(f"[books] Garden baseline: {self._last_garden_path.name}") print(f"[books] Watching {BOOKS_DIR.name} for exported books, and " f"{PLAYER_LOG.name} for books opened in-game (no export needed) ...") # ── Player.log tail (deaths · loot · aurest · agrashab · blood · tips · books) ── def _init_player_log_state(self): """Baseline the Player.log tail position so the live loop only processes lines written from now on (forward-only, like the chat tail). Isolated in try/except so a missing or locked Player.log degrades only this feature, never aborts the whole monitor at construction time.""" try: self._player_log_pos = PLAYER_LOG.stat().st_size if PLAYER_LOG.exists() else 0 print(f"[player] Watching {PLAYER_LOG.name} for deaths / loot / world reports " f"(pos={self._player_log_pos:,}) ...") except Exception as e: # setup must never crash the monitor print(f"[player] init failed (Player.log watch disabled this run): {e}", file=sys.stderr) def _check_player_log(self): """Poll Player.log for newly written lines and dispatch them to the deaths / loot / aurest / agrashab / blood / tips / current-fight / world-report-book scanners. Whole body is wrapped: a parse/IO/network error here must never break the monitor loop (it runs every second alongside chat + book polling).""" now = time.time() if now - self._last_player_log_check < 5: return self._last_player_log_check = now try: if not PLAYER_LOG.exists(): return size = PLAYER_LOG.stat().st_size if size < self._player_log_pos: # log rotated/truncated on game restart self._player_log_pos = 0 self.reset_loot_state() # new session reuses entity ids if size <= self._player_log_pos: return with open(PLAYER_LOG, 'rb') as f: f.seek(self._player_log_pos) data = f.read() # Only process up to the last complete line; a half-written final line # stays buffered (pos not advanced past it) and is re-read next poll. # Splitting on b'\n' is UTF-8 safe — 0x0A never occurs mid-codepoint. nl = data.rfind(b'\n') if nl == -1: # no complete line buffered yet return self._player_log_pos += nl + 1 chunk = data[:nl + 1].decode('utf-8', errors='replace') # Active character (login), so mob-drop observations below are attributed # to the toon at the keyboard. Last login in the chunk wins. logins = LOGIN_RE.findall(chunk) if logins: self._active_char = logins[-1] # Each scanner below is isolated in its own try/except so a parse/merge/IO # error in one can't reach the outer handler and starve the others of this # chunk after the log cursor has already advanced past it. # Death XP: recorded locally AND pushed to the site death-XP overlay. try: self._record_deaths(self._scan_deaths(chunk)) except Exception as e: print(f"[death] record failed (skipping this chunk): {e}", file=sys.stderr) # Mob drops → community bestiary observed-drop DB (POST /api/loot). try: self._scan_loot(chunk) except Exception as e: print(f"[loot] scan failed (skipping this chunk): {e}", file=sys.stderr) # Daily community world reports: Aurest mission, Agrashab bounty, blood # target, casino hot tips, current arena fight. try: self._scan_aurest(chunk) except Exception as e: print(f"[aurest] scan failed (skipping this chunk): {e}", file=sys.stderr) try: self._scan_agrashab(chunk) except Exception as e: print(f"[agrashab] scan failed (skipping this chunk): {e}", file=sys.stderr) try: self._scan_blood(chunk) except Exception as e: print(f"[blood] scan failed (skipping this chunk): {e}", file=sys.stderr) try: self._scan_hot_tips(chunk) except Exception as e: print(f"[hot-tip] scan failed (skipping this chunk): {e}", file=sys.stderr) try: self._scan_next_fight(chunk) except Exception as e: print(f"[fight] scan failed (skipping this chunk): {e}", file=sys.stderr) try: self._scan_bets(chunk) except Exception as e: print(f"[bet] scan failed (skipping this chunk): {e}", file=sys.stderr) # Weather / garden / warden books, straight from the in-game read — # no Books/ export needed. try: self._scan_books(chunk) except Exception as e: print(f"[books] scan failed (skipping this chunk): {e}", file=sys.stderr) # Save wanted books to Books/ so the in-game export click can be skipped. # Separate from _scan_books: saving is local and must keep working on a # world the site doesn't track (where world_reports is off). try: self._save_books(chunk) except Exception as e: print(f"[books] save failed (skipping this chunk): {e}", file=sys.stderr) except Exception as e: # the Player.log tail must never break the monitor loop print(f"[player] check failed (skipping this poll): {e}", file=sys.stderr) # ── Which world am I on? (Player.log connection → server name) ────────────── def _detect_server(self): """Work out which Project Gorgon world you're on from Player.log: the client logs its server list ("Url"/"Name"), then the host it connects to (connected, url=sN.projectgorgon.com). Map host -> Name. Returns a lowercase server name (strekios/miraverre/…), or None if it can't tell. Reads only up to the first connection (near the top of a session's log), so it's cheap.""" try: if not PLAYER_LOG.exists(): return None url_to_name = {} with open(PLAYER_LOG, 'r', encoding='utf-8', errors='replace') as f: for line in f: if '"Url"' in line and '"Name"' in line: m = re.search(r'"Url"\s*:\s*"([^"]+)".*?"Name"\s*:\s*"([^"]+)"', line) if m: url_to_name[m.group(1)] = m.group(2) elif 'connected, url=' in line: m = re.search(r'connected, url=([^,\s]+)', line) if m and m.group(1) in url_to_name: return url_to_name[m.group(1)].lower() return None except Exception: return None def _reporting_schema(self): """GET /api/reporting-schema once per run, cached; {} if the site is unreachable. Two things need it — which worlds the site tracks, and the arena roster — and a player's connection may be slow, so it is fetched once rather than per question. """ if self._schema is None: self._schema = {} try: url = self.server_url.rstrip('/') + '/api/reporting-schema' req = urllib.request.Request( url, headers={'User-Agent': f'GorgonChatMonitor/{__version__}'}) with urllib.request.urlopen(req, timeout=10) as resp: got = json.loads(resp.read().decode('utf-8')) if isinstance(got, dict): self._schema = got except Exception: pass # unreachable/older server: every caller falls back return self._schema def _refresh_fighters(self): """Take the arena roster from the site, falling back to the built-in maps.""" if _apply_fighter_schema(self._reporting_schema().get('fighters')): print(f"[arena] roster from the site: {', '.join(sorted(set(_FIGHTERS.values())))}") def _world_is_tracked(self): """Does the site keep per-server data for the world we're logged into? Beta (beta.projectgorgon.com) is a real world the client connects to and NOT one the site tracks, so every per-server endpoint answers `?server=beta` with a bare `abort(400)` — reporting anyway costs a rejected request per book read or blood scan, and beta values would be plain WRONG for live players if they ever did land. /api/reporting-schema is the authority (a world added there needs no client edit); VALID_GAME_SERVERS is the offline fallback.""" worlds = VALID_GAME_SERVERS live = self._reporting_schema().get('servers') if isinstance(live, list) and live: worlds = {str(s).lower() for s in live} if self.game_server in worlds: return True print(f"[server] '{self.game_server}' is not a world this site tracks " f"({', '.join(sorted(worlds))}). World reporting is OFF for this " f"session (weather, garden, warden, blood, aurest, agrashab, casino " f"tips). Everything else still runs.", file=sys.stderr) return False def _world_report_off(self, kind): """True when a per-world report must be skipped, logging once per event.""" if self.world_reports: return False print(f"[{kind}] not reported: {self.game_server} is not tracked by the site") return True # ── Aurest mission (Lady Alethina's conversation → /nb/aurest/set) ────────── def _aurest_from_preface(self, qid): """A preface quest id -> its Aurest mission name, or None if it isn't an Aurest mission. Resolves through quests.json (works for every quest / any future mission), falling back to the built-in map when the catalogue is unavailable.""" qid = int(qid) name = (self._quest_catalogue().get(f'quest_{qid}') or {}).get('Name', '') if name.startswith(AUREST_PREFIX): return name[len(AUREST_PREFIX):].strip() return AUREST_MISSIONS.get(qid) def _scan_aurest(self, chunk): """Report the day's Aurest mission if Lady Alethina's preface appears in this chunk. Last match wins; de-duped so the same mission is reported once.""" mission = None for m in AUREST_PREFACE_RE.finditer(chunk): found = self._aurest_from_preface(m.group(1)) if found: mission = found if mission and mission != self._last_aurest: self._last_aurest = mission print(f"[aurest] Lady Alethina's mission today: {mission}") self._report_aurest(mission) def _init_aurest_state(self): """Backfill: find the most recent Aurest preface already in Player.log and report it, so the current day's mission is picked up even if Lady Alethina was visited before the monitor started. Isolated: never aborts construction.""" self._last_aurest = None try: if not PLAYER_LOG.exists(): return last = None with open(PLAYER_LOG, 'r', encoding='utf-8', errors='replace') as f: for line in f: if '_Preface>>>' not in line: continue m = AUREST_PREFACE_RE.search(line) if m: found = self._aurest_from_preface(m.group(1)) if found: last = found if last: print(f"[aurest] Today's mission (from Lady Alethina's log): {last}") self._last_aurest = last self._report_aurest(last) except Exception as e: print(f"[aurest] init failed (skipping): {e}", file=sys.stderr) def _report_aurest(self, mission): """Community-report the mission to /nb/aurest/set (validated + rate-limited server-side). GET, per game server.""" if self._world_report_off('aurest'): return url = (self.server_url.rstrip('/') + '/nb/aurest/set?m=' + urllib.parse.quote(mission) + '&server=' + urllib.parse.quote(self.game_server)) try: with urllib.request.urlopen(url, timeout=15) as resp: print(f"[aurest] reported -> {resp.read().decode('utf-8', 'replace').strip()}") global REPORTS_MADE # these two report by GET, not post_json REPORTS_MADE += 1 except Exception as e: print(f"[aurest] report failed: {e}", file=sys.stderr) # ── Agrashab's daily bounty (Fae Realm god's preface → /nb/agrashab/set) ──── def _agrashab_from_preface(self, qid): """A preface quest id -> its Agrashab bounty target, or None if it isn't an Agrashab bounty. Resolves through quests.json (works for any future target), falling back to the built-in map when the catalogue is unavailable.""" qid = int(qid) name = (self._quest_catalogue().get(f'quest_{qid}') or {}).get('Name', '') if name.startswith(AGRASHAB_PREFIX): return name[len(AGRASHAB_PREFIX):].strip() return AGRASHAB_TARGETS.get(qid) def _scan_agrashab(self, chunk): """Report the day's Agrashab bounty if the god's preface appears in this chunk. Last match wins; de-duped so the same target is reported once.""" target = None for m in AUREST_PREFACE_RE.finditer(chunk): # generic <<<quest_N_Preface>>> found = self._agrashab_from_preface(m.group(1)) if found: target = found if target and target != self._last_agrashab: self._last_agrashab = target print(f"[agrashab] Today's bounty: {target}") self._report_agrashab(target) def _init_agrashab_state(self): """Backfill: find the most recent Agrashab preface already in Player.log and report it, so today's bounty is picked up even if the god was visited before the monitor started. Isolated: never aborts construction.""" self._last_agrashab = None try: if not PLAYER_LOG.exists(): return last = None with open(PLAYER_LOG, 'r', encoding='utf-8', errors='replace') as f: for line in f: if '_Preface>>>' not in line: continue m = AUREST_PREFACE_RE.search(line) if m: found = self._agrashab_from_preface(m.group(1)) if found: last = found if last: print(f"[agrashab] Today's bounty (from the log): {last}") self._last_agrashab = last self._report_agrashab(last) except Exception as e: print(f"[agrashab] init failed (skipping): {e}", file=sys.stderr) def _report_agrashab(self, target): """Community-report the bounty to /nb/agrashab/set (validated + rate-limited server-side). GET, per game server.""" if self._world_report_off('agrashab'): return url = (self.server_url.rstrip('/') + '/nb/agrashab/set?m=' + urllib.parse.quote(target) + '&server=' + urllib.parse.quote(self.game_server)) try: with urllib.request.urlopen(url, timeout=15) as resp: print(f"[agrashab] reported -> {resp.read().decode('utf-8', 'replace').strip()}") global REPORTS_MADE # these two report by GET, not post_json REPORTS_MADE += 1 except Exception as e: print(f"[agrashab] report failed: {e}", file=sys.stderr) # ── Vampire blood target (Player.log "Current Blood Season" → /api/blood-target) ── def _scan_blood(self, chunk): """Report the day's blood target from the 'Current Blood Season' talk screen. Last match wins; de-duped so the same target reports once.""" target = None for m in BLOOD_SEASON_RE.finditer(chunk): target = m.group(1).strip() if target and target != self._last_blood: self._last_blood = target print(f"[blood] Today's blood target: {target}") self._update_blood_target(target) def _init_blood_state(self): """Backfill: report the most recent blood target already in Player.log, so today's is picked up even if seen before the monitor started.""" self._last_blood = None try: if not PLAYER_LOG.exists(): return last = None with open(PLAYER_LOG, 'r', encoding='utf-8', errors='replace') as f: for line in f: if 'Current Blood Season' not in line: continue m = BLOOD_SEASON_RE.search(line) if m: last = m.group(1).strip() if last: print(f"[blood] Today's blood target (from log): {last}") self._last_blood = last self._update_blood_target(last) except Exception as e: print(f"[blood] init failed (skipping): {e}", file=sys.stderr) # ── Casino hot tips (Player.log NPC reads → /api/gambling/tip) ───────────── @staticmethod def _eastern_date(dt_utc): """Eastern (America/New_York) calendar date of a UTC datetime, DST-aware. Mirrors server.py's _eastern_date_of, with the same crude fallback.""" try: from zoneinfo import ZoneInfo return dt_utc.astimezone(ZoneInfo('America/New_York')).date() except Exception: off = -4 if 3 <= dt_utc.month <= 11 else -5 return (dt_utc + datetime.timedelta(hours=off)).date() def _tip_casino_day(self, hms): """Eastern casino-day for a Player.log UTC 'HH:MM:SS' stamp, or None. Player.log stamps are UTC (confirmed: its last line matches `date -u` to the second, while ChatLogs are local). The casino rolls over at midnight Eastern (= 04:00 UTC in EDT, 05:00 in EST). We rebuild the full UTC instant from the current UTC date — rolling back a day when the time-of-day is in the future, i.e. it belongs to the previous UTC date — then read its Eastern date. A pre-turnover read (e.g. 00:42 UTC) lands on the PREVIOUS Eastern day, so it is never mistaken for a current tip. This is the "don't look past 04:00" cutoff.""" try: t = datetime.datetime.strptime(hms, "%H:%M:%S").time() except ValueError: return None now = datetime.datetime.now(datetime.timezone.utc) cand = now.replace(hour=t.hour, minute=t.minute, second=t.second, microsecond=0) if cand > now + datetime.timedelta(minutes=5): cand -= datetime.timedelta(days=1) return self._eastern_date(cand) def _parse_hot_tip_key(self, text): """(gambling-tip key, human label) for a casino hot-tip read, or (None, None). Edge tips (upper-hand matchups) → 'edge:winner:loser'; solo tips → 'plus5:X' / 'minus5:X' by sign. Edge is tried first; an edge-shaped line we can't fully parse returns None rather than being misread as a solo tip.""" m = _HOT_TIP_EDGE_RE.search(text) if m: a = _FIGHTERS[m.group(1).lower()] b = _FIGHTERS[m.group(2).lower()] winner = _FIGHTERS[m.group(3).lower()] loser = b if winner == a else a if loser == winner: return None, None return f"edge:{winner}:{loser}", f"{winner} over {loser} +10%" if 'upper hand' in text.lower(): return None, None # edge-shaped but unparsed — don't solo-misfire m = _HOT_TIP_SOLO_RE.search(text) if m: name = _FIGHTERS[m.group(1).lower()] pct = int(m.group(2)) if pct > 0: return f"plus5:{name}", f"{name} +5%" if pct < 0: return f"minus5:{name}", f"{name} -5%" return None, None def _scan_hot_tips(self, chunk): """Parse casino 'hot tip' NPC reads in this Player.log chunk and report the ones on the CURRENT casino day. Tips stamped before today's midnight-Eastern turnover (00:00-03:59 UTC in summer) belong to yesterday's casino and are skipped — the server stamps tips on receipt, so a stale one would be mis-dated onto today. De-duped per casino day.""" today = self._eastern_date(datetime.datetime.now(datetime.timezone.utc)) if self._tips_day != today: # new casino day → fresh dedup set self._tips_day = today self._tips_seen = set() for line in chunk.splitlines(): if not _HOT_TIP_LINE_RE.search(line): continue ts = _PLAYERLOG_TS_RE.match(line) if not ts or self._tip_casino_day(ts.group(1)) != today: continue # stale: before today's turnover key, label = self._parse_hot_tip_key(line) if not key or key in self._tips_seen: continue self._tips_seen.add(key) print(f"[hot-tip] casino read: {label} → {key}") self._report_gambling_tip(key) def _init_hot_tips_state(self): """Backfill: report today's casino hot tips already in the tail of Player.log, so a read seen just before the monitor started still reaches the board. Bounded to the recent tail (a time-of-day stamp alone can't date multi-day-old lines) and filtered by the same casino-day cutoff. Isolated: never aborts construction.""" try: if not PLAYER_LOG.exists(): return size = PLAYER_LOG.stat().st_size window = 262144 with open(PLAYER_LOG, 'rb') as f: f.seek(max(0, size - window)) tail = f.read().decode('utf-8', errors='replace') if size > window: tail = tail.split('\n', 1)[-1] # drop the partial leading line self._scan_hot_tips(tail) except Exception as e: print(f"[hot-tip] init failed (skipping): {e}", file=sys.stderr) def _report_gambling_tip(self, key): """POST an authoritative casino tip to /api/gambling/tip (community board, not auth-gated; the server de-dups, budgets, and stamps the Eastern date). Per game server. Isolated: a report failure must never break the monitor loop.""" if self._world_report_off('hot-tip'): return url = (self.server_url.rstrip('/') + '/api/gambling/tip?server=' + urllib.parse.quote(self.game_server)) body = json.dumps({'key': key, 'value': True}).encode('utf-8') req = urllib.request.Request( url, data=body, headers={'Content-Type': 'application/json'}, method='POST') try: with urllib.request.urlopen(req, timeout=15) as resp: resp.read() print(f"[hot-tip] reported → {key}") except Exception as e: print(f"[hot-tip] report failed ({key}): {e}", file=sys.stderr) self._mirror_redwing_tip(key) def _mirror_redwing_tip(self, key): """Forward a reported casino tip to redwing.bet (POST /api/v1/tips, Bearer GORGON_KEY). OFF unless the token is set. Isolated: any failure here is logged and swallowed so it can never affect the gorgonzola report or the monitor loop.""" if not REDWING_TOKEN: return payload = _redwing_payload(key) if payload is None: return try: res = post_json(REDWING_URL, '/api/v1/tips', payload, headers={'Authorization': f'Bearer {REDWING_TOKEN}', 'User-Agent': REDWING_UA}) if res is None or res.get('status', 200) >= 400: print(f"[hot-tip] redwing mirror failed ({key}): {res}", file=sys.stderr) else: print(f"[hot-tip] redwing ← {key}") except Exception as e: print(f"[hot-tip] redwing mirror error ({key}): {e}", file=sys.stderr) # ── Current arena fight (Kyuroma the bookie → /api/current-fight) ─────────── def _scan_next_fight(self, chunk): """Parse the bookie's 'next fight will be between X and Y … begins in N' talk screen and report the live matchup. Forward-only (no tail backfill) — a stale countdown would resolve to a fight that's already over. Re-reports only on a matchup change or after the last report has aged out, so an unchanged screen refreshing every ~30s doesn't spam the endpoint.""" for line in chunk.splitlines(): m = _NEXT_FIGHT_RE.search(line) if not m: continue a = _FIGHTERS[m.group(1).lower()] b = _FIGHTERS[m.group(2).lower()] if a == b: continue secs = _parse_fight_countdown(line) now = time.monotonic() if (a, b) == self._fight_last and (now - self._fight_last_at) < 300: continue self._fight_last, self._fight_last_at = (a, b), now print(f"[fight] Current fight: {a} vs {b} (starts in {secs // 60}m {secs % 60}s)") self._report_current_fight(a, b, secs) def _report_current_fight(self, a, b, secs): """POST the bookie's live matchup to /api/current-fight (community-set, per game server; the server resolves the countdown to a start time on receipt). Isolated: a report failure — including the expected 429 cooldown — never breaks the loop.""" if self._world_report_off('fight'): return url = (self.server_url.rstrip('/') + '/api/current-fight?server=' + urllib.parse.quote(self.game_server)) body = json.dumps({'a': a, 'b': b, 'starts_in_secs': secs}).encode('utf-8') req = urllib.request.Request( url, data=body, headers={'Content-Type': 'application/json'}, method='POST') try: with urllib.request.urlopen(req, timeout=15) as resp: resp.read() print(f"[fight] reported → {a} vs {b}") except Exception as e: if getattr(e, 'code', None) != 429: # 429 = per-server cooldown, expected print(f"[fight] report failed ({a} vs {b}): {e}", file=sys.stderr) self._mirror_redwing_live_fight(a, b, secs) def _mirror_redwing_live_fight(self, a, b, secs): """Report the live matchup to redwing.bet (POST /api/live/report, Bearer LIVE_REPORT_SECRET) using its numeric fighter IDs + an epoch-ms start time (now + the bookie's countdown). OFF unless the secret is set. Isolated: any failure is logged and swallowed so it never touches the gorgonzola report.""" if not LIVE_REPORT_SECRET: return aid, bid = REDWING_FIGHTER_IDS.get(a), REDWING_FIGHTER_IDS.get(b) if not aid or not bid: print(f"[fight] redwing live: no id for {a}/{b}", file=sys.stderr) return payload = {'fighter_a_id': aid, 'fighter_b_id': bid, 'started_at': int((time.time() + secs) * 1000), 'reporter': REDWING_REPORTER or self.game_server} try: res = post_json(REDWING_URL, '/api/live/report', payload, headers={'Authorization': f'Bearer {LIVE_REPORT_SECRET}', 'User-Agent': REDWING_UA}) if res is None or res.get('status', 200) >= 400: print(f"[fight] redwing live failed ({a} vs {b}): {res}", file=sys.stderr) else: print(f"[fight] redwing live ← {a} vs {b}") except Exception as e: print(f"[fight] redwing live error ({a} vs {b}): {e}", file=sys.stderr) # ── Quest catalogue (shared by the Aurest / Agrashab preface resolvers) ───── def _quest_catalogue(self): """Lazy quests.json ({quest_<id>: {...}}) for resolving preface quest ids.""" if self._quests_cat is None: try: self._quests_cat = json.loads( (Path(__file__).parent / 'quests.json').read_text(encoding='utf-8')) except Exception: self._quests_cat = {} return self._quests_cat # ── Arena betting (Kyuroma → community win/loss ratios, POST /api/bets) ───── def _scan_bets(self, chunk): """Detect placed-and-resolved arena bets in a Player.log chunk and report each resolved bet. Processed line-by-line in log order so the Confirm -> Placed -> Outcome sequence is respected. Only the matchup + win/loss is kept.""" for line in chunk.splitlines(): m = BET_CONFIRM_RE.search(line) if m: self._bet_pending = {'fighter': m.group(1).strip(), 'opponent': m.group(2).strip()} continue m = BET_PLACED_RE.search(line) if m: fighter = m.group(1).strip() p = self._bet_pending self._bet_open = (dict(p) if p and p.get('fighter') == fighter else {'fighter': fighter, 'opponent': None}) self._bet_pending = None print(f"[bet] placed on {fighter}" + (f" vs {self._bet_open['opponent']}" if self._bet_open.get('opponent') else '')) continue if BET_WIN_RE.search(line): self._resolve_bet(True) continue if BET_LOSS_RE.search(line): self._resolve_bet(False) def _resolve_bet(self, won): """Resolve the open placed bet with its win/loss outcome and report it. An outcome with no known placed bet is skipped — the outcome screen names neither the fighter nor the matchup, so it can't be attributed or bucketed by edge.""" b = self._bet_open self._bet_open = None if not b: # The bet was placed before this monitor run began (started mid-session) or # the client restarted (state reset on log rotation). Logged, not recorded, # so every stored bet keeps a known matchup. print(f"[bet] {'won' if won else 'lost'} outcome seen but no placed bet on " f"record, skipped (monitor started after the bet, or client restarted)") return ts = datetime.datetime.now().isoformat(timespec='seconds') rec = {'fighter': b['fighter'], 'opponent': b.get('opponent'), 'won': won, 'ts': ts, 'id': f"{ts}|{b['fighter']}|{b.get('opponent')}|{int(won)}"} print(f"[bet] {'WON' if won else 'LOST'}: {b['fighter']}" + (f" vs {b['opponent']}" if b.get('opponent') else '')) self._report_bets([rec]) def _report_bets(self, records): """Push resolved arena bets to the site (/api/bets) for community win/loss ratios. Skipped on a world the site doesn't track. Wrapped: a network problem never breaks the monitor (a rare missed bet is acceptable).""" if self._world_report_off('bet'): return payload = {'char': self._active_char, 'bets': records} headers = {'Authorization': f'Bearer {BETS_TOKEN}'} if BETS_TOKEN else None try: res = post_json(self.server_url, f'/api/bets?server={self.game_server}', payload, headers=headers) if res is not None and res.get('ok'): print(f"[bet] reported {res.get('added', 0)} resolved bet(s)") except Exception as e: print(f"[bet] report failed: {e}", file=sys.stderr) # ── Mob-drop capture (Player.log → community bestiary observed-drop DB) ───── # Stateful, line-by-line: track the open corpse (mob name + entity id), pair each # looted item's ProcessAddItem(..,True) with its confirming ProcessRemoveLoot(sameId), # and buffer: # {type:'corpse', mob, corpse_id} once per distinct looted corpse (the rate # denominator; re-opening a corpse is skipped) # {type:'loot', mob, corpse_id, item} per confirmed drop, attributed to the # current corpse (RemoveLoot id-match is the # real guard; LIFO handles multi-corpse melee) # Events are flushed in batches by _flush_loot. The server aggregates by mob into # drop counts + rate. Never per-line push. def reset_loot_state(self): """Clear cross-line loot state — called on log rotation (a new session reuses entity ids, so a stale open-corpse/seen-eid set would misattribute). Also clears the cross-line arena-bet state so an outcome from a pre-restart bet can't attach to a stale pending bet.""" self._loot_corpse = None self._loot_pending = {} self._loot_window_adds = {} self._loot_seen_corpses = set() self._loot_seen_harvest = set() self._bet_pending = None self._bet_open = None @staticmethod def _harvest_action(line): """The harvest action named in a corpse RESULT summary, or None for the plain opening/refresh screen. 'extracted X from the corpse' / 'butchered the corpse' / 'skinned the corpse'.""" for phrase, action in HARVEST_VERBS: if phrase in line: return action if 'extracted ' in line and 'from the corpse' in line: return 'extract' return None def _emit_harvest(self, mob, eid, action): """A harvest RESULT summary fired: the AddItem(True) items still open in this window (i.e. not taken as base loot via RemoveLoot) are this action's output. Emit one 'action' marker (the per-action rate denominator) + a 'loot' event per item tagged with the action, then clear the window buffer so the next action on the same corpse starts fresh.""" self._loot_buffer.append({ 'type': 'action', 'char': self._active_char, 'mob': mob, 'corpse_id': eid, 'action': action, }) for iname in self._loot_window_adds.values(): self._loot_buffer.append({ 'type': 'loot', 'action': action, 'char': self._active_char, 'mob': mob, 'corpse_id': eid, 'item': iname, 'zone': self._loot_zone, 'at': int(time.time()), # see the base-loot event for why }) self._loot_window_adds = {} def _scan_loot(self, chunk): """Detect mob drops in a Player.log chunk and buffer them (batched by _flush_loot). Processed in log order so item→corpse attribution is temporal. Base loot is AddItem(...,True) confirmed by a matching RemoveLoot -> action 'loot'. Skin/butcher/extract never emit RemoveLoot, so they're captured from the corpse RESULT summary verb: every AddItem in the open window that WASN'T taken as base loot is attributed to that harvest action (see _emit_harvest).""" for line in chunk.splitlines(): z = ZONE_RE.search(line) if z: self._loot_zone = z.group(1) continue m = CORPSE_RE.search(line) if m: eid, mob = m.group(1), m.group(2).strip() action = self._harvest_action(line) if action: # RESULT summary -> harvest output key = (eid, action) # the summary line can repeat; count once if key not in self._loot_seen_harvest: self._loot_seen_harvest.add(key) self._emit_harvest(mob, eid, action) else: # opening/refresh -> (re)open window prev = self._loot_corpse[0] if self._loot_corpse else None self._loot_corpse = (eid, mob) if eid != prev: # a NEW corpse gets a fresh window; self._loot_window_adds = {} # a refresh of the same one keeps it if eid not in self._loot_seen_corpses: self._loot_seen_corpses.add(eid) self._loot_buffer.append({ 'type': 'corpse', 'char': self._active_char, 'mob': mob, 'corpse_id': eid, 'zone': self._loot_zone, }) continue m = LOOT_ADD_RE.search(line) if m: iname, iid = m.group(1), m.group(2) self._loot_pending[iid] = iname self._loot_window_adds[iid] = iname # candidate harvest output continue m = LOOT_REMOVE_RE.search(line) if m: iid = m.group(1) iname = self._loot_pending.pop(iid, None) self._loot_window_adds.pop(iid, None) # taken as base loot, not a harvest if iname and self._loot_corpse: eid, mob = self._loot_corpse self._loot_buffer.append({ 'type': 'loot', 'action': 'loot', 'char': self._active_char, 'mob': mob, 'corpse_id': eid, 'item': iname, 'zone': self._loot_zone, # When this was SEEN. A failed push keeps its batch and retries, # so a buffer can cross midnight; without this the site would # file yesterday's kills under today's blood season. 'at': int(time.time()), }) continue if len(self._loot_buffer) >= LOOT_MAX_BUFFER: self._flush_loot(force=True) def _flush_loot(self, force=False): """Send buffered mob-drop events as one batch to /api/loot. Throttled to LOOT_FLUSH_SECS unless forced/full. On failure the buffer is kept (capped) and retried, so a brief outage never loses a drop. LOOT_TOKEN is sent as a Bearer header when set (placeholder — the endpoint is open for now). The world rides along because vampire loot names the day's blood target, and the site joins the two to correct a creature's blood type. It is a query param, not a payload field, so the endpoint keeps accepting an older monitor's batch unchanged; a world the site does not track is sent anyway and simply infers nothing there.""" if not self._loot_buffer: return now = time.time() if not force and (now - self._last_loot_flush) < LOOT_FLUSH_SECS \ and len(self._loot_buffer) < LOOT_MAX_BUFFER: return self._last_loot_flush = now payload = {'char': self._active_char, 'events': self._loot_buffer} headers = {'Authorization': f'Bearer {LOOT_TOKEN}'} if LOOT_TOKEN else None try: path = '/api/loot?server=' + urllib.parse.quote(self.game_server) res = post_json(self.server_url, path, payload, headers=headers) except Exception as e: res = None print(f"[loot] push failed (will retry): {e}", file=sys.stderr) if res is not None and res.get('ok'): print(f"[loot] pushed {res.get('accepted', 0)} drop event(s)") self._loot_buffer = [] elif len(self._loot_buffer) > LOOT_BUFFER_CAP: self._loot_buffer = self._loot_buffer[-LOOT_BUFFER_CAP:] # ── Death XP research (Player.log → site death-XP overlay) ───────────────── # Records the death-XP the game awards per cause of death and pushes it to the # site so the unknown xp/rxp values in deaths.json can be filled in (live overlay # on the Dying page; deaths.json stays the hand-curated source of truth). Kept # locally too. deaths_observed.json maps each cause_of_death key -> # {name, first_xp, repeat_xp, xp_values, last_seen}: first_xp is the canonical # award (only ProcessDeathMessage firstTime=True carries it); repeat_xp is the # HIGHEST repeat award seen (firstTime=False, decays toward 0 so max is the useful # value). See _report_deaths_xp for the push. def _deaths_catalogue(self): """Lazy {cause_of_death key -> {name, xp}} from deaths.json, to label observations and flag which ones fill a currently-unknown (null) XP.""" if self._deaths_cat is None: self._deaths_cat = {} try: data = json.loads((Path(__file__).parent / 'deaths.json').read_text(encoding='utf-8')) self._deaths_cat = {d['key']: d for d in data if d.get('key')} except Exception: pass return self._deaths_cat @staticmethod def _scan_deaths(text): """Death observations in `text` as [{key, xp, first}] (log order).""" return [{'key': key.lower(), 'xp': int(xp), 'first': first == 'True'} for key, xp, first in DEATH_RE.findall(text)] def _merge_deaths(self, observations): """Fold observations into self._deaths_map, tracking the canonical first-death XP (firstTime=True, set once) and the HIGHEST repeat award (firstTime=False — repeat XP decays toward 0, so max is the value worth keeping). Idempotent: re-scanning the full logs never lowers a value or corrupts the xp_values set. Returns the changed keys as push-ready aggregates [{key, first_xp, repeat_xp}] (a newly-learned first_xp or a new higher repeat_xp).""" cat = self._deaths_catalogue() changed = [] for o in observations: ent = self._deaths_map.setdefault(o['key'], { 'name': (cat.get(o['key']) or {}).get('name', ''), 'first_xp': None, 'repeat_xp': None, 'xp_values': [], 'last_seen': '', }) ent.setdefault('repeat_xp', None) # backfill records written before this field note = False if o['xp'] not in ent['xp_values']: ent['xp_values'] = sorted(set(ent['xp_values']) | {o['xp']}) note = True # a value nobody had recorded is research in its own # right: the ladder of repeat awards (50, 5, 1, 0) is # what deaths.json's rxp is trying to describe, and it # is invisible if only the highest is ever reported if o['first']: if ent['first_xp'] is None: ent['first_xp'] = o['xp']; note = True elif ent['repeat_xp'] is None or o['xp'] > ent['repeat_xp']: ent['repeat_xp'] = o['xp']; note = True ent['last_seen'] = datetime.datetime.now().isoformat(timespec='seconds') if note: changed.append(o['key']) seen, out = set(), [] for k in changed: if k in seen: continue seen.add(k) ent = self._deaths_map[k] out.append({'key': k, 'first_xp': ent['first_xp'], 'repeat_xp': ent['repeat_xp'], 'xp_values': list(ent['xp_values'])}) return out def _write_deaths(self): """Save the local death-XP research file (deaths_observed.json), sorted so that opening it in an editor, or diffing it, stays readable.""" try: ordered = dict(sorted(self._deaths_map.items())) self._deaths_file.write_text(json.dumps(ordered, indent=2, ensure_ascii=False), encoding='utf-8') except Exception as e: print(f"[death] could not write {self._deaths_file.name}: {e}", file=sys.stderr) def _record_deaths(self, observations): """Merge live death observations, persist, log each, and push changed XP values to the site overlay.""" if not observations: return changed = self._merge_deaths(observations) self._write_deaths() cat = self._deaths_catalogue() for o in observations: meta = cat.get(o['key']) or {} name = meta.get('name') or o['key'] if o['first']: gap = ', FILLS A GAP in deaths.json' if meta.get('xp') is None else '' print(f"[death] FIRST death: {name} = {o['xp']} XP (canonical){gap}") else: print(f"[death] {name}: {o['xp']} XP (repeat)") self._report_deaths_xp(changed) def _report_deaths_xp(self, observations): """Contribute observed death XP to the community overlay (POST /api/death-xp), which fills deaths.json's null xp/rxp on the Dying page. Three numbers go up per cause: the canonical first-death XP, the highest repeat award, and `xp_values`, the whole set of awards seen for that cause. That last one is the research: repeat XP decays toward zero over successive deaths, so the ladder (50, 5, 1, 0) is the thing worth knowing, and reporting only the highest throws it away. Nobody dies of everything, which is exactly why this is worth pooling. The server keeps the max of each number and the union of the set, so re-pushing the same history, restarting, or two people reporting the same cause can never lower or lose a value. Wrapped so a network problem never breaks the monitor: the local record already stands, and a missed push goes out with the next one.""" observations = [o for o in observations if o and (o.get('first_xp') is not None or o.get('repeat_xp') is not None or o.get('xp_values'))] if not observations: return try: res = post_json(self.server_url, '/api/death-xp', {'observations': observations}) if res is not None and res.get('ok'): print(f"[death] pushed {res.get('updated', 0)} XP value(s) to the overlay") except Exception as e: print(f"[death] push failed (recorded locally, will retry): {e}", file=sys.stderr) def _init_deaths_state(self): """Backfill the death-XP record from the full Player logs (idempotent), so history survives restarts and log rotation. Wrapped: never crash the monitor.""" try: if self._deaths_file.exists(): self._deaths_map = json.loads(self._deaths_file.read_text(encoding='utf-8')) for p in (PLAYER_PREV_LOG, PLAYER_LOG): if p.exists(): self._merge_deaths(self._scan_deaths(p.read_text(encoding='utf-8', errors='replace'))) self._write_deaths() known = sum(1 for e in self._deaths_map.values() if e.get('first_xp') is not None) cat = self._deaths_catalogue() null_in_cat = sum(1 for k, e in self._deaths_map.items() if e.get('first_xp') is not None and (cat.get(k) or {}).get('xp') is None) print(f"[death] Death-XP log: {len(self._deaths_map)} cause(s) seen, " f"{known} with a canonical first-death XP " f"({null_in_cat} fill a null in deaths.json) -> {self._deaths_file.name}") # Re-push everything (server keeps the max, so this can't corrupt), so any # value whose live push failed on a previous run reaches the overlay now. self._report_deaths_xp([ {'key': k, 'first_xp': e.get('first_xp'), 'repeat_xp': e.get('repeat_xp')} for k, e in self._deaths_map.items()]) except Exception as e: print(f"[death] init failed (death logging disabled this run): {e}", file=sys.stderr) # ── Main loop ──────────────────────────────────────────────────────────── def monitor_chat(self): """The main loop. Runs until you press Ctrl-C. Once a second: read whatever is new in today's chat log, then check the other watchers (exported books, Player.log, the queue of mob drops waiting to be sent). At midnight the game starts a new chat log, so the outer loop notices the date changed and picks up the new file. """ while True: log_date = datetime.datetime.now().date() log_file = LOG_FOLDER / ("Chat-" + log_date.strftime("%y-%m-%d") + ".log") last_pos = log_file.stat().st_size if log_file.exists() else 0 print(f"[monitor] Watching {log_file.name} (pos={last_pos:,}) ...") while datetime.datetime.now().date() == log_date: if log_file.exists(): with open(log_file, "r", encoding="utf-8", errors="replace") as f: f.seek(last_pos) new_lines = f.readlines() last_pos = f.tell() for line in new_lines: self._process_line(line.strip()) self._check_books() self._check_player_log() self._flush_loot() # time-based batch flush (throttled) time.sleep(1) print(f"[monitor] Day rolled over.") self._current_fighters = [] # ── Per-line processing ────────────────────────────────────────────────── def _process_line(self, line): """Handle one line of chat. Most lines are of no interest and fall straight through. The ones that matter are the casino's HotTips broadcasts and Kuzavek's arena commentary, which arrives a line at a time ("the far side is OTIS", then blows traded, then "GLOZ has fallen"), so who is fighting has to be remembered between lines. A fight that goes quiet for FIGHT_TIMEOUT is forgotten, otherwise a disconnection would leave two fighters standing in the ring forever. """ if not line: return # Continuation line (no timestamp) if not TS_RE.match(line): if self._tip_block is not None: self._tip_block += '\n' + line if line.rstrip().endswith('"'): self._parse_bulk_tips(self._tip_block) self._tip_block = None return # New timestamped line — clear any unfinished tip block self._tip_block = None bt_m = BLOOD_TARGET_RE.search(line) if bt_m: self._update_blood_target(bt_m.group(1).strip()) return if "[HotTips]" in line: msg = re.sub(r"^.*\[HotTips\]\s+\S+\s*", "", line) if BULK_TIP_START_RE.match(msg): if msg.rstrip().endswith('"'): self._parse_bulk_tips(msg) else: self._tip_block = msg return tip = _parse_fight_tip(msg) if tip: print(f"[fight-tip] {tip}") else: print(f"[tip] {line}") return if "[NPC Chatter] Kuzavek:" not in line: return line_dt = _parse_line_ts(line) # Timeout: discard stale fight state if self._fight_start and line_dt: elapsed = (line_dt - self._fight_start).total_seconds() if elapsed > FIGHT_TIMEOUT: if self._current_fighters: print(f"[fight] Timeout ({elapsed:.0f}s), discarding {self._current_fighters}") self._current_fighters = [] self._fight_start = None fm = INTRO_RE.search(line) if fm: name = _canon_fighter(fm.group(1)) if name and name not in self._current_fighters: if not self._current_fighters and line_dt: self._fight_start = line_dt self._current_fighters.append(name) return winner = _extract_winner(line) if winner: loser = next((f for f in self._current_fighters if f != winner), "?") print(f"[fight] {winner} beat {loser}") self._current_fighters = [] self._fight_start = None return loser = _extract_loser(line) if loser and self._current_fighters: winner = next((f for f in self._current_fighters if f != loser), None) if winner: print(f"[fight] {winner} beat {loser}") self._current_fighters = [] self._fight_start = None # ── Blood target ───────────────────────────────────────────────────────── def _update_blood_target(self, name): """Report today's vampire blood target to the site for this world.""" if self._world_report_off('blood'): return target = name if name else None result = post_json(self.server_url, f'/api/blood-target?server={self.game_server}', {'target': target}) if result and result.get('ok'): print(f"[blood] Blood target → {name or '(cleared)'}") else: print(f"[blood] Update failed: {result}", file=sys.stderr) def _parse_bulk_tips(self, block): """Parse a multi-line HotTips block and emit fight-tip lines. Format 1 (sectioned): section headers "+5%:", "-10%:", "+10%:" followed by fighter names or "Winner > Loser" lines. Format 2 (freeform): lines like "Otis over Gloz +10%" with embedded percentages. Both formats can appear in the same block; Format 2 lines are caught as fallback. """ lines = [l.strip() for l in block.split('\n') if l.strip()] if not lines: return if lines[0].startswith('"'): lines[0] = lines[0][1:] if lines[-1].endswith('"'): lines[-1] = lines[-1][:-1] current_pct = None for line in lines: if not line: continue # Header: "Hot Tips (N/M):" / "Hoht Tips (N/M):" etc. if re.match(r'\w[\w\s]*[Tt]ips\w*\s*\(\d+/\d+\):', line, re.IGNORECASE): continue # Separator if re.match(r'^-{3,}$', line): continue # Section header: "+5%:", "-10%:", "+10%:" m = re.match(r'^([+-]?\d+)%?\s*:', line) if m: current_pct = int(m.group(1)) continue # Matchup: "Winner > Loser1, Loser2" — check before _parse_fight_tip # so multi-loser lines and section percentages are handled correctly. # Strip leading junk (e.g. "+15 Cor > Otis" → winner "Cor") for embedded-pct lines. if '>' in line: parts = line.split('>', 1) raw_w = re.sub(r'^[^A-Za-z]+', '', parts[0]).strip() winner = _FIGHTERS.get(raw_w.lower()) if winner: losers = [_FIGHTERS.get(re.sub(r"[^A-Za-z']+$", '', l.strip()).lower()) for l in parts[1].split(',')] losers = [l for l in losers if l] pct_str = '' if current_pct is not None: pct_str = f" +{current_pct}%" if current_pct >= 0 else f" {current_pct}%" for loser in losers: print(f"[fight-tip] {winner} over {loser}{pct_str}") else: # fallback: line may have embedded pct in Format 2 style tip = _parse_fight_tip(line) if tip: print(f"[fight-tip] {tip}") continue # Solo fighter: individual name in +5%/-5% section name = _FIGHTERS.get(line.lower()) if name and current_pct is not None: sign = "+" if current_pct >= 0 else "" print(f"[fight-tip] {name} {sign}{current_pct}%") elif not name: # fallback for any Format 2 freeform lines outside sections tip = _parse_fight_tip(line) if tip: print(f"[fight-tip] {tip}") # ── Book watching ──────────────────────────────────────────────────────── def _find_latest_book(self, pattern): """The most recently modified book in the Books folder matching a name pattern, e.g. TodaysWeather_*.txt, or None if you have never exported one.""" files = sorted(BOOKS_DIR.glob(pattern), key=lambda p: p.stat().st_mtime) return files[-1] if files else None def _watch_book(self, kind, path): """Handle one newly-seen Books/ export; True once it needs no further attention — reported, or too stale to ever report. False leaves the path unbaselined so a rejected push is re-attempted (on _report_book's backoff).""" if not self.world_reports: return True # world untracked: baseline it, don't poll it forever when = book_timestamp(path) if self._stale_book(path, when): return True return self._report_book(kind, path.read_text(encoding='utf-8'), when, path.name) def _stale_book(self, path, when): """True if this Books/ export is too old to report (and should be baselined). These books state their times RELATIVE to when they were read ("ends in 2 hours"), and both this monitor and the server resolve that against the moment of upload — so a two-day-old export doesn't merely add nothing, it overwrites live garden/warden state with seasons that already ended. Only the Books/ watcher needs this: a book read in-game reaches _scan_books from Player.log within seconds. An unparseable filename is treated as fresh — the timestamp is the only age signal, and dropping a book we can't date would silently lose real reports.""" if not when: return False age = (datetime.datetime.now() - when).total_seconds() if age <= BOOK_MAX_AGE_SECS: return False print(f"[books] Skipping {path.name}: exported {age / 3600:.1f}h ago, " f"its relative times are no longer meaningful", file=sys.stderr) return True def _check_books(self): """Every ten seconds, look for a newly exported book in the Books folder. This is the second-best source and it is kept only for completeness: opening a book in game is already caught from Player.log (see _scan_books), which means you do not have to export anything at all. A book that arrives both ways is recognised as the same content and sent once. """ now = time.time() if now - self._last_book_check < 10: return self._last_book_check = now # The Books/ watcher is now the SECOND source: opening a book in-game is # already caught from Player.log by _scan_books, and exporting it is an # extra manual step. It stays because an export is still a valid input — # _report_book de-dupes by content, so a book that arrives both ways is # only pushed once. w = self._find_latest_book("TodaysWeather_*.txt") if w and w != self._last_weather_path and self._watch_book('weather', w): self._last_weather_path = w g = self._find_latest_book("GardeningAlmanac_*.txt") if g and g != self._last_garden_path and self._watch_book('garden', g): self._last_garden_path = g wa = self._find_latest_book("ActiveWardenAlerts_*.txt") if wa and wa != self._last_warden_path and self._watch_book('warden', wa): self._last_warden_path = wa # ── World-report books: Player.log ProcessBook OR Books/*.txt ─────────────── @staticmethod def _unescape_book(body): """Turn a ProcessBook body's literal backslash-escapes into real text. The log writes one book as ONE physical line, so its newlines arrive as the two characters \\ and n. Unescaped in a single pass — a naive .replace('\\\\n', '\\n') chain would also rewrite a genuine backslash followed by an n. """ return re.sub(r'\\(.)', lambda m: {'n': '\n', 't': '\t', 'r': '\r'}.get(m.group(1), m.group(1)), body) def _report_book(self, kind, text, when, source): """Push one world-report book, whichever source it came from. Single funnel for the log scanner and the Books/ watcher, so the two can never drift in what they parse or how they report. De-dupes on the body itself: opening a book and then exporting it yields identical content, and that is one observation, not two. Content is the right key rather than a timestamp — a genuinely new weather book always differs, and a re-read warden alert differs too (its countdown moved), which is a fresher snapshot worth sending. For warden, last-sig alone is not enough: a delayed Books/ export of an OLDER snapshot arriving after a fresher in-game read would otherwise overwrite the server's fresher state. So warden also skips any snapshot whose sig was EVER sent (see _warden_sigs) — safe because a warden read never legitimately repeats (its countdown strictly decreases). A push the server REJECTS is retried, but on a backoff, never on the caller's own 10s poll: the Books/ watcher leaves a failed book's path unbaselined, so an unthrottled retry means ~360 identical rejected POSTs an hour (which is what a rate-limited garden book did). The wait is the server's own `retry_after` where it sends one, so the first retry lands as the cooldown expires rather than blind-polling through it. """ sig = hashlib.sha1(text.strip().encode('utf-8', 'replace')).hexdigest() if self._book_sigs.get(kind) == sig: return False if kind == 'warden' and sig in self._warden_sigs: return False if time.time() < self._book_retry_at.get(kind, 0): return False handler = {'weather': self._upload_weather, 'garden': self._upload_garden, 'warden': self._upload_warden_alerts}[kind] ok = (handler(text, source) if kind == 'warden' else handler(text, when, source)) if ok: self._book_sigs[kind] = sig self._book_retry_at.pop(kind, None) if kind == 'warden': self._warden_sigs.append(sig) else: wait = self._retry_hint or BOOK_RETRY_SECS self._book_retry_at[kind] = time.time() + wait print(f"[books] {kind} push rejected, not retrying for {int(wait)}s", file=sys.stderr) self._retry_hint = 0 return ok def _post_book(self, endpoint, payload): """POST one book payload, noting how long the server asked us to wait. The upload handlers return a bool, so the 429's `retry_after` reaches _report_book through `_retry_hint` — set here, read and cleared there. One wrapper is the only place that side channel is written.""" result = post_json(self.server_url, endpoint, payload) if isinstance(result, dict) and not result.get('ok'): try: self._retry_hint = max(0, int(result.get('retry_after') or 0)) except (TypeError, ValueError): self._retry_hint = 0 return result # ── Book auto-save (Player.log ProcessBook -> Books/*.txt) ───────────────── def _init_book_saving(self, save_books=None): """Resolve which books to auto-save and load the already-written set. Precedence: the --save-books argument, else GORGON_SAVE_BOOKS, else the SAVE_BOOKS_DEFAULT, which is every book. 'off'/'none' disables auto-saving. Wrapped like the other init helpers: a bad value or unreadable state file disables auto-saving for this run rather than crashing the monitor.""" try: raw = save_books if save_books is not None else os.environ.get('GORGON_SAVE_BOOKS', '') raw = (raw or '').strip() if raw.lower() in ('off', 'none', '0', 'false'): self._save_titles = [] elif raw.lower() == 'all': self._save_titles = ['*'] elif raw: self._save_titles = [t.strip() for t in raw.split(',') if t.strip()] else: self._save_titles = ['*'] # SAVE_BOOKS_DEFAULT if self._books_written_file.exists(): self._books_written = json.loads( self._books_written_file.read_text(encoding='utf-8')) # What is being saved is announced by _print_summary, which runs next. except Exception as e: self._save_titles = [] print(f"[books] auto-save init failed (disabled this run): {e}", file=sys.stderr) def _write_books_state(self): """Persist the written-book signatures, newest kept if the cap is hit.""" try: items = sorted(self._books_written.items(), key=lambda kv: kv[1].get('seen', ''))[-_BOOKS_WRITTEN_MAX:] self._books_written = dict(items) self._books_written_file.write_text( json.dumps(self._books_written, indent=1, ensure_ascii=False), encoding='utf-8') except Exception as e: print(f"[books] could not write {self._books_written_file.name}: {e}", file=sys.stderr) @staticmethod def _book_filename(title, when, export_name=''): """Name a saved book exactly as the game would. 'Order Fulfilled' -> 'OrderFulfilled_260902_200023.txt'. The EXPORT NAME wins when the book carries one, because it is not always the title with the spaces taken out: both "Today's Shop Logs" and "Yesterday's Shop Logs" export as PlayerShopLog, and shop_pnl.html reads the year out of exactly that prefix (gorgon-shoplog.js parseYear), so a title-derived name would be a file it does not recognise. Books with no export name (the work orders) keep the title-derived name they have always had. """ stem = re.sub(r'[^A-Za-z0-9]', '', export_name or '') or \ re.sub(r'[^A-Za-z0-9]', '', title) or 'Book' return f"{stem}_{when.strftime('%y%m%d_%H%M%S')}.txt" def _wants_saving(self, title): """Is this book one you asked to have saved to disk? Matching is on a piece of the title, so 'Shop Logs' catches both today's and yesterday's.""" titles = getattr(self, '_save_titles', []) if not titles: return False if titles == ['*']: return True low = title.lower() return any(t.lower() in low for t in titles) def _write_book_file(self, title, body, when, export_name=''): """Write one book exactly as the in-game save would: body only (no title, no timestamp — both live in the filename), CRLF line ends, markup intact. Returns the Path written, or None when it was already written or the write failed. Never overwrites: two books saved in the same second get _2, _3, … """ name = self._book_filename(title, when, export_name) base, path, n = name[:-4], BOOKS_DIR / name, 1 while path.exists(): # two books read in the same second n += 1 path = BOOKS_DIR / f'{base}_{n}.txt' try: BOOKS_DIR.mkdir(parents=True, exist_ok=True) with open(path, 'w', encoding='utf-8', newline='\r\n') as f: f.write(body) return path except Exception as e: print(f"[books] could not save {name}: {e}", file=sys.stderr) return None def _save_books(self, chunk): """Save every wanted book opened in this Player.log chunk to Books/. This is the save-step equivalent of what _scan_books does for world reports: reading the book in game is enough, no export click. The body is de-duped on its own content because re-opening a book logs an identical ProcessBook line, and for a work order a second file would double-book the cost downstream. "Now" is the book's time: the tail runs within seconds of the read, and the log line's own stamp is UTC while every saved book on disk is named in local time — deriving one from the other would misdate the file by the UTC offset. """ if not getattr(self, '_save_titles', None): return wrote = False for m in BOOK_LOG_RE.finditer(chunk): title, body, export_name = m.group(1), m.group(2), (m.group(3) or '') if not self._wants_saving(title): continue text = self._unescape_book(body) sig = hashlib.sha1(f'{title}\n{text}'.strip().encode('utf-8', 'replace')).hexdigest() if sig in self._books_written: continue path = self._write_book_file(title, text, datetime.datetime.now(), export_name) if not path: continue self._books_written[sig] = { 'title': title, 'file': path.name, 'seen': datetime.datetime.now().strftime('%Y-%m-%dT%H:%M:%S'), } wrote = True print(f"[books] saved {path.name}: {title}") if wrote: self._write_books_state() def _scan_books(self, chunk): """Report any world-report book OPENED in this Player.log chunk. This is what removes the export step: reading the book in-game is enough. """ if not self.world_reports: return for m in BOOK_LOG_RE.finditer(chunk): title, body = m.group(1), m.group(2) kind = next((k for k, pat in BOOK_KINDS if pat.search(title)), None) if not kind: continue # The tail runs within seconds of the read, so "now" is the book's # reference time — and it is already naive LOCAL, which is what # parse_almanac_text wants. Deriving it from the log's own stamp # would mean rebuilding a UTC date the line doesn't carry. self._report_book(kind, self._unescape_book(body), datetime.datetime.now(), f'Player.log ({title})') # ── Warden Alerts (Books/ActiveWardenAlerts_*.txt → /api/warden-alerts) ───── @staticmethod def _ends_at(ends): """'23 hours 48 minutes' -> absolute ISO-UTC end time, or '' if not parseable. Lets the site show a live countdown instead of the stale 'Ends in …' snapshot.""" secs = 0 for n, unit in re.findall(r'(\d+)\s*(day|hour|hr|minute|min)', ends or '', re.IGNORECASE): secs += int(n) * {'day': 86400, 'hour': 3600, 'hr': 3600, 'minute': 60, 'min': 60}[unit.lower()] if not secs: return '' return (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=secs)).strftime('%Y-%m-%dT%H:%M:%SZ') def _parse_warden_alerts(self, text): """Parse an ActiveWardenAlerts book into [{npc, zone, threat, ends, ends_at}]. Each alert is an <h2>Aid Request from NPC in Zone</h2> plus a threat sentence and an 'Ends in …' line. Threat is the phrase after 'assistance with'.""" alerts = [] for m in re.finditer(r'<h2>Aid Request from (.+?) in (.+?)</h2>(.*?)(?=<h2>|$)', text, re.DOTALL): npc, zone, rest = m.group(1).strip(), m.group(2).strip(), m.group(3) tm = re.search(r'assistance with (.+?)\.?\s*</indent>', rest, re.DOTALL) if tm: threat = tm.group(1).strip() else: bm = re.search(r'<indent=\d+>(.+?)</indent>', rest, re.DOTALL) threat = bm.group(1).strip() if bm else '' em = re.search(r'Ends in ([^<]+?)\s*</i>', rest) ends = em.group(1).strip() if em else '' alerts.append({'npc': npc, 'zone': zone, 'threat': threat, 'ends': ends, 'ends_at': self._ends_at(ends)}) return alerts def _upload_warden_alerts(self, text, source): """Send the Warden Alerts book to the site. True if it was accepted.""" alerts = self._parse_warden_alerts(text) if not alerts: print(f"[books] no warden alerts parsed from {source}", file=sys.stderr) return False result = self._post_book(f'/api/warden-alerts?server={self.game_server}', {'alerts': alerts}) if result and result.get('ok'): print(f"[books] Warden alerts uploaded: {result['alerts']} active " f"({', '.join(a['zone'] for a in alerts)})") return True print(f"[books] Warden alert upload failed: {result}", file=sys.stderr) return False def _upload_weather(self, text, when, source): """Send today's weather (one line per zone) to the site. True if accepted.""" ts = when.strftime('%Y-%m-%d %H:%M:%S') if when else '?' zones = parse_weather_text(text) if not zones: print(f"[books] ERROR: no weather zones parsed from {source}", file=sys.stderr) return False result = self._post_book( f'/api/weather/save?server={self.game_server}', {'zones': [{'zone': z, 'weather': w} for z, w in zones]}, ) if result and result.get('ok'): print(f"[books] Weather uploaded: {result['zones']} zones " f"({result['danger']} danger, {result['safe']} safe) [{ts}]") return True print(f"[books] Weather upload failed: {result}", file=sys.stderr) return False def _upload_garden(self, text, when, source): """Send the gardening almanac to the site. True if accepted. The book states its times relative to when you read it ("ends in 2 hours"), never as a clock time, so the moment it was read has to travel with it. That is what `when` is. """ # parse_almanac_text resolves the book's relative times ("ends in 2h") # against this reference, and book_timestamp yields a NAIVE LOCAL time # from the filename — so the log path must hand it local time too, never # the raw UTC log stamp, or every garden window lands hours off. file_time = when or datetime.datetime.now() ts = file_time.strftime('%Y-%m-%d %H:%M:%S') events = parse_almanac_text(text, file_time) if not events: print(f"[books] WARNING: no garden events parsed from {source}", file=sys.stderr) return False now_ms = time.time() * 1000 result = self._post_book( f'/api/garden/save?server={self.game_server}', {'events': [ { 'crop': e['crop'].lower(), 'zone': e['zone'], 'active': e['active'], 'dur_secs': max(0, int((e['abs_ts_ms'] - now_ms) / 1000)), } for e in events ]}, ) if result and result.get('ok'): print(f"[books] Garden uploaded: {result['active']} active, " f"{result['upcoming']} upcoming [{ts}]") if result.get('skipped'): print(f"[books] Skipped crops: {result['skipped']}", file=sys.stderr) return True print(f"[books] Garden upload failed: {result}", file=sys.stderr) return False # ============================================================================== # PART 6 | COMMAND LINE # # What runs when you type `python3 gorgon_monitor.py`. Everything has a sensible # default, so the options exist only for the cases where the guess is wrong. # ============================================================================== if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( description=f'Gorgon Monitor {__version__}. Leave it running while you play. ' 'It reads the logs Project Gorgon already writes on your PC and ' 'reports public world data (weather, garden, warden alerts, Aurest, ' "Agrashab, blood season, casino tips and fights, arena results, mob " 'drops, death XP) to the Gorgonzola community site. It sends nothing ' 'private and never touches the game. Full details, including exactly ' 'what is and is not sent, are in the header of this file, and at ' 'https://gorgonzola.spellboundtavern.com/monitor.html', epilog='Press Ctrl-C to stop.') parser.add_argument('--server', default=DEFAULT_SERVER, help=f'Site base URL to report to (default: {DEFAULT_SERVER})') parser.add_argument('--game-server', default=None, dest='game_server', metavar='NAME', help='Project Gorgon world (default: auto-detect ' f'from Player.log, else {FALLBACK_GAME_SERVER})') parser.add_argument('--data-dir', default=None, dest='data_dir', metavar='DIR', help='Directory for local research files (default: script directory)') parser.add_argument('--save-books', default=None, dest='save_books', metavar='LIST', help='Books to auto-save to the Books folder when read in game, ' 'skipping the in-game save click: "off", "all", or a ' 'comma-separated list of title substrings ' f'(default: {SAVE_BOOKS_DEFAULT}, every book you open; ' 'env GORGON_SAVE_BOOKS)') parser.add_argument('--game-dir', default=None, dest='game_dir', metavar='DIR', help="Project Gorgon's folder, the one holding ChatLogs and " 'Books (default: found automatically; env GORGON_GAME_DIR)') parser.add_argument('--version', action='version', version=f'Gorgon Monitor {__version__}') args = parser.parse_args() # Windows writes console output in the local code page, so the arrows and ellipses # in the lines below would end the run the moment somebody redirected output to a # file. Ask for utf-8 and, failing that, for unprintable characters to be replaced # rather than raised. Guarded: some environments hand over a stream that cannot be # reconfigured, and losing an accent is not worth losing the program over. for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(encoding='utf-8', errors='replace') except Exception: pass if args.game_dir: use_game_dir(args.game_dir) if not LOG_FOLDER.exists() and not PLAYER_LOG.exists(): # Nothing to read. Say so and stop: running on regardless would look healthy # and report nothing, which is the one failure nobody can diagnose. print(game_dir_help(args.game_dir or os.environ.get('GORGON_GAME_DIR')), file=sys.stderr) sys.exit(1) monitor = ChatMonitor(server_url=args.server, game_server=args.game_server, data_dir=args.data_dir, save_books=args.save_books) try: monitor.monitor_chat() except KeyboardInterrupt: # Ctrl-C is how the header tells people to stop it, so it is a normal exit and # must not look like a crash. monitor._flush_loot(force=True) # don't drop drops already seen print(f"\n\nStopped. {REPORTS_MADE} report(s) accepted this session. " f"Thanks for keeping the site current.") sys.exit(0)
It reads log files that the game writes and that anyone can open in Notepad. It does not modify, inject into, automate or communicate with the game. Reading your own logs is the same thing every combat parser and every "what did I loot" spreadsheet does.
Yes. It prints every single report to its window as it makes it. Nothing is sent silently.
The reports are not individually switchable yet. If you want one of them off, say so on
the feedback form and it will get a switch. Book saving already
has one (--save-books off).
Yes, and the endpoints this uses are documented for you on the API page, with the live list of valid values.
It is the shareable twin of the monitor that has been feeding this site since it
started. It lives in Gorgonzola's source as gorgon_monitor.py, the same
file you download. It is standard library only and deliberately one file, so that it
stays something a person can read in one sitting.