watchy/fetch_watch.py at main · nithis-057/watchy · 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
111
112
113
114
115
116
117
118
119
120
121
122
import asyncio
import os
import urllib.parse
import random
import re
from dotenv import load_dotenv
from supabase import create_client
from crawlee.crawlers import PlaywrightCrawler, PlaywrightCrawlingContext
from crawlee.router import Router
from crawlee.fingerprint_suite import DefaultFingerprintGenerator, HeaderGeneratorOptions
from crawlee import Request

load_dotenv()
supa_url = os.getenv('SUPABASE_URL')
supa_key = os.getenv('SUPABASE_KEY')
supabase = create_client(supa_url, supa_key)

router = Router[PlaywrightCrawlingContext]()

@router.handler("swisstimehouse")
async def swisstimehouse_handler(context: PlaywrightCrawlingContext):
    page = context.page
    await page.wait_for_selector("h1", timeout=15000)
    total_name = await page.locator('h1').first.inner_text()
    t = total_name.split()
    brand = t[0]
    model = ''.join(t[1:])
    price = await page.locator('div.ce-product-price span').first.inner_text()
    details = {"brand": brand.strip(), "model": model.strip(), "price": price, "url": context.request.url, "label": "swisstimehouse"}
    await context.push_data(details)

@router.handler("timexoff")
async def timex_off_handler(context: PlaywrightCrawlingContext):
    page = context.page
    await page.wait_for_selector("div.mini-details",timeout = 15000)
    tot_name = await page.locator("div.mini-details").inner_text()


@router.handler("helios")
async def helios_handler(context: PlaywrightCrawlingContext):
    page = context.page
    await page.wait_for_selector("span.special-price", timeout=15000)
    price = await page.locator("span.special-price").first.inner_text()
    brand = await page.locator("div.custom-subname h3").first.inner_text()
    type_label = await page.locator(".product-info-stock-sku .type").first.inner_text()
    value_text = await page.locator(".product-info-stock-sku .value").first.inner_text()
    model = f"{type_label.strip()}: {value_text.strip()}"
    await context.push_data({"brand": brand.strip(), "model": model.strip(), "price": price, "url": context.request.url, "label": "helios"})

@router.handler("ethos")
async def ethos_handling(context: PlaywrightCrawlingContext):
    page = context.page
    await page.wait_for_selector("h1.ethos_title", timeout=15000)
    price = await page.locator("span.price").first.inner_text()
    brand = await page.locator("h1.ethos_title a").first.inner_text()
    model = await page.locator("h1.ethos_title span").first.inner_text()
    await context.push_data({"brand": brand.strip(), "model": model.strip(), "price": price, "url": context.request.url, "label": "ethos"})

@router.handler("kapoor")
async def kapoor_handler(context: PlaywrightCrawlingContext):
    page = context.page
    await page.wait_for_selector("h1", timeout=15000)
    price = await page.locator("p.proPrice").first.inner_text()
    brand = await page.locator("h1 span.proName").inner_text()
    model = await page.locator("h1 span.proCode").inner_text()
    await context.push_data({"brand": brand.strip(), "model": model.strip(), "price": price, "url": context.request.url, "label": "kapoor"})

@router.handler("search_results")
async def search_results_handler(context: PlaywrightCrawlingContext):
    page = context.page
    label = context.request.label
    grid_selector = ".product-items, .products-grid, .ce-product-list, .product-listing"
    await page.wait_for_selector(grid_selector, timeout=15000)
    link_selector = ".product-item-link, .product-title a, .ce-product-name a, .proName a"
    first_link = page.locator(link_selector).first
    if await first_link.count() > 0:
        product_url = await first_link.get_attribute("href")
        if product_url.startswith('/'):
            product_url = urllib.parse.urljoin(context.request.url, product_url)
        await context.enqueue_links(urls=[product_url], label=label)

async def live_search_all_sites(watch_name: str):
    query = urllib.parse.quote_plus(watch_name)
    requests = [
        Request.from_url(f"https://www.ethoswatches.com/catalogsearch/result/?q={query}", label="ethos", user_data={"label": "search_results"}),
        Request.from_url(f"https://www.helioswatchstore.com/catalogsearch/result/?q={query}", label="helios", user_data={"label": "search_results"}),
        Request.from_url(f"https://www.swisstimehouse.com/search?search_query={query}", label="swisstimehouse", user_data={"label": "search_results"}),
        Request.from_url(f"https://www.kapoorwatch.com/catalogsearch/result/?q={query}", label="kapoor", user_data={"label": "search_results"})
    ]

    crawler = PlaywrightCrawler(
        request_handler=router,
        max_requests_per_crawl=10,
        browser_launch_options={
            "args": ["--disable-blink-features=AutomationControlled"]
        },
        headless=True
    )

    await asyncio.sleep(random.uniform(2, 5))
    await crawler.run(requests)
    data = await crawler.get_data()
    return data.items

def bridge_function(url):
    data_object = asyncio.run(use_crawlee(url))
    if data_object and data_object.items:
        supabase.table("watches").insert(data_object.items).execute()
        return data_object.items
    return "No data retrieved."

async def use_crawlee(url):
    if "kapoorwatch" in url: label = "kapoor"
    elif "ethoswatches" in url: label = "ethos"
    elif "swisstimehouse" in url: label = "swisstimehouse"
    elif "helioswatchstore" in url: label = "helios"
    else: return None
    crawler = PlaywrightCrawler(request_handler=router, max_requests_per_crawl=1, headless=True)
    await crawler.run([Request.from_url(url, label=label)])
    return await crawler.get_data()