#!/usr/bin/env python3 """ CLI для парсинга логов HTTP API с использованием discriminated union в LangChain """ import sys import argparse from typing import List from tabulate import tabulate from models import ApiEvent, HttpOkEvent, HttpErrorEvent from parser import LogEventParser # Пример лога (разные форматы для демонстрации гибкости) EXAMPLE_LOG = """GET /api/users 200 duration=45ms POST /api/products/123 404 error="Product not found" time=12ms DELETE /api/sessions 500 Internal Server Error (duration=230ms) GET /health 200 2ms PUT /api/config 400 Invalid request body, duration=67""" class LogCLI: def __init__(self): self.parser = LogEventParser() def display_results(self, events: List[ApiEvent]) -> None: """Выводит результаты в читаемом виде""" if not events: print("❌ Нет успешно обработанных событий") return print("\n" + "="*80) print(f"📊 Обработано событий: {len(events)}") print("="*80) # Таблица для компактного отображения table_data = [] for event in events: if isinstance(event, HttpOkEvent): table_data.append([ event.kind.upper(), event.path, event.status, f"{event.duration_ms}ms", "-" ]) elif isinstance(event, HttpErrorEvent): table_data.append([ event.kind.upper(), event.path, event.status, "-", event.error_message[:50] + ("..." if len(event.error_message) > 50 else "") ]) headers = ["Kind", "Path", "Status", "Duration", "Error Message"] print(tabulate(table_data, headers=headers, tablefmt="grid")) # Подробный вывод print("\n📋 Детализация:") for i, event in enumerate(events, 1): print(f"{i}. {event}") def run_interactive(self): """Интерактивный режим""" print("🔄 HTTP Log Parser (Structured Output with Discriminated Union)") print("Введите лог (пустая строка для завершения):") lines = [] while True: line = input("> ") if not line: break lines.append(line) if lines: log_text = "\n".join(lines) events = self.parser.parse_multiple_events(log_text) self.display_results(events) def run_example(self): """Запуск на примере""" print("🚀 Запуск на примере лога:") print("─" * 80) print(EXAMPLE_LOG) print("─" * 80) events = self.parser.parse_multiple_events(EXAMPLE_LOG) self.display_results(events) def run_from_file(self, filename: str): """Парсинг лога из файла""" try: with open(filename, 'r', encoding='utf-8') as f: log_text = f.read() events = self.parser.parse_multiple_events(log_text) self.display_results(events) except FileNotFoundError: print(f"❌ Файл не найден: {filename}") sys.exit(1) except Exception as e: print(f"❌ Ошибка чтения файла: {e}") sys.exit(1) def main(): parser = argparse.ArgumentParser( description="Парсинг логов HTTP API с discriminated union в LangChain", formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument( "-f", "--file", help="Файл с логами (каждое событие на новой строке)" ) parser.add_argument( "-i", "--interactive", action="store_true", help="Интерактивный режим ввода" ) parser.add_argument( "-e", "--example", action="store_true", help="Запустить на встроенном примере" ) parser.add_argument( "--model", default="gpt-3.5-turbo", help="Модель OpenAI (по умолчанию: gpt-3.5-turbo)" ) args = parser.parse_args() # Если нет аргументов, показываем help if len(sys.argv) == 1: parser.print_help() print("\n" + "="*80) print("Пример использования:") print(" python cli.py --example # Запуск на примере") print(" python cli.py -i # Интерактивный режим") print(" python cli.py -f logs.txt # Парсинг файла") return cli = LogCLI() if args.example: cli.run_example() elif args.interactive: cli.run_interactive() elif args.file: cli.run_from_file(args.file) else: parser.print_help() if __name__ == "__main__": main()