Yelp Search

MIT License

Copyright (c) 2026 Letta, Inc.

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

#!/usr/bin/env python3
"""Get detailed information about a Yelp business."""

import argparse
import json
import os
import sys
import urllib.request

def load_api_key():
    """Load API key from .env file."""
    env_path = os.path.join(os.path.dirname(__file__), '..', '..', '..', '.env')
    env_path = os.path.abspath(env_path)
    
    if os.path.exists(env_path):
        with open(env_path) as f:
            for line in f:
                if line.startswith('YELP_API_KEY='):
                    return line.strip().split('=', 1)[1]
    
    return os.environ.get('YELP_API_KEY')

def get_business_details(business_id):
    """Get detailed business information."""
    api_key = load_api_key()
    if not api_key:
        print("Error: YELP_API_KEY not found", file=sys.stderr)
        sys.exit(1)
    
    url = f"https://api.yelp.com/v3/businesses/{business_id}"
    
    req = urllib.request.Request(url)
    req.add_header('Authorization', f'Bearer {api_key}')
    
    try:
        with urllib.request.urlopen(req) as response:
            return json.loads(response.read().decode())
    except urllib.error.HTTPError as e:
        print(f"API Error: {e.code} - {e.read().decode()}", file=sys.stderr)
        sys.exit(1)

def format_hours(hours_data):
    """Format business hours."""
    if not hours_data:
        return []
    
    days = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
    hours_list = hours_data[0].get('open', [])
    
    # Group by day
    day_hours = {i: [] for i in range(7)}
    for entry in hours_list:
        day_hours[entry['day']].append(entry)
    
    formatted = []
    for day_num, entries in day_hours.items():
        if entries:
            times = []
            for e in entries:
                start = f"{e['start'][:2]}:{e['start'][2:]}"
                end = f"{e['end'][:2]}:{e['end'][2:]}"
                times.append(f"{start}-{end}")
            formatted.append(f"{days[day_num]}: {', '.join(times)}")
        else:
            formatted.append(f"{days[day_num]}: Closed")
    
    return formatted

def main():
    parser = argparse.ArgumentParser(description='Get Yelp business details')
    parser.add_argument('business_id', help='Business ID or alias')
    parser.add_argument('--json', action='store_true', help='Output raw JSON')
    
    args = parser.parse_args()
    
    data = get_business_details(args.business_id)
    
    if args.json:
        print(json.dumps(data, indent=2))
        return
    
    print(f"=== {data.get('name', 'Unknown')} ===\n")
    
    print(f"⭐ Rating: {data.get('rating', 'N/A')} ({data.get('review_count', 0)} reviews)")
    print(f"💰 Price: {data.get('price', 'N/A')}")
    print(f"📞 Phone: {data.get('display_phone', data.get('phone', 'N/A'))}")
    
    location = data.get('location', {})
    address = '\n         '.join(location.get('display_address', ['N/A']))
    print(f"📍 Address: {address}")
    
    if location.get('cross_streets'):
        print(f"   Cross streets: {location['cross_streets']}")
    
    categories = ', '.join([c['title'] for c in data.get('categories', [])])
    print(f"📂 Categories: {categories}")
    
    hours = data.get('hours', [])
    if hours:
        is_open = hours[0].get('is_open_now', False)
        print(f"\n🕐 Currently: {'Open' if is_open else 'Closed'}")
        print("\nHours:")
        for line in format_hours(hours):
            print(f"  {line}")
    
    print(f"\n🔗 Yelp URL: {data.get('url', 'N/A')}")

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""Fetch Yelp reviews using browser automation."""

import asyncio
import argparse
import os
import sys

# Set up environment for browser-use
os.environ.setdefault("ANONYMIZED_TELEMETRY", "false")

async def get_reviews(business_name: str, location: str, num_reviews: int = 5):
    """Get reviews for a business using browser automation."""
    from browser_use import Agent, ChatOpenAI
    
    # Check for API key
    api_key = os.environ.get("OPENAI_API_KEY")
    if not api_key:
        # Try loading from .env
        env_path = os.path.join(os.path.dirname(__file__), '..', '..', '..', '.env')
        if os.path.exists(env_path):
            with open(env_path) as f:
                for line in f:
                    if line.startswith('OPENAI_API_KEY='):
                        api_key = line.strip().split('=', 1)[1]
                        os.environ["OPENAI_API_KEY"] = api_key
                        break
    
    if not api_key:
        print("Error: OPENAI_API_KEY not found. Set it in environment or .env file.", file=sys.stderr)
        sys.exit(1)
    
    # Use browser-use's ChatOpenAI wrapper
    llm = ChatOpenAI(model="gpt-4o-mini")
    
    task = f"""
    Go to Yelp.com and search for "{business_name}" in "{location}".
    Click on the business in the search results.
    Find and read the reviews section.
    Extract the first {num_reviews} reviews, including:
    - Reviewer name
    - Star rating
    - Date
    - Review text (full text if possible)
    
    Return the reviews in a structured format.
    """
    
    agent = Agent(
        task=task,
        llm=llm,
    )
    
    result = await agent.run()
    return result

def main():
    parser = argparse.ArgumentParser(description='Get Yelp reviews using browser automation')
    parser.add_argument('business', help='Business name to search for')
    parser.add_argument('--location', '-l', default='San Francisco', help='Location (default: San Francisco)')
    parser.add_argument('--num-reviews', '-n', type=int, default=5, help='Number of reviews to fetch (default: 5)')
    
    args = parser.parse_args()
    
    print(f"Fetching reviews for '{args.business}' in {args.location}...")
    print("(This may take a minute as the browser navigates Yelp)\n")
    
    result = asyncio.run(get_reviews(args.business, args.location, args.num_reviews))
    
    print("\n=== Reviews ===\n")
    print(result)

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""Search Yelp for a business by phone number."""

import argparse
import json
import os
import sys
import urllib.request

def load_api_key():
    """Load API key from .env file."""
    env_path = os.path.join(os.path.dirname(__file__), '..', '..', '..', '.env')
    env_path = os.path.abspath(env_path)
    
    if os.path.exists(env_path):
        with open(env_path) as f:
            for line in f:
                if line.startswith('YELP_API_KEY='):
                    return line.strip().split('=', 1)[1]
    
    return os.environ.get('YELP_API_KEY')

def search_by_phone(phone):
    """Search for a business by phone number."""
    api_key = load_api_key()
    if not api_key:
        print("Error: YELP_API_KEY not found", file=sys.stderr)
        sys.exit(1)
    
    # Ensure phone is in E.164 format
    phone = phone.replace(' ', '').replace('-', '').replace('(', '').replace(')', '')
    if not phone.startswith('+'):
        phone = '+1' + phone  # Assume US if no country code
    
    url = f"https://api.yelp.com/v3/businesses/search/phone?phone={phone}"
    
    req = urllib.request.Request(url)
    req.add_header('Authorization', f'Bearer {api_key}')
    
    try:
        with urllib.request.urlopen(req) as response:
            return json.loads(response.read().decode())
    except urllib.error.HTTPError as e:
        print(f"API Error: {e.code} - {e.read().decode()}", file=sys.stderr)
        sys.exit(1)

def main():
    parser = argparse.ArgumentParser(description='Search Yelp by phone number')
    parser.add_argument('phone', help='Phone number (e.g., +14155551234)')
    parser.add_argument('--json', action='store_true', help='Output raw JSON')
    
    args = parser.parse_args()
    
    data = search_by_phone(args.phone)
    
    if args.json:
        print(json.dumps(data, indent=2))
        return
    
    businesses = data.get('businesses', [])
    
    if not businesses:
        print(f"No business found for phone: {args.phone}")
        return
    
    print(f"Found {len(businesses)} business(es) for {args.phone}:\n")
    
    for biz in businesses:
        name = biz.get('name', 'Unknown')
        rating = biz.get('rating', 'N/A')
        review_count = biz.get('review_count', 0)
        
        location = biz.get('location', {})
        address = ', '.join(location.get('display_address', ['N/A']))
        
        categories = ', '.join([c['title'] for c in biz.get('categories', [])])
        
        print(f"📍 {name}")
        print(f"   ⭐ {rating} ({review_count} reviews)")
        print(f"   📍 {address}")
        print(f"   📂 {categories}")
        print(f"   🔗 {biz.get('url', 'N/A')}")
        print()

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""Scrape Yelp reviews using Browserbase (handles CAPTCHA with proxies)."""

import argparse
import json
import os
import sys

def load_env():
    """Load environment variables from .env file."""
    # Try multiple paths
    for env_path in [
        os.path.join(os.path.dirname(__file__), '..', '..', '..', '.env'),
        os.path.join(os.getcwd(), '.env'),
    ]:
        if os.path.exists(env_path):
            with open(env_path) as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#') and '=' in line:
                        key, value = line.split('=', 1)
                        os.environ[key] = value
            return True
    return False

def scrape_reviews(url: str, num_reviews: int = 5):
    """Scrape reviews from a Yelp business page using Browserbase."""
    from playwright.sync_api import sync_playwright
    from browserbase import Browserbase
    
    load_env()
    
    api_key = os.environ.get("BROWSERBASE_API_KEY")
    project_id = os.environ.get("BROWSERBASE_PROJECT_ID")
    
    if not api_key or not project_id:
        print("Error: BROWSERBASE_API_KEY and BROWSERBASE_PROJECT_ID required", file=sys.stderr)
        return None
    
    bb = Browserbase(api_key=api_key)
    
    with sync_playwright() as p:
        print("Creating Browserbase session with proxies...", file=sys.stderr)
        session = bb.sessions.create(
            project_id=project_id,
            proxies=True
        )
        
        browser = p.chromium.connect_over_cdp(session.connect_url)
        context = browser.contexts[0]
        page = context.pages[0]
        
        try:
            print(f"Loading {url}...", file=sys.stderr)
            page.goto(url, timeout=45000)
            page.wait_for_timeout(3000)
            
            # Extract business name
            name = "Unknown"
            h1 = page.locator("h1")
            if h1.count() > 0:
                name = h1.first.text_content() or "Unknown"
            
            # Extract rating
            rating = None
            rating_el = page.locator("[aria-label*='star rating']")
            if rating_el.count() > 0:
                rating = rating_el.first.get_attribute("aria-label")
            
            # Extract reviews
            reviews = []
            seen_texts = set()
            
            # Try p[class*='comment'] first
            review_els = page.locator("p[class*='comment']")
            count = review_els.count()
            
            for i in range(count):
                if len(reviews) >= num_reviews:
                    break
                try:
                    text = review_els.nth(i).text_content()
                    if text and len(text) > 50:
                        text_key = text[:100]
                        if text_key not in seen_texts:
                            seen_texts.add(text_key)
                            reviews.append({"text": text.strip()})
                except:
                    continue
            
            print(f"Extracted {len(reviews)} reviews", file=sys.stderr)
            
        finally:
            browser.close()
            print(f"Session: https://browserbase.com/sessions/{session.id}", file=sys.stderr)
        
        return {"business": name.strip(), "rating": rating, "reviews": reviews}

def main():
    parser = argparse.ArgumentParser(description='Scrape Yelp reviews via Browserbase')
    parser.add_argument('url', help='Yelp business URL')
    parser.add_argument('--num-reviews', '-n', type=int, default=5)
    parser.add_argument('--json', action='store_true', help='Output as JSON')
    
    args = parser.parse_args()
    
    result = scrape_reviews(args.url, args.num_reviews)
    
    if not result:
        sys.exit(1)
    
    if args.json:
        print(json.dumps(result, indent=2))
    else:
        print(f"\n=== {result['business']} ===")
        if result['rating']:
            print(f"Rating: {result['rating']}")
        print()
        for i, review in enumerate(result['reviews'], 1):
            text = review['text']
            if len(text) > 300:
                text = text[:300] + "..."
            print(f"[{i}] {text}\n")

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""Search Yelp for businesses by term and location."""

import argparse
import json
import os
import sys
from urllib.parse import urlencode
import urllib.request

def load_api_key():
    """Load API key from .env file."""
    env_path = os.path.join(os.path.dirname(__file__), '..', '..', '..', '.env')
    env_path = os.path.abspath(env_path)
    
    if os.path.exists(env_path):
        with open(env_path) as f:
            for line in f:
                if line.startswith('YELP_API_KEY='):
                    return line.strip().split('=', 1)[1]
    
    # Fall back to environment variable
    return os.environ.get('YELP_API_KEY')

def search_businesses(term, location=None, latitude=None, longitude=None, 
                     limit=5, sort_by='best_match', price=None):
    """Search for businesses on Yelp."""
    api_key = load_api_key()
    if not api_key:
        print("Error: YELP_API_KEY not found in .env or environment", file=sys.stderr)
        sys.exit(1)
    
    params = {'term': term, 'limit': limit, 'sort_by': sort_by}
    
    if location:
        params['location'] = location
    elif latitude and longitude:
        params['latitude'] = latitude
        params['longitude'] = longitude
    else:
        print("Error: Must provide --location or --latitude/--longitude", file=sys.stderr)
        sys.exit(1)
    
    if price:
        params['price'] = price
    
    url = f"https://api.yelp.com/v3/businesses/search?{urlencode(params)}"
    
    req = urllib.request.Request(url)
    req.add_header('Authorization', f'Bearer {api_key}')
    
    try:
        with urllib.request.urlopen(req) as response:
            data = json.loads(response.read().decode())
    except urllib.error.HTTPError as e:
        print(f"API Error: {e.code} - {e.read().decode()}", file=sys.stderr)
        sys.exit(1)
    
    return data

def format_hours(hours_data):
    """Format business hours for display."""
    if not hours_data:
        return "Hours not available"
    
    days = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
    hours_list = hours_data[0].get('open', [])
    
    formatted = []
    for entry in hours_list:
        day = days[entry['day']]
        start = f"{entry['start'][:2]}:{entry['start'][2:]}"
        end = f"{entry['end'][:2]}:{entry['end'][2:]}"
        formatted.append(f"{day}: {start}-{end}")
    
    return ', '.join(formatted) if formatted else "Hours not available"

def main():
    parser = argparse.ArgumentParser(description='Search Yelp for businesses')
    parser.add_argument('term', help='Search term (e.g., "dog groomer", "pizza")')
    parser.add_argument('--location', '-l', help='Location (e.g., "San Francisco, CA")')
    parser.add_argument('--latitude', type=float, help='Latitude for location-based search')
    parser.add_argument('--longitude', type=float, help='Longitude for location-based search')
    parser.add_argument('--limit', '-n', type=int, default=5, help='Number of results (default: 5)')
    parser.add_argument('--sort-by', choices=['best_match', 'rating', 'review_count', 'distance'],
                       default='best_match', help='Sort order')
    parser.add_argument('--price', help='Price filter (1,2,3,4 for $-$$$$)')
    parser.add_argument('--json', action='store_true', help='Output raw JSON')
    
    args = parser.parse_args()
    
    data = search_businesses(
        term=args.term,
        location=args.location,
        latitude=args.latitude,
        longitude=args.longitude,
        limit=args.limit,
        sort_by=args.sort_by,
        price=args.price
    )
    
    if args.json:
        print(json.dumps(data, indent=2))
        return
    
    businesses = data.get('businesses', [])
    total = data.get('total', 0)
    
    print(f"Found {total} results for \"{args.term}\"")
    print(f"Showing top {len(businesses)}:\n")
    
    for i, biz in enumerate(businesses, 1):
        name = biz.get('name', 'Unknown')
        rating = biz.get('rating', 'N/A')
        review_count = biz.get('review_count', 0)
        price = biz.get('price', 'N/A')
        phone = biz.get('phone', 'N/A')
        display_phone = biz.get('display_phone', phone)
        
        location = biz.get('location', {})
        address = ', '.join(location.get('display_address', ['Address not available']))
        
        distance = biz.get('distance')
        distance_str = f"{distance/1609.34:.1f} mi" if distance else "N/A"
        
        hours = biz.get('business_hours', [])
        is_open = hours[0].get('is_open_now', False) if hours else None
        open_status = "Open now" if is_open else ("Closed" if is_open is False else "")
        
        categories = ', '.join([c['title'] for c in biz.get('categories', [])])
        
        print(f"[{i}] {name}")
        print(f"    ⭐ {rating} ({review_count} reviews) | {price} | {distance_str}")
        print(f"    📍 {address}")
        print(f"    📞 {display_phone}")
        if open_status:
            print(f"    🕐 {open_status}")
        if categories:
            print(f"    📂 {categories}")
        print()

if __name__ == '__main__':
    main()