Download logs to .logs/gh/ directory for analysis. Returns: Path to downloaded log file, or None if download failed
(self)
| 254 | return None |
| 255 | |
| 256 | def download_logs(self) -> Optional[Path]: |
| 257 | """Download logs to .logs/gh/ directory for analysis. |
| 258 | |
| 259 | Returns: |
| 260 | Path to downloaded log file, or None if download failed |
| 261 | """ |
| 262 | # Create .logs/gh/ directory |
| 263 | log_dir = Path(".logs/gh") |
| 264 | log_dir.mkdir(parents=True, exist_ok=True) |
| 265 | |
| 266 | log_file = log_dir / f"run_{self.run_id}.txt" |
| 267 | |
| 268 | # Check if log already exists and is recent (< 1 hour old) |
| 269 | if log_file.exists(): |
| 270 | age_seconds: float = log_file.stat().st_ctime |
| 271 | if time.time() - age_seconds < 3600: |
| 272 | print(f"Using cached log file: {log_file}") |
| 273 | return log_file |
| 274 | |
| 275 | print(f"Downloading logs to: {log_file}") |
| 276 | |
| 277 | try: |
| 278 | # Download logs using gh run view |
| 279 | result = subprocess.run( |
| 280 | ["gh", "run", "view", self.run_id, "--log"], |
| 281 | capture_output=True, |
| 282 | text=True, |
| 283 | encoding="utf-8", |
| 284 | errors="replace", |
| 285 | check=True, |
| 286 | timeout=300, # 5 minute timeout |
| 287 | ) |
| 288 | |
| 289 | # Save to file |
| 290 | with open(log_file, "w", encoding="utf-8") as f: |
| 291 | f.write(result.stdout) |
| 292 | |
| 293 | print(f"Log downloaded successfully ({len(result.stdout)} bytes)\n") |
| 294 | return log_file |
| 295 | |
| 296 | except subprocess.TimeoutExpired: |
| 297 | print("Error: Log download timed out", file=sys.stderr) |
| 298 | return None |
| 299 | except KeyboardInterrupt as ki: |
| 300 | handle_keyboard_interrupt(ki) |
| 301 | raise |
| 302 | except Exception as e: |
| 303 | print(f"Error downloading logs: {e}", file=sys.stderr) |
| 304 | return None |
| 305 | |
| 306 | def analyze_logs(self, log_file: Path) -> None: |
| 307 | """Analyze downloaded logs for errors. |