Yellow Pages Malaysia

"""YellowPages.my Business Scraper — Apify Actor."""

import asyncio import html import json import random import re from urllib.parse import urljoin, urlparse

import httpx from playwright_stealth import Stealth

from apify import Actor from crawlee import ConcurrencySettings, Request from crawlee.crawlers import PlaywrightCrawler, PlaywrightCrawlingContext from crawlee.proxy_configuration import ProxyConfiguration

BASE_URL = "https://www.yellowpages.my" API_URL = "https://api.yellowpages.my"

TRANSFERSTATE_DUMPED = False

async def discover_categories() -> list[dict]: """Fetch categories from the unprotected /home API endpoint.""" categories = [] async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client: resp = await client.get( f"{API_URL}/home", headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}, ) if resp.status_code == 200: data = resp.json().get("data", {}) # API uses 'categories' key with nested structure for cat in data.get("categories", []): slug = cat.get("slug", "") name = cat.get("name", "") if slug: categories.append({ "name": name, "slug": slug, "url": f"{BASE_URL}/category/{slug}", "count": cat.get("post_count", 0), }) Actor.log.info(f"Discovered {len(categories)} categories from /home API") else: Actor.log.warning(f"Failed to fetch /home API: {resp.status_code}")

return categories

def _decode_transferstate(raw: str) -> dict: """Decode TransferState content with &q; encoding and HTML entities.""" # Replace &q; with double quotes, &a; with & decoded = raw.replace("&q;", '"').replace("&a;", "&") # Unescape HTML entities like … etc. decoded = html.unescape(decoded) return json.loads(decoded)

def _deep_find_profile(data: dict) -> dict | None: """Find profile data in TransferState dict.

Looks for key matching pattern like 'profile-data' or similar,
then returns body.data from that entry.
"""
for key, value in data.items():
    if isinstance(value, dict) and "body" in value:
        body = value["body"]
        if isinstance(body, dict) and "data" in body:
            profile = body["data"]
            if isinstance(profile, dict) and "title" in profile:
                return profile
return None

def extract_from_transfer_state(data: dict, url: str) -> dict: """Parse Angular TransferState JSON for business data.""" record = { "url": url, "name": None, "store_name": None, "phone": None, "phone_masked": None, "extra_phones": [], "fax_number": None, "address": None, "categories": [], "website_url": None, "hours": None, "structured_hours": [], "description": None, "email": None, "lat": None, "lon": None, "rating": None, "rating_count": None, "photo": None, "all_images": [], "company_no": None, "plan": None, "is_claimed": None, "social_media": {}, "view_count": None, }

try:
    profile = _deep_find_profile(data)

    if profile:
        record["name"] = profile.get("title")
        record["view_count"] = profile.get("view_count")

        locations = profile.get("locations", [])
        profile_nested = profile.get("profile", {}) if isinstance(profile.get("profile"), dict) else {}

        # Phone: masked from first location
        for loc in locations:
            phone = loc.get("company_phone_number")
            if phone:
                record["phone_masked"] = phone
                break

        # Fax number from first location
        if locations:
            fax = locations[0].get("fax_number")
            if fax:
                record["fax_number"] = fax

        # Extra phones from additional locations
        for loc in locations[1:]:
            extra_phone = loc.get("company_phone_number")
            if extra_phone:
                record["extra_phones"].append(extra_phone)

        # Email from first location
        if locations:
            email = locations[0].get("email")
            if email:
                record["email"] = email

        # Address: combine from first location
        if locations:
            loc = locations[0]
            addr_parts = [
                loc.get("address_line1"),
                loc.get("address_line2"),
                f"{loc.get('postcode', '')} {loc.get('city_state', '')}".strip()
            ]
            record["address"] = ", ".join(p for p in addr_parts if p) or None

        # Categories: all categories as array
        for cat in profile.get("categories", []):
            if isinstance(cat, dict):
                cat_name = cat.get("name")
                if cat_name:
                    parent = cat.get("parent_category")
                    if parent and isinstance(parent, dict):
                        parent_name = parent.get("name")
                        if parent_name:
                            cat_name = f"{parent_name} > {cat_name}"
                    record["categories"].append(cat_name)
            elif isinstance(cat, str):
                record["categories"].append(cat)

        # Coordinates: skip if "0"
        if locations:
            loc = locations[0]
            lat = loc.get("latitude")
            lon = loc.get("longitude")
            if lat and lat != "0":
                try:
                    record["lat"] = float(lat)
                except (ValueError, TypeError):
                    pass
            if lon and lon != "0":
                try:
                    record["lon"] = float(lon)
                except (ValueError, TypeError):
                    pass

        # Description: prefer profile.profile.about over profile.summary
        about_text = profile_nested.get("about") if profile_nested else None
        summary = profile.get("summary")
        if about_text:
            record["description"] = html.unescape(about_text)
        elif summary:
            record["description"] = html.unescape(summary)

        # Extract unmasked phone from about text
        source_text = about_text or ""
        phone_patterns = [
            r'(?:Telephone|Tel|Phone|Call)\s*[:\-]?\s*(?:\(?\+?60\)?[\s\-]?)?([\d][\d\s\-]{8,14}\d)',
            r'(?:\+?60[\s\-])?(\d{2,3}[\s\-]\d{3,4}[\s\-]\d{4})',
            r'(\d{3}[\s\-]\d{4}[\s\-]\d{4})',
        ]
        for pat in phone_patterns:
            match = re.search(pat, source_text, re.IGNORECASE)
            if match:
                raw_phone = match.group(0).strip()
                phone_match = re.search(r'(\+?60[\s\-]?\d[\d\s\-]{7,12}\d)', raw_phone)
                if phone_match:
                    record["phone"] = phone_match.group(1).strip()
                else:
                    record["phone"] = re.search(r'([\d][\d\s\-]{7,12}\d)', raw_phone).group(1).strip()
                break

        # Extract operating hours from about text (fallback)
        hours_patterns = [
            r'(?:Opening\s*Hours?|Business\s*Hours?|Operating\s*Hours?|Office\s*Hours?)\s*[:\-]?\s*(.+?)(?:\n\n|\n[A-Z]|$)',
            r'((?:Monday|Mon)[\s:].+?(?:\n\n|\n[A-Z]|$))',
        ]
        for pat in hours_patterns:
            match = re.search(pat, source_text, re.IGNORECASE | re.DOTALL)
            if match:
                hours = match.group(1) if match.lastindex else match.group(0)
                hours = hours.strip()[:200]
                hours = re.split(r'\n(?:Licence|MEMBERSHIP|ASSOCIATES|Product)', hours)[0].strip()
                if hours:
                    record["hours"] = hours
                break

        # Fallback: masked phone from locations if no unmasked found
        if not record["phone"] and record["phone_masked"]:
            record["phone"] = record["phone_masked"]

        # Structured hours from profile.profile.business_hours
        business_hours = profile_nested.get("business_hours", [])
        if isinstance(business_hours, list):
            for entry in business_hours:
                if isinstance(entry, dict):
                    record["structured_hours"].append({
                        "day": entry.get("day"),
                        "open": entry.get("open"),
                        "close": entry.get("close"),
                    })

        # Rating: from profile.profile.rating
        rating = profile_nested.get("rating")
        if rating:
            try:
                record["rating"] = float(rating)
            except (ValueError, TypeError):
                pass

        rating_count = profile_nested.get("rating_count")
        if rating_count:
            try:
                record["rating_count"] = int(rating_count)
            except (ValueError, TypeError):
                pass

        # Photo: from profile.profile.image.thumbnail
        image = profile_nested.get("image", {})
        if isinstance(image, dict):
            thumbnail = image.get("thumbnail")
            if thumbnail:
                if thumbnail.startswith("/"):
                    record["photo"] = f"{BASE_URL}{thumbnail}"
                elif not thumbnail.startswith("http"):
                    record["photo"] = f"{BASE_URL}/{thumbnail}"
                else:
                    record["photo"] = thumbnail

        # All images from profile.profile.images[]
        for img in profile_nested.get("images", []):
            if isinstance(img, dict):
                img_data = {}
                full = img.get("full") or img.get("image", {}).get("full") if isinstance(img.get("image"), dict) else img.get("full")
                thumb = img.get("thumbnail") or img.get("image", {}).get("thumbnail") if isinstance(img.get("image"), dict) else img.get("thumbnail")
                if full:
                    img_data["full"] = f"{BASE_URL}{full}" if full.startswith("/") else full
                if thumb:
                    img_data["thumbnail"] = f"{BASE_URL}{thumb}" if thumb.startswith("/") else thumb
                if img_data:
                    record["all_images"].append(img_data)

        # Website URL
        url_val = profile_nested.get("url")
        if url_val:
            record["website_url"] = url_val

        # Company number (SSM)
        company_no = profile_nested.get("company_no")
        if company_no:
            record["company_no"] = company_no

        # Business plan/tier
        plan = profile_nested.get("plan")
        if plan:
            record["plan"] = plan

        # Store name (alternate business name)
        store_name = profile_nested.get("store_name")
        if store_name:
            record["store_name"] = store_name

        # Claimed status
        claimed = profile_nested.get("claimed")
        if claimed is not None:
            record["is_claimed"] = int(claimed) if claimed else 0

        # Social media links
        social_fields = ["facebook", "instagram", "twitter", "linkedin", "youtube", "whatsapp", "wechat"]
        for field in social_fields:
            val = profile_nested.get(field)
            if val:
                record["social_media"][field] = val

except Exception as e:
    Actor.log.warning(f"TransferState parse error: {e}")

return record

async def extract_from_dom(page, url: str) -> dict: """Fallback: extract data from visible DOM elements.""" record = { "url": url, "name": None, "phone": None, "address": None, "category": None, "website": None, "hours": None, "description": None, "lat": None, "lon": None, "rating": None, "photo": None, }

try:
    record["name"] = await page.text_content("h1") or None

    phone_el = await page.query_selector('a[href^="tel:"]')
    if phone_el:
        href = await phone_el.get_attribute("href")
        record["phone"] = href.replace("tel:", "") if href else None

    addr_el = await page.query_selector(
        '[class*="address"], .listing-address, address'
    )
    if addr_el:
        record["address"] = (await addr_el.text_content()).strip()

    desc_el = await page.query_selector(
        '[class*="description"], .listing-desc, .summary'
    )
    if desc_el:
        record["description"] = (await desc_el.text_content()).strip()

    rating_el = await page.query_selector('[class*="rating"], .star-rating')
    if rating_el:
        rating_text = await rating_el.text_content()
        match = re.search(r"(\d+\.?\d*)", rating_text or "")
        if match:
            record["rating"] = float(match.group(1))

    breadcrumbs = await page.query_selector_all(
        '.breadcrumb a, nav[aria-label="breadcrumb"] a'
    )
    if breadcrumbs:
        cats = [await b.text_content() for b in breadcrumbs]
        cats = [c.strip() for c in cats if c and c.strip() and c.strip() != "Home"]
        record["category"] = " > ".join(cats) if cats else None

    map_link = await page.query_selector(
        'a[href*="maps.google"], a[href*="google.com/maps"]'
    )
    if map_link:
        href = await map_link.get_attribute("href")
        match = re.search(r"@(-?\d+\.?\d*),(-?\d+\.?\d*)", href or "")
        if match:
            record["lat"] = float(match.group(1))
            record["lon"] = float(match.group(2))

except Exception as e:
    Actor.log.warning(f"DOM extraction error on {url}: {e}")

return record

async def _detect_cloudflare(page) -> bool: """Return True if page is a Cloudflare challenge.""" try: title = (await page.title()).lower() if "just a moment" in title or "attention required" in title: return True cf_el = await page.query_selector("#challenge-running, #challenge-stage, .cf-browser-verification") if cf_el: return True body_text = (await page.text_content("body") or "").lower() if "checking your browser" in body_text and "cloudflare" in body_text: return True except Exception: pass return False

async def main() -> None: async with Actor: Actor.log.info("Starting YellowPages.my scraper") actor_input = await Actor.get_input() or {}

input_categories = actor_input.get("categories", [])
    max_pages = actor_input.get("maxPagesPerCategory", 0)
    max_concurrency = actor_input.get("maxConcurrency", 5)
    proxy_input = actor_input.get("proxy")

    # Phase 1: Discover categories
    if input_categories:
        categories = [
            {"slug": c, "url": f"{BASE_URL}/category/{c}"}
            for c in input_categories
        ]
        Actor.log.info(f"Using {len(categories)} user-specified categories")
    else:
        categories = await discover_categories()
        Actor.log.info(f"Auto-discovered {len(categories)} categories")

    if not categories:
        Actor.log.error("No categories found. Aborting.")
        return

    await Actor.set_value("categories", categories)
    Actor.log.info(f"Ready to scrape {len(categories)} categories")

    stats = {
        "categories_crawled": 0,
        "listings_found": 0,
        "listings_scraped": 0,
        "errors": 0,
        "cf_blocked": 0,
    }
    category_page_counts: dict[str, int] = {}

    def _get_category_key(url: str) -> str:
        """Extract base category slug from URL for pagination tracking."""
        path = urlparse(url).path.rstrip("/")
        parts = path.split("/")
        # /category/{slug} or /category/{parent}/{sub}
        if "category" in parts:
            idx = parts.index("category")
            return "/".join(parts[idx : idx + 2]) if idx + 1 < len(parts) else path
        return path

    async def handle_listing(context: PlaywrightCrawlingContext) -> None:
        page = context.page
        request = context.request
        await stealth.apply_stealth_async(page)

        await asyncio.sleep(random.uniform(0.5, 2.0))
        await page.wait_for_load_state("networkidle", timeout=30000)

        if await _detect_cloudflare(page):
            stats["cf_blocked"] += 1
            Actor.log.warning(f"CF challenge on listing {request.url}, will retry")
            raise RuntimeError("Cloudflare challenge detected")

        # Strategy 1: Extract Angular TransferState JSON
        # TransferState uses &q; encoding, need to extract from HTML directly
        transfer_data = None
        try:
            # Get full HTML and extract script content via string matching
            html_content = await page.content()
            
            # Find script tag with id="serverApp-state"
            match = re.search(
                r'<script[^>]*id="serverApp-state"[^>]*>(.*?)</script>',
                html_content,
                re.DOTALL
            )
            if match:
                raw_content = match.group(1)
                # Decode &q; to " and unescape HTML entities
                decoded = raw_content.replace("&q;", '"').replace("&a;", "&")
                decoded = html.unescape(decoded)
                transfer_data = json.loads(decoded)
                
                global TRANSFERSTATE_DUMPED
                if not TRANSFERSTATE_DUMPED:
                    await Actor.set_value("transferstate_debug", transfer_data)
        except Exception as e:
            Actor.log.warning(f"TransferState extraction failed: {e}")

        if transfer_data:
            record = extract_from_transfer_state(transfer_data, request.url)
        else:
            record = await extract_from_dom(page, request.url)

        if record and record.get("name"):
            await Actor.push_data(record)
            stats["listings_scraped"] += 1
            Actor.log.info(f"Scraped: {record['name']}")
        else:
            stats["errors"] += 1
            Actor.log.warning(f"No data extracted from {request.url}")

    async def handle_category(context: PlaywrightCrawlingContext) -> None:
        page = context.page
        request = context.request
        await stealth.apply_stealth_async(page)

        await asyncio.sleep(random.uniform(0.5, 2.0))
        await page.wait_for_load_state("networkidle", timeout=30000)

        if await _detect_cloudflare(page):
            stats["cf_blocked"] += 1
            Actor.log.warning(f"CF challenge on category {request.url}, will retry")
            raise RuntimeError("Cloudflare challenge detected")

        # Extract listing links (/profile/...)
        listing_urls = await page.eval_on_selector_all(
            'a[href*="/profile/"]',
            "elements => elements.map(el => el.href)",
        )

        seen: set[str] = set()
        for url in listing_urls:
            if url not in seen and "/profile/" in url:
                seen.add(url)
                await context.add_requests(
                    [Request.from_url(url, label="listing")]
                )

        stats["listings_found"] += len(seen)
        stats["categories_crawled"] += 1
        Actor.log.info(
            f"Category page {request.url}: found {len(seen)} listing URLs"
        )

        # Pagination — respect maxPagesPerCategory
        category_key = _get_category_key(request.url)
        category_page_counts[category_key] = (
            category_page_counts.get(category_key, 0) + 1
        )

        if max_pages > 0 and category_page_counts[category_key] >= max_pages:
            Actor.log.info(
                f"Reached max pages ({max_pages}) for {category_key}"
            )
            return

        next_page = await page.query_selector(
            'a[rel="next"], .pagination .next a, a:has-text("Next")'
        )
        if next_page:
            next_url = await next_page.get_attribute("href")
            if next_url:
                absolute_url = urljoin(request.url, next_url)
                await context.add_requests(
                    [Request.from_url(absolute_url, label="category")]
                )

    # Proxy configuration
    proxy_configuration = None
    if proxy_input:
        proxy_configuration = ProxyConfiguration()

    # Phase 2+3+4: Crawl categories and extract listings
    stealth = Stealth()
    crawler = PlaywrightCrawler(
        max_requests_per_crawl=None,
        concurrency_settings=ConcurrencySettings(max_concurrency=max_concurrency),
        headless=True,
        request_handler_timeout=timedelta(seconds=120),
        max_request_retries=3,
        proxy_configuration=proxy_configuration,
    )

    @crawler.router.default_handler
    async def default_handler(context: PlaywrightCrawlingContext) -> None:
        label = context.request.label
        if label == "listing":
            await handle_listing(context)
        else:
            await handle_category(context)

    seed_requests = [
        Request.from_url(cat["url"], label="category") for cat in categories
    ]
    await crawler.run(seed_requests)

    total_attempts = stats["listings_scraped"] + stats["errors"]
    error_rate = stats["errors"] / total_attempts if total_attempts > 0 else 0
    if error_rate > 0.5 and total_attempts > 10:
        Actor.log.error(
            f"High error rate: {error_rate:.1%} ({stats['errors']}/{total_attempts}). "
            f"CF blocks: {stats['cf_blocked']}"
        )

    await Actor.set_value("run_stats", stats)
    Actor.log.info(f"Run complete: {stats}")