storage.py

import csv
import os
import threading
from pathlib import Path
from models import BusinessCard, Review

BUSINESSES_FIELDS = [
    "card_id", "source", "name", "address", "city", "category",
    "phones", "mobile_phones", "rating", "review_count", "working_hours",
    "has_website", "website_url", "card_url", "lat", "lon",
    "whatsapp", "telegram",
    "has_booking", "booking_url",
    "pain_point", "pain_point_quote",
    "review_summary", "first_message",
    "negative_reviews_count", "pain_point_match", "pain_point_evidence",
    "rating_min_used", "rating_max_used",
    "script_dialog",
]

REVIEWS_FIELDS = ["card_id", "author", "date", "rating", "text"]

_lock = threading.Lock()


def _ensure_file(path: str, fieldnames: list[str]):
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    if not p.exists():
        with open(p, "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=fieldnames)
            writer.writeheader()


def load_saved_urls(businesses_csv: str) -> set[str]:
    p = Path(businesses_csv)
    if not p.exists():
        return set()
    with open(p, "r", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        return {row["card_url"] for row in reader if "card_url" in row}


def append_card(card: BusinessCard, path: str):
    _ensure_file(path, BUSINESSES_FIELDS)
    with _lock:
        with open(path, "a", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=BUSINESSES_FIELDS)
            writer.writerow(card.to_csv_row())


def append_reviews(reviews: list[Review], path: str):
    if not reviews:
        return
    _ensure_file(path, REVIEWS_FIELDS)
    with _lock:
        with open(path, "a", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=REVIEWS_FIELDS)
            for r in reviews:
                writer.writerow({
                    "card_id": r.card_id,
                    "author": r.author,
                    "date": r.date,
                    "rating": r.rating,
                    "text": r.text,
                })