maciejpi/nordabiz
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
3d462d52f8
nordabiz/blueprints/api/routes_seo_audit.py
Maciej Pienczyn 3d462d52f8
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
fix: handle proxy timeout in SEO audit and return JSON on 404 
- Check content-type before parsing JSON in runAudit() to show
  helpful message when NPM proxy times out (returns HTML)
- Replace abort(404) with jsonify in audit trigger endpoint

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-18 20:12:44 +01:00

552 lines
21 KiB
Python
Raw Blame History

"""
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 abort, jsonify, request, current_app
from flask_login import current_user, login_required
from utils.decorators import is_audit_owner
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/<slug> 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
"""
if not is_audit_owner():
return jsonify({'success': False, 'error': 'Nie znaleziono'}), 404
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/<slug>')
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
"""
if not is_audit_owner():
return jsonify({'success': False, 'error': 'Nie znaleziono'}), 404
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
"""
if not is_audit_owner():
return jsonify({'success': False, 'error': 'Nie znaleziono'}), 404
# 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()
Reference in New Issue View Git Blame Copy Permalink