alexpolo1
968c75fefd
New modules: - test/auto/log_analyzer.py: parses bot logs and event JSONL files, categorizes failures (approach, maintenance, battle, chicken, timeout, crash, OCR) into structured BotFailure objects - test/auto/auto_fixer.py: maps failure patterns to targeted code fixes, checks if fixes are already applied, reports what needs work - test/auto/test_self_healing.py: orchestrator that launches the bot, monitors for failures in real-time, analyzes logs, applies fixes, and retries up to 3 rounds Supporting changes: - test/auto/test_log_analyzer.py: 8 tests against historical run data - pytest.ini: added repo root to pythonpath for test package imports - test/__init__.py: new, enables test/ as importable package All 95 existing tests pass. 8 new auto-test tests pass.
98 lines
3.1 KiB
Python
98 lines
3.1 KiB
Python
#!/usr/bin/env python
|
|
"""Hermes bot control — send commands to botty via TCP socket on 127.0.0.1:18899.
|
|
|
|
Usage:
|
|
python scripts/hermes_bot_control.py start # start/pause bot
|
|
python scripts/hermes_bot_control.py pause # toggle pause
|
|
python scripts/hermes_bot_control.py stop # stop bot
|
|
python scripts/hermes_bot_control.py status # get bot status
|
|
python scripts/hermes_bot_control.py logs [n] # last n log lines
|
|
python scripts/hermes_bot_control.py errors [n] # last n error lines
|
|
python scripts/hermes_bot_control.py runs # run stats
|
|
"""
|
|
import sys
|
|
import os
|
|
import time
|
|
import subprocess
|
|
import glob
|
|
import socket
|
|
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
PROJECT_ROOT = os.path.dirname(SCRIPT_DIR)
|
|
LOG_FILE = os.path.join(PROJECT_ROOT, 'log', 'log.txt')
|
|
SOCKET_PORT = 18899
|
|
|
|
def send_command(cmd):
|
|
"""Send a command to the bot's control socket."""
|
|
try:
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(3.0)
|
|
s.connect(('127.0.0.1', SOCKET_PORT))
|
|
s.sendall(cmd.encode())
|
|
response = s.recv(1024).decode().strip()
|
|
s.close()
|
|
if response:
|
|
print(response)
|
|
else:
|
|
print(f"OK: command '{cmd}' sent")
|
|
except socket.timeout:
|
|
print(f"ERROR: no response from bot (socket timeout)")
|
|
except ConnectionRefusedError:
|
|
print(f"ERROR: bot not listening on port {SOCKET_PORT} (is it running?)")
|
|
except Exception as e:
|
|
print(f"ERROR: {e}")
|
|
|
|
def read_logs(n=20):
|
|
if not os.path.exists(LOG_FILE):
|
|
print("No log file found")
|
|
return
|
|
result = subprocess.run(["tail", "-n", str(n), LOG_FILE],
|
|
capture_output=True, text=True)
|
|
print(result.stdout)
|
|
|
|
def read_errors(n=10):
|
|
if not os.path.exists(LOG_FILE):
|
|
print("No log file found")
|
|
return
|
|
result = subprocess.run(["grep", "-E", "ERROR|WARNING|Failed|failed|ERROR.*step", LOG_FILE],
|
|
capture_output=True, text=True)
|
|
lines = result.stdout.strip().split('\n')
|
|
for line in lines[-n:]:
|
|
print(line)
|
|
|
|
def run_stats():
|
|
stats_dir = os.path.join(PROJECT_ROOT, 'log', 'stats')
|
|
stats_files = glob.glob(os.path.join(stats_dir, 'stats_*.log'))
|
|
if stats_files:
|
|
latest = max(stats_files, key=os.path.getmtime)
|
|
with open(latest) as f:
|
|
print(f.read())
|
|
else:
|
|
print("No stats files found")
|
|
|
|
def check_status():
|
|
send_command('status')
|
|
|
|
if __name__ == '__main__':
|
|
if len(sys.argv) < 2:
|
|
print(__doc__)
|
|
sys.exit(1)
|
|
|
|
cmd = sys.argv[1]
|
|
|
|
if cmd in ('start', 'pause', 'stop'):
|
|
send_command(cmd)
|
|
elif cmd == 'status':
|
|
check_status()
|
|
elif cmd == 'logs':
|
|
n = int(sys.argv[2]) if len(sys.argv) > 2 else 20
|
|
read_logs(n)
|
|
elif cmd == 'errors':
|
|
n = int(sys.argv[2]) if len(sys.argv) > 2 else 10
|
|
read_errors(n)
|
|
elif cmd == 'runs':
|
|
run_stats()
|
|
else:
|
|
print(f"Unknown command: {cmd}")
|
|
print(__doc__)
|
|
sys.exit(1) |