auto-claude: subtask-2-1 - Add GooglePlacesSearcher class to social_media_audit.py
Implements GooglePlacesSearcher class with: - find_place() method: searches for business by name and city using Google Places findplacefromtext API - get_place_details() method: retrieves rating, review count, opening hours, business status, phone, and website Features: - Uses GOOGLE_PLACES_API_KEY environment variable - Comprehensive error handling (timeout, request errors) - Polish language locale support - Follows existing BraveSearcher class pattern Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
1426a73e0d
commit
4110ef63b5
@ -439,6 +439,207 @@ class WebsiteAuditor:
|
||||
return result
|
||||
|
||||
|
||||
class GooglePlacesSearcher:
|
||||
"""Search for Google Business profiles using Google Places API."""
|
||||
|
||||
# Google Places API configuration
|
||||
FIND_PLACE_URL = 'https://maps.googleapis.com/maps/api/place/findplacefromtext/json'
|
||||
PLACE_DETAILS_URL = 'https://maps.googleapis.com/maps/api/place/details/json'
|
||||
|
||||
def __init__(self, api_key: Optional[str] = None):
|
||||
"""
|
||||
Initialize GooglePlacesSearcher.
|
||||
|
||||
Args:
|
||||
api_key: Google Places API key. Falls back to GOOGLE_PLACES_API_KEY env var.
|
||||
"""
|
||||
self.api_key = api_key or os.getenv('GOOGLE_PLACES_API_KEY')
|
||||
self.session = requests.Session()
|
||||
self.session.headers.update({'User-Agent': USER_AGENT})
|
||||
|
||||
def find_place(self, company_name: str, city: str = 'Wejherowo') -> Optional[str]:
|
||||
"""
|
||||
Find a place by company name and city.
|
||||
|
||||
Uses Google Places findplacefromtext API to search for a business
|
||||
and returns the place_id if found.
|
||||
|
||||
Args:
|
||||
company_name: Name of the company to search for.
|
||||
city: City to narrow down the search (default: Wejherowo).
|
||||
|
||||
Returns:
|
||||
place_id string if found, None otherwise.
|
||||
"""
|
||||
if not self.api_key:
|
||||
logger.warning('Google Places API key not configured')
|
||||
return None
|
||||
|
||||
try:
|
||||
# Construct search query with company name and city
|
||||
search_query = f'{company_name} {city}'
|
||||
|
||||
params = {
|
||||
'input': search_query,
|
||||
'inputtype': 'textquery',
|
||||
'fields': 'place_id,name,formatted_address',
|
||||
'language': 'pl',
|
||||
'key': self.api_key,
|
||||
}
|
||||
|
||||
response = self.session.get(
|
||||
self.FIND_PLACE_URL,
|
||||
params=params,
|
||||
timeout=REQUEST_TIMEOUT
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
if data.get('status') == 'OK' and data.get('candidates'):
|
||||
candidate = data['candidates'][0]
|
||||
place_id = candidate.get('place_id')
|
||||
logger.info(
|
||||
f"Found place for '{company_name}': {candidate.get('name')} "
|
||||
f"at {candidate.get('formatted_address')}"
|
||||
)
|
||||
return place_id
|
||||
elif data.get('status') == 'ZERO_RESULTS':
|
||||
logger.info(f"No Google Business Profile found for '{company_name}' in {city}")
|
||||
return None
|
||||
else:
|
||||
logger.warning(
|
||||
f"Google Places API returned status: {data.get('status')} "
|
||||
f"for '{company_name}'"
|
||||
)
|
||||
return None
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
logger.error(f"Timeout searching for '{company_name}' on Google Places")
|
||||
return None
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"Request error searching for '{company_name}': {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Error finding place for '{company_name}': {e}")
|
||||
return None
|
||||
|
||||
def get_place_details(self, place_id: str) -> Dict[str, Any]:
|
||||
"""
|
||||
Get detailed information about a place.
|
||||
|
||||
Retrieves rating, review count, opening hours, and other business details
|
||||
from Google Places API.
|
||||
|
||||
Args:
|
||||
place_id: Google Place ID returned from find_place().
|
||||
|
||||
Returns:
|
||||
Dict containing:
|
||||
- google_rating: Decimal rating (1.0-5.0) or None
|
||||
- google_reviews_count: Integer review count or None
|
||||
- opening_hours: Dict with weekday_text and open_now, or None
|
||||
- business_status: String like 'OPERATIONAL', 'CLOSED_TEMPORARILY', etc.
|
||||
- formatted_phone: Phone number or None
|
||||
- website: Website URL or None
|
||||
"""
|
||||
result = {
|
||||
'google_rating': None,
|
||||
'google_reviews_count': None,
|
||||
'opening_hours': None,
|
||||
'business_status': None,
|
||||
'formatted_phone': None,
|
||||
'website': None,
|
||||
}
|
||||
|
||||
if not self.api_key:
|
||||
logger.warning('Google Places API key not configured')
|
||||
return result
|
||||
|
||||
if not place_id:
|
||||
return result
|
||||
|
||||
try:
|
||||
# Request fields we need for the audit
|
||||
fields = [
|
||||
'rating',
|
||||
'user_ratings_total',
|
||||
'opening_hours',
|
||||
'business_status',
|
||||
'formatted_phone_number',
|
||||
'website',
|
||||
'name',
|
||||
]
|
||||
|
||||
params = {
|
||||
'place_id': place_id,
|
||||
'fields': ','.join(fields),
|
||||
'language': 'pl',
|
||||
'key': self.api_key,
|
||||
}
|
||||
|
||||
response = self.session.get(
|
||||
self.PLACE_DETAILS_URL,
|
||||
params=params,
|
||||
timeout=REQUEST_TIMEOUT
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
if data.get('status') == 'OK' and data.get('result'):
|
||||
place = data['result']
|
||||
|
||||
# Extract rating
|
||||
if 'rating' in place:
|
||||
result['google_rating'] = round(float(place['rating']), 1)
|
||||
|
||||
# Extract review count
|
||||
if 'user_ratings_total' in place:
|
||||
result['google_reviews_count'] = int(place['user_ratings_total'])
|
||||
|
||||
# Extract opening hours
|
||||
if 'opening_hours' in place:
|
||||
hours = place['opening_hours']
|
||||
result['opening_hours'] = {
|
||||
'weekday_text': hours.get('weekday_text', []),
|
||||
'open_now': hours.get('open_now'),
|
||||
'periods': hours.get('periods', []),
|
||||
}
|
||||
|
||||
# Extract business status
|
||||
if 'business_status' in place:
|
||||
result['business_status'] = place['business_status']
|
||||
|
||||
# Extract phone
|
||||
if 'formatted_phone_number' in place:
|
||||
result['formatted_phone'] = place['formatted_phone_number']
|
||||
|
||||
# Extract website
|
||||
if 'website' in place:
|
||||
result['website'] = place['website']
|
||||
|
||||
logger.info(
|
||||
f"Retrieved details for {place.get('name')}: "
|
||||
f"rating={result['google_rating']}, "
|
||||
f"reviews={result['google_reviews_count']}"
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
f"Google Places API returned status: {data.get('status')} "
|
||||
f"for place_id: {place_id}"
|
||||
)
|
||||
|
||||
except requests.exceptions.Timeout:
|
||||
logger.error(f"Timeout getting details for place_id: {place_id}")
|
||||
except requests.exceptions.RequestException as e:
|
||||
logger.error(f"Request error getting place details: {e}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting place details for {place_id}: {e}")
|
||||
|
||||
return result
|
||||
|
||||
|
||||
class BraveSearcher:
|
||||
"""Search for social media profiles and Google reviews using Brave Search."""
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user