""" pyserve logs - View service logs """ import asyncio import time from pathlib import Path from typing import Any, Optional import click @click.command("logs") @click.argument("services", nargs=-1) @click.option( "-f", "--follow", is_flag=True, help="Follow log output", ) @click.option( "--tail", "tail", default=100, type=int, help="Number of lines to show from the end", ) @click.option( "--since", "since", default=None, help="Show logs since timestamp (e.g., '10m', '1h', '2024-01-01')", ) @click.option( "--until", "until_time", default=None, help="Show logs until timestamp", ) @click.option( "-t", "--timestamps", is_flag=True, help="Show timestamps", ) @click.option( "--no-color", is_flag=True, help="Disable colored output", ) @click.option( "--filter", "filter_pattern", default=None, help="Filter logs by pattern", ) @click.pass_obj def logs_cmd( ctx: Any, services: tuple[str, ...], follow: bool, tail: int, since: Optional[str], until_time: Optional[str], timestamps: bool, no_color: bool, filter_pattern: Optional[str], ) -> None: """ View service logs. If no services are specified, shows logs from all services. \b Examples: pyserve logs # All logs pyserve logs api # Logs from api service pyserve logs api admin # Logs from multiple services pyserve logs -f # Follow logs pyserve logs --tail 50 # Last 50 lines pyserve logs --since "10m" # Logs from last 10 minutes """ from ..output import print_info from ..state import StateManager state_manager = StateManager(Path(".pyserve"), ctx.project) if services: log_files = [(name, state_manager.get_service_log_file(name)) for name in services] else: all_services = state_manager.get_all_services() if not all_services: main_log = Path("logs/pyserve.log") if main_log.exists(): log_files = [("pyserve", main_log)] else: print_info("No logs available. Start services with 'pyserve up'") return else: log_files = [(name, state_manager.get_service_log_file(name)) for name in all_services] existing_logs = [(name, path) for name, path in log_files if path.exists()] if not existing_logs: print_info("No log files found") return since_time = _parse_time(since) if since else None until_timestamp = _parse_time(until_time) if until_time else None colors = ["cyan", "green", "yellow", "blue", "magenta"] service_colors = {name: colors[i % len(colors)] for i, (name, _) in enumerate(existing_logs)} if follow: asyncio.run( _follow_logs( existing_logs, service_colors, timestamps, no_color, filter_pattern, ) ) else: _read_logs( existing_logs, service_colors, tail, since_time, until_timestamp, timestamps, no_color, filter_pattern, ) def _parse_time(time_str: str) -> Optional[float]: import re from datetime import datetime # Relative time (e.g., "10m", "1h", "2d") match = re.match(r"^(\d+)([smhd])$", time_str) if match: value = int(match.group(1)) unit = match.group(2) units = {"s": 1, "m": 60, "h": 3600, "d": 86400} return time.time() - (value * units[unit]) # Relative phrase (e.g., "10m ago") match = re.match(r"^(\d+)([smhd])\s+ago$", time_str) if match: value = int(match.group(1)) unit = match.group(2) units = {"s": 1, "m": 60, "h": 3600, "d": 86400} return time.time() - (value * units[unit]) # ISO format try: dt = datetime.fromisoformat(time_str) return dt.timestamp() except ValueError: pass return None def _read_logs( log_files: list[tuple[str, Path]], service_colors: dict[str, str], tail: int, since_time: Optional[float], until_time: Optional[float], timestamps: bool, no_color: bool, filter_pattern: Optional[str], ) -> None: import re from ..output import console all_lines = [] for service_name, log_path in log_files: try: with open(log_path) as f: lines = f.readlines() # Take last N lines lines = lines[-tail:] if tail else lines for line in lines: line = line.rstrip() if not line: continue if filter_pattern and filter_pattern not in line: continue line_time = None timestamp_match = re.match(r"^(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2})", line) if timestamp_match: try: from datetime import datetime line_time = datetime.fromisoformat(timestamp_match.group(1).replace(" ", "T")).timestamp() except ValueError: pass if since_time and line_time and line_time < since_time: continue if until_time and line_time and line_time > until_time: continue all_lines.append((line_time or 0, service_name, line)) except Exception as e: console.print(f"[red]Error reading {log_path}: {e}[/red]") all_lines.sort(key=lambda x: x[0]) for _, service_name, line in all_lines: if len(log_files) > 1: # Multiple services - prefix with service name if no_color: console.print(f"{service_name} | {line}") else: color = service_colors.get(service_name, "white") console.print(f"[{color}]{service_name}[/{color}] | {line}") else: console.print(line) async def _follow_logs( log_files: list[tuple[str, Path]], service_colors: dict[str, str], timestamps: bool, no_color: bool, filter_pattern: Optional[str], ) -> None: from ..output import console positions = {} for service_name, log_path in log_files: if log_path.exists(): positions[service_name] = log_path.stat().st_size else: positions[service_name] = 0 console.print("[dim]Following logs... Press Ctrl+C to stop[/dim]\n") try: while True: for service_name, log_path in log_files: if not log_path.exists(): continue current_size = log_path.stat().st_size if current_size > positions[service_name]: with open(log_path) as f: f.seek(positions[service_name]) new_content = f.read() positions[service_name] = f.tell() for line in new_content.splitlines(): if filter_pattern and filter_pattern not in line: continue if len(log_files) > 1: if no_color: console.print(f"{service_name} | {line}") else: color = service_colors.get(service_name, "white") console.print(f"[{color}]{service_name}[/{color}] | {line}") else: console.print(line) await asyncio.sleep(0.5) except KeyboardInterrupt: console.print("\n[dim]Stopped following logs[/dim]")