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 <noreply@anthropic.com>
This commit is contained in:
parent
8fa23bc77e
commit
47c415a63b
@ -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()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user