From 47c415a63bbad873ab40d8cb8270cf684f3c9d4e Mon Sep 17 00:00:00 2001 From: Maciej Pienczyn Date: Thu, 8 Jan 2026 19:20:25 +0100 Subject: [PATCH] auto-claude: subtask-3-4 - Integrate AI recommendations using Gemini service Added AI-powered recommendation generation to GBP audit service: - Import gemini_service module for AI integration - generate_ai_recommendations(): Main method to generate personalized recommendations using Gemini with proper cost tracking - _build_ai_recommendation_prompt(): Builds context-aware prompt with company info, audit results, and field statuses in Polish - _parse_ai_recommendations(): Parses JSON response from Gemini with robust error handling and fallback to static recommendations - audit_with_ai(): Convenience method for running audit with AI - audit_company_with_ai(): Module-level convenience function Features: - Recommendations are personalized to company industry/category - Includes action_steps and expected_impact for each recommendation - Graceful fallback to static recommendations if AI unavailable - Cost tracking via 'gbp_audit_ai' feature tag - Updated test runner with --ai flag for testing AI mode Co-Authored-By: Claude Opus 4.5 --- gbp_audit_service.py | 320 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 317 insertions(+), 3 deletions(-) diff --git a/gbp_audit_service.py b/gbp_audit_service.py index 6328236..dd26919 100644 --- a/gbp_audit_service.py +++ b/gbp_audit_service.py @@ -14,6 +14,7 @@ Author: Norda Biznes Development Team Created: 2026-01-08 """ +import json import logging from dataclasses import dataclass, field from datetime import datetime @@ -23,6 +24,7 @@ from typing import Dict, List, Optional, Any from sqlalchemy.orm import Session from database import Company, GBPAudit, CompanyWebsiteAnalysis, SessionLocal +import gemini_service # Configure logging logger = logging.getLogger(__name__) @@ -644,6 +646,266 @@ class GBPAuditService: return 'low' + # === AI-Powered Recommendations === + + def generate_ai_recommendations( + self, + company: Company, + result: AuditResult, + user_id: Optional[int] = None + ) -> List[Dict[str, Any]]: + """ + Generate AI-powered recommendations using Gemini. + + Args: + company: Company being audited + result: AuditResult from the audit + user_id: Optional user ID for cost tracking + + Returns: + List of AI-generated recommendation dicts with keys: + - priority: 'high', 'medium', 'low' + - field: field name this applies to + - recommendation: AI-generated recommendation text + - action_steps: list of specific action steps + - expected_impact: description of expected improvement + """ + service = gemini_service.get_gemini_service() + if not service: + logger.warning("Gemini service not available - using static recommendations") + return result.recommendations + + try: + # Build context for AI + prompt = self._build_ai_recommendation_prompt(company, result) + + # Call Gemini with cost tracking + response_text = service.generate_text( + prompt=prompt, + feature='gbp_audit_ai', + user_id=user_id, + temperature=0.7, + max_tokens=2000 + ) + + # Parse AI response + ai_recommendations = self._parse_ai_recommendations(response_text, result) + + logger.info( + f"AI recommendations generated for company {company.id}: " + f"{len(ai_recommendations)} recommendations" + ) + + return ai_recommendations + + except Exception as e: + logger.error(f"AI recommendation generation failed: {e}") + # Fall back to static recommendations + return result.recommendations + + def _build_ai_recommendation_prompt( + self, + company: Company, + result: AuditResult + ) -> str: + """ + Build prompt for Gemini to generate personalized recommendations. + + Args: + company: Company being audited + result: AuditResult with field statuses + + Returns: + Formatted prompt string + """ + # Build field status summary + field_summary = [] + for field_name, field_status in result.fields.items(): + status_emoji = { + 'complete': '✅', + 'partial': '⚠️', + 'missing': '❌' + }.get(field_status.status, '❓') + + field_summary.append( + f"- {field_name}: {status_emoji} {field_status.status} " + f"({field_status.score:.1f}/{field_status.max_score:.1f} pkt)" + ) + + # Get category info + category_name = company.category.name if company.category else 'Nieznana' + + prompt = f"""Jesteś ekspertem od Google Business Profile (Wizytówki Google) i lokalnego SEO. + +FIRMA: {company.name} +BRANŻA: {category_name} +MIASTO: {company.address_city or 'Nieznane'} +WYNIK AUDYTU: {result.completeness_score}/100 + +STATUS PÓL WIZYTÓWKI: +{chr(10).join(field_summary)} + +LICZBA ZDJĘĆ: {result.photo_count} +LICZBA OPINII: {result.review_count} +OCENA: {result.average_rating or 'Brak'} + +ZADANIE: +Wygeneruj 3-5 spersonalizowanych rekomendacji dla tej firmy, aby poprawić jej wizytówkę Google. + +WYMAGANIA: +1. Każda rekomendacja powinna być konkretna i dostosowana do branży firmy +2. Skup się na polach z najniższymi wynikami +3. Podaj praktyczne kroki do wykonania +4. Używaj języka polskiego + +ZWRÓĆ ODPOWIEDŹ W FORMACIE JSON (TYLKO JSON, BEZ MARKDOWN): +[ + {{ + "priority": "high|medium|low", + "field": "nazwa_pola", + "recommendation": "Krótki opis co poprawić", + "action_steps": ["Krok 1", "Krok 2", "Krok 3"], + "expected_impact": "Opis spodziewanej poprawy" + }} +] + +Priorytety: +- high: kluczowe pola (name, address, categories, description) +- medium: ważne pola (phone, website, photos, services) +- low: dodatkowe pola (hours, reviews) + +Odpowiedź (TYLKO JSON):""" + + return prompt + + def _parse_ai_recommendations( + self, + response_text: str, + fallback_result: AuditResult + ) -> List[Dict[str, Any]]: + """ + Parse AI response into structured recommendations. + + Args: + response_text: Raw text from Gemini + fallback_result: AuditResult to use for fallback + + Returns: + List of recommendation dicts + """ + try: + # Clean up response - remove markdown code blocks if present + cleaned = response_text.strip() + if cleaned.startswith('```'): + # Remove markdown code block markers + lines = cleaned.split('\n') + # Find JSON content between ``` markers + json_lines = [] + in_json = False + for line in lines: + if line.startswith('```') and not in_json: + in_json = True + continue + elif line.startswith('```') and in_json: + break + elif in_json: + json_lines.append(line) + cleaned = '\n'.join(json_lines) + + # Parse JSON + recommendations = json.loads(cleaned) + + # Validate and enhance recommendations + valid_recommendations = [] + valid_priorities = {'high', 'medium', 'low'} + valid_fields = set(FIELD_WEIGHTS.keys()) + + for rec in recommendations: + if not isinstance(rec, dict): + continue + + # Validate priority + priority = rec.get('priority', 'medium') + if priority not in valid_priorities: + priority = 'medium' + + # Validate field + field = rec.get('field', 'general') + if field not in valid_fields: + field = 'general' + + # Get impact score from field weights + impact = FIELD_WEIGHTS.get(field, 5) + + valid_recommendations.append({ + 'priority': priority, + 'field': field, + 'recommendation': rec.get('recommendation', ''), + 'action_steps': rec.get('action_steps', []), + 'expected_impact': rec.get('expected_impact', ''), + 'impact': impact, + 'source': 'ai' + }) + + if valid_recommendations: + # Sort by priority and impact + priority_order = {'high': 0, 'medium': 1, 'low': 2} + valid_recommendations.sort( + key=lambda x: (priority_order.get(x['priority'], 3), -x['impact']) + ) + return valid_recommendations + + except json.JSONDecodeError as e: + logger.warning(f"Failed to parse AI recommendations JSON: {e}") + except Exception as e: + logger.warning(f"Error processing AI recommendations: {e}") + + # Return fallback recommendations with source marker + fallback = [] + for rec in fallback_result.recommendations: + rec_copy = dict(rec) + rec_copy['source'] = 'static' + rec_copy['action_steps'] = [] + rec_copy['expected_impact'] = '' + fallback.append(rec_copy) + + return fallback + + def audit_with_ai( + self, + company_id: int, + user_id: Optional[int] = None + ) -> AuditResult: + """ + Run full GBP audit with AI-powered recommendations. + + Args: + company_id: ID of the company to audit + user_id: Optional user ID for cost tracking + + Returns: + AuditResult with AI-enhanced recommendations + """ + # Run standard audit + result = self.audit_company(company_id) + + # Get company for AI context + company = self.db.query(Company).filter(Company.id == company_id).first() + if not company: + return result + + # Generate AI recommendations + ai_recommendations = self.generate_ai_recommendations( + company=company, + result=result, + user_id=user_id + ) + + # Replace static recommendations with AI-generated ones + result.recommendations = ai_recommendations + + return result + # === Convenience Functions === @@ -683,6 +945,33 @@ def get_company_audit(db: Session, company_id: int) -> Optional[GBPAudit]: return service.get_latest_audit(company_id) +def audit_company_with_ai( + db: Session, + company_id: int, + save: bool = True, + user_id: Optional[int] = None +) -> AuditResult: + """ + Audit a company's GBP completeness with AI-powered recommendations. + + Args: + db: Database session + company_id: Company ID to audit + save: Whether to save audit to database + user_id: Optional user ID for cost tracking + + Returns: + AuditResult with AI-enhanced recommendations + """ + service = GBPAuditService(db) + result = service.audit_with_ai(company_id, user_id=user_id) + + if save: + service.save_audit(result, source='ai') + + return result + + def batch_audit_companies( db: Session, company_ids: Optional[List[int]] = None, @@ -722,9 +1011,14 @@ def batch_audit_companies( # === Main for Testing === if __name__ == '__main__': + import sys + # Test the service logging.basicConfig(level=logging.INFO) + # Check for --ai flag to test AI recommendations + use_ai = '--ai' in sys.argv + db = SessionLocal() try: # Get first active company @@ -733,19 +1027,39 @@ if __name__ == '__main__': print(f"\nAuditing company: {company.name} (ID: {company.id})") print("-" * 50) - result = audit_company(db, company.id, save=False) + if use_ai: + print("\n[AI MODE] Generating AI-powered recommendations...") + result = audit_company_with_ai(db, company.id, save=False) + else: + result = audit_company(db, company.id, save=False) print(f"\nCompleteness Score: {result.completeness_score}/100") print(f"\nField Status:") for name, field in result.fields.items(): - status_icon = {'complete': '[check mark]', 'partial': '~', 'missing': '[X]'}.get(field.status, '?') + status_icon = {'complete': '✅', 'partial': '⚠️', 'missing': '❌'}.get(field.status, '?') print(f" {status_icon} {name}: {field.status} ({field.score:.1f}/{field.max_score:.1f})") print(f"\nRecommendations ({len(result.recommendations)}):") for rec in result.recommendations[:5]: - print(f" [{rec['priority'].upper()}] {rec['field']}: {rec['recommendation'][:80]}...") + source = rec.get('source', 'static') + source_label = '[AI]' if source == 'ai' else '[STATIC]' + print(f"\n {source_label} [{rec['priority'].upper()}] {rec['field']}:") + print(f" {rec['recommendation']}") + + # Print AI-specific fields if present + if rec.get('action_steps'): + print(" Action steps:") + for step in rec['action_steps']: + print(f" • {step}") + + if rec.get('expected_impact'): + print(f" Expected impact: {rec['expected_impact']}") else: print("No active companies found") + print("\n" + "-" * 50) + print("Usage: python gbp_audit_service.py [--ai]") + print(" --ai Generate AI-powered recommendations using Gemini") + finally: db.close()