# Yellow Pages Malaysia (`devjonas/yellow-pages-malaysia`) Actor

Extract high-quality B2B leads effortlessly. This Yellow Pages Scraper automates data collection from business listings, capturing essential details including contact info, locations, ratings, and URLs. Perfect for lead generation, market research, and sales outreach.

- **URL**: https://apify.com/devjonas/yellow-pages-malaysia.md
- **Developed by:** [Jai](https://apify.com/devjonas) (community)
- **Categories:** Lead generation, Integrations
- **Stats:** 2 total users, 1 monthly users, 100.0% runs succeeded, 0 bookmarks
- **User rating**: No ratings yet

## Pricing

from $0.05 / actor start

This Actor is paid per event. You are not charged for the Apify platform usage, but only a fixed price for specific events.

Learn more: https://docs.apify.com/platform/actors/running/actors-in-store#pay-per-event

## What's an Apify Actor?

Actors are web data automations that power AI and operations. They run on the Apify platform to scrape websites, process data, connect APIs, and automate workflows.
In Batch mode, an Actor accepts a well-defined JSON input, performs an action which can take anything from a few seconds to a few hours,
and optionally produces a well-defined JSON output, datasets with results, or files in key-value store.
In Standby mode, an Actor provides a web server which can be used as a website, API, or an MCP server.
Actors are written with capital "A".

## How to integrate an Actor?

If asked about integration, you help developers integrate Actors into their projects.
You adapt to their stack and deliver integrations that are safe, well-documented, and production-ready.
The best way to integrate Actors is as follows.

- **AI agents and MCP clients** — the [Apify MCP server](https://docs.apify.com/integrations/mcp.md) at `https://mcp.apify.com` (remote, streamable HTTP, OAuth on first use).
- **Agentic workflows and local Actor development** — [Agent Skills](https://apify.com/.well-known/agent-skills/index.json) with the [Apify CLI](https://docs.apify.com/cli/docs.md): `npm install -g apify-cli`, then `apify login`.
- **JavaScript/TypeScript projects** — the official [JS/TS client](https://docs.apify.com/api/client/js/docs.md): `npm install apify-client`.
- **Python projects** — the official [Python client](https://docs.apify.com/api/client/python/docs.md): `pip install apify-client`.
- **Any other language** — the [REST API](https://docs.apify.com/api/v2.md).

For usage examples, see the [API](#api) section below.

For more details, see Apify documentation as [Markdown index](https://docs.apify.com/llms.txt) and [Markdown full-text](https://docs.apify.com/llms-full.txt).

# README

"""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}")
```

# Actor input Schema

## `categories` (type: `array`):

List of category slugs to scrape. Leave empty to scrape ALL categories discovered from the site.

## `maxPagesPerCategory` (type: `integer`):

Maximum number of listing pages to crawl per category (0 = unlimited). Use to limit scope for testing.

## `maxConcurrency` (type: `integer`):

Maximum number of parallel browser pages (1-20). Lower values are stealthier.

## `proxy` (type: `object`):

Proxy settings for bypassing Cloudflare rate limits. STRONGLY RECOMMENDED for large runs (100+ listings).

## Actor input object example

```json
{
  "categories": [],
  "maxPagesPerCategory": 0,
  "maxConcurrency": 5,
  "proxy": {
    "useApifyProxy": true,
    "apifyProxyGroups": [
      "RESIDENTIAL"
    ]
  }
}
```

# API

You can run this Actor programmatically using our API. Below are code examples in JavaScript, Python, and CLI, as well as the OpenAPI specification and MCP server setup.

## JavaScript example

```javascript
import { ApifyClient } from 'apify-client';

// Initialize the ApifyClient with your Apify API token
// Replace the '<YOUR_API_TOKEN>' with your token
const client = new ApifyClient({
    token: '<YOUR_API_TOKEN>',
});

// Prepare Actor input
const input = {
    "categories": []
};

// Run the Actor and wait for it to finish
const run = await client.actor("devjonas/yellow-pages-malaysia").call(input);

// Fetch and print Actor results from the run's dataset (if any)
console.log('Results from dataset');
console.log(`💾 Check your data here: https://console.apify.com/storage/datasets/${run.defaultDatasetId}`);
const { items } = await client.dataset(run.defaultDatasetId).listItems();
items.forEach((item) => {
    console.dir(item);
});

// 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/js/docs

```

## Python example

```python
from apify_client import ApifyClient

# Initialize the ApifyClient with your Apify API token
# Replace '<YOUR_API_TOKEN>' with your token.
client = ApifyClient("<YOUR_API_TOKEN>")

# Prepare the Actor input
run_input = { "categories": [] }

# Run the Actor and wait for it to finish
run = client.actor("devjonas/yellow-pages-malaysia").call(run_input=run_input)

# Fetch and print Actor results from the run's dataset (if there are any)
print("💾 Check your data here: https://console.apify.com/storage/datasets/" + run["defaultDatasetId"])
for item in client.dataset(run["defaultDatasetId"]).iterate_items():
    print(item)

# 📚 Want to learn more 📖? Go to → https://docs.apify.com/api/client/python/docs/quick-start

```

## CLI example

```bash
echo '{
  "categories": []
}' |
apify call devjonas/yellow-pages-malaysia --silent --output-dataset

```

## MCP server setup

```json
{
    "mcpServers": {
        "apify": {
            "command": "npx",
            "args": [
                "mcp-remote",
                "https://mcp.apify.com/?tools=devjonas/yellow-pages-malaysia",
                "--header",
                "Authorization: Bearer <YOUR_API_TOKEN>"
            ]
        }
    }
}

```

## OpenAPI specification

Download the OpenAPI definition: https://api.apify.com/v2/acts/48jAKH9kOIPsZ6Flb/builds/8Xj44gkx9vqMJH4l0/openapi.json