agents/common.py

from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, get_args, get_origin

from pydantic import BaseModel


APP_VERSION = "1.0.0"
SEED = 42
ROOT_DIR = Path(__file__).resolve().parents[1]
DATA_DIR = ROOT_DIR / "data"
ARTIFACT_DIR = ROOT_DIR / "model_artifacts"
METADATA_PATH = ARTIFACT_DIR / "metadata.json"


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def ensure_runtime_dirs() -> None:
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)


def load_metadata() -> dict[str, Any]:
    if not METADATA_PATH.exists():
        return {}
    return json.loads(METADATA_PATH.read_text(encoding="utf-8"))


def save_metadata(payload: dict[str, Any]) -> None:
    ensure_runtime_dirs()
    METADATA_PATH.write_text(json.dumps(payload, indent=2), encoding="utf-8")


def _annotation_to_name(annotation: Any) -> str:
    origin = get_origin(annotation)
    if origin is None:
        if annotation in {str, int, float, bool}:
            return annotation.__name__
        return str(annotation).replace("typing.", "")

    if origin is list:
        args = get_args(annotation)
        inner = _annotation_to_name(args[0]) if args else "object"
        return f"list[{inner}]"

    if origin is dict:
        args = get_args(annotation)
        key_name = _annotation_to_name(args[0]) if args else "object"
        value_name = _annotation_to_name(args[1]) if len(args) > 1 else "object"
        return f"dict[{key_name}, {value_name}]"

    if origin is tuple:
        return "tuple"

    if str(origin).endswith("Literal"):
        values = ", ".join(repr(value) for value in get_args(annotation))
        return f"Literal[{values}]"

    if str(origin).endswith("Union"):
        args = ", ".join(_annotation_to_name(arg) for arg in get_args(annotation))
        return f"Union[{args}]"

    return str(annotation).replace("typing.", "")


def schema_fields(model_cls: type[BaseModel]) -> list[dict[str, Any]]:
    fields = []
    for name, field in model_cls.model_fields.items():
        fields.append(
            {
                "name": name,
                "type": _annotation_to_name(field.annotation),
                "required": field.is_required(),
            }
        )
    return fields