import logging import requests from datetime import datetime, timedelta, timezone from odoo import models, fields, api, _ from odoo.exceptions import UserError _logger = logging.getLogger(__name__) RATING_MAP = { 'ONE': 1, 'TWO': 2, 'THREE': 3, 'FOUR': 4, 'FIVE': 5, } STAR_DISPLAY = { 1: '★☆☆☆☆', 2: '★★☆☆☆', 3: '★★★☆☆', 4: '★★★★☆', 5: '★★★★★', } class GoogleMapReview(models.Model): _name = 'google.map.review' _description = 'Google Business Profile Review' _order = 'publish_date desc' _rec_name = 'name' name = fields.Char( string='Review', compute='_compute_name', store=True, ) company_id = fields.Many2one( 'res.company', string='Branch / Company', required=True, ondelete='cascade', index=True, ) review_id = fields.Char( string='Google Review ID', index=True, copy=False, help='Unique review identifier returned by the Google Business Profile API.', ) author_name = fields.Char(string='Reviewer', index=True) rating = fields.Float( string='Average Rating', group_operator='avg', digits=(16, 1), help='Star rating 1–5.', ) rating_display = fields.Char( string='Stars', compute='_compute_rating_display', store=True, ) review_text = fields.Text(string='Review Text') reply_text = fields.Text( string="Owner's Reply", help='The business owner reply fetched from Google.', ) publish_date = fields.Datetime(string='Published', index=True) language = fields.Char(string='Language', size=10) @api.depends('author_name', 'publish_date') def _compute_name(self): for rec in self: date_str = rec.publish_date.strftime('%Y-%m-%d') if rec.publish_date else '' rec.name = f"{rec.author_name or 'Unknown'} – {date_str}" @api.depends('rating') def _compute_rating_display(self): for rec in self: rec.rating_display = STAR_DISPLAY.get(int(rec.rating or 0), '') # ------------------------------------------------------------------ # OAuth2 helpers # ------------------------------------------------------------------ def _get_api_credentials(self): """Return (client_id, client_secret, refresh_token) from system params.""" ICP = self.env['ir.config_parameter'].sudo() client_id = ICP.get_param('google_map_review.client_id', '') client_secret = ICP.get_param('google_map_review.client_secret', '') refresh_token = ICP.get_param('google_map_review.refresh_token', '') if not all([client_id, client_secret, refresh_token]): raise UserError(_( 'Google Business Profile credentials are not configured. ' 'Please go to Settings → General Settings → Google Business Reviews ' 'and fill in the Client ID, Client Secret and Refresh Token.' )) return client_id, client_secret, refresh_token def _get_access_token(self): """Exchange refresh_token for a fresh access_token.""" client_id, client_secret, refresh_token = self._get_api_credentials() resp = requests.post( 'https://oauth2.googleapis.com/token', data={ 'grant_type': 'refresh_token', 'client_id': client_id, 'client_secret': client_secret, 'refresh_token': refresh_token, }, timeout=30, ) resp.raise_for_status() token_data = resp.json() access_token = token_data.get('access_token') if not access_token: raise UserError(_('Failed to obtain access token from Google: %s') % resp.text) return access_token def _get_auth_headers(self, access_token): return { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json', 'Accept-Language': 'id', } def _get_account_name(self, headers): url = 'https://mybusinessaccountmanagement.googleapis.com/v1/accounts' resp = requests.get(url, headers=headers, timeout=30) resp.raise_for_status() accounts = resp.json().get('accounts', []) if not accounts: raise UserError(_('No Google Business accounts found for the provided credentials.')) return accounts[0]['name'] # ------------------------------------------------------------------ # Text cleaning helpers # ------------------------------------------------------------------ @staticmethod def _clean_review_text(comment): """Strip Google-translated prefixes/suffixes from review text.""" if not comment: return comment if '(Original)' in comment: parts = comment.split('(Original)') if len(parts) > 1: return parts[-1].strip() if '(Diterjemahkan oleh Google)' in comment: parts = comment.split('\n\n(Diterjemahkan oleh Google)') if len(parts) > 1: return parts[0].strip() parts = comment.split('(Diterjemahkan oleh Google)') return parts[0].strip() return comment # ------------------------------------------------------------------ # Main crawl method (called by cron) # ------------------------------------------------------------------ def action_fetch_all_reviews(self): """ Cron entry-point. Fetches reviews from the Google Business Profile API for every res.company that has a google_business_id set. Only reviews published within the last 90 days are processed. """ try: access_token = self._get_access_token() except UserError as e: _logger.error('Google Map Review: %s', e) return except Exception as e: _logger.error('Google Map Review: Failed to get access token: %s', e) return headers = self._get_auth_headers(access_token) try: account_name = self._get_account_name(headers) except Exception as e: _logger.error('Google Map Review: Failed to get account name: %s', e) return companies = self.env['res.company'].sudo().search([ ('google_business_id', '!=', False), ('google_business_id', '!=', ''), ]) if not companies: _logger.info('Google Map Review: No companies with google_business_id found.') return cutoff = datetime.now(timezone.utc) - timedelta(days=90) _logger.info('Google Map Review: Processing %d companies. Cutoff: %s UTC', len(companies), cutoff.strftime('%Y-%m-%d')) total_upserted = 0 for company in companies: location_id = company.google_business_id.strip() base_url = ( f'https://mybusiness.googleapis.com/v4/' f'{account_name}/locations/{location_id}/reviews' ) page_token = None company_upserted = 0 company_skipped = 0 stop_pagination = False _logger.info('Google Map Review: Crawling "%s" (location: %s)', company.name, location_id) while not stop_pagination: params = {} if page_token: params['pageToken'] = page_token try: response = requests.get(base_url, headers=headers, params=params, timeout=30) response.raise_for_status() data = response.json() except Exception as e: _logger.warning( 'Google Map Review: Error fetching reviews for "%s": %s', company.name, e, ) break reviews = data.get('reviews', []) if not reviews: break for review in reviews: # Parse publish time create_time_str = review.get('createTime', '') publish_time = None if create_time_str: try: clean = create_time_str.split('.')[0].replace('Z', '') publish_time = datetime.strptime( clean, '%Y-%m-%dT%H:%M:%S' ).replace(tzinfo=timezone.utc) except ValueError: pass # 90-day filter if publish_time and publish_time < cutoff: company_skipped += 1 stop_pagination = True continue review_id = review.get('reviewId') comment = self._clean_review_text(review.get('comment', '')) author_name = review.get('reviewer', {}).get('displayName', 'Unknown') rating_str = review.get('starRating', '') rating = RATING_MAP.get(rating_str, 0) # Owner reply reply_comment = review.get('reviewReply', {}).get('comment', '') reply_text = reply_comment.strip() if reply_comment else False # Upsert publish_dt = publish_time.replace(tzinfo=None) if publish_time else False existing = self.sudo().search([ ('review_id', '=', review_id), ('company_id', '=', company.id), ], limit=1) vals = { 'company_id': company.id, 'review_id': review_id, 'author_name': author_name, 'rating': rating, 'review_text': comment, 'reply_text': reply_text, 'publish_date': publish_dt, } if existing: existing.sudo().write(vals) else: self.sudo().create(vals) company_upserted += 1 total_upserted += 1 page_token = data.get('nextPageToken') if not page_token: break self.env.cr.commit() _logger.info( 'Google Map Review: "%s" → %d upserted, %d skipped (>90 days)', company.name, company_upserted, company_skipped, ) _logger.info('Google Map Review: Crawl complete. Total upserted: %d', total_upserted)