maciejpi/nordabiz
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
4ff386fa7d
nordabiz/audit_ai_service.py
Maciej Pienczyn 4ff386fa7d
Some checks are pending
NordaBiz Tests / Unit & Integration Tests (push) Waiting to run
Details
NordaBiz Tests / E2E Tests (Playwright) (push) Blocked by required conditions
Details
NordaBiz Tests / Smoke Tests (Production) (push) Blocked by required conditions
Details
NordaBiz Tests / Send Failure Notification (push) Blocked by required conditions
Details
debug(cache): Log per-field hashes to identify unstable field 
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-07 13:54:03 +01:00

922 lines
35 KiB
Python
Raw Blame History

"""
Audit AI Service
=================
Centralny serwis AI do analizy wyników audytów i generowania
priorytetowanych akcji z treścią gotową do wdrożenia.
Obsługiwane typy audytów:
- SEO (PageSpeed, on-page, technical, local SEO)
- GBP (Google Business Profile completeness)
- Social Media (presence across platforms)
Używa Gemini API (via gemini_service.py) do generowania analiz.
Author: Norda Biznes Development Team
Created: 2026-02-07
"""
import hashlib
import json
import logging
from datetime import datetime, timedelta
from html import unescape
from database import (
SessionLocal, Company, CompanyWebsiteAnalysis, CompanySocialMedia,
CompanyCitation, AuditAction, AuditAICache
)
logger = logging.getLogger(__name__)
# Cache expiry: 7 days
CACHE_EXPIRY_DAYS = 7
def _get_gemini_service():
"""Get the initialized Gemini service instance."""
from gemini_service import get_gemini_service
service = get_gemini_service()
if not service:
raise RuntimeError("Gemini service not initialized")
return service
def _hash_data(data: dict) -> str:
"""Generate SHA256 hash of audit data for cache invalidation."""
serialized = json.dumps(data, sort_keys=True, default=str)
return hashlib.sha256(serialized.encode()).hexdigest()
# ============================================================
# SEO AUDIT DATA COLLECTION
# ============================================================
def _collect_seo_data(db, company) -> dict:
"""Collect SEO audit data for AI analysis."""
analysis = db.query(CompanyWebsiteAnalysis).filter(
CompanyWebsiteAnalysis.company_id == company.id
).order_by(CompanyWebsiteAnalysis.seo_audited_at.desc()).first()
if not analysis or not analysis.seo_audited_at:
return {}
citations = db.query(CompanyCitation).filter(
CompanyCitation.company_id == company.id
).all()
return {
'company_name': company.name,
'company_category': company.category,
'website': company.website,
'city': company.address_city,
# PageSpeed scores
'seo_score': analysis.pagespeed_seo_score,
'performance_score': analysis.pagespeed_performance_score,
'accessibility_score': analysis.pagespeed_accessibility_score,
'best_practices_score': analysis.pagespeed_best_practices_score,
# On-page
'meta_title': unescape(analysis.meta_title or ''),
'meta_description': unescape(analysis.meta_description or ''),
'h1_count': analysis.h1_count,
'h1_text': unescape(analysis.h1_text or ''),
'h2_count': analysis.h2_count,
'h3_count': analysis.h3_count,
'total_images': analysis.total_images,
'images_without_alt': analysis.images_without_alt,
# Technical
'has_ssl': analysis.has_ssl,
'has_sitemap': analysis.has_sitemap,
'has_robots_txt': analysis.has_robots_txt,
'has_canonical': analysis.has_canonical,
'is_indexable': analysis.is_indexable,
'is_mobile_friendly': getattr(analysis, 'is_mobile_friendly', None),
'load_time_ms': analysis.load_time_ms,
# Structured data
'has_structured_data': analysis.has_structured_data,
'structured_data_types': analysis.structured_data_types,
'has_local_business_schema': analysis.has_local_business_schema,
# Social/analytics
'has_og_tags': analysis.has_og_tags,
'has_twitter_cards': analysis.has_twitter_cards,
'has_google_analytics': analysis.has_google_analytics,
'has_google_tag_manager': analysis.has_google_tag_manager,
# Local SEO
'local_seo_score': analysis.local_seo_score,
'has_google_maps_embed': analysis.has_google_maps_embed,
'has_local_keywords': analysis.has_local_keywords,
'nap_on_website': analysis.nap_on_website,
# Core Web Vitals
'lcp_ms': analysis.largest_contentful_paint_ms,
'fid_ms': analysis.first_input_delay_ms,
'cls': float(analysis.cumulative_layout_shift) if analysis.cumulative_layout_shift else None,
# Content
'content_freshness_score': analysis.content_freshness_score,
'word_count_homepage': analysis.word_count_homepage,
# Links
'internal_links_count': analysis.internal_links_count,
'external_links_count': analysis.external_links_count,
'broken_links_count': analysis.broken_links_count,
# Citations
'citations_count': len(citations),
'citations_found': len([c for c in citations if c.status == 'found']),
}
def _collect_gbp_data(db, company) -> dict:
"""Collect GBP audit data for AI analysis."""
try:
from gbp_audit_service import get_company_audit
audit = get_company_audit(db, company.id)
except ImportError:
audit = None
if not audit:
return {}
return {
'company_name': company.name,
'company_category': company.category,
'city': company.address_city,
'completeness_score': audit.completeness_score,
# Field presence
'has_name': audit.has_name,
'has_address': audit.has_address,
'has_phone': audit.has_phone,
'has_website': audit.has_website,
'has_hours': audit.has_hours,
'has_categories': audit.has_categories,
'has_photos': audit.has_photos,
'has_description': audit.has_description,
'has_services': audit.has_services,
'has_reviews': audit.has_reviews,
# Reviews
'review_count': audit.review_count,
'average_rating': float(audit.average_rating) if audit.average_rating else None,
'reviews_with_response': audit.reviews_with_response,
'reviews_without_response': audit.reviews_without_response,
'review_response_rate': float(audit.review_response_rate) if audit.review_response_rate else None,
# Activity
'has_posts': audit.has_posts,
'posts_count_30d': audit.posts_count_30d,
'has_products': audit.has_products,
'has_qa': audit.has_qa,
# Photos
'photo_count': audit.photo_count,
'logo_present': audit.logo_present,
'cover_photo_present': audit.cover_photo_present,
# NAP
'nap_consistent': audit.nap_consistent,
'nap_issues': audit.nap_issues,
}
def _collect_social_data(db, company) -> dict:
"""Collect social media audit data for AI analysis."""
profiles = db.query(CompanySocialMedia).filter(
CompanySocialMedia.company_id == company.id
).all()
all_platforms = ['facebook', 'instagram', 'linkedin', 'youtube', 'twitter', 'tiktok']
profiles_dict = {}
for p in profiles:
profiles_dict[p.platform] = {
'url': p.url,
'is_valid': p.is_valid,
'followers_count': p.followers_count,
'has_bio': p.has_bio,
'has_profile_photo': p.has_profile_photo,
'has_cover_photo': p.has_cover_photo,
'posts_count_30d': p.posts_count_30d,
'last_post_date': str(p.last_post_date) if p.last_post_date else None,
'posting_frequency_score': p.posting_frequency_score,
'engagement_rate': float(p.engagement_rate) if p.engagement_rate else None,
'profile_completeness_score': p.profile_completeness_score,
}
present = [p for p in all_platforms if p in profiles_dict]
missing = [p for p in all_platforms if p not in profiles_dict]
return {
'company_name': company.name,
'company_category': company.category,
'city': company.address_city,
'platforms_present': present,
'platforms_missing': missing,
'profiles': profiles_dict,
'total_platforms': len(all_platforms),
'platforms_found': len(present),
'score': int((len(present) / len(all_platforms)) * 100) if all_platforms else 0,
}
# ============================================================
# GEMINI PROMPTS
# ============================================================
def _build_seo_prompt(data: dict) -> str:
"""Build Gemini prompt for SEO audit analysis."""
return f"""Jesteś ekspertem SEO analizującym stronę internetową lokalnej firmy w Polsce.
DANE FIRMY:
- Nazwa: {data.get('company_name', 'N/A')}
- Branża: {data.get('company_category', 'N/A')}
- Miasto: {data.get('city', 'N/A')}
- Strona: {data.get('website', 'N/A')}
WYNIKI AUDYTU SEO:
- Wynik SEO (PageSpeed): {data.get('seo_score', 'brak')}/100
- Wydajność: {data.get('performance_score', 'brak')}/100
- Dostępność: {data.get('accessibility_score', 'brak')}/100
- Best Practices: {data.get('best_practices_score', 'brak')}/100
Core Web Vitals:
- LCP: {data.get('lcp_ms', 'brak')} ms
- FID: {data.get('fid_ms', 'brak')} ms
- CLS: {data.get('cls', 'brak')}
On-Page SEO:
- Meta title: {data.get('meta_title', 'brak')}
- Meta description: {'tak' if data.get('meta_description') else 'BRAK'}
- H1: {data.get('h1_count', 0)} (treść: {data.get('h1_text', 'brak')})
- H2: {data.get('h2_count', 0)}, H3: {data.get('h3_count', 0)}
- Obrazy: {data.get('total_images', 0)} (bez alt: {data.get('images_without_alt', 0)})
- Linki wewnętrzne: {data.get('internal_links_count', 0)}, zewnętrzne: {data.get('external_links_count', 0)}, uszkodzone: {data.get('broken_links_count', 0)}
Technical SEO:
- SSL: {'tak' if data.get('has_ssl') else 'NIE'}
- Sitemap: {'tak' if data.get('has_sitemap') else 'NIE'}
- Robots.txt: {'tak' if data.get('has_robots_txt') else 'NIE'}
- Canonical: {'tak' if data.get('has_canonical') else 'NIE'}
- Indeksowalna: {'tak' if data.get('is_indexable') else 'NIE'}
- Mobile-friendly: {'tak' if data.get('is_mobile_friendly') else 'NIE/brak danych'}
Dane strukturalne:
- Schema.org: {'tak' if data.get('has_structured_data') else 'NIE'} (typy: {data.get('structured_data_types', [])})
- LocalBusiness Schema: {'tak' if data.get('has_local_business_schema') else 'NIE'}
Social & Analytics:
- Open Graph: {'tak' if data.get('has_og_tags') else 'NIE'}
- Twitter Cards: {'tak' if data.get('has_twitter_cards') else 'NIE'}
- Google Analytics: {'tak' if data.get('has_google_analytics') else 'NIE'}
- GTM: {'tak' if data.get('has_google_tag_manager') else 'NIE'}
Local SEO (wynik: {data.get('local_seo_score', 'brak')}/100):
- Mapa Google: {'tak' if data.get('has_google_maps_embed') else 'NIE'}
- Lokalne słowa kluczowe: {'tak' if data.get('has_local_keywords') else 'NIE'}
- NAP na stronie: {'tak' if data.get('nap_on_website') else 'NIE'}
- Cytacje: {data.get('citations_found', 0)}/{data.get('citations_count', 0)} znalezionych
Treść:
- Świeżość: {data.get('content_freshness_score', 'brak')}/100
- Słów na stronie głównej: {data.get('word_count_homepage', 'brak')}
ZADANIE:
Przygotuj analizę w formacie JSON z dwoma kluczami:
1. "summary" - krótki akapit (2-4 zdania) podsumowujący stan SEO strony, co jest dobrze, a co wymaga poprawy. Pisz bezpośrednio do właściciela firmy, po polsku.
2. "actions" - lista od 3 do 8 priorytetowanych akcji do podjęcia. Każda akcja to obiekt:
{{
"action_type": "typ akcji z listy: generate_schema_org, generate_meta_description, suggest_heading_fix, generate_alt_texts, seo_roadmap, add_analytics, add_sitemap, fix_ssl, add_og_tags, improve_performance, add_local_keywords, add_nap, fix_broken_links",
"title": "krótki tytuł po polsku",
"description": "opis co trzeba zrobić i dlaczego, 1-2 zdania",
"priority": "critical/high/medium/low",
"impact_score": 1-10,
"effort_score": 1-10,
"platform": "website"
}}
Priorytetyzuj wg: impact_score / effort_score (wyższy stosunek = wyższy priorytet).
NIE sugeruj akcji dla rzeczy, które firma już ma poprawnie.
Odpowiedz WYŁĄCZNIE poprawnym JSON-em, bez markdown, bez komentarzy."""
def _build_gbp_prompt(data: dict) -> str:
"""Build Gemini prompt for GBP audit analysis."""
return f"""Jesteś ekspertem Google Business Profile analizującym wizytówkę lokalnej firmy w Polsce.
DANE FIRMY:
- Nazwa: {data.get('company_name', 'N/A')}
- Branża: {data.get('company_category', 'N/A')}
- Miasto: {data.get('city', 'N/A')}
WYNIKI AUDYTU GBP (kompletność: {data.get('completeness_score', 'brak')}/100):
- Nazwa: {'✓' if data.get('has_name') else '✗'}
- Adres: {'✓' if data.get('has_address') else '✗'}
- Telefon: {'✓' if data.get('has_phone') else '✗'}
- Strona WWW: {'✓' if data.get('has_website') else '✗'}
- Godziny otwarcia: {'✓' if data.get('has_hours') else '✗'}
- Kategorie: {'✓' if data.get('has_categories') else '✗'}
- Zdjęcia: {'✓' if data.get('has_photos') else '✗'} ({data.get('photo_count', 0)} zdjęć)
- Opis: {'✓' if data.get('has_description') else '✗'}
- Usługi: {'✓' if data.get('has_services') else '✗'}
- Logo: {'✓' if data.get('logo_present') else '✗'}
- Zdjęcie w tle: {'✓' if data.get('cover_photo_present') else '✗'}
Opinie:
- Liczba opinii: {data.get('review_count', 0)}
- Średnia ocena: {data.get('average_rating', 'brak')}
- Z odpowiedzią: {data.get('reviews_with_response', 0)}
- Bez odpowiedzi: {data.get('reviews_without_response', 0)}
- Wskaźnik odpowiedzi: {data.get('review_response_rate', 'brak')}%
Aktywność:
- Posty: {'✓' if data.get('has_posts') else '✗'} ({data.get('posts_count_30d', 0)} w ostatnich 30 dniach)
- Produkty: {'✓' if data.get('has_products') else '✗'}
- Pytania i odpowiedzi: {'✓' if data.get('has_qa') else '✗'}
NAP:
- Spójność NAP: {'✓' if data.get('nap_consistent') else '✗'}
- Problemy NAP: {data.get('nap_issues', 'brak')}
ZADANIE:
Przygotuj analizę w formacie JSON z dwoma kluczami:
1. "summary" - krótki akapit (2-4 zdania) podsumowujący stan wizytówki Google, co jest dobrze, a co wymaga poprawy. Pisz bezpośrednio do właściciela firmy, po polsku.
2. "actions" - lista od 3 do 8 priorytetowanych akcji. Każda akcja:
{{
"action_type": "typ z listy: generate_gbp_description, generate_gbp_post, respond_to_review, suggest_categories, gbp_improvement_plan, add_photos, add_hours, add_services, add_products",
"title": "krótki tytuł po polsku",
"description": "opis co trzeba zrobić i dlaczego",
"priority": "critical/high/medium/low",
"impact_score": 1-10,
"effort_score": 1-10,
"platform": "google"
}}
NIE sugeruj akcji dla pól, które firma już ma poprawnie uzupełnione.
Odpowiedz WYŁĄCZNIE poprawnym JSON-em, bez markdown, bez komentarzy."""
def _build_social_prompt(data: dict) -> str:
"""Build Gemini prompt for social media audit analysis."""
profiles_info = ""
for platform, info in data.get('profiles', {}).items():
profiles_info += f"\n {platform}: followers={info.get('followers_count', '?')}, "
profiles_info += f"bio={'✓' if info.get('has_bio') else '✗'}, "
profiles_info += f"photo={'✓' if info.get('has_profile_photo') else '✗'}, "
profiles_info += f"posty_30d={info.get('posts_count_30d', '?')}, "
profiles_info += f"kompletność={info.get('profile_completeness_score', '?')}%"
return f"""Jesteś ekspertem social media analizującym obecność lokalnej firmy w Polsce w mediach społecznościowych.
DANE FIRMY:
- Nazwa: {data.get('company_name', 'N/A')}
- Branża: {data.get('company_category', 'N/A')}
- Miasto: {data.get('city', 'N/A')}
OBECNOŚĆ W SOCIAL MEDIA (wynik: {data.get('score', 0)}/100):
- Platformy znalezione ({data.get('platforms_found', 0)}/{data.get('total_platforms', 6)}): {', '.join(data.get('platforms_present', []))}
- Platformy brakujące: {', '.join(data.get('platforms_missing', []))}
Szczegóły profili:{profiles_info or ' brak profili'}
ZADANIE:
Przygotuj analizę w formacie JSON z dwoma kluczami:
1. "summary" - krótki akapit (2-4 zdania) podsumowujący obecność firmy w social media. Pisz po polsku, do właściciela firmy.
2. "actions" - lista od 3 do 8 priorytetowanych akcji. Każda akcja:
{{
"action_type": "typ z listy: generate_social_post, generate_bio, content_calendar, content_strategy, create_profile, improve_profile, increase_engagement",
"title": "krótki tytuł po polsku",
"description": "opis co trzeba zrobić i dlaczego",
"priority": "critical/high/medium/low",
"impact_score": 1-10,
"effort_score": 1-10,
"platform": "facebook/instagram/linkedin/youtube/twitter/tiktok/all"
}}
Dla firm lokalnych priorytetyzuj: Facebook > Instagram > LinkedIn > reszta.
NIE sugeruj tworzenia profili na platformach nieistotnych dla branży.
Odpowiedz WYŁĄCZNIE poprawnym JSON-em, bez markdown, bez komentarzy."""
# ============================================================
# CONTENT GENERATION PROMPTS
# ============================================================
CONTENT_PROMPTS = {
'generate_schema_org': """Wygeneruj kompletny JSON-LD Schema.org LocalBusiness dla firmy:
- Nazwa: {company_name}
- Branża: {category}
- Adres: {address}
- Miasto: {city}
- Telefon: {phone}
- Strona: {website}
- Email: {email}
Wygeneruj WYŁĄCZNIE poprawny tag <script type="application/ld+json"> z JSON-LD. Bez komentarzy.""",
'generate_meta_description': """Napisz meta description (150-160 znaków) dla strony firmy:
- Nazwa: {company_name}
- Branża: {category}
- Miasto: {city}
- Usługi: {services}
Opis powinien zawierać lokalne słowa kluczowe, CTA i zachęcać do kliknięcia.
Odpowiedz WYŁĄCZNIE tekstem meta description, bez cudzysłowów, bez komentarzy.""",
'suggest_heading_fix': """Zaproponuj poprawioną strukturę nagłówków (H1, H2, H3) dla strony firmy:
- Nazwa: {company_name}
- Branża: {category}
- Obecny H1: {h1_text}
- Liczba H1: {h1_count}, H2: {h2_count}, H3: {h3_count}
Zaproponuj strukturę z jednym H1 i logiczną hierarchią H2/H3. Po polsku.
Format odpowiedzi - lista nagłówków z poziomami, np.:
H1: ...
H2: ...
H3: ...
H2: ...""",
'generate_alt_texts': """Zaproponuj teksty alternatywne (alt) dla obrazów na stronie firmy:
- Nazwa: {company_name}
- Branża: {category}
- Liczba obrazów bez alt: {images_without_alt}
Napisz {images_without_alt} propozycji tekstów alt (max 125 znaków każdy).
Uwzględnij lokalne słowa kluczowe i opis kontekstowy. Po polsku.
Format: po jednym alt tekście na linię.""",
'generate_gbp_description': """Napisz opis firmy na Google Business Profile (max 750 znaków):
- Nazwa: {company_name}
- Branża: {category}
- Miasto: {city}
- Usługi: {services}
Opis powinien:
- Zawierać lokalne słowa kluczowe (miasto, region)
- Wymieniać główne usługi/produkty
- Zawierać CTA (zachętę do kontaktu)
- Być profesjonalny ale przyjazny
Odpowiedz WYŁĄCZNIE tekstem opisu, po polsku.""",
'generate_gbp_post': """Napisz post na Google Business Profile dla firmy:
- Nazwa: {company_name}
- Branża: {category}
- Miasto: {city}
Post powinien:
- Mieć 150-300 słów
- Zawierać CTA (np. "Zadzwoń", "Odwiedź stronę")
- Być angażujący i profesjonalny
- Dotyczyć aktualnej oferty/promocji/wydarzenia
Odpowiedz WYŁĄCZNIE tekstem postu, po polsku.""",
'respond_to_review': """Napisz profesjonalną odpowiedź na opinię Google:
- Firma: {company_name}
- Ocena: {review_rating}/5
- Treść opinii: {review_text}
Odpowiedź powinna:
- Podziękować za opinię
- Odnieść się do konkretnych punktów
- {'Przeprosić i zaproponować rozwiązanie' if int(str({review_rating}).replace('None','3')) <= 3 else 'Zachęcić do ponownych odwiedzin'}
- Być profesjonalna i osobista
- Max 200 słów
Odpowiedz WYŁĄCZNIE tekstem odpowiedzi, po polsku.""",
'generate_social_post': """Napisz post na {platform} dla firmy:
- Nazwa: {company_name}
- Branża: {category}
- Miasto: {city}
Wymagania dla {platform}:
- Facebook: 100-250 słów, może zawierać emoji, CTA
- Instagram: 100-150 słów, 5-10 hashtagów, emoji
- LinkedIn: 150-300 słów, profesjonalny ton, 3-5 hashtagów
- Twitter: max 280 znaków, 2-3 hashtagi
Odpowiedz WYŁĄCZNIE tekstem postu z hashtagami, po polsku.""",
'generate_bio': """Napisz bio/opis profilu na {platform} dla firmy:
- Nazwa: {company_name}
- Branża: {category}
- Miasto: {city}
- Strona: {website}
Limity znaków:
- Facebook: 255 znaków
- Instagram: 150 znaków
- LinkedIn: 2000 znaków (opis firmy)
- Twitter: 160 znaków
Odpowiedz WYŁĄCZNIE tekstem bio, po polsku.""",
'content_calendar': """Przygotuj kalendarz treści na tydzień (pon-pt) dla firmy w social media:
- Nazwa: {company_name}
- Branża: {category}
- Miasto: {city}
- Platformy: {platforms}
Dla każdego dnia podaj:
- Dzień tygodnia
- Platforma
- Temat postu
- Krótki zarys treści (1-2 zdania)
- Sugerowany typ: tekst/zdjęcie/wideo/karuzela
Format JSON array:
[{{"day": "Poniedziałek", "platform": "...", "topic": "...", "outline": "...", "content_type": "..."}}]
Odpowiedz WYŁĄCZNIE JSON-em, po polsku.""",
'content_strategy': """Przygotuj krótką strategię obecności w social media dla firmy:
- Nazwa: {company_name}
- Branża: {category}
- Miasto: {city}
- Obecne platformy: {platforms_present}
- Brakujące platformy: {platforms_missing}
Strategia powinna zawierać:
1. Rekomendowane platformy (z uzasadnieniem)
2. Częstotliwość publikacji per platforma
3. Typy treści per platforma
4. Ton komunikacji
5. Cele na 3 miesiące
Max 500 słów, po polsku. Pisz bezpośrednio do właściciela firmy.""",
}
# ============================================================
# MAIN SERVICE FUNCTIONS
# ============================================================
def generate_analysis(company_id: int, audit_type: str, user_id: int = None, force: bool = False) -> dict:
"""
Generate AI analysis for an audit.
Returns cached version if available and data hasn't changed.
Args:
company_id: Company ID
audit_type: 'seo', 'gbp', or 'social'
user_id: Current user ID for cost tracking
force: Force regeneration even if cache is valid
Returns:
dict with 'summary' and 'actions' keys
"""
db = SessionLocal()
try:
company = db.query(Company).filter_by(id=company_id, status='active').first()
if not company:
return {'error': 'Firma nie znaleziona'}
# Collect audit data
collectors = {
'seo': _collect_seo_data,
'gbp': _collect_gbp_data,
'social': _collect_social_data,
}
collector = collectors.get(audit_type)
if not collector:
return {'error': f'Nieznany typ audytu: {audit_type}'}
data = collector(db, company)
if not data:
return {'error': f'Brak danych audytu {audit_type} dla tej firmy'}
# Exclude volatile fields from hash to improve cache hit rate
hash_data = {k: v for k, v in data.items() if k not in ('citations_count', 'citations_found')}
data_hash = _hash_data(hash_data)
# Check cache
if not force:
cache = db.query(AuditAICache).filter_by(
company_id=company_id,
audit_type=audit_type
).first()
if cache:
logger.info(f"Cache check: stored_hash={cache.audit_data_hash[:12]}... current_hash={data_hash[:12]}... match={cache.audit_data_hash == data_hash} expires={cache.expires_at}")
if cache.audit_data_hash != data_hash:
# Debug: find which fields changed
for k, v in sorted(hash_data.items()):
field_hash = hashlib.sha256(json.dumps({k: v}, default=str).encode()).hexdigest()[:8]
logger.info(f" field_hash: {k}={field_hash} type={type(v).__name__} val={str(v)[:80]}")
if cache and cache.audit_data_hash == data_hash and cache.expires_at and cache.expires_at > datetime.now():
logger.info(f"AI analysis cache hit for company {company_id} audit_type={audit_type}")
return {
'summary': cache.analysis_summary,
'actions': cache.actions_json or [],
'cached': True,
'generated_at': cache.generated_at.isoformat() if cache.generated_at else None,
}
# Build prompt
prompt_builders = {
'seo': _build_seo_prompt,
'gbp': _build_gbp_prompt,
'social': _build_social_prompt,
}
prompt = prompt_builders[audit_type](data)
# Call Gemini
gemini = _get_gemini_service()
response_text = gemini.generate_text(
prompt=prompt,
temperature=0.3,
thinking_level='low',
feature='audit_analysis',
user_id=user_id,
company_id=company_id,
related_entity_type=f'{audit_type}_audit',
)
if not response_text:
return {'error': 'Gemini nie zwrócił odpowiedzi'}
# Parse JSON response
try:
# Clean possible markdown code blocks
cleaned = response_text.strip()
if cleaned.startswith('```'):
cleaned = cleaned.split('\n', 1)[1] if '\n' in cleaned else cleaned[3:]
if cleaned.endswith('```'):
cleaned = cleaned[:-3]
cleaned = cleaned.strip()
result = json.loads(cleaned)
except json.JSONDecodeError as e:
logger.error(f"Failed to parse Gemini response as JSON: {e}\nResponse: {response_text[:500]}")
return {'error': 'Nie udało się przetworzyć odpowiedzi AI', 'raw_response': response_text}
summary = result.get('summary', '')
actions = result.get('actions', [])
# Save to cache (upsert)
cache = db.query(AuditAICache).filter_by(
company_id=company_id,
audit_type=audit_type
).first()
if cache:
cache.analysis_summary = summary
cache.actions_json = actions
cache.audit_data_hash = data_hash
cache.generated_at = datetime.now()
cache.expires_at = datetime.now() + timedelta(days=CACHE_EXPIRY_DAYS)
else:
cache = AuditAICache(
company_id=company_id,
audit_type=audit_type,
analysis_summary=summary,
actions_json=actions,
audit_data_hash=data_hash,
generated_at=datetime.now(),
expires_at=datetime.now() + timedelta(days=CACHE_EXPIRY_DAYS),
)
db.add(cache)
# Save individual actions to audit_actions table
for action_data in actions:
action = AuditAction(
company_id=company_id,
audit_type=audit_type,
action_type=action_data.get('action_type', 'unknown'),
title=action_data.get('title', 'Akcja'),
description=action_data.get('description', ''),
priority=action_data.get('priority', 'medium'),
impact_score=action_data.get('impact_score'),
effort_score=action_data.get('effort_score'),
platform=action_data.get('platform', 'website'),
ai_model=gemini.model_name,
status='suggested',
created_by=user_id,
)
db.add(action)
db.commit()
return {
'summary': summary,
'actions': actions,
'cached': False,
'generated_at': datetime.now().isoformat(),
}
except Exception as e:
db.rollback()
logger.error(f"Error generating AI analysis: {e}", exc_info=True)
return {'error': f'Błąd generowania analizy: {str(e)}'}
finally:
db.close()
def generate_content(company_id: int, action_type: str, context: dict = None, user_id: int = None) -> dict:
"""
Generate specific content for an audit action.
Args:
company_id: Company ID
action_type: Content type (e.g. 'generate_schema_org', 'generate_gbp_post')
context: Additional context (e.g. platform, review_text)
user_id: Current user ID for cost tracking
Returns:
dict with 'content' key
"""
db = SessionLocal()
try:
company = db.query(Company).filter_by(id=company_id, status='active').first()
if not company:
return {'error': 'Firma nie znaleziona'}
# Build context for prompt
prompt_context = {
'company_name': company.name,
'category': company.category or 'Usługi',
'city': company.address_city or 'Polska',
'website': company.website or '',
'phone': company.phone or '',
'email': company.email or '',
'address': f"{company.address_street or ''}, {company.address_city or ''}".strip(', '),
'services': ', '.join([s.service.name for s in company.services[:5] if s.service]) if hasattr(company, 'services') and company.services else 'brak danych',
}
# Merge in extra context
if context:
prompt_context.update(context)
# Add audit-specific data
if action_type in ('suggest_heading_fix', 'generate_alt_texts'):
analysis = db.query(CompanyWebsiteAnalysis).filter(
CompanyWebsiteAnalysis.company_id == company.id
).order_by(CompanyWebsiteAnalysis.seo_audited_at.desc()).first()
if analysis:
prompt_context.update({
'h1_text': analysis.h1_text or 'brak',
'h1_count': analysis.h1_count or 0,
'h2_count': analysis.h2_count or 0,
'h3_count': analysis.h3_count or 0,
'images_without_alt': analysis.images_without_alt or 0,
})
if action_type in ('content_calendar', 'content_strategy'):
profiles = db.query(CompanySocialMedia).filter(
CompanySocialMedia.company_id == company.id
).all()
all_platforms = ['facebook', 'instagram', 'linkedin', 'youtube', 'twitter', 'tiktok']
present = [p.platform for p in profiles]
missing = [p for p in all_platforms if p not in present]
prompt_context.update({
'platforms': ', '.join(present) if present else 'brak',
'platforms_present': ', '.join(present) if present else 'brak',
'platforms_missing': ', '.join(missing) if missing else 'brak',
})
# Build prompt from template or generate dynamic fallback
if action_type in CONTENT_PROMPTS:
prompt_template = CONTENT_PROMPTS[action_type]
else:
# Dynamic fallback — look up action title/description from DB
existing_action = db.query(AuditAction).filter_by(
company_id=company_id,
action_type=action_type,
status='suggested'
).order_by(AuditAction.created_at.desc()).first()
action_title = existing_action.title if existing_action else action_type.replace('_', ' ')
action_desc = existing_action.description if existing_action else ''
prompt_template = f"""Jesteś ekspertem od marketingu cyfrowego i SEO dla lokalnych firm w Polsce.
Firma: {{company_name}}
Branża: {{category}}
Miasto: {{city}}
Strona: {{website}}
Usługi: {{services}}
ZADANIE: {action_title}
{('KONTEKST: ' + action_desc) if action_desc else ''}
Przygotuj konkretną, gotową do wdrożenia treść lub instrukcję krok po kroku dla tego zadania.
Pisz po polsku, bezpośrednio do właściciela firmy.
Bądź konkretny — podaj gotowe przykłady kodu, tekstów lub konfiguracji, które właściciel może skopiować i użyć.
Max 500 słów."""
try:
prompt = prompt_template.format(**prompt_context)
except KeyError as e:
prompt = prompt_template # Fall back to raw template if context is incomplete
logger.warning(f"Missing prompt context key: {e}")
# Call Gemini
gemini = _get_gemini_service()
content = gemini.generate_text(
prompt=prompt,
temperature=0.5,
thinking_level='low',
feature='audit_content_generation',
user_id=user_id,
company_id=company_id,
related_entity_type=f'audit_action_{action_type}',
)
if not content:
return {'error': 'Gemini nie zwrócił odpowiedzi'}
# Update action record if it exists
action = db.query(AuditAction).filter_by(
company_id=company_id,
action_type=action_type,
status='suggested'
).order_by(AuditAction.created_at.desc()).first()
if action:
action.ai_content = content
action.ai_model = gemini.model_name
db.commit()
return {
'content': content,
'action_type': action_type,
'model': gemini.model_name,
}
except Exception as e:
db.rollback()
logger.error(f"Error generating content: {e}", exc_info=True)
return {'error': f'Błąd generowania treści: {str(e)}'}
finally:
db.close()
def get_actions_for_company(company_id: int, audit_type: str = None) -> list:
"""
Get all audit actions for a company, optionally filtered by audit type.
Returns list of action dicts sorted by priority.
"""
db = SessionLocal()
try:
query = db.query(AuditAction).filter_by(company_id=company_id)
if audit_type:
query = query.filter_by(audit_type=audit_type)
actions = query.filter(
AuditAction.status.in_(['suggested', 'approved'])
).order_by(AuditAction.created_at.desc()).all()
# Deduplicate by action_type (keep newest)
seen = set()
unique_actions = []
for a in actions:
if a.action_type not in seen:
seen.add(a.action_type)
unique_actions.append(a)
# Sort by priority
priority_order = {'critical': 0, 'high': 1, 'medium': 2, 'low': 3}
unique_actions.sort(key=lambda a: priority_order.get(a.priority, 2))
return [{
'id': a.id,
'action_type': a.action_type,
'title': a.title,
'description': a.description,
'priority': a.priority,
'impact_score': a.impact_score,
'effort_score': a.effort_score,
'ai_content': a.ai_content,
'status': a.status,
'platform': a.platform,
'created_at': a.created_at.isoformat() if a.created_at else None,
} for a in unique_actions]
finally:
db.close()
def update_action_status(action_id: int, new_status: str) -> dict:
"""Update the status of an audit action."""
valid_statuses = ['suggested', 'approved', 'implemented', 'dismissed']
if new_status not in valid_statuses:
return {'error': f'Niepoprawny status. Dozwolone: {", ".join(valid_statuses)}'}
db = SessionLocal()
try:
action = db.query(AuditAction).filter_by(id=action_id).first()
if not action:
return {'error': 'Akcja nie znaleziona'}
action.status = new_status
if new_status == 'implemented':
action.implemented_at = datetime.now()
db.commit()
return {'success': True, 'id': action.id, 'status': new_status}
except Exception as e:
db.rollback()
logger.error(f"Error updating action status: {e}")
return {'error': str(e)}
finally:
db.close()
Reference in New Issue View Git Blame Copy Permalink