""" SEO Audit API Routes - API blueprint Migrated from app.py as part of the blueprint refactoring. Contains API routes for SEO audit functionality. """ import logging import sys from datetime import datetime from pathlib import Path from flask import jsonify, request, current_app from flask_login import current_user, login_required from database import SessionLocal, Company, CompanyWebsiteAnalysis from . import bp logger = logging.getLogger(__name__) # Check if SEO audit service is available try: scripts_dir = Path(__file__).parent.parent.parent / 'scripts' if str(scripts_dir) not in sys.path: sys.path.insert(0, str(scripts_dir)) from seo_audit import SEOAuditor SEO_AUDIT_AVAILABLE = True SEO_AUDIT_VERSION = '2.0' except ImportError as e: logger.warning(f"SEO audit service not available: {e}") SEO_AUDIT_AVAILABLE = False SEO_AUDIT_VERSION = None def get_limiter(): """Get rate limiter from current app.""" return current_app.extensions.get('limiter') # ============================================================ # SEO AUDIT HELPER FUNCTIONS # ============================================================ def _build_seo_audit_response(company, analysis): """ Helper function to build SEO audit response JSON. Used by both /api/seo/audit and /api/seo/audit/ endpoints. """ # Build issues list from various checks issues = [] # Check for images without alt if analysis.images_without_alt and analysis.images_without_alt > 0: issues.append({ 'severity': 'warning', 'message': f'{analysis.images_without_alt} obrazów nie ma atrybutu alt', 'category': 'accessibility' }) # Check for missing meta description if not analysis.meta_description: issues.append({ 'severity': 'warning', 'message': 'Brak meta description', 'category': 'on_page' }) # Check H1 count (should be exactly 1) if analysis.h1_count is not None: if analysis.h1_count == 0: issues.append({ 'severity': 'error', 'message': 'Brak nagłówka H1 na stronie', 'category': 'on_page' }) elif analysis.h1_count > 1: issues.append({ 'severity': 'warning', 'message': f'Strona zawiera {analysis.h1_count} nagłówków H1 (zalecany: 1)', 'category': 'on_page' }) # Check SSL if analysis.has_ssl is False: issues.append({ 'severity': 'error', 'message': 'Strona nie używa HTTPS (brak certyfikatu SSL)', 'category': 'security' }) # Check robots.txt if analysis.has_robots_txt is False: issues.append({ 'severity': 'info', 'message': 'Brak pliku robots.txt', 'category': 'technical' }) # Check sitemap if analysis.has_sitemap is False: issues.append({ 'severity': 'info', 'message': 'Brak pliku sitemap.xml', 'category': 'technical' }) # Check indexability if analysis.is_indexable is False: issues.append({ 'severity': 'error', 'message': f'Strona nie jest indeksowalna: {analysis.noindex_reason or "nieznana przyczyna"}', 'category': 'technical' }) # Check structured data if analysis.has_structured_data is False: issues.append({ 'severity': 'info', 'message': 'Brak danych strukturalnych (Schema.org)', 'category': 'on_page' }) # Check Open Graph tags if analysis.has_og_tags is False: issues.append({ 'severity': 'info', 'message': 'Brak tagów Open Graph (ważne dla udostępniania w social media)', 'category': 'social' }) # Check mobile-friendliness if analysis.is_mobile_friendly is False: issues.append({ 'severity': 'warning', 'message': 'Strona nie jest przyjazna dla urządzeń mobilnych', 'category': 'technical' }) # Add issues from seo_issues JSONB field if available if analysis.seo_issues: stored_issues = analysis.seo_issues if isinstance(analysis.seo_issues, list) else [] for issue in stored_issues: if isinstance(issue, dict): issues.append(issue) # Build response return { 'success': True, 'company_id': company.id, 'company_name': company.name, 'website': company.website, 'seo_audit': { 'audited_at': analysis.seo_audited_at.isoformat() if analysis.seo_audited_at else None, 'audit_version': analysis.seo_audit_version, 'overall_score': analysis.seo_overall_score, 'pagespeed': { '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': analysis.meta_title, 'meta_description': analysis.meta_description, 'h1_count': analysis.h1_count, 'h1_text': analysis.h1_text, 'h2_count': analysis.h2_count, 'h3_count': analysis.h3_count, 'total_images': analysis.total_images, 'images_without_alt': analysis.images_without_alt, 'images_with_alt': analysis.images_with_alt, 'internal_links_count': analysis.internal_links_count, 'external_links_count': analysis.external_links_count, 'has_structured_data': analysis.has_structured_data, 'structured_data_types': analysis.structured_data_types }, 'technical': { 'has_ssl': analysis.has_ssl, 'ssl_issuer': analysis.ssl_issuer, 'ssl_expires_at': analysis.ssl_expires_at.isoformat() if analysis.ssl_expires_at else None, 'has_sitemap': analysis.has_sitemap, 'has_robots_txt': analysis.has_robots_txt, 'has_canonical': analysis.has_canonical, 'canonical_url': analysis.canonical_url, 'is_indexable': analysis.is_indexable, 'noindex_reason': analysis.noindex_reason, 'is_mobile_friendly': analysis.is_mobile_friendly, 'viewport_configured': analysis.viewport_configured, 'load_time_ms': analysis.load_time_ms, 'http_status_code': analysis.http_status_code }, 'core_web_vitals': { 'largest_contentful_paint_ms': analysis.largest_contentful_paint_ms, 'interaction_to_next_paint_ms': analysis.interaction_to_next_paint_ms, 'cumulative_layout_shift': float(analysis.cumulative_layout_shift) if analysis.cumulative_layout_shift else None }, 'social': { 'has_og_tags': analysis.has_og_tags, 'og_title': analysis.og_title, 'og_description': analysis.og_description, 'og_image': analysis.og_image, 'has_twitter_cards': analysis.has_twitter_cards }, 'language': { 'html_lang': analysis.html_lang, 'has_hreflang': analysis.has_hreflang }, 'issues': issues } } def _get_seo_audit_for_company(db, company): """ Helper function to get SEO audit data for a company. Returns tuple of (response_dict, status_code). """ # Get latest SEO audit for this company analysis = db.query(CompanyWebsiteAnalysis).filter_by( company_id=company.id ).order_by(CompanyWebsiteAnalysis.analyzed_at.desc()).first() if not analysis: return { 'success': True, 'company_id': company.id, 'company_name': company.name, 'website': company.website, 'seo_audit': None, 'message': 'Brak danych SEO dla tej firmy. Audyt nie został jeszcze przeprowadzony.' }, 200 # Check if SEO audit was performed (seo_audited_at is set) if not analysis.seo_audited_at: return { 'success': True, 'company_id': company.id, 'company_name': company.name, 'website': company.website, 'seo_audit': None, 'message': 'Audyt SEO nie został jeszcze przeprowadzony dla tej firmy.' }, 200 # Build full response return _build_seo_audit_response(company, analysis), 200 # ============================================================ # SEO AUDIT API ROUTES # ============================================================ @bp.route('/seo/audit') def api_seo_audit(): """ API: Get SEO audit results for a company. Query parameters: - company_id: Company ID (integer) - slug: Company slug (string) At least one of company_id or slug must be provided. Returns JSON with: - pagespeed scores (seo, performance, accessibility, best_practices) - on_page metrics (meta tags, headings, images, links, structured data) - technical checks (ssl, sitemap, robots.txt, mobile-friendly) - issues list with severity levels """ company_id = request.args.get('company_id', type=int) slug = request.args.get('slug', type=str) if not company_id and not slug: return jsonify({ 'success': False, 'error': 'Podaj company_id lub slug firmy' }), 400 db = SessionLocal() try: # Find company by ID or slug if company_id: company = db.query(Company).filter_by(id=company_id, status='active').first() else: company = db.query(Company).filter_by(slug=slug, status='active').first() if not company: return jsonify({ 'success': False, 'error': 'Firma nie znaleziona' }), 404 response, status_code = _get_seo_audit_for_company(db, company) return jsonify(response), status_code finally: db.close() @bp.route('/seo/audit/') def api_seo_audit_by_slug(slug): """ API: Get SEO audit results for a company by slug. Convenience endpoint that uses slug from URL path. Example: GET /api/seo/audit/pixlab-sp-z-o-o """ db = SessionLocal() try: # Find company by slug company = db.query(Company).filter_by(slug=slug, status='active').first() if not company: return jsonify({ 'success': False, 'error': 'Firma nie znaleziona' }), 404 response, status_code = _get_seo_audit_for_company(db, company) return jsonify(response), status_code finally: db.close() @bp.route('/seo/audit', methods=['POST']) @login_required def api_seo_audit_trigger(): """ API: Trigger SEO audit for a company (admin-only). This endpoint runs a full SEO audit including: - Google PageSpeed Insights analysis - On-page SEO analysis (meta tags, headings, images, links) - Technical SEO checks (robots.txt, sitemap, canonical URLs) Request JSON body: - company_id: Company ID (integer) OR - slug: Company slug (string) Returns: - Success: Full SEO audit results saved to database - Error: Error message with status code """ # Check admin panel access if not current_user.can_access_admin_panel(): return jsonify({ 'success': False, 'error': 'Brak uprawnień. Tylko administrator może uruchamiać audyty SEO.' }), 403 # Check if SEO audit service is available if not SEO_AUDIT_AVAILABLE: return jsonify({ 'success': False, 'error': 'Usługa audytu SEO jest niedostępna. Sprawdź konfigurację serwera.' }), 503 # Parse request data data = request.get_json() if not data: return jsonify({ 'success': False, 'error': 'Brak danych w żądaniu. Podaj company_id lub slug.' }), 400 company_id = data.get('company_id') slug = data.get('slug') if not company_id and not slug: return jsonify({ 'success': False, 'error': 'Podaj company_id lub slug firmy do audytu.' }), 400 db = SessionLocal() try: # Find company by ID or slug if company_id: company = db.query(Company).filter_by(id=company_id, status='active').first() else: company = db.query(Company).filter_by(slug=slug, status='active').first() if not company: return jsonify({ 'success': False, 'error': 'Firma nie znaleziona lub nieaktywna.' }), 404 # Check if company has a website if not company.website: return jsonify({ 'success': False, 'error': f'Firma "{company.name}" nie ma zdefiniowanej strony internetowej.', 'company_id': company.id, 'company_name': company.name }), 400 logger.info(f"SEO audit triggered by admin {current_user.email} for company: {company.name} (ID: {company.id})") # Initialize SEO auditor and run audit try: auditor = SEOAuditor() # Prepare company dict for auditor company_dict = { 'id': company.id, 'name': company.name, 'slug': company.slug, 'website': company.website, 'address_city': company.address_city } # Run the audit audit_result = auditor.audit_company(company_dict) # Check for errors if audit_result.get('errors') and not audit_result.get('onpage') and not audit_result.get('pagespeed'): return jsonify({ 'success': False, 'error': f'Audyt nie powiódł się: {", ".join(audit_result["errors"])}', 'company_id': company.id, 'company_name': company.name, 'website': company.website }), 422 # Save result to database saved = auditor.save_audit_result(audit_result) 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 # Enrich with OAuth data (Search Console) if available try: from oauth_service import OAuthService from search_console_service import SearchConsoleService oauth = OAuthService() gsc_token = oauth.get_valid_token(db, company.id, 'google', 'search_console') if gsc_token and company.website: gsc = SearchConsoleService(gsc_token) gsc_data = gsc.get_search_analytics(company.website, days=28) if gsc_data: # Update the analysis record with GSC data analysis_record = db.query(CompanyWebsiteAnalysis).filter_by( company_id=company.id ).first() if analysis_record: # Basic metrics analysis_record.gsc_clicks = gsc_data.get('clicks') analysis_record.gsc_impressions = gsc_data.get('impressions') analysis_record.gsc_ctr = gsc_data.get('ctr') analysis_record.gsc_avg_position = gsc_data.get('position') analysis_record.gsc_top_queries = gsc_data.get('top_queries', []) analysis_record.gsc_top_pages = gsc_data.get('top_pages', []) analysis_record.gsc_period_days = gsc_data.get('period_days', 28) # Extended GSC data collection try: # Device breakdown device_data = gsc.get_device_breakdown(company.website, days=28) if device_data: analysis_record.gsc_device_breakdown = device_data # Country breakdown country_data = gsc.get_country_breakdown(company.website, days=28) if country_data: analysis_record.gsc_country_breakdown = country_data # Search type breakdown type_data = gsc.get_search_type_breakdown(company.website, days=28) if type_data: analysis_record.gsc_search_type_breakdown = type_data # Trend data (period-over-period) trend_data = gsc.get_trend_data(company.website, days=28) if trend_data: analysis_record.gsc_trend_data = trend_data # URL Inspection (for homepage) homepage = company.website if homepage and not homepage.endswith('/'): homepage += '/' inspection = gsc.inspect_url(company.website, homepage) if inspection: analysis_record.gsc_index_status = inspection.get('index_status') last_crawl = inspection.get('last_crawl') if last_crawl: try: from datetime import datetime as dt analysis_record.gsc_last_crawl = dt.fromisoformat(last_crawl.replace('Z', '+00:00')) except (ValueError, TypeError): pass analysis_record.gsc_crawled_as = inspection.get('crawled_as') # Sitemaps sitemaps = gsc.get_sitemaps(company.website) if sitemaps: analysis_record.gsc_sitemaps = sitemaps except Exception as ext_err: logger.warning(f"Extended GSC data collection failed for company {company.id}: {ext_err}") db.commit() logger.info(f"GSC data saved for company {company.id}: {gsc_data.get('clicks', 0)} clicks") except ImportError: pass except Exception as e: logger.warning(f"GSC enrichment failed for company {company.id}: {e}") # Get the updated analysis record to return db.expire_all() # Refresh the session to get updated data analysis = db.query(CompanyWebsiteAnalysis).filter_by( company_id=company.id ).order_by(CompanyWebsiteAnalysis.analyzed_at.desc()).first() # Build response using the existing helper function response = _build_seo_audit_response(company, analysis) return jsonify({ 'success': True, 'message': f'Audyt SEO dla firmy "{company.name}" został zakończony pomyślnie.', 'audit_version': SEO_AUDIT_VERSION, 'triggered_by': current_user.email, 'triggered_at': datetime.now().isoformat(), **response }), 200 except Exception as e: logger.error(f"SEO audit error for company {company.id}: {e}") return jsonify({ 'success': False, 'error': f'Błąd podczas wykonywania audytu: {str(e)}', 'company_id': company.id, 'company_name': company.name }), 500 finally: db.close()