main.py

"""
Клиент API прямого геокодинга Atlorium — координаты по адресу.

Запуск (работает сразу, без регистрации — на демо-ключе):
    pip install -r requirements.txt
    python main.py "Казань, улица Баумана, 13"

Боевой ключ: получить на https://atlorium.com и положить в переменную окружения
ATLORIUM_API_KEY. Код при этом не меняется.
"""

import os
import re
import sys
from dataclasses import dataclass, field

import requests

# Публичный демо-ключ. С ним API отвечает МОКАМИ (не данными реестра) — чтобы можно
# было встроить и протестировать интеграцию до оплаты. Ответы детерминированы:
# один и тот же запрос всегда даёт один и тот же результат, поэтому на них можно
# писать стабильные тесты.
SANDBOX_KEY = "ak_sandbox_demo_mockdata_v1"

API_KEY = os.environ.get("ATLORIUM_API_KEY", SANDBOX_KEY)
BASE_URL = os.environ.get("ATLORIUM_BASE_URL", "https://atlorium.com")

TIMEOUT = 30


class AtloriumError(RuntimeError):
    """Ошибка API. Код HTTP разложен в человекочитаемую причину."""

    REASONS = {
        400: "Некорректный запрос: адресная строка пуста, слишком коротка или слишком длинна",
        401: "API-ключ отсутствует, просрочен или недействителен",
        402: "Недостаточно кредитов на балансе — пополните на https://atlorium.com",
        404: "Адрес не найден либо найденное грубее уровня улицы (такой ответ не тарифицируется)",
        429: "Превышен лимит запросов — повторите позже",
        503: "Геокодер временно недоступен (за сбой на своей стороне мы не списываем деньги)",
    }

    def __init__(self, status: int, body: str):
        reason = self.REASONS.get(status, "Неизвестная ошибка")
        super().__init__(f"HTTP {status}: {reason}. Ответ сервера: {body[:200]}")
        self.status = status


def _request(method: str, path: str, params: dict | None = None, json_body: dict | None = None) -> dict:
    response = requests.request(
        method,
        f"{BASE_URL}{path}",
        params=params,
        json=json_body,
        headers={
            "Authorization": f"Bearer {API_KEY}",
            "Accept": "application/json",
        },
        timeout=TIMEOUT,
    )
    if not response.ok:
        raise AtloriumError(response.status_code, response.text)
    return response.json()


# ── Вызовы API ────────────────────────────────────────────────────────────────


def geocode(text: str, size: int = 5, exact_only: bool = False) -> dict:
    """Координаты по адресу, записанному одной строкой в свободной форме."""
    return _request(
        "GET",
        "/api/geocodeforward",
        {"text": text, "size": size, "exactOnly": str(exact_only).lower()},
    )


def geocode_structured(
    address: str | None = None,
    locality: str | None = None,
    region: str | None = None,
    postal_code: str | None = None,
    size: int = 5,
) -> dict:
    """Координаты по адресу, разложенному на поля.

    Устойчивее свободной строки, когда адрес уже хранится по полям: снимается
    неопределённость, где город, а где улица. Тарифицируется по отдельной ставке.
    """
    params: dict = {"size": size}
    if address:
        params["address"] = address
    if locality:
        params["locality"] = locality
    if region:
        params["region"] = region
    if postal_code:
        params["postalCode"] = postal_code
    return _request("GET", "/api/geocodeforward/structured", params)


def geocode_batch(queries: list[str], size_per_query: int = 1) -> dict:
    """Пакетная обработка: до 1000 адресов за один вызов.

    Порядок ответов совпадает с порядком запроса, ненайденная строка получает
    пустой список вариантов и не прерывает пакет.
    """
    return _request(
        "POST",
        "/api/geocodeforward/batch",
        json_body={"queries": queries, "sizePerQuery": size_per_query},
    )


def get_place(place_id: str, include_boundary: bool = False) -> dict:
    """Карточка объекта по идентификатору из предыдущего ответа геокодинга.

    Идентификатор присвоен государственным реестром и не меняется, поэтому
    повторное обращение по нему дешевле и надёжнее нового поиска.
    """
    return _request(
        "GET",
        "/api/geocodeforward/place",
        {"placeId": place_id, "includeBoundary": str(include_boundary).lower()},
    )


def get_postcode(code: str, size: int = 20) -> dict:
    """Что относится к почтовому индексу: населённый пункт и улицы.

    Дома не возвращаются — их на индекс десятки тысяч.
    """
    return _request("GET", "/api/geocodeforward/postcode", {"code": code, "size": size})


# ── Применение данных: годна ли координата для привязки к зданию ──────────────
# Ответ геокодера сам по себе — просто точка. Практический вопрос другой: можно
# ли ставить на этой точке метку конкретного дома. Отвечают на него два поля,
# читаемые вместе: layer («что нашли») и precision («чья это координата»).

USE = "USE"          # координату можно использовать как есть
VERIFY = "VERIFY"    # координата есть, но её стоит показать человеку
REJECT = "REJECT"    # для привязки к зданию не годится


@dataclass
class Verdict:
    decision: str
    reasons: list[str] = field(default_factory=list)


# Номер дома отделяется от названия улицы справа налево, по последнему отдельно
# стоящему числу — то же правило, по которому разбирает строку сам сервис.
# Поэтому «улица 8 Марта, 5» разбирается верно: частью названия остаётся то,
# что ему принадлежит.
_HOUSE_TAIL = re.compile(r"(\d+[а-яёa-z]?(?:\s*к\s*\d+)?)\s*$", re.IGNORECASE)


def requested_house(text: str) -> str | None:
    """Достаёт номер дома из исходной строки запроса, если он там был."""
    match = _HOUSE_TAIL.search(text.strip())
    return match.group(1) if match else None


def _normalize(house: str | None) -> str:
    return re.sub(r"[\s.\-]", "", house or "").lower()


def assess_coordinate(query: str, match: dict) -> Verdict:
    """Выносит вердикт: можно ли привязывать этот вариант к конкретному зданию."""
    precision = match.get("precision")
    wanted = requested_house(query)
    got = match.get("houseNumber")

    if precision == "exact":
        # Точное попадание в дом. Остаётся сверить, тот ли это дом, который просили:
        # сервис мог вернуть «13», когда спрашивали «13к2» — см. ветку range ниже.
        if wanted and got and _normalize(wanted) != _normalize(got):
            return Verdict(VERIFY, [f"Запрошен дом {wanted}, найден дом {got}"])
        return Verdict(USE, [f"Точная координата дома {got}" if got else "Точная координата объекта"])

    if precision == "range":
        # range означает одно из двух: номер вычислен между соседними домами либо
        # совпадение было неполным. И то и другое — «найдено с оговоркой»: точка
        # рядом с нужным зданием, но не обязательно на нём.
        reasons = ["Номер вычислен между соседними домами либо совпадение неполное"]
        if wanted and got and _normalize(wanted) != _normalize(got):
            reasons.append(f"запрошен дом {wanted}, найден дом {got}")
        return Verdict(VERIFY, reasons)

    if precision == "street":
        return Verdict(REJECT, ["Дома в данных нет — отдан центр улицы"])

    # locality и area до клиента обычно не доходят: такие результаты сервис считает
    # слишком грубыми и отвечает 404, не списывая денег. Ветка оставлена, чтобы
    # неизвестное значение не превратилось молча в «годится».
    return Verdict(REJECT, [f"Уровень точности «{precision}» слишком груб для привязки к зданию"])


# ── Печать ────────────────────────────────────────────────────────────────────


def print_match(index: int, query: str, match: dict) -> None:
    point = match.get("point") or {}
    print(f"{index}. {match.get('label')}")
    print(
        f"   {point.get('latitude')}, {point.get('longitude')}"
        f" · layer={match.get('layer')} · precision={match.get('precision')}"
    )
    print(f"   Индекс {match.get('postalCode')} · placeId {match.get('placeId')}")

    verdict = assess_coordinate(query, match)
    print(f"   [{verdict.decision}] {'; '.join(verdict.reasons)}")


def main() -> int:
    if API_KEY == SANDBOX_KEY:
        print("Демо-ключ: ответы сгенерированы (моки), не данные реестра.\n")

    query = sys.argv[1] if len(sys.argv) > 1 else "Казань, улица Баумана, 13"

    try:
        result = geocode(query)
    except AtloriumError as error:
        print(f"Ошибка: {error}", file=sys.stderr)
        return 1

    matches = result.get("matches") or []
    print(f"Запрос: {query}")
    print(f"Найдено вариантов: {len(matches)}\n")

    for index, match in enumerate(matches, start=1):
        print_match(index, query, match)
        print()

    # Пакетный режим — то, ради чего сервис обычно и подключают: обогатить
    # координатами накопленную выгрузку. Показываем главное: строки, по которым
    # ничего не нашлось, не прерывают пакет и не тарифицируются.
    batch_queries = [query, "Москва, Тверская 1", "улицы с таким названием не существует"]
    try:
        batch = geocode_batch(batch_queries)
    except AtloriumError as error:
        print(f"Пакетный режим недоступен: {error}", file=sys.stderr)
        return 0

    print(f"Пакетная выгрузка ({batch.get('requested')} строки):")
    for index, item in enumerate(batch.get("items") or [], start=1):
        item_matches = item.get("matches") or []
        if item_matches:
            point = item_matches[0].get("point") or {}
            found = f"{point.get('latitude')}, {point.get('longitude')} ({item_matches[0].get('precision')})"
        else:
            found = "не найдено"
        print(f"  {index}. {(item.get('query') or '')[:42]:<42} → {found}")

    print(
        f"  Отправлено {batch.get('requested')}, найдено {batch.get('found')}, "
        f"списано единиц {batch.get('billedUnits')} — ненайденные строки не тарифицируются."
    )

    # Показывать источник данных обязательно: этого требует лицензия исходных наборов.
    print(f"\n{result.get('attribution')}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())