lead-parser/main.py at master · feewwee39-dev/lead-parser · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
#!/usr/bin/env python3
"""
Entry point for 2GIS / Yandex Maps business scraper.

Usage:
    python main.py                        # uses config.json
    python main.py --config my.json
    python main.py --city Москва --category "маникюрный салон" --min-cards 30
"""

import argparse
import asyncio
import json
import logging
import sys
from pathlib import Path

import storage
from scraper_2gis import Scraper2GIS
from scraper_google import ScraperGoogle
from scraper_yandex import ScraperYandex

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)


def load_config(path: str = "config.json") -> dict:
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def apply_cli_overrides(cfg: dict, args: argparse.Namespace) -> dict:
    if args.city and args.category:
        cfg["tasks"] = [{
            "city": args.city,
            "category": args.category,
            "min_cards": args.min_cards or 20,
        }]
    if args.rating_min is not None:
        cfg.setdefault("rating_filter", {})["min"] = args.rating_min
    if args.rating_max is not None:
        cfg.setdefault("rating_filter", {})["max"] = args.rating_max
    if args.headless is not None:
        cfg.setdefault("browser", {})["headless"] = args.headless
    return cfg


async def main_async(cfg: dict):
    source = cfg.get("source", "2gis").lower()
    out = cfg.get("output", {})
    businesses_csv = out.get("businesses_csv", "output/businesses.csv")
    reviews_csv = out.get("reviews_csv", "output/reviews.csv")
    tasks = cfg.get("tasks", [])

    if not tasks:
        print("Нет задач в конфиге (tasks: []). Укажите город и категорию.")
        sys.exit(1)

    # load already-collected URLs for deduplication
    saved_urls = storage.load_saved_urls(businesses_csv)
    logger.info(f"Уже собрано карточек: {len(saved_urls)}")

    def on_card(card):
        storage.append_card(card, businesses_csv)

    def on_reviews(reviews):
        storage.append_reviews(reviews, reviews_csv)

    if source == "2gis":
        scraper = Scraper2GIS(cfg)
        await scraper.run(tasks, saved_urls, on_card, on_reviews)
    elif source == "yandex":
        scraper = ScraperYandex(cfg)
        await scraper.run(tasks, saved_urls, on_card, on_reviews)
    elif source == "google":
        scraper = ScraperGoogle(cfg)
        await scraper.run(tasks, saved_urls, on_card, on_reviews)
    else:
        print(f"Источник '{source}' не поддерживается. Используйте '2gis' или 'yandex'.")
        sys.exit(1)

    print(f"\nГотово. Данные сохранены:")
    print(f"  Карточки:  {businesses_csv}")
    print(f"  Отзывы:    {reviews_csv}")


def main():
    parser = argparse.ArgumentParser(description="2GIS business scraper")
    parser.add_argument("--config", default="config.json", help="Путь к конфигу")
    parser.add_argument("--city", help="Город (переопределяет tasks в конфиге)")
    parser.add_argument("--category", help="Категория бизнеса")
    parser.add_argument("--min-cards", type=int, help="Минимум карточек")
    parser.add_argument("--rating-min", type=float)
    parser.add_argument("--rating-max", type=float)
    parser.add_argument("--headless", action=argparse.BooleanOptionalAction,
                        default=None, help="--headless / --no-headless")
    args = parser.parse_args()

    cfg = load_config(args.config)
    cfg = apply_cli_overrides(cfg, args)

    asyncio.run(main_async(cfg))


if __name__ == "__main__":
    main()