360 lines
14 KiB
Python
360 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Self-installing / self-updating entry point — `python update.py`.
|
|
|
|
Run it once (the install one-liner does this for you). From then on a Windows
|
|
Scheduled Task runs it automatically **at logon and once a day**, so the watchers
|
|
stay up to date and running with zero further thought. Each run it:
|
|
|
|
1. makes config.ini on first run (webhook prefilled; gitignored so your edits
|
|
and OCR snip survive every update),
|
|
2. `git pull`s the latest watcher code,
|
|
3. installs/upgrades deps only when something changed,
|
|
4. (re)starts the watchers ONLY if the code changed or they aren't running —
|
|
so the daily run never interrupts an active session for nothing,
|
|
5. ensures the daily+logon Scheduled Task exists (idempotent).
|
|
|
|
First time on a new PC (or just paste the install one-liner):
|
|
git clone https://git.armoredarmadillo.com/brockdarnold/eve-watcher.git
|
|
cd eve-watcher
|
|
python update.py
|
|
python eve_orehold_watcher.py --snip # one-time GUI step for OCR hold alerts
|
|
"""
|
|
import configparser
|
|
import glob
|
|
import os
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
|
|
# Suppress flashing console windows from child processes (powershell/git/pip/tesseract).
|
|
if os.name == "nt":
|
|
_CREATE_NO_WINDOW = 0x08000000
|
|
_orig_popen_init = subprocess.Popen.__init__
|
|
|
|
def _popen_no_window(self, *a, **k):
|
|
k["creationflags"] = k.get("creationflags", 0) | _CREATE_NO_WINDOW
|
|
_orig_popen_init(self, *a, **k)
|
|
|
|
subprocess.Popen.__init__ = _popen_no_window
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
WIN = os.name == "nt"
|
|
TASK = "EveWatcher"
|
|
WATCHER_SCRIPTS = ("eve_combat_watcher.py", "eve_chat_watcher.py",
|
|
"eve_orehold_watcher.py", "eve_rock_watcher.py",
|
|
"eve_audio_watcher.py") # eve_local_watcher disabled (false threats)
|
|
|
|
|
|
def run(cmd, **kw):
|
|
print(f" $ {' '.join(cmd)}")
|
|
return subprocess.run(cmd, cwd=HERE, **kw)
|
|
|
|
|
|
def ps(script):
|
|
"""Run a PowerShell snippet, return (rc, stdout+stderr)."""
|
|
r = subprocess.run(["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass",
|
|
"-Command", script], cwd=HERE, capture_output=True, text=True)
|
|
return r.returncode, ((r.stdout or "") + (r.stderr or "")).strip()
|
|
|
|
|
|
def pythonw():
|
|
p = os.path.join(os.path.dirname(sys.executable), "pythonw.exe")
|
|
return p if os.path.exists(p) else "pythonw.exe"
|
|
|
|
|
|
def ensure_config():
|
|
cfg = os.path.join(HERE, "config.ini")
|
|
first = not os.path.exists(cfg)
|
|
if first:
|
|
shutil.copyfile(os.path.join(HERE, "config.ini.example"), cfg)
|
|
print("• created config.ini from example (webhook prefilled).")
|
|
else:
|
|
print("• config.ini exists — your settings + OCR snip left untouched.")
|
|
# heal a UTF-8 BOM left by the old PowerShell installer (breaks configparser)
|
|
with open(cfg, "rb") as fh:
|
|
raw = fh.read()
|
|
if raw.startswith(b"\xef\xbb\xbf"):
|
|
with open(cfg, "wb") as fh:
|
|
fh.write(raw[3:])
|
|
print("• stripped a UTF-8 BOM from config.ini (old-installer leftover).")
|
|
return cfg, first
|
|
|
|
|
|
ZIP_URL = "https://git.armoredarmadillo.com/brockdarnold/eve-watcher/archive/main.zip"
|
|
|
|
|
|
def _git(*a):
|
|
env = dict(os.environ, GIT_TERMINAL_PROMPT="0") # never block on an interactive prompt
|
|
return subprocess.run(["git", "-C", HERE, *a], capture_output=True, text=True, env=env)
|
|
|
|
|
|
def _have_git():
|
|
import shutil as _sh
|
|
return os.path.isdir(os.path.join(HERE, ".git")) and _sh.which("git") is not None
|
|
|
|
|
|
def update_code():
|
|
"""Pull the latest watcher code. Uses git when this is a clone; otherwise
|
|
downloads the repo zip and overwrites files (config.ini is never touched).
|
|
Returns True if anything changed."""
|
|
if _have_git():
|
|
before = _git("rev-parse", "HEAD").stdout.strip()
|
|
if _git("fetch", "-q", "origin").returncode != 0:
|
|
print("! git fetch failed — falling back to zip update.")
|
|
return _zip_update()
|
|
if _git("reset", "--hard", "-q", "origin/main").returncode != 0:
|
|
print("! git reset failed (locked .git pack?) — falling back to zip update.")
|
|
return _zip_update() # zip writes files directly, no .git pack to unlink
|
|
after = _git("rev-parse", "HEAD").stdout.strip()
|
|
if before and after and before != after:
|
|
print(f" updated: {before[:7]} -> {after[:7]}")
|
|
return True
|
|
print(" already on latest code.")
|
|
return False
|
|
return _zip_update()
|
|
|
|
|
|
def _zip_update():
|
|
"""git-less update: download the repo zip and write any changed files."""
|
|
import io
|
|
import urllib.request
|
|
import zipfile
|
|
try:
|
|
req = urllib.request.Request(ZIP_URL, headers={"User-Agent": "eve-watcher-updater"})
|
|
data = urllib.request.urlopen(req, timeout=60).read()
|
|
z = zipfile.ZipFile(io.BytesIO(data))
|
|
except Exception as e:
|
|
print(f"• update check failed (offline?): {e}")
|
|
return False
|
|
changed = 0
|
|
for n in z.namelist():
|
|
if n.endswith("/"):
|
|
continue
|
|
rel = n.split("/", 1)[1] if "/" in n else n # strip top 'eve-watcher/'
|
|
if not rel or rel == "config.ini": # never clobber local config
|
|
continue
|
|
dest = os.path.join(HERE, rel.replace("/", os.sep))
|
|
new = z.read(n)
|
|
old = open(dest, "rb").read() if os.path.exists(dest) else None
|
|
if old != new:
|
|
os.makedirs(os.path.dirname(dest) or HERE, exist_ok=True)
|
|
with open(dest, "wb") as f:
|
|
f.write(new)
|
|
changed += 1
|
|
print(f" code updated (zip): {changed} file(s)." if changed
|
|
else " already on latest code (zip).")
|
|
return changed > 0
|
|
|
|
|
|
def deps():
|
|
req = os.path.join(HERE, "requirements.txt")
|
|
if os.path.exists(req):
|
|
run([sys.executable, "-m", "pip", "install", "--quiet", "-r", req])
|
|
|
|
|
|
def ensure_tesseract(cfg_path):
|
|
"""OCR needs the Tesseract binary. Detect it; if missing, try a silent winget
|
|
install (best-effort — may raise a one-time UAC prompt). Writes tesseract_cmd
|
|
into config.ini when found. Non-fatal: timer/combat/chat work without it."""
|
|
if not WIN:
|
|
return
|
|
import shutil as _sh
|
|
std = r"C:\Program Files\Tesseract-OCR\tesseract.exe"
|
|
path = _sh.which("tesseract") or (std if os.path.exists(std) else None)
|
|
if not path:
|
|
print("• Tesseract not found — installing (one-time, may prompt UAC)...")
|
|
try:
|
|
subprocess.run(["winget", "install", "--id", "UB-Mannheim.TesseractOCR",
|
|
"-e", "--silent", "--accept-package-agreements",
|
|
"--accept-source-agreements"],
|
|
capture_output=True, text=True, timeout=600)
|
|
except Exception as ex:
|
|
print(f" (winget install skipped: {ex})")
|
|
path = _sh.which("tesseract") or (std if os.path.exists(std) else None)
|
|
if not path:
|
|
print("• Tesseract still missing — live OCR % off until it's installed "
|
|
"(combat/chat/phone alerts still work).")
|
|
return
|
|
# record the path so pytesseract finds it even if not on PATH
|
|
cp = configparser.ConfigParser()
|
|
cp.read(cfg_path, encoding="utf-8-sig")
|
|
if not cp.has_section("ocr"):
|
|
cp.add_section("ocr")
|
|
if cp.get("ocr", "tesseract_cmd", fallback="").strip() != path:
|
|
cp["ocr"]["tesseract_cmd"] = path
|
|
with open(cfg_path, "w") as f:
|
|
cp.write(f)
|
|
print(f"• Tesseract ready: {path}")
|
|
|
|
|
|
def _kill_scripts(names):
|
|
"""Force-stop pythonw processes whose command line names any of `names`."""
|
|
if not WIN:
|
|
return
|
|
pats = " -or ".join(f"$_.CommandLine -like '*{n}*'" for n in names)
|
|
ps("Get-CimInstance Win32_Process -Filter \"Name='pythonw.exe'\" | "
|
|
f"Where-Object {{ {pats} }} | "
|
|
"ForEach-Object { Stop-Process -Id $_.ProcessId -Force }")
|
|
|
|
|
|
def _start_script(name):
|
|
if not WIN:
|
|
return
|
|
ps(f"Start-Process pythonw -ArgumentList '\"{os.path.join(HERE, name)}\"' "
|
|
f"-WorkingDirectory '{HERE}' -WindowStyle Hidden")
|
|
|
|
|
|
def restart_mining_watchers():
|
|
"""Restart only the data watchers (not the autoupdate poller) — used by the
|
|
poller to bring the watchers onto freshly-pulled code without killing itself."""
|
|
_kill_scripts(WATCHER_SCRIPTS)
|
|
for n in WATCHER_SCRIPTS:
|
|
_start_script(n)
|
|
|
|
|
|
def watchers_running():
|
|
if not WIN:
|
|
return 0
|
|
_, out = ps("(Get-CimInstance Win32_Process -Filter \"Name='pythonw.exe'\" "
|
|
"| Where-Object { $_.CommandLine -like '*eve_*' }).Count")
|
|
try:
|
|
return int(out or "0")
|
|
except ValueError:
|
|
return 0
|
|
|
|
|
|
def running_scripts():
|
|
"""Which watcher scripts are actually running right now (by command line)."""
|
|
if not WIN:
|
|
return []
|
|
_, out = ps("Get-CimInstance Win32_Process -Filter \"Name='pythonw.exe'\" | "
|
|
"Select-Object -ExpandProperty CommandLine")
|
|
allw = WATCHER_SCRIPTS + ("eve_autoupdate.py",)
|
|
return [s for s in allw if s in (out or "")]
|
|
|
|
|
|
def _diagnostics(cp):
|
|
"""Self-check the things that make watchers silent, so we can see the root cause."""
|
|
import shutil as _sh
|
|
checks = []
|
|
# Tesseract (hold % + rock readers need it)
|
|
std = r"C:\Program Files\Tesseract-OCR\tesseract.exe"
|
|
tess = _sh.which("tesseract") or (std if os.path.exists(std) else
|
|
(cp.get("ocr", "tesseract_cmd", fallback="").strip() or None))
|
|
checks.append(("tesseract", bool(tess and os.path.exists(tess) if tess and ":" in str(tess) else tess)))
|
|
# audio deps
|
|
for mod in ("numpy", "soundcard"):
|
|
try:
|
|
__import__(mod); checks.append((mod, True))
|
|
except Exception:
|
|
checks.append((mod, False))
|
|
# EVE log dirs (combat/chat need them)
|
|
docs = os.path.expanduser("~/Documents/EVE/logs")
|
|
one = os.path.expanduser("~/OneDrive/Documents/EVE/logs")
|
|
checks.append(("EVE Gamelogs", os.path.isdir(docs + "/Gamelogs") or os.path.isdir(one + "/Gamelogs")))
|
|
checks.append(("EVE Chatlogs", os.path.isdir(docs + "/Chatlogs") or os.path.isdir(one + "/Chatlogs")))
|
|
# calibrated regions
|
|
checks.append(("hold region", bool(cp.get("ocr", "region", fallback="").strip())))
|
|
checks.append(("survey region", bool(cp.has_section("rock") and cp.get("rock", "region", fallback="").strip())))
|
|
return checks
|
|
|
|
|
|
def announce(cfg_path):
|
|
"""Post a definitive install/update report to Discord — always fires (not deduped):
|
|
which watchers are up + a self-diagnostic of what's missing, so silence is explainable."""
|
|
try:
|
|
import eve_orehold_watcher as w
|
|
cp = configparser.ConfigParser()
|
|
cp.read(cfg_path, encoding="utf-8-sig")
|
|
run_now = running_scripts()
|
|
short = ", ".join(s.replace("eve_", "").replace("_watcher.py", "").replace(".py", "")
|
|
for s in run_now) or "none"
|
|
diag = _diagnostics(cp)
|
|
diag_line = " ".join(f"{'✅' if ok else '❌'}{name}" for name, ok in diag)
|
|
host = socket.gethostname()
|
|
w._discord(cp, "🟢 eve-watcher updated",
|
|
f"On **{host}** — {len(run_now)} up: {short}\n{diag_line}\n"
|
|
"`!mining on` starts the live feed. ❌ items above are why something's quiet.")
|
|
print(f"announced: {len(run_now)} running — {short}")
|
|
except Exception as e:
|
|
print(f"announce skipped: {e}")
|
|
|
|
|
|
def stop_watchers():
|
|
# all of ours run as pythonw on an eve_*.py script (watchers + autoupdate poller)
|
|
_kill_scripts(("eve_",))
|
|
|
|
|
|
def start_watchers(cfg_path):
|
|
if not WIN:
|
|
print("• non-Windows — start watchers manually (this is a Windows tool).")
|
|
return
|
|
stop_watchers()
|
|
run(["powershell", "-ExecutionPolicy", "Bypass", "-File",
|
|
os.path.join(HERE, "start-all.ps1")])
|
|
cp = configparser.ConfigParser()
|
|
cp.read(cfg_path, encoding="utf-8-sig")
|
|
if cp.get("watcher", "mode", fallback="ocr") == "ocr" and \
|
|
not cp.get("ocr", "region", fallback="").strip():
|
|
print("\n >> The live hold % reader auto-calibrates: just have your in-game\n"
|
|
" Ore Hold window OPEN and it finds the readout itself (no snip).")
|
|
|
|
|
|
def ensure_logon_autostart():
|
|
"""Drop a launcher in the Startup folder so this runs at every logon.
|
|
No admin needed — most reliable auto-start mechanism on Windows."""
|
|
if not WIN:
|
|
return False
|
|
rc, startup = ps("[Environment]::GetFolderPath('Startup')")
|
|
startup = startup.strip()
|
|
if rc != 0 or not startup or not os.path.isdir(startup):
|
|
print("! couldn't find Startup folder for logon auto-start")
|
|
return False
|
|
launcher = os.path.join(startup, "EveWatcher.cmd")
|
|
body = ('@echo off\r\n'
|
|
f'start "" "{pythonw()}" "{os.path.join(HERE, "update.py")}"\r\n')
|
|
with open(launcher, "w", newline="") as f:
|
|
f.write(body)
|
|
print(f"• logon auto-start installed: {launcher}")
|
|
return True
|
|
|
|
|
|
def ensure_daily_task():
|
|
"""Best-effort daily Scheduled Task (covers PCs left on for days). Uses
|
|
schtasks (works for the current user without admin). Idempotent."""
|
|
if not WIN:
|
|
return
|
|
q = subprocess.run(["schtasks", "/query", "/tn", TASK],
|
|
capture_output=True, text=True)
|
|
if q.returncode == 0:
|
|
print(f"• daily task '{TASK}' already set.")
|
|
return
|
|
tr = f'"{pythonw()}" "{os.path.join(HERE, "update.py")}"'
|
|
c = subprocess.run(["schtasks", "/create", "/tn", TASK, "/tr", tr,
|
|
"/sc", "DAILY", "/st", "05:00", "/f"],
|
|
capture_output=True, text=True)
|
|
if c.returncode == 0:
|
|
print(f"• daily auto-update task '{TASK}' created (05:00).")
|
|
else:
|
|
print(f"• daily task not created ({(c.stdout + c.stderr).strip() or 'unknown'});"
|
|
" logon auto-start still covers updates.")
|
|
|
|
|
|
def main():
|
|
print("=== Eve watcher setup / auto-update ===")
|
|
cfg, first = ensure_config()
|
|
try:
|
|
update_code()
|
|
deps() # ensure deps every explicit run (cheap if satisfied)
|
|
ensure_tesseract(cfg)
|
|
start_watchers(cfg)
|
|
ensure_logon_autostart()
|
|
ensure_daily_task()
|
|
finally:
|
|
announce(cfg) # ALWAYS report to Discord (even if a step above failed)
|
|
print("\nDone. Hands-off from here: it updates + keeps itself running automatically.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|