Add src/main.py

src/main.py

@@ -0,0 +1,106 @@
+"""CLI tool to parse raw API logs into structured events using LangChain and Pydantic.
+
+Usage:
+    python -m src.main [--log LOG_TEXT]
+
+If LOG_TEXT is omitted, a default example log is used.
+"""
+import argparse
+import os
+from typing import List
+
+from langchain_openai import ChatOpenAI
+from pydantic import ValidationError
+
+from .models import ApiEvent
+
+# Default example log: mix of 200 and error lines
+DEFAULT_LOG = (
+    "GET /api/users 200 120ms\n"
+    "POST /api/login 404 Not Found\n"
+    "GET /api/orders 500 Internal Server Error\n"
+    "PUT /api/users/42 200 45ms"
+)
+
+def parse_event(line: str, llm) -> ApiEvent:
+    """Parse a single log line into an ApiEvent using the LLM's structured output.
+
+    Parameters
+    ----------
+    line: str
+        Raw log line.
+    llm: ChatOpenAI
+        LLM instance configured with structured output.
+    """
+    # Prepare a prompt that instructs the LLM to output a JSON matching the ApiEvent schema.
+    prompt = (
+        "Parse the following raw log line into a JSON object that matches one of the following schemas:\n"
+        "1. Ok event: {\n  \"kind\": \"ok\", \"status\": 200, \"path\": string, \"duration_ms\": int\n}\n"
+        "2. Error event: {\n  \"kind\": \"error\", \"status\": int, \"path\": string, \"error_message\": string\n}\n"
+        "Return only the JSON object, nothing else.\n"
+        f"Log line: {line.strip()}"
+    )
+    response = llm.invoke(prompt)
+    # The LLM is configured with structured output, so response is a dict
+    try:
+        event = ApiEvent(**response)
+    except ValidationError as e:
+        raise ValueError(f"LLM output could not be parsed into ApiEvent: {e}")
+    return event
+
+def parse_log(log_text: str, llm) -> List[ApiEvent]:
+    """Parse a multiline log into a list of ApiEvent objects."""
+    events: List[ApiEvent] = []
+    for line in log_text.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            event = parse_event(line, llm)
+            events.append(event)
+        except Exception as exc:
+            print(f"Failed to parse line: {line}\nError: {exc}")
+    return events
+
+def print_table(events: List[ApiEvent]):
+    """Print a simple table of the parsed events."""
+    header = f"{'Kind':<6} | {'Path':<20} | {'Status':<6} | {'Details'}"
+    print(header)
+    print('-' * len(header))
+    for ev in events:
+        if ev.kind == "ok":
+            details = f"duration_ms={ev.duration_ms}"
+        else:
+            details = f"error_message={ev.error_message}"
+        print(f"{ev.kind:<6} | {ev.path:<20} | {ev.status:<6} | {details}")
+
+def main():
+    parser = argparse.ArgumentParser(description="Parse raw API logs into structured events.")
+    parser.add_argument(
+        "--log",
+        type=str,
+        help="Raw log text. If omitted, a default example is used.",
+    )
+    args = parser.parse_args()
+
+    log_text = args.log or DEFAULT_LOG
+
+    # Load OpenAI API key from environment
+    openai_api_key = os.getenv("OPENAI_API_KEY")
+    if not openai_api_key:
+        raise RuntimeError("OPENAI_API_KEY environment variable not set.")
+
+    # Configure LLM with structured output
+    llm = ChatOpenAI(api_key=openai_api_key, temperature=0.0).with_structured_output(ApiEvent)
+
+    events = parse_log(log_text, llm)
+
+    print("\nParsed events:\n")
+    for ev in events:
+        print(ev.model_dump())
+
+    print("\nTable:\n")
+    print_table(events)
+
+if __name__ == "__main__":
+    main()