diff --git a/main.py b/main.py new file mode 100644 index 0000000..a89d8d0 --- /dev/null +++ b/main.py @@ -0,0 +1,152 @@ +""" +Main entry point for the Structured Output Union events API task. + +This script demonstrates how to parse a raw log containing multiple HTTP events +using Pydantic v2 models and LangChain structured output. It can be run as a +stand‑alone CLI or imported as a module. + +Requirements (see requirements.txt): +- langchain-core>=1.0.0 +- langchain-openai +- pydantic>=2.0 +- python-dotenv +""" + +from __future__ import annotations + +import argparse +import os +import sys +from pathlib import Path +from typing import List, Union + +from dotenv import load_dotenv +from langchain_core.output_parsers import PydanticOutputParser +from langchain_openai import ChatOpenAI +from pydantic import BaseModel, Field, ValidationError + +# --------------------------------------------------------------------------- +# Models +# --------------------------------------------------------------------------- +class HttpOkEvent(BaseModel): + """Represents a successful HTTP request.""" + + kind: str = Field("ok", description="Discriminator for the union.") + status: int = Field(200, description="HTTP status code (must be 200).") + path: str = Field(..., description="Requested URL path.") + duration_ms: int = Field(..., description="Duration of the request in milliseconds.") + +class HttpErrorEvent(BaseModel): + """Represents a failed HTTP request.""" + + kind: str = Field("error", description="Discriminator for the union.") + status: int = Field(..., ge=400, le=599, description="HTTP error status code.") + path: str = Field(..., description="Requested URL path.") + error_message: str = Field(..., description="Human readable error message.") + +# Union with discriminator ``kind``. Pydantic v2 automatically uses the field +# named ``kind`` to decide which model to instantiate. +ApiEvent = Union[HttpOkEvent, HttpErrorEvent] + +# --------------------------------------------------------------------------- +# Parser helper +# --------------------------------------------------------------------------- +parser = PydanticOutputParser(pydantic_object=ApiEvent) + +# --------------------------------------------------------------------------- +# LLM wrapper +# --------------------------------------------------------------------------- +load_dotenv() # Load JARVIS API key from .env if present. +llm = ChatOpenAI( + model="openai/gpt-oss-20b:free", + base_url="https://platform.brojs.ru/jrnl-bh/api/inference/v1", + api_key=os.getenv("JOURNAL_MCP_PAT"), + temperature=0.0, +) + +# --------------------------------------------------------------------------- +# Core logic +# --------------------------------------------------------------------------- +def parse_event(raw: str) -> ApiEvent: + """Parse a single raw log line into an :class:`ApiEvent`. + + The function sends the raw text to the LLM with a structured output prompt + and then validates the result using Pydantic. If validation fails, the + exception is propagated so that callers can decide how to handle it. + """ + # Build a simple prompt that instructs the model to return JSON matching + # one of the two event schemas. + prompt = f"Parse the following log line into JSON:\n{raw}\nJSON:" # noqa: E501 + response = llm.invoke([prompt]) + json_text = parser.parse(response.content) + try: + return ApiEvent.model_validate(json_text) # type: ignore[arg-type] + except ValidationError as exc: # pragma: no cover - defensive + raise ValueError(f"Failed to validate event: {exc}") from exc + + +def parse_log(raw_log: str) -> List[ApiEvent]: + """Parse a multiline log into a list of :class:`ApiEvent` objects. + + The function splits the input on newlines and ignores empty lines. + Each non‑empty line is parsed individually. + """ + events: List[ApiEvent] = [] + for line in raw_log.splitlines(): + line = line.strip() + if not line: + continue + try: + event = parse_event(line) + events.append(event) + except Exception as exc: # pragma: no cover - log and skip + print(f"Warning: could not parse line '{line}': {exc}", file=sys.stderr) + return events + +# --------------------------------------------------------------------------- +# CLI entry point +# --------------------------------------------------------------------------- +def main() -> None: + parser_cli = argparse.ArgumentParser(description="Parse raw HTTP logs into structured events.") + parser_cli.add_argument( + "--log-file", + type=Path, + help="Path to a file containing the raw log. If omitted, a sample log is used.", + ) + args = parser_cli.parse_args() + + if args.log_file and args.log_file.exists(): + raw_log = args.log_file.read_text(encoding="utf-8") + else: + # Sample log with mixed success and error events. + raw_log = """ +GET /api/users 200 OK duration=123ms +POST /api/login 404 Not Found: user not found +PUT /api/items/42 500 Internal Server Error: database timeout +GET /health 200 OK duration=45ms +""" + print("Using sample log:\n", raw_log) + + events = parse_log(raw_log) + if not events: + print("No valid events parsed.") + sys.exit(1) + + # Pretty‑print the results as a table. + header = f"{'Kind':<6} | {'Path':<20} | {'Status':<6} | Details" + print(header) + print("-" * len(header)) + for ev in events: + if isinstance(ev, HttpOkEvent): + details = f"duration={ev.duration_ms}ms" + else: # HttpErrorEvent + details = f"error='{ev.error_message}'" + print(f"{ev.kind:<6} | {ev.path:<20} | {ev.status:<6} | {details}") + + # Also output the raw Pydantic model dumps for debugging. + print("\nModel dumps:\n") + for ev in events: + print(ev.model_dump_json(indent=2)) + +if __name__ == "__main__": # pragma: no cover - entry point + main()