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.
323 lines
14 KiB
Python
323 lines
14 KiB
Python
"""
|
|
Log analyzer for botty. Reads bot logs and event files, extracts
|
|
structured failure information for automated diagnosis and fixing.
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import List
|
|
|
|
|
|
@dataclass
|
|
class BotFailure:
|
|
"""A single failure event extracted from bot logs."""
|
|
failure_type: str # "approach_failed", "maintenance_failed", "battle_failed", "chicken", "timeout", "ocr_error", "crash"
|
|
run_name: str = ""
|
|
step: str = ""
|
|
reason: str = ""
|
|
game_number: int = 0
|
|
run_number: int = 0
|
|
elapsed_seconds: float = 0.0
|
|
timestamp: str = ""
|
|
source_file: str = ""
|
|
|
|
|
|
@dataclass
|
|
class LogAnalysisResult:
|
|
"""Results of analyzing one bot session."""
|
|
failures: List[BotFailure] = field(default_factory=list)
|
|
total_runs: int = 0
|
|
successful_runs: int = 0
|
|
failed_runs: int = 0
|
|
session_duration_seconds: float = 0.0
|
|
log_file: str = ""
|
|
# Failure counts by type
|
|
approach_failures: int = 0
|
|
maintenance_failures: int = 0
|
|
battle_failures: int = 0
|
|
chicken_triggers: int = 0
|
|
timeouts: int = 0
|
|
crashes: int = 0
|
|
ocr_errors: int = 0
|
|
|
|
|
|
class LogAnalyzer:
|
|
"""Analyzes botty log files and event JSONL files for failures."""
|
|
|
|
# Patterns for parsing log lines
|
|
PATTERNS = {
|
|
"approach_failed": re.compile(
|
|
r"Approach failed for (\w+)\s*\[step:\s*(\w+)\]"
|
|
),
|
|
"maintenance_failed": re.compile(
|
|
r"Maintenance failed\s*\[step:\s*(\w+)\]\s*(?:—\s*(.+))?"
|
|
),
|
|
"battle_failed": re.compile(
|
|
r"Battle failed for (\w+)"
|
|
),
|
|
"chicken": re.compile(
|
|
r"Health chicken triggered"
|
|
),
|
|
"timeout": re.compile(
|
|
r"Maintenance timeout after (\d+)s before \[(\w+)\](?:\s*—\s*(.+))?"
|
|
),
|
|
"ocr_error": re.compile(
|
|
r"Neither tesserocr nor pytesseract"
|
|
),
|
|
"crash": re.compile(
|
|
r"Traceback|Uncaught exception"
|
|
),
|
|
}
|
|
|
|
def analyze_log_file(self, log_path: str) -> LogAnalysisResult:
|
|
"""Analyze a bot log.txt file for failures."""
|
|
result = LogAnalysisResult(log_file=log_path)
|
|
|
|
try:
|
|
with open(log_path, 'r', encoding='utf-8', errors='replace') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
# Check for approach failures
|
|
m = self.PATTERNS["approach_failed"].search(line)
|
|
if m:
|
|
result.approach_failures += 1
|
|
result.failures.append(BotFailure(
|
|
failure_type="approach_failed",
|
|
run_name=m.group(1),
|
|
step=m.group(2),
|
|
reason=line,
|
|
source_file=log_path,
|
|
))
|
|
continue
|
|
|
|
# Check for maintenance failures
|
|
m = self.PATTERNS["maintenance_failed"].search(line)
|
|
if m:
|
|
result.maintenance_failures += 1
|
|
reason = m.group(2) if m.group(2) else ""
|
|
result.failures.append(BotFailure(
|
|
failure_type="maintenance_failed",
|
|
step=m.group(1),
|
|
reason=reason,
|
|
source_file=log_path,
|
|
))
|
|
continue
|
|
|
|
# Check for battle failures
|
|
m = self.PATTERNS["battle_failed"].search(line)
|
|
if m:
|
|
result.battle_failures += 1
|
|
result.failures.append(BotFailure(
|
|
failure_type="battle_failed",
|
|
run_name=m.group(1),
|
|
reason=line,
|
|
source_file=log_path,
|
|
))
|
|
continue
|
|
|
|
# Check for chicken triggers
|
|
if self.PATTERNS["chicken"].search(line):
|
|
result.chicken_triggers += 1
|
|
result.failures.append(BotFailure(
|
|
failure_type="chicken",
|
|
reason="Health chicken triggered",
|
|
source_file=log_path,
|
|
))
|
|
continue
|
|
|
|
# Check for timeouts
|
|
m = self.PATTERNS["timeout"].search(line)
|
|
if m:
|
|
result.timeouts += 1
|
|
step = m.group(2) if m.group(2) else ""
|
|
reason = m.group(3) if m.group(3) else ""
|
|
result.failures.append(BotFailure(
|
|
failure_type="timeout",
|
|
step=step,
|
|
reason=reason,
|
|
elapsed_seconds=float(m.group(1)),
|
|
source_file=log_path,
|
|
))
|
|
continue
|
|
|
|
# Check for OCR errors
|
|
if self.PATTERNS["ocr_error"].search(line):
|
|
result.ocr_errors += 1
|
|
result.failures.append(BotFailure(
|
|
failure_type="ocr_error",
|
|
reason=line,
|
|
source_file=log_path,
|
|
))
|
|
continue
|
|
|
|
# Check for crashes
|
|
if self.PATTERNS["crash"].search(line):
|
|
result.crashes += 1
|
|
result.failures.append(BotFailure(
|
|
failure_type="crash",
|
|
reason=line,
|
|
source_file=log_path,
|
|
))
|
|
continue
|
|
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
return result
|
|
|
|
def analyze_event_file(self, event_path: str) -> LogAnalysisResult:
|
|
"""Analyze an events_*.jsonl file for failures."""
|
|
result = LogAnalysisResult(log_file=event_path)
|
|
|
|
try:
|
|
with open(event_path, 'r', encoding='utf-8') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
try:
|
|
event = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
event_type = event.get("event", "")
|
|
|
|
if event_type == "game_ended":
|
|
result.total_runs += 1
|
|
if event.get("failed", False):
|
|
result.failed_runs += 1
|
|
reason = event.get("reason", "")
|
|
run_name = event.get("location", "")
|
|
|
|
if "Approach failed" in reason:
|
|
result.approach_failures += 1
|
|
m = re.search(r"Approach failed for (\w+)\s*\[step:\s*(\w+)\]", reason)
|
|
step = m.group(2) if m else ""
|
|
run = m.group(1) if m else run_name
|
|
result.failures.append(BotFailure(
|
|
failure_type="approach_failed",
|
|
run_name=run,
|
|
step=step,
|
|
reason=reason,
|
|
game_number=event.get("game", 0),
|
|
run_number=event.get("run", 0),
|
|
elapsed_seconds=event.get("elapsed_seconds", 0),
|
|
timestamp=event.get("ts", ""),
|
|
source_file=event_path,
|
|
))
|
|
elif "Maintenance failed" in reason:
|
|
result.maintenance_failures += 1
|
|
m = re.search(r"step:\s*(\w+)", reason)
|
|
step = m.group(1) if m else ""
|
|
result.failures.append(BotFailure(
|
|
failure_type="maintenance_failed",
|
|
run_name=run_name,
|
|
step=step,
|
|
reason=reason,
|
|
game_number=event.get("game", 0),
|
|
run_number=event.get("run", 0),
|
|
elapsed_seconds=event.get("elapsed_seconds", 0),
|
|
timestamp=event.get("ts", ""),
|
|
source_file=event_path,
|
|
))
|
|
elif "Battle failed" in reason:
|
|
result.battle_failures += 1
|
|
m = re.search(r"Battle failed for (\w+)", reason)
|
|
run = m.group(1) if m else run_name
|
|
result.failures.append(BotFailure(
|
|
failure_type="battle_failed",
|
|
run_name=run,
|
|
reason=reason,
|
|
game_number=event.get("game", 0),
|
|
run_number=event.get("run", 0),
|
|
elapsed_seconds=event.get("elapsed_seconds", 0),
|
|
timestamp=event.get("ts", ""),
|
|
source_file=event_path,
|
|
))
|
|
elif "Maintenance timeout" in reason:
|
|
result.timeouts += 1
|
|
m = re.search(r"timeout after (\d+)s before \[(\w+)\]", reason)
|
|
elapsed = int(m.group(1)) if m else 0
|
|
step = m.group(2) if m else ""
|
|
result.failures.append(BotFailure(
|
|
failure_type="timeout",
|
|
run_name=run_name,
|
|
step=step,
|
|
reason=reason,
|
|
elapsed_seconds=elapsed,
|
|
game_number=event.get("game", 0),
|
|
run_number=event.get("run", 0),
|
|
timestamp=event.get("ts", ""),
|
|
source_file=event_path,
|
|
))
|
|
elif "Health chicken" in reason:
|
|
result.chicken_triggers += 1
|
|
result.failures.append(BotFailure(
|
|
failure_type="chicken",
|
|
run_name=run_name,
|
|
reason=reason,
|
|
game_number=event.get("game", 0),
|
|
run_number=event.get("run", 0),
|
|
elapsed_seconds=event.get("elapsed_seconds", 0),
|
|
timestamp=event.get("ts", ""),
|
|
source_file=event_path,
|
|
))
|
|
else:
|
|
result.failures.append(BotFailure(
|
|
failure_type="unknown_failure",
|
|
run_name=run_name,
|
|
reason=reason,
|
|
game_number=event.get("game", 0),
|
|
run_number=event.get("run", 0),
|
|
elapsed_seconds=event.get("elapsed_seconds", 0),
|
|
timestamp=event.get("ts", ""),
|
|
source_file=event_path,
|
|
))
|
|
else:
|
|
result.successful_runs += 1
|
|
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
return result
|
|
|
|
def analyze_all(self, log_dir: str, stats_dir: str = None) -> LogAnalysisResult:
|
|
"""Analyze all log files in a directory."""
|
|
combined = LogAnalysisResult(log_file=log_dir)
|
|
|
|
# Analyze log.txt
|
|
log_path = os.path.join(log_dir, "log.txt")
|
|
log_result = self.analyze_log_file(log_path)
|
|
self._merge(combined, log_result)
|
|
|
|
# Analyze all event files
|
|
if stats_dir is None:
|
|
stats_dir = os.path.join(log_dir, "stats")
|
|
|
|
if os.path.isdir(stats_dir):
|
|
for fname in os.listdir(stats_dir):
|
|
if fname.startswith("events_") and fname.endswith(".jsonl"):
|
|
event_result = self.analyze_event_file(os.path.join(stats_dir, fname))
|
|
self._merge(combined, event_result)
|
|
|
|
return combined
|
|
|
|
@staticmethod
|
|
def _merge(target: LogAnalysisResult, source: LogAnalysisResult):
|
|
"""Merge one analysis result into another."""
|
|
target.failures.extend(source.failures)
|
|
target.total_runs += source.total_runs
|
|
target.successful_runs += source.successful_runs
|
|
target.failed_runs += source.failed_runs
|
|
target.approach_failures += source.approach_failures
|
|
target.maintenance_failures += source.maintenance_failures
|
|
target.battle_failures += source.battle_failures
|
|
target.chicken_triggers += source.chicken_triggers
|
|
target.timeouts += source.timeouts
|
|
target.crashes += source.crashes
|
|
target.ocr_errors += source.ocr_errors |