gogo-telefon/gogo/dashboard/parse.py

"""'Zalijepi cjenovnik' paste-to-parse (§10.2): LLM turns a free-text price list
(from a website, Word, Facebook…) into structured services the owner reviews.
Same approach for working hours."""

from __future__ import annotations

import json
import logging

from gogo.agent.llm import LLMProvider, get_llm

log = logging.getLogger("gogo.parse")

_SERVICES_TOOL = {
    "name": "save_services",
    "description": "Save the parsed price list as structured services.",
    "input_schema": {
        "type": "object",
        "properties": {
            "services": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "duration_min": {
                            "type": "integer",
                            "description": "estimate a realistic duration if not stated",
                        },
                        "price_min": {"type": "number"},
                        "price_max": {
                            "type": "number",
                            "description": "same as price_min when a single price",
                        },
                        "home_visit": {"type": "boolean", "default": False},
                        "note": {"type": "string"},
                    },
                    "required": ["name", "duration_min"],
                },
            }
        },
        "required": ["services"],
    },
}

_HOURS_TOOL = {
    "name": "save_hours",
    "description": "Save the parsed working hours.",
    "input_schema": {
        "type": "object",
        "properties": {
            day: {
                "type": "array",
                "description": f"open intervals for {day} as [\"HH:MM\",\"HH:MM\"] pairs; empty if closed",
                "items": {
                    "type": "array",
                    "items": {"type": "string"},
                    "minItems": 2,
                    "maxItems": 2,
                },
            }
            for day in ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]
        },
        "required": ["mon", "tue", "wed", "thu", "fri", "sat", "sun"],
    },
}


async def parse_price_list(text: str, llm: LLMProvider | None = None) -> list[dict]:
    """Returns a list of service dicts for the owner to review (never saved directly)."""
    llm = llm or get_llm()
    response = await llm.complete(
        system=(
            "You parse beauty-salon price lists written in Bosnian/Serbian/Croatian "
            "into structured data. Prices are in KM (BAM). A range like '60-90 KM' "
            "means price_min=60, price_max=90. Estimate realistic durations when "
            "missing (šišanje ~45, farbanje ~90, manikir ~45, pedikir ~60, "
            "depilacija ~30 minutes). Set home_visit only when the text explicitly "
            "mentions coming to the client's home. Keep names as written (fix only "
            "obvious typos). Call save_services exactly once."
        ),
        messages=[{"role": "user", "content": text[:8000]}],
        tools=[_SERVICES_TOOL],
        max_tokens=4096,
    )
    for tu in response.tool_uses:
        if tu.name == "save_services":
            return list(tu.input.get("services", []))
    log.warning("price-list parse produced no tool call")
    return []


async def parse_working_hours(text: str, llm: LLMProvider | None = None) -> dict | None:
    llm = llm or get_llm()
    response = await llm.complete(
        system=(
            "You parse salon working hours written in Bosnian/Serbian/Croatian into "
            "structured JSON. 'pon-pet 9-18' means mon..fri [[\"09:00\",\"18:00\"]]. "
            "'pauza 13-14' splits the day into two intervals. Closed days get []. "
            "Call save_hours exactly once."
        ),
        messages=[{"role": "user", "content": text[:2000]}],
        tools=[_HOURS_TOOL],
        max_tokens=2048,
    )
    for tu in response.tool_uses:
        if tu.name == "save_hours":
            hours = {k: tu.input.get(k, []) for k in
                     ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]}
            return _validate_hours(hours)
    return None


def _validate_hours(hours: dict) -> dict | None:
    try:
        for _day, intervals in hours.items():
            for iv in intervals:
                a, b = iv
                for t in (a, b):
                    h, m = t.split(":")
                    assert 0 <= int(h) <= 23 and 0 <= int(m) <= 59
        return hours
    except (ValueError, AssertionError, TypeError):
        log.warning("parsed hours failed validation: %s", json.dumps(hours)[:200])
        return None