feat: solution for 'Повторный экзамен: Structured output — Union событий API'

.gitignore

@@ -0,0 +1,5 @@
+node_modules/
+.env
+dist/
+build/
+*.log

README.md

@@ -0,0 +1,95 @@
+# Structured Log Parser
+
+This project demonstrates how to parse raw log lines into typed events using **Pydantic v2** and **LangChain**'s structured output.  
+It supports two event types:
+
+| Event | Fields |
+|-------|--------|
+| `HttpOkEvent` | `kind="ok"`, `status=200`, `path`, `duration_ms` |
+| `HttpErrorEvent` | `kind="error"`, `status` (4xx/5xx), `path`, `error_message` |
+
+The parser uses a LangChain prompt to convert each log line into a JSON object that matches one of the schemas. The output is then validated with Pydantic.
+
+## Features
+
+- **Discriminated Union**: `ApiEvent` is a union of `HttpOkEvent` and `HttpErrorEvent` with a `kind` discriminator.
+- **Structured Output**: Uses LangChain's `PydanticOutputParser` to enforce schema.
+- **Batch Parsing**: Handles multiple log lines in a single run.
+- **CLI**: Accepts example logs, a file, or custom text and prints a table of parsed events.
+
+## Installation
+
+```bash
+# Clone the repo
+git clone https://github.com/yourusername/structured-log-parser.git
+cd structured-log-parser
+
+# Create a virtual environment (optional but recommended)
+python -m venv .venv
+source .venv/bin/activate  # On Windows: .venv\Scripts\activate
+
+# Install dependencies
+pip install -r requirements.txt
+```
+
+## Usage
+
+### 1. Using the built‑in example
+
+```bash
+python -m src.main --example
+```
+
+### 2. Parsing a log file
+
+```bash
+python -m src.main --file path/to/log.txt
+```
+
+### 3. Parsing custom text
+
+```bash
+python -m src.main --text "2026-08-31 12:00:01 INFO /api/users 200 123ms"
+```
+
+The output will be a Markdown‑style table:
+
+```
+| kind | path          | status | duration_ms / error_message |
+|------|---------------|--------|-----------------------------|
+| ok   | /api/users    | 200    | 123 ms                      |
+| error| /api/orders   | 404    | Not Found                   |
+| error| /api/payments | 500    | Internal Server Error       |
+| ok   | /api/products | 200    | 45 ms                       |
+```
+
+## Environment Variables
+
+The project uses OpenAI's API. Set the following variable before running:
+
+```bash
+export OPENAI_API_KEY="sk-..."
+```
+
+Alternatively, create a `.env` file in the project root:
+
+```
+OPENAI_API_KEY=sk-...
+```
+
+The `python-dotenv` package will load it automatically.
+
+## Project Structure
+
+```
+src/
+├── __init__.py
+├── main.py          # CLI entry point
+├── cli.py           # Argument parsing and table rendering
+├── parser.py        # Log parsing logic
+└── models.py        # Pydantic event models
+```
+
+## License
+
+MIT License

requirements.txt

@@ -0,0 +1,5 @@
+langchain-core>=0.2.0
+langchain-openai>=0.2.0
+pydantic>=2.0
+tabulate>=0.9.0
+python-dotenv>=1.0.0

src/cli.py

@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import argparse
+import textwrap
+from pathlib import Path
+from typing import List
+
+from tabulate import tabulate
+
+from .parser import parse_log
+from .models import ApiEvent
+
+EXAMPLE_LOG = textwrap.dedent(
+    """\
+    2026-08-31 12:00:01 INFO /api/users 200 123ms
+    2026-08-31 12:00:02 ERROR /api/orders 404 Not Found
+    2026-08-31 12:00:03 WARN /api/payments 500 Internal Server Error
+    2026-08-31 12:00:04 INFO /api/products 200 45ms
+    """
+)
+
+def build_table(events: List[ApiEvent]) -> str:
+    headers = ["kind", "path", "status", "duration_ms / error_message"]
+    rows = []
+    for ev in events:
+        if ev.kind == "ok":
+            rows.append([ev.kind, ev.path, ev.status, f"{ev.duration_ms} ms"])
+        else:
+            rows.append([ev.kind, ev.path, ev.status, ev.error_message])
+    return tabulate(rows, headers=headers, tablefmt="github")
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Parse raw log lines into structured API events using LangChain."
+    )
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument(
+        "--example",
+        action="store_true",
+        help="Use built-in example log",
+    )
+    group.add_argument(
+        "--file",
+        type=Path,
+        help="Path to a file containing raw log lines",
+    )
+    group.add_argument(
+        "--text",
+        type=str,
+        help="Custom log text (enclosed in quotes)",
+    )
+    args = parser.parse_args()
+
+    if args.example:
+        log_text = EXAMPLE_LOG
+    elif args.file:
+        log_text = args.file.read_text(encoding="utf-8")
+    elif args.text:
+        log_text = args.text
+    else:
+        log_text = EXAMPLE_LOG
+
+    events = parse_log(log_text)
+    print(build_table(events))
+
+if __name__ == "__main__":
+    main()

src/main.py

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+"""Entry point for the log parser CLI."""
+from .cli import main
+
+if __name__ == "__main__":
+    main()

src/models.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import Literal, Annotated, Union
+
+from pydantic import BaseModel, Field, FieldValidationError, ValidationError, field_validator
+
+# Base event model with discriminator field
+class BaseEvent(BaseModel):
+    kind: Literal["ok", "error"] = Field(..., description="Event kind discriminator")
+
+# OK event
+class HttpOkEvent(BaseEvent):
+    kind: Literal["ok"] = Field("ok", description="OK event kind")
+    status: Literal[200] = Field(..., description="HTTP status code")
+    path: str = Field(..., description="Request path")
+    duration_ms: int = Field(..., description="Duration in milliseconds")
+
+# Error event
+class HttpErrorEvent(BaseEvent):
+    kind: Literal["error"] = Field("error", description="Error event kind")
+    status: int = Field(..., description="HTTP status code (4xx or 5xx)")
+    path: str = Field(..., description="Request path")
+    error_message: str = Field(..., description="Error message")
+
+# Union with discriminator
+ApiEvent = Annotated[
+    Union[HttpOkEvent, HttpErrorEvent],
+    Field(discriminator="kind")
+]

src/parser.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import os
+from typing import Iterable, List
+
+from langchain_core.output_parsers import PydanticOutputParser
+from langchain_openai import ChatOpenAI
+from langchain_core.prompts import PromptTemplate
+from langchain_core.messages import HumanMessage
+
+from .models import ApiEvent
+
+# Load OpenAI 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")
+
+# LLM
+llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, api_key=OPENAI_API_KEY)
+
+# Structured output parser
+parser = PydanticOutputParser(pydantic_object=ApiEvent)
+
+# Prompt template for parsing a single log line
+prompt_template = PromptTemplate(
+    input_variables=["log_line"],
+    template=(
+        "You are a log parser. Convert the following raw log line into a structured JSON object "
+        "matching one of the following schemas:\n\n"
+        "1. OK event: {{ ok_schema }}\n"
+        "2. Error event: {{ error_schema }}\n\n"
+        "The output must be valid JSON and must include a field `kind` with value `ok` or `error`.\n\n"
+        "Raw log line:\n"
+        "{{ log_line }}\n\n"
+        "Output JSON:"
+    ),
+)
+
+# Fill in schema descriptions
+prompt_template = prompt_template.partial(
+    ok_schema=parser.get_format_instructions(),
+    error_schema=parser.get_format_instructions(),
+)
+
+def parse_line(line: str) -> ApiEvent:
+    """
+    Parse a single log line into an ApiEvent using LangChain structured output.
+    """
+    # Build prompt
+    prompt = prompt_template.format(log_line=line.strip())
+    # Send to LLM
+    response = llm.invoke([HumanMessage(content=prompt)])
+    # Parse JSON
+    try:
+        event = parser.parse(response.content)
+    except Exception as exc:
+        raise ValueError(f"Failed to parse line: {line!r}\nLLM response: {response.content}\nError: {exc}") from exc
+    return event
+
+def parse_log(text: str) -> List[ApiEvent]:
+    """
+    Parse a multiline log string into a list of ApiEvent objects.
+    """
+    events: List[ApiEvent] = []
+    for line in text.strip().splitlines():
+        if not line.strip():
+            continue
+        events.append(parse_line(line))
+    return events