fix(social-audit): Use SocialMediaAuditor for comprehensive audit

- Import and use SocialMediaAuditor from scripts/social_media_audit.py
- Audit now scans website HTML for social media links
- Discovers profiles via Brave Search API (if configured)
- Fetches Google Business Profile data via Google Places API
- Saves discovered profiles to database

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Maciej Pienczyn 2026-01-09 05:12:25 +01:00
parent cadf91b481
commit deed279521

135
app.py
View File

@ -4166,8 +4166,11 @@ def api_social_audit_trigger():
""" """
API: Trigger Social Media audit for a company. API: Trigger Social Media audit for a company.
This endpoint verifies social media profile URLs and updates their status. This endpoint performs a comprehensive social media audit:
It checks if URLs are valid and accessible. - Scans company website for social media links
- Searches for profiles via Brave Search API (if configured)
- Fetches Google Business Profile data
- Updates database with discovered profiles
Request JSON body: Request JSON body:
- company_id: Company ID (integer) OR - company_id: Company ID (integer) OR
@ -4179,9 +4182,22 @@ def api_social_audit_trigger():
Rate limited to 10 requests per hour per user. Rate limited to 10 requests per hour per user.
""" """
import requests as http_requests # Import the SocialMediaAuditor from scripts
try:
import sys
from pathlib import Path
scripts_dir = Path(__file__).parent / 'scripts'
if str(scripts_dir) not in sys.path:
sys.path.insert(0, str(scripts_dir))
from social_media_audit import SocialMediaAuditor
except ImportError as e:
logger.error(f"Failed to import SocialMediaAuditor: {e}")
return jsonify({
'success': False,
'error': 'Usługa audytu Social Media jest niedostępna. Sprawdź konfigurację serwera.'
}), 503
# Admin or company owner check # Parse request data
data = request.get_json() data = request.get_json()
if not data: if not data:
return jsonify({ return jsonify({
@ -4222,73 +4238,66 @@ def api_social_audit_trigger():
logger.info(f"Social Media audit triggered by {current_user.email} for company: {company.name} (ID: {company.id})") logger.info(f"Social Media audit triggered by {current_user.email} for company: {company.name} (ID: {company.id})")
# Get existing social media profiles # Prepare company dict for auditor
profiles = db.query(CompanySocialMedia).filter( company_dict = {
CompanySocialMedia.company_id == company.id 'id': company.id,
).all() 'name': company.name,
'slug': company.slug,
'website': company.website,
'address_city': company.address_city or 'Wejherowo'
}
verified_count = 0 # Initialize auditor and run audit
errors = [] try:
auditor = SocialMediaAuditor()
audit_result = auditor.audit_company(company_dict)
# Verify each profile URL # Check for errors
for profile in profiles: if audit_result.get('errors') and not audit_result.get('social_media') and not audit_result.get('website'):
try: return jsonify({
response = http_requests.head( 'success': False,
profile.url, 'error': f'Audyt nie powiódł się: {", ".join(audit_result["errors"][:3])}',
timeout=10, 'company_id': company.id,
allow_redirects=True, 'company_name': company.name
headers={'User-Agent': 'Mozilla/5.0 (compatible; NordaBizBot/1.0)'} }), 422
)
if response.status_code == 200: # Save result to database
profile.is_valid = True saved = auditor.save_audit_result(audit_result)
profile.check_status = 'ok'
verified_count += 1
elif response.status_code in [301, 302, 303, 307, 308]:
profile.is_valid = True
profile.check_status = 'redirect'
verified_count += 1
elif response.status_code == 404:
profile.is_valid = False
profile.check_status = '404'
else:
profile.is_valid = True # Assume valid if we get any response
profile.check_status = f'http_{response.status_code}'
verified_count += 1
profile.last_checked_at = datetime.now() if not saved:
return jsonify({
'success': False,
'error': 'Audyt został wykonany, ale nie udało się zapisać wyników do bazy danych.',
'company_id': company.id,
'company_name': company.name
}), 500
except http_requests.exceptions.Timeout: # Get count of social media profiles found
profile.check_status = 'timeout' social_media_found = audit_result.get('social_media', {})
profile.last_checked_at = datetime.now() platforms_count = len(social_media_found)
errors.append(f'{profile.platform}: timeout')
except http_requests.exceptions.ConnectionError:
profile.check_status = 'connection_error'
profile.last_checked_at = datetime.now()
errors.append(f'{profile.platform}: connection error')
except Exception as e:
profile.check_status = 'error'
profile.last_checked_at = datetime.now()
errors.append(f'{profile.platform}: {str(e)[:50]}')
db.commit() # Calculate score
all_platforms = ['facebook', 'instagram', 'linkedin', 'youtube', 'twitter', 'tiktok']
score = int((platforms_count / len(all_platforms)) * 100)
# Calculate score return jsonify({
all_platforms = ['facebook', 'instagram', 'linkedin', 'youtube', 'twitter', 'tiktok'] 'success': True,
profiles_dict = {p.platform: p for p in profiles} 'message': f'Audyt Social Media zakończony. Znaleziono {platforms_count} profili.',
platforms_with_profiles = len([p for p in all_platforms if p in profiles_dict]) 'company_id': company.id,
score = int((platforms_with_profiles / len(all_platforms)) * 100) 'company_name': company.name,
'profiles_found': platforms_count,
'platforms': list(social_media_found.keys()),
'score': score,
'google_reviews': audit_result.get('google_reviews', {}),
'errors': audit_result.get('errors') if audit_result.get('errors') else None
}), 200
return jsonify({ except Exception as e:
'success': True, logger.error(f"Social Media audit error for company {company.id}: {e}")
'message': f'Audyt Social Media zakończony. Zweryfikowano {verified_count} z {len(profiles)} profili.', return jsonify({
'company_id': company.id, 'success': False,
'company_name': company.name, 'error': f'Błąd podczas audytu: {str(e)}'
'profiles_count': len(profiles), }), 500
'verified_count': verified_count,
'score': score,
'errors': errors if errors else None
}), 200
except Exception as e: except Exception as e:
logger.error(f"Social Media audit error for company {slug or company_id}: {e}") logger.error(f"Social Media audit error for company {slug or company_id}: {e}")