Files
mostovik-backend/src/apps/parsers/clients/checko/datasets/base.py
Aleksandr Meshchriakov c36c7b9ba9 feat(parsers): добавлен API клиент для checko.ru
- Реализован CheckoClient с поддержкой всех 10 эндпоинтов API v2
- Frozen dataclass модели для запросов и ответов
- Справочники ОКВЭД2, ОКФС, ОКОПФ, ОКПД, статусы компаний
- Маппинг русских полей API на английские имена
- Unit тесты с моками
- E2E тесты с реальными запросами
- Настройка CHECKO_API_KEY в settings.py
2026-02-03 17:00:19 +01:00

180 lines
5.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Базовые классы для работы со справочниками Checko.
Справочники загружаются лениво (lazy loading) при первом обращении.
Данные кэшируются на уровне класса (singleton pattern).
"""
import json
from dataclasses import dataclass
from pathlib import Path
from typing import ClassVar, Generic, TypeVar
T = TypeVar("T")
# Путь к директории с JSON файлами
DATA_DIR = Path(__file__).parent / "data"
@dataclass(frozen=True)
class DatasetItem:
"""Базовый элемент справочника."""
code: str
name: str
class BaseDataset(Generic[T]):
"""
Базовый класс для работы со справочниками.
Реализует:
- Lazy loading данных при первом обращении
- Кэширование на уровне класса
- Поиск по коду и названию
Наследники должны определить:
- _json_filename: имя JSON файла
- _parse_item: метод парсинга элемента из JSON
"""
_data: ClassVar[dict[str, T] | None] = None
_json_filename: ClassVar[str] = ""
@classmethod
def _get_json_path(cls) -> Path:
"""Получить путь к JSON файлу."""
return DATA_DIR / cls._json_filename
@classmethod
def _parse_item(cls, raw: dict) -> T:
"""Распарсить элемент из JSON. Переопределяется в наследниках."""
raise NotImplementedError
@classmethod
def _get_item_code(cls, raw: dict) -> str:
"""Получить код элемента из JSON."""
return raw.get("code", "")
@classmethod
def _ensure_loaded(cls) -> None:
"""Загрузить данные если ещё не загружены."""
if cls._data is not None:
return
json_path = cls._get_json_path()
if not json_path.exists():
cls._data = {}
return
with open(json_path, encoding="utf-8") as f:
raw_data = json.load(f)
cls._data = {}
for raw in raw_data:
item = cls._parse_item(raw)
code = cls._get_item_code(raw)
cls._data[code] = item
@classmethod
def get(cls, code: str) -> T | None:
"""
Получить элемент по коду.
Args:
code: Код элемента справочника.
Returns:
Элемент справочника или None если не найден.
"""
cls._ensure_loaded()
return cls._data.get(code)
@classmethod
def get_name(cls, code: str) -> str | None:
"""
Получить название по коду.
Args:
code: Код элемента справочника.
Returns:
Название элемента или None если не найден.
"""
item = cls.get(code)
if item is None:
return None
return getattr(item, "name", None) or getattr(item, "full_name", None)
@classmethod
def all(cls) -> list[T]:
"""
Получить все элементы справочника.
Returns:
Список всех элементов.
"""
cls._ensure_loaded()
return list(cls._data.values())
@classmethod
def codes(cls) -> list[str]:
"""
Получить все коды справочника.
Returns:
Список всех кодов.
"""
cls._ensure_loaded()
return list(cls._data.keys())
@classmethod
def search(cls, query: str) -> list[T]:
"""
Поиск по названию (регистронезависимый).
Args:
query: Поисковый запрос.
Returns:
Список найденных элементов.
"""
cls._ensure_loaded()
q = query.lower()
results = []
for item in cls._data.values():
name = getattr(item, "name", "") or getattr(item, "full_name", "")
if name and q in name.lower():
results.append(item)
return results
@classmethod
def exists(cls, code: str) -> bool:
"""
Проверить существование кода.
Args:
code: Код элемента справочника.
Returns:
True если код существует.
"""
return cls.get(code) is not None
@classmethod
def count(cls) -> int:
"""
Получить количество элементов в справочнике.
Returns:
Количество элементов.
"""
cls._ensure_loaded()
return len(cls._data)
@classmethod
def reload(cls) -> None:
"""Перезагрузить данные из JSON файла."""
cls._data = None
cls._ensure_loaded()