154 lines
5.3 KiB
Python
154 lines
5.3 KiB
Python
#!/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() |