SpoolBeacon/seed_shops.py at master · SytxLabs/SpoolBeacon · 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
"""
Shop seed — ShopRules only, no filament/inventory data.
Safe to run standalone or alongside seed.py. Idempotent (skips existing domains).

Domains with a registered adapter (see app.shop_adapters.registry.registered_domains())
are intentionally NOT seeded here — the adapter always takes priority over a ShopRule
for the same domain, so a rule for it would be dead weight and confusing in the UI.
"""
import asyncio
from dotenv import load_dotenv

load_dotenv()

# pylint: disable=wrong-import-position
# Must be imported after load_dotenv() — app.config reads DB_* env vars at import time.
from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

from app.config import _build_database_url
from app.models.shop_rule import ShopRule  # noqa: F401 — import ensures model is registered
# pylint: enable=wrong-import-position

RULES = [
    # ── Rule-only shops (no adapter, httpx/Playwright fetch) ─────────────────
    {
        "domain": "amazon.de",
        "price_selector": ".a-price .a-offscreen",
        "price_regex": r"\d+[,\.]\d{2}",
        "title_selector": None,
        "availability_selector": "#availability span",
        "availability_regex": None,
        "currency": "EUR",
        "test_url": "",
        "is_active": True,
        "notes": "BLOCKED — ASIN pages return 404 without session. Future: Amazon Product Advertising API.",
    },
    {
        "domain": "ebay.de",
        "price_selector": ".x-price-primary .ux-textspans",
        "price_regex": r"\d+[,\.]\d{2}",
        "title_selector": None,
        "availability_selector": None,
        "availability_regex": None,
        "currency": "EUR",
        "test_url": "",
        "is_active": True,
        "notes": "BLOCKED — Cloudflare (httpx + Playwright + cloudscraper). Future: eBay Browse API.",
    },
    {
        "domain": "sunlu.com",
        "price_selector": "[class*='price']",
        "price_regex": r"\d+[,\.]\d{2}",
        "title_selector": None,
        "availability_selector": None,
        "availability_regex": None,
        "currency": "USD",
        "test_url": (
            "https://store.sunlu.com/products/"
            "over-6kg-of-pla-pla-meta-3d-filaments-1kg-2-2lbs-fit-most-of-fdm-printer"
        ),
        "is_active": True,
        "notes": "HTTP 500 on collection pages. Server instability or geo-blocking.",
    },
    {
        "domain": "shop.polymaker.com",
        "price_selector": ".price-item--regular",
        "price_regex": r"\d+[,\.]\d{2}",
        "title_selector": None,
        "availability_selector": None,
        "availability_regex": None,
        "currency": "USD",
        "test_url": "https://shop.polymaker.com/en-eu/products/panchroma-matte?variant=43631458549817",
        "is_active": True,
        "notes": "Marketing/product-info site only — no purchasable prices. Sold via distributors.",
    },
]


async def run() -> None:
    engine = create_async_engine(_build_database_url(), echo=False)
    async_session = async_sessionmaker(engine, expire_on_commit=False)

    async with async_session() as session:
        added = 0
        skipped = 0
        for rule_data in RULES:
            existing = (await session.execute(
                select(ShopRule).where(ShopRule.domain == rule_data["domain"])
            )).scalar_one_or_none()
            if existing:
                skipped += 1
                continue
            session.add(ShopRule(**rule_data))
            added += 1

        await session.commit()

    await engine.dispose()
    print(f"seed_shops: {added} rules added, {skipped} already existed.")


if __name__ == "__main__":
    asyncio.run(run())