python/mcp/mappers.py - master - ksilisk/spbtechrun_hack

python/mcp/mappers.py
182 строки5 KB
Shaliko Salimov
documents for RAG and supported polyclinics + dou
10 дек 2025, 20:51
10 дек 2025, 20:517cd073f
from __future__ import annotations

from datetime import datetime
from typing import Any, Iterable

from yazzh import (
    AfishaEvent,
    BeautifulPlace,
    MfcOffice,
    PetShelter,
    PetVetClinic,
    PetWalkingArea,
    SportEvent,
    Sportground, Building, Dou, Polyclinic,
)


def _extract_items(payload: Any, keys: Iterable[str] = ("data", "items", "mfc_list")) -> list[Any]:
    if payload is None:
        return []
    if isinstance(payload, list):
        return payload
    if isinstance(payload, dict):
        for key in keys:
            candidate = payload.get(key)
            if isinstance(candidate, list):
                return candidate
    return []


def _split_dt(value: str | None) -> tuple[str | None, str | None]:
    if not value:
        return None, None
    try:
        dt = datetime.fromisoformat(value)
        return dt.date().isoformat(), dt.time().isoformat()
    except ValueError:
        return value, None


def _extract_address(item: dict[str, Any]) -> str | None:
    return (
            item.get("address")
            or (item.get("location") or {}).get("address")
            or item.get("location_place")
    )


def map_afisha_events(payload: Any) -> list[AfishaEvent]:
    events: list[AfishaEvent] = []
    for item in _extract_items(payload):
        place = item.get("place")
        start_date = place["start_date"]
        end_date = place["end_date"]
        events.append(
            AfishaEvent(
                title=place.get("title") or place.get("name", ""),
                start_time=start_date or place.get("start_time"),
                end_time=end_date or place.get("end_time"),
                address=place.get("address"),
                location_title=place.get("location_title"),
            )
        )
    return events


def map_beautiful_places(payload: Any) -> list[BeautifulPlace]:
    return [
        BeautifulPlace(
            title=item["place"].get("title") or item["place"].get("name", ""),
            description=item["place"].get("description") or item["place"].get("note"),
            address=item["place"].get("address"),
        )
        for item in _extract_items(payload)
    ]


def map_sport_events(payload: Any) -> list[SportEvent]:
    events: list[SportEvent] = []
    for item in _extract_items(payload["data"]):
        date_str, time_str = item.get("start_date"), item.get("start_time")
        events.append(
            SportEvent(
                title=item.get("name") or item.get("title", ""),
                description=item.get("description"),
                start_date=date_str,
                start_time=time_str,
                address=item.get("address"),
            )
        )
    return events


def map_sportgrounds(payload: Any) -> list[Sportground]:
    return [
        Sportground(
            name=item["place"].get("title") or item["place"].get("name", ""),
            address=item["place"]["address"],
        )
        for item in _extract_items(payload)
    ]


def map_vet_clinics(payload: Any) -> list[PetVetClinic]:
    return [
        PetVetClinic(
            title=item["place"].get("title") or item.get("name", ""),
            address=item["place"]["address"],
        )
        for item in _extract_items(payload)[0:4]
    ]


def map_pet_shelters(payload: Any) -> list[PetShelter]:
    return [
        PetShelter(
            title=item["place"].get("title") or item.get("name", ""),
            address=item["place"]["address"],
        )
        for item in _extract_items(payload)[0:4]
    ]


def map_pet_walking_areas(payload: Any) -> list[PetWalkingArea]:
    return [
        PetWalkingArea(
            title=item["place"].get("title") or item.get("name", ""),
            address=item["place"]["address"],
        )
        for item in _extract_items(payload)[0:4]
    ]


def map_mfc(payload: Any) -> list[MfcOffice]:
    return [
        MfcOffice(
            name=item.get("name") or item.get("title", ""),
            nearest_metro=item.get("nearest_metro") or item.get("metro"),
            address=item["address"],
        )
        for item in _extract_items(payload, keys=("mfc_list", "data", "items"))[0:4]
    ]


def map_single_building(payload: Any) -> Building:
    return Building(
        address=payload["data"].get("short_address"),
        district=payload["data"].get("district"),
        id=payload["data"].get("id"),
    )


def map_buildings(payload: Any) -> list[Building]:
    return [
        Building(
            address=item.get("short_address"),
            district=item.get("district"),
            id=item.get("id"),
        )
        for item in _extract_items(payload, keys=("mfc_list", "data", "items"))
    ]


def map_dou(payload: Any) -> [Dou]:
    return [
        Dou(
            name=item.get("doo_short"),
            building_id=item.get("building_id"),
        )
        for item in _extract_items(payload, keys=("mfc_list", "data", "items"))[0:5]
    ]


def map_polyclinic(payload: Any) -> [Polyclinic]:
    return [
        Polyclinic(
            name=item["clinic_name"],
            address=item["clinic_address"],
            url=item["url"] or "",
        )
        for item in _extract_items(payload, keys=("mfc_list", "data", "items"))[0:5]
    ]