serp useless with num_results params

[ycq0125]

import asyncio
from brightdata import BrightDataClient

token = "xxxxx"


async def brightdata_google_search():
    async with BrightDataClient(token=token) as client:
        result = await client.search.google(query="homeless", num_results=100, language="en", location="United States")
        print(result.to_json())
        for item in result.data:
            print(item)

(.venv) xxx in ~/Desktop/serp-service on main ● ● ● λ pip show brightdata-sdk
Name: brightdata-sdk
Version: 2.1.1

[ycq0125]

import asyncio
from brightdata import BrightDataClient

token = "xxxxx"


async def brightdata_google_search():
    async with BrightDataClient(token=token) as client:
        result = await client.search.google(query="homeless", num_results=100, language="en", location="United States")
        print(result.to_json())
        for item in result.data:
            print(item)

(.venv) xxx in ~/Desktop/serp-service on main ● ● ● λ pip show brightdata-sdk
Name: brightdata-sdk
Version: 2.1.1

patch：

#!/usr/bin/env python 
# -*- coding: utf-8 -*-
"""
@File    : test.py
@Project : serp-service
@Software: PyCharm
@Time    : 2026/1/22 17:02
"""
import time
from brightdata.models import SearchEngineType, SearchResult
from loguru import logger
from urllib.parse import urlencode, urlparse, parse_qs, urlunparse
from brightdata.api.serp.data_normalizer import GoogleDataNormalizer
from brightdata.api.serp.url_builder import BaseURLBuilder
from brightdata.api.serp.base import BaseSERPService
import os
import aiohttp
import json
from typing import Union, List, Dict, Any
from datetime import datetime, timezone
from brightdata.utils.retry import retry_with_backoff
from brightdata.utils.validation import validate_zone_name
from brightdata.utils.function_detection import get_caller_function_name
from brightdata.core.engine import AsyncEngine
from brightdata.core.zone_manager import ZoneManager
from brightdata.constants import (
    HTTP_OK,
)
from brightdata.exceptions import ValidationError, AuthenticationError
from typing import Optional
from urllib.parse import quote_plus
from brightdata.utils.location import LocationService, LocationFormat


class FixGoogleURLBuilder(BaseURLBuilder):
    """
    URL builder for Google search.
    Handles standard parameters and Bright Data specific flags.
    """

    def build(
            self,
            query: str,
            location: Optional[str] = None,
            language: str = "en",
            device: str = "desktop",
            num_results: int = 10,
            start: int = 0,
            **kwargs,
    ) -> str:
        """Build Google search URL with Bright Data parsing enabled."""
        encoded_query = quote_plus(query)
        url = f"https://www.google.com/search?q={encoded_query}"

        # Google's default is 10 results per page.
        # We append the 'start' parameter to handle pagination offsets.
        if start > 0:
            url += f"&start={start}"

        # Enable Bright Data SERP parsing (returns JSON)
        url += "&brd_json=1"

        if language:
            url += f"&hl={language}"

        if location:
            location_code = LocationService.parse_location(location, LocationFormat.GOOGLE)
            if location_code:
                url += f"&gl={location_code}"

        if device == "mobile":
            url += "&mobileaction=1"

        if "safe_search" in kwargs:
            url += f"&safe={'active' if kwargs['safe_search'] else 'off'}"

        if "time_range" in kwargs:
            url += f"&tbs=qdr:{kwargs['time_range']}"

        return url

    @staticmethod
    def build_next_page_url(next_link: str, language: str, location: Optional[str]) -> str:
        """
        Injects necessary parameters into the next page link safely.
        Ensures the 'brd_json=1' flag persists across pagination.
        """
        parsed_url = urlparse(next_link)
        query_params = parse_qs(parsed_url.query)

        # Ensure Bright Data parsing is enabled for the next link
        query_params["brd_json"] = ["1"]

        if language:
            query_params["hl"] = [language]
        if location:
            code = LocationService.parse_location(location, LocationFormat.GOOGLE)
            if code:
                query_params["gl"] = [code]

        new_query = urlencode(query_params, doseq=True)
        return urlunparse((
            parsed_url.scheme, parsed_url.netloc, parsed_url.path,
            parsed_url.params, new_query, parsed_url.fragment
        ))


class FixGoogleSERPService(BaseSERPService):
    """
    Google Search Engine Results Page service.

    Enhanced with 'concurrent_pagination' to fetch multiple SERP pages in parallel
    rather than sequentially following 'next' links.
    """

    SEARCH_ENGINE: SearchEngineType = "google"

    def __init__(
            self,
            engine: AsyncEngine,
            timeout: Optional[int] = None,
            max_retries: int = 3,
    ):
        """Initialize Google SERP service."""
        url_builder = FixGoogleURLBuilder()
        data_normalizer = GoogleDataNormalizer()
        super().__init__(
            engine=engine,
            url_builder=url_builder,
            data_normalizer=data_normalizer,
            timeout=timeout,
            max_retries=max_retries,
        )

    async def search(
            self,
            query: Union[str, List[str]],
            zone: str,
            location: Optional[str] = None,
            language: str = "en",
            device: str = "desktop",
            num_results: int = 10,
            concurrent_pagination: bool = False,
            **kwargs,
    ) -> Union[SearchResult, List[SearchResult]]:
        """
        Perform search asynchronously with optional concurrent pagination.
        """
        is_single = isinstance(query, str)
        query_list = [query] if is_single else query

        self._validate_zone(zone)
        self._validate_queries(query_list)

        if len(query_list) == 1:
            return await self._search_single_async(
                query=query_list[0],
                zone=zone,
                location=location,
                language=language,
                device=device,
                num_results=num_results,
                concurrent_pagination=concurrent_pagination,
                **kwargs,
            )
        else:
            return await self._search_multiple_async(
                queries=query_list,
                zone=zone,
                location=location,
                language=language,
                device=device,
                num_results=num_results,
                concurrent_pagination=concurrent_pagination,
                **kwargs,
            )

    async def _search_single_async(
            self,
            query: str,
            zone: str,
            location: Optional[str],
            language: str,
            device: str,
            num_results: int,
            concurrent_pagination: bool = False,
            **kwargs,
    ) -> SearchResult:
        """
        Execute a single search query.

        Logic Flow:
        1. Request Page 1 sequentially.
        2. Analyze Page 1 metadata to determine the 'step' (results per page).
        3. If concurrent_pagination is True: Calculate offsets and fetch remaining pages in parallel.
        4. If concurrent_pagination is False: Fallback to sequential 'next link' following.
        """
        trigger_sent_at = datetime.now(timezone.utc)

        logger.info(
            f"Starting search for query='{query}'. Target results={num_results}, Concurrent={concurrent_pagination}"
        )

        # 1. Build Initial URL for Page 1 (start=0)
        # We initially request the full num_results, though Google usually caps per page.
        initial_url = self.url_builder.build(
            query=query,
            location=location,
            language=language,
            device=device,
            num_results=num_results,
            start=0,
            **kwargs,
        )

        base_payload = {
            "zone": zone,
            "format": "json",
            "method": "GET",
        }

        sdk_function = get_caller_function_name()
        if sdk_function:
            base_payload["sdk_function"] = sdk_function

        aggregated_data: List[Dict[str, Any]] = []
        total_organic_count = 0

        # --- STEP 1: Fetch Page 1 Sequentially ---
        # Essential to determine the actual layout and pagination step size.

        logger.debug(f"Fetching Page 1 for query='{query}'...")

        first_page_payload = {**base_payload, "url": initial_url}
        result = await retry_with_backoff(
            lambda: self._execute_http_request(
                url=f"{self.engine.BASE_URL}{self.ENDPOINT}",
                payload=first_page_payload,
                query_context={"q": query, "location": location, "language": language},
                trigger_time=trigger_sent_at
            ),
            max_retries=self.max_retries,
        )

        if not result.success:
            logger.error(f"Failed to fetch Page 1 for query='{query}'. Error: {result.error}")
            return result

        page_data = result.data or {}
        organic_results = page_data.get("organic", [])
        aggregated_data.append(page_data)
        count_this_page = len(organic_results)
        total_organic_count += count_this_page

        logger.info(
            f"Page 1 fetched for query='{query}'. Found {count_this_page} items. "
            f"Total so far: {total_organic_count}/{num_results}."
        )

        # If we have enough results or no results were found, stop here.
        if total_organic_count >= num_results or count_this_page == 0:
            logger.info(f"Target count reached or no results for query='{query}'. Finishing.")
            return self._build_final_result(query, location, language, aggregated_data, total_organic_count,
                                            trigger_sent_at, num_results)

        # --- STEP 2: Concurrent Pagination Logic ---
        if concurrent_pagination:
            # 2a. Determine Pagination Step (Results Per Page)
            pagination_info = page_data.get("pagination", {})
            next_start = pagination_info.get("next_page_start")

            step = 10  # Fallback default
            if next_start:
                try:
                    step = int(next_start)
                    # If step is 0 (unlikely), fallback to the count of items on current page or 10
                    if step == 0:
                        step = count_this_page or 10
                except (ValueError, TypeError):
                    step = count_this_page or 10
            else:
                step = count_this_page or 10

            logger.info(f"Concurrent pagination engaged for query='{query}'. Step size: {step}")

            # 2b. Generate Tasks for Remaining Pages
            tasks = []
            current_offset = step  # Start from Page 2 (offset = step)

            # Loop until we cover the requested num_results
            while current_offset < num_results:
                # We request 'step' number of results for subsequent pages to mimic natural browsing
                page_req_num = step

                page_url = self.url_builder.build(
                    query=query,
                    location=location,
                    language=language,
                    device=device,
                    num_results=page_req_num,
                    start=current_offset,  # Explicit offset
                    **kwargs,
                )

                payload = {**base_payload, "url": page_url}

                logger.debug(
                    f"Queueing task for query='{query}': offset={current_offset}, limit={page_req_num}"
                )

                tasks.append(
                    retry_with_backoff(
                        lambda: self._execute_http_request(
                            url=f"{self.engine.BASE_URL}{self.ENDPOINT}",
                            payload=payload,
                            query_context={"q": query, "offset": current_offset},
                            trigger_time=trigger_sent_at
                        ),
                        max_retries=self.max_retries
                    )
                )

                current_offset += step

            # 2c. Execute Tasks Concurrently
            if tasks:
                logger.info(f"Firing {len(tasks)} concurrent requests for query='{query}'...")
                results = await asyncio.gather(*tasks, return_exceptions=True)

                success_count = 0
                for i, res in enumerate(results):
                    if isinstance(res, Exception):
                        logger.error(f"Concurrent task {i} for query='{query}' failed with exception: {res}")
                        continue

                    if res.success:
                        p_data = res.data or {}
                        count = len(p_data.get("organic", []))
                        if count > 0:
                            aggregated_data.append(p_data)
                            total_organic_count += count
                            success_count += 1
                        else:
                            logger.debug(f"Concurrent task {i} for query='{query}' returned 0 results.")
                    else:
                        logger.warning(f"Concurrent task {i} for query='{query}' failed: {res.error}")

                logger.info(
                    f"Concurrent batch finished for query='{query}'. "
                    f"Successful pages: {success_count}. Total items: {total_organic_count}"
                )

        # --- STEP 3: Sequential Fallback (If Concurrent is OFF) ---
        elif total_organic_count < num_results:
            logger.info(f"Concurrent pagination OFF for query='{query}'. Falling back to sequential.")

            current_url = None
            next_link = page_data.get("pagination", {}).get("next_page_link")

            if next_link:
                # Safely reconstruct the next link ensuring Bright Data params are present
                current_url = self.url_builder.build_next_page_url(next_link, language, location)  # noqa

            loop_limit = 20  # Safety break

            while current_url and total_organic_count < num_results and loop_limit > 0:
                loop_limit -= 1
                logger.debug(f"Fetching next sequential page for query='{query}'. Limit remaining: {loop_limit}")

                current_payload = {**base_payload, "url": current_url}

                res = await retry_with_backoff(
                    lambda: self._execute_http_request(
                        url=f"{self.engine.BASE_URL}{self.ENDPOINT}",
                        payload=current_payload,
                        query_context={"q": query, "page_seq": "next"},
                        trigger_time=trigger_sent_at
                    ),
                    max_retries=self.max_retries,
                )

                if not res.success:
                    logger.warning(f"Sequential page fetch failed for query='{query}': {res.error}")
                    break

                p_data = res.data or {}
                items = p_data.get("organic", [])
                if not items:
                    logger.debug(f"Sequential page returned no items for query='{query}'. Stopping.")
                    break

                aggregated_data.append(p_data)
                total_organic_count += len(items)

                next_link = p_data.get("pagination", {}).get("next_page_link")
                if not next_link:
                    logger.debug(f"No next page link found for query='{query}'. Stopping.")
                    break

                current_url = self.url_builder.build_next_page_url(next_link, language, location)  # noqa

        # Build and return the final aggregated result
        return self._build_final_result(query, location, language, aggregated_data, total_organic_count,
                                        trigger_sent_at, num_results)

    def _build_final_result(self, query, location, language, data, count, trigger_time, num_results):
        """Helper to construct the final SearchResult object from aggregated data."""
        return SearchResult(
            success=True,
            query={"q": query, "location": location, "language": language},
            data=data,
            total_found=count,
            search_engine=self.SEARCH_ENGINE,
            country=location,
            results_per_page=num_results,
            trigger_sent_at=trigger_time,
            data_fetched_at=datetime.now(timezone.utc),
        )

    async def _execute_http_request(
            self,
            url: str,
            payload: Dict[str, Any],
            query_context: Dict[str, Any],
            trigger_time: datetime
    ) -> SearchResult:
        """
        Executes the actual HTTP POST request to Bright Data and handles response parsing.
        """
        async with self.engine.post_to_url(
                url,
                json_data=payload,
                timeout=aiohttp.ClientTimeout(total=self.timeout),
        ) as response:
            data_fetched_at = datetime.now(timezone.utc)

            if response.status != 200:
                error_text = await response.text()
                return SearchResult(
                    success=False,
                    query=query_context,
                    error=f"HTTP {response.status}: {error_text}",
                    search_engine=self.SEARCH_ENGINE,
                    trigger_sent_at=trigger_time,
                    data_fetched_at=data_fetched_at,
                )

            raw_text = await response.text()
            parsed_data = self._parse_api_response(raw_text)

            # Note: total_found here refers only to the count on this specific page
            return SearchResult(
                success=True,
                query=query_context,
                data=parsed_data,
                total_found=len(parsed_data.get("organic", [])),
                search_engine=self.SEARCH_ENGINE,
                trigger_sent_at=trigger_time,
                data_fetched_at=data_fetched_at,
            )

    @staticmethod
    def _parse_api_response(text: str):
        """
        Parses the raw API response.
        Handles cases where Bright Data wraps the response in a JSON object
        containing a 'body' field (which might be HTML or nested JSON).
        """
        try:
            data = json.loads(text)
        except json.JSONDecodeError:
            # Fallback: treat distinct non-JSON response as raw HTML wrapper
            return {"raw_html": text}

        # Check for wrapped response format (e.g. Bright Data wrapping)
        if isinstance(data, dict) and "body" in data and "status_code" in data:
            body = data.get("body", "")
            status_code = data.get("status_code")

            if isinstance(body, str):
                body_stripped = body.strip()
                if body_stripped.startswith("<") or body_stripped.startswith("<!DOCTYPE"):
                    # It's HTML string inside JSON
                    return {"body": body, "status_code": status_code}

                # Try parsing body as nested JSON string
                try:
                    return json.loads(body)
                except (json.JSONDecodeError, TypeError):
                    pass

            # If body isn't a string or failed to parse as JSON, return as is
            return {"body": body, "status_code": status_code}

        return data

    async def _search_multiple_async(
            self,
            queries: List[str],
            zone: str,
            location: Optional[str],
            language: str,
            device: str,
            num_results: int,
            **kwargs,
    ) -> List[SearchResult]:
        """Execute multiple search queries concurrently."""

        # Extract concurrent_pagination param, defaulting to False
        concurrent_pagination = kwargs.pop("concurrent_pagination", False)

        tasks = [
            self._search_single_async(
                query=q,
                zone=zone,
                location=location,
                language=language,
                device=device,
                num_results=num_results,
                concurrent_pagination=concurrent_pagination,
                **kwargs,
            )
            for q in queries
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        processed_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                logger.error(f"Search failed for query '{queries[i]}': {result}")
                processed_results.append(
                    SearchResult(
                        success=False,
                        query={"q": queries[i]},
                        error=f"Exception: {str(result)}",
                        search_engine=self.SEARCH_ENGINE,
                        trigger_sent_at=datetime.now(timezone.utc),
                        data_fetched_at=datetime.now(timezone.utc),
                    )
                )
            else:
                processed_results.append(result)

        return processed_results

    def _validate_queries(self, queries: List[str]) -> None:
        """Validate search queries."""
        if not queries:
            raise ValidationError("Query list cannot be empty")

        for query in queries:
            if not query or not isinstance(query, str):
                raise ValidationError(f"Invalid query: {query}. Must be non-empty string.")

    def _validate_zone(self, zone: str) -> None:
        """
        Validate zone name format.
        Does not check if the zone actually exists in the account, only format.
        """
        validate_zone_name(zone)


class FixSearchService:
    """
    Search service namespace (SERP API).

    Provides access to search engine result scrapers with normalized
    data across different search engines.
    """

    def __init__(self, client: "FixBrightDataClient"):
        """Initialize search service with client reference."""
        self._client = client
        self._google_service: Optional["FixGoogleSERPService"] = None

    async def google(
            self,
            query: Union[str, List[str]],
            location: Optional[str] = None,
            language: str = "en",
            device: str = "desktop",
            num_results: int = 10,
            zone: Optional[str] = None,
            **kwargs,
    ) -> Union[SearchResult, List[SearchResult]]:
        """
        Search Google asynchronously.

        This base method typically delegates to the specific SERP service.
        """

        if self._google_service is None:
            self._google_service = FixGoogleSERPService(
                engine=self._client.engine,
                timeout=self._client.timeout,
            )

        zone = zone or self._client.serp_zone
        return await self._google_service.search(
            query=query,
            zone=zone,
            location=location,
            language=language,
            device=device,
            num_results=num_results,
            **kwargs,
        )


class FixedSearchService(FixSearchService):
    """
    Custom SearchService wrapper.
    Overrides the google method to support the 'concurrent_pagination' flag.
    """

    async def google(
            self,
            query: Union[str, List[str]],
            location: Optional[str] = None,
            language: str = "en",
            device: str = "desktop",
            num_results: int = 10,
            zone: Optional[str] = None,
            concurrent_pagination: bool = False,
            **kwargs,
    ) -> Union[SearchResult, List[SearchResult]]:
        if self._google_service is None:
            self._google_service = FixGoogleSERPService(
                engine=self._client.engine,
                timeout=self._client.timeout,
            )

        zone = zone or self._client.serp_zone

        return await self._google_service.search(
            query=query,
            zone=zone,
            location=location,
            language=language,
            device=device,
            num_results=num_results,
            concurrent_pagination=concurrent_pagination,
            **kwargs,
        )


class FixBrightDataClient:
    """
    Main entry point for Bright Data SDK.

    Provides unified access to:
    - Web Unlocker
    - SERP API (Search)
    - Browser API
    """

    # Default configuration
    DEFAULT_TIMEOUT = 15
    DEFAULT_WEB_UNLOCKER_ZONE = "sdk_unlocker"
    DEFAULT_SERP_ZONE = "sdk_serp"
    DEFAULT_BROWSER_ZONE = "sdk_browser"

    # Environment variable name for API token
    TOKEN_ENV_VAR = "BRIGHTDATA_API_TOKEN"

    def __init__(
            self,
            token: Optional[str] = None,
            timeout: int = DEFAULT_TIMEOUT,
            web_unlocker_zone: Optional[str] = None,
            serp_zone: Optional[str] = None,
            browser_zone: Optional[str] = None,
            auto_create_zones: bool = True,
            validate_token: bool = False,
            rate_limit: Optional[float] = None,
            rate_period: float = 1.0,
    ):
        """
        Initialize the Client.

        Args:
            token: API Token (optional if ENV var is set).
            auto_create_zones: If True, attempts to create missing zones on startup.
            validate_token: If True, tests connection upon initialization.
        """
        self.token = self._load_token(token)
        self.timeout = timeout
        self.web_unlocker_zone = web_unlocker_zone or self.DEFAULT_WEB_UNLOCKER_ZONE
        self.serp_zone = serp_zone or self.DEFAULT_SERP_ZONE
        self.browser_zone = browser_zone or self.DEFAULT_BROWSER_ZONE
        self.auto_create_zones = auto_create_zones

        self.engine = AsyncEngine(
            self.token, timeout=timeout, rate_limit=rate_limit, rate_period=rate_period
        )

        self._search_service: Optional[FixedSearchService] = None
        self._zone_manager: Optional[ZoneManager] = None
        self._is_connected = False
        self._zones_ensured = False

        self._validate_token_on_enter = validate_token

    def _ensure_initialized(self) -> None:
        """Ensure client is properly initialized (used as context manager)."""
        if self.engine._session is None:
            raise RuntimeError(
                "BrightDataClient not initialized. "
                "Use: async with BrightDataClient() as client: ..."
            )

    def _load_token(self, token: Optional[str]) -> str:
        """Load token from parameter or environment variable."""
        if token:
            if not isinstance(token, str) or len(token.strip()) < 10:
                raise ValidationError(
                    f"Invalid token format. Token must be a string with at least 10 characters."
                )
            return token.strip()

        env_token = os.getenv(self.TOKEN_ENV_VAR)
        if env_token:
            return env_token.strip()

        raise ValidationError(
            f"API token required but not found.\n"
            f"Set environment variable: {self.TOKEN_ENV_VAR}"
        )

    @property
    def search(self) -> FixedSearchService:
        """
        Access search services (SERP API).
        """
        if self._search_service is None:
            self._search_service = FixedSearchService(self)
        return self._search_service

    async def test_connection(self) -> bool:
        """
        Test API connection and token validity.
        Returns False on failure instead of raising exceptions.
        """
        self._ensure_initialized()
        try:
            async with self.engine.get_from_url(
                    f"{self.engine.BASE_URL}/zone/get_active_zones"
            ) as response:
                if response.status == HTTP_OK:
                    self._is_connected = True
                    return True
                else:
                    self._is_connected = False
                    return False

        except (asyncio.TimeoutError, OSError, Exception):
            self._is_connected = False
            return False

    async def _ensure_zones(self) -> None:
        """Ensure required zones exist if auto_create_zones is enabled."""
        if self._zones_ensured or not self.auto_create_zones:
            return

        if self._zone_manager is None:
            self._zone_manager = ZoneManager(self.engine)

        await self._zone_manager.ensure_required_zones(
            web_unlocker_zone=self.web_unlocker_zone,
            serp_zone=self.serp_zone,
            browser_zone=None,
        )
        self._zones_ensured = True

    async def __aenter__(self):
        """Async context manager entry."""
        await self.engine.__aenter__()

        if self._validate_token_on_enter:
            is_valid = await self.test_connection()
            if not is_valid:
                await self.engine.__aexit__(None, None, None)
                raise AuthenticationError("Token validation failed. Please check your API token.")

        await self._ensure_zones()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit."""
        await self.engine.__aexit__(exc_type, exc_val, exc_tb)

    def __repr__(self) -> str:
        token_preview = f"{self.token[:10]}...{self.token[-5:]}" if self.token else "None"
        status = "Connected" if self._is_connected else "Not tested"
        return f"<BrightDataClient token={token_preview} status='{status}'>"


if __name__ == '__main__':
    import asyncio

    token = "xxx"
    serp_zone = "xxxx"


    async def brightdata_google_search():
        async with FixBrightDataClient(token=token, serp_zone=serp_zone) as client:
            start = time.time()
            query = "homeless"
            result = await client.search.google(
                query=query, num_results=40, language="en", location="United States", concurrent_pagination=True
            )
            print(result.to_json())
            # for item in result.data:
            #     print(item)
            end = time.time()
            print(end - start)


    asyncio.run(brightdata_google_search())