card_cache.py

# -*- coding: utf-8 -*-
"""
Кэш спарсенных карточек.

Хранится в output/cache_parsed.json.
Ключ — нормализованный URL карточки (без trailing slash, без query).
Значение — dict с полями карточки + список отзывов.

Использование:
    from card_cache import CardCache
    cache = CardCache()

    hit = cache.get(url)          # None если нет в кэше
    cache.put(url, card, reviews) # сохранить после парсинга
"""

import json
import re
from pathlib import Path
from typing import Optional

from models import BusinessCard, Phone, Review

_CACHE_PATH = Path("output/cache_parsed.json")


def _norm(url: str) -> str:
    """Strip trailing slash and query string for stable key."""
    url = url.split("?")[0].split("#")[0].rstrip("/")
    # yandex.ru и yandex.com — один и тот же объект
    url = re.sub(r"yandex\.[a-z]+", "yandex.com", url)
    return url.lower()


class CardCache:
    def __init__(self, path: Path = _CACHE_PATH):
        self._path = path
        self._path.parent.mkdir(parents=True, exist_ok=True)
        self._data: dict = {}
        if self._path.exists():
            try:
                self._data = json.loads(self._path.read_text(encoding="utf-8"))
            except Exception:
                self._data = {}

    # ── public API ────────────────────────────────────────────────────────────

    def get(self, url: str) -> Optional[tuple["BusinessCard", list["Review"]]]:
        """Return (card, reviews) from cache, or None if not cached."""
        entry = self._data.get(_norm(url))
        if not entry:
            return None
        try:
            card = _card_from_dict(entry["card"])
            reviews = [_review_from_dict(r) for r in entry.get("reviews", [])]
            return card, reviews
        except Exception:
            return None

    def put(self, url: str, card: "BusinessCard", reviews: list["Review"]) -> None:
        """Save card + reviews to cache and flush to disk."""
        self._data[_norm(url)] = {
            "card": card.to_csv_row(),
            "reviews": [
                {"card_id": r.card_id, "author": r.author,
                 "date": r.date, "rating": r.rating, "text": r.text}
                for r in reviews
            ],
        }
        self._flush()

    def has(self, url: str) -> bool:
        return _norm(url) in self._data

    def size(self) -> int:
        return len(self._data)

    # ── internal ──────────────────────────────────────────────────────────────

    def _flush(self):
        self._path.write_text(
            json.dumps(self._data, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )


# ── helpers ───────────────────────────────────────────────────────────────────

def _card_from_dict(d: dict) -> "BusinessCard":
    phones = [Phone(p.strip()) for p in (d.get("phones") or "").split(";") if p.strip()]
    return BusinessCard(
        card_id=d.get("card_id", ""),
        name=d.get("name", ""),
        address=d.get("address", ""),
        city=d.get("city", ""),
        category=d.get("category", ""),
        phones=phones,
        rating=_float(d.get("rating")),
        review_count=_int(d.get("review_count")),
        working_hours=d.get("working_hours", ""),
        has_website=bool(d.get("has_website")),
        website_url=d.get("website_url") or None,
        card_url=d.get("card_url", ""),
        lat=_float(d.get("lat")),
        lon=_float(d.get("lon")),
        source=d.get("source", "yandex"),
        whatsapp=d.get("whatsapp") or None,
        telegram=d.get("telegram") or None,
        has_booking=bool(d.get("has_booking")),
        booking_url=d.get("booking_url") or None,
        pain_point=bool(d.get("pain_point")),
        pain_point_quote=d.get("pain_point_quote", ""),
        review_summary=d.get("review_summary", ""),
        first_message=d.get("first_message", ""),
        negative_reviews_count=_int(d.get("negative_reviews_count")) or 0,
        pain_point_match=bool(d.get("pain_point_match")),
        pain_point_evidence=d.get("pain_point_evidence", ""),
        rating_min_used=_float(d.get("rating_min_used")),
        rating_max_used=_float(d.get("rating_max_used")),
        script_dialog=d.get("script_dialog", ""),
    )


def _review_from_dict(d: dict) -> "Review":
    return Review(
        card_id=d.get("card_id", ""),
        author=d.get("author", ""),
        date=d.get("date", ""),
        rating=_float(d.get("rating")),
        text=d.get("text", ""),
    )


def _float(v) -> Optional[float]:
    try:
        return float(v) if v not in (None, "", "None") else None
    except Exception:
        return None


def _int(v) -> Optional[int]:
    try:
        return int(v) if v not in (None, "", "None") else None
    except Exception:
        return None