#!/usr/bin/env python3 """ Unified Audit Report Generator for NordaBiz ============================================= Generates comprehensive audit reports combining: - Social Media audit data - Google Business Profile audit data - SEO audit data - Competitor monitoring data (if available) Usage: python generate_audit_report.py --company-id 26 python generate_audit_report.py --all python generate_audit_report.py --company-id 26 --type social Author: NordaBiz Development Team Created: 2026-02-06 """ import os import sys import json import argparse import logging from datetime import datetime, date, timedelta from typing import Optional, Dict, List, Any from pathlib import Path # Load .env file try: from dotenv import load_dotenv script_dir = Path(__file__).resolve().parent project_root = script_dir.parent env_path = project_root / '.env' if env_path.exists(): load_dotenv(env_path) except ImportError: pass sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker from database import ( Company, CompanySocialMedia, GBPAudit, CompanyWebsiteAnalysis, CompanyCompetitor, CompetitorSnapshot, AuditReport, SessionLocal ) logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) DATABASE_URL = os.getenv( 'DATABASE_URL', 'postgresql://nordabiz_app:CHANGE_ME@127.0.0.1:5432/nordabiz' ) class AuditReportGenerator: """Generates unified audit reports for companies.""" def __init__(self, database_url: str = DATABASE_URL): self.engine = create_engine(database_url) self.Session = sessionmaker(bind=self.engine) def generate_report(self, company_id: int, report_type: str = 'full') -> Dict[str, Any]: """Generate a comprehensive audit report for a company.""" with self.Session() as session: company = session.query(Company).filter(Company.id == company_id).first() if not company: raise ValueError(f"Company {company_id} not found") logger.info(f"Generating {report_type} report for: {company.name} (ID: {company_id})") report_data = { 'company': { 'id': company.id, 'name': company.name, 'slug': company.slug, 'website': company.website, 'city': company.address_city, 'category': company.category.name if company.category else None, }, 'generated_at': datetime.now().isoformat(), 'report_type': report_type, 'sections': {}, 'scores': {}, } sections_included = {'social': False, 'gbp': False, 'seo': False, 'competitors': False} # Social Media section if report_type in ('full', 'social'): social_data = self._get_social_data(session, company_id) if social_data: report_data['sections']['social'] = social_data report_data['scores']['social'] = social_data.get('average_completeness', 0) sections_included['social'] = True # GBP section if report_type in ('full', 'gbp'): gbp_data = self._get_gbp_data(session, company_id) if gbp_data: report_data['sections']['gbp'] = gbp_data report_data['scores']['gbp'] = gbp_data.get('completeness_score', 0) sections_included['gbp'] = True # SEO section if report_type in ('full', 'seo'): seo_data = self._get_seo_data(session, company_id) if seo_data: report_data['sections']['seo'] = seo_data report_data['scores']['seo'] = seo_data.get('overall_score', 0) sections_included['seo'] = True # Competitors section if report_type == 'full': competitor_data = self._get_competitor_data(session, company_id) if competitor_data: report_data['sections']['competitors'] = competitor_data sections_included['competitors'] = True # Calculate overall score scores = [v for v in report_data['scores'].values() if v and v > 0] overall = int(sum(scores) / len(scores)) if scores else 0 report_data['scores']['overall'] = overall # Save report report = AuditReport( company_id=company_id, report_type=report_type, period_start=date.today() - timedelta(days=30), period_end=date.today(), overall_score=overall, social_score=report_data['scores'].get('social'), gbp_score=report_data['scores'].get('gbp'), seo_score=report_data['scores'].get('seo'), sections=sections_included, data=report_data, generated_by='system', status='draft', ) session.add(report) session.commit() session.refresh(report) report_data['report_id'] = report.id logger.info(f"Report #{report.id} generated. Overall score: {overall}/100") return report_data def _get_social_data(self, session, company_id: int) -> Optional[Dict]: """Get social media audit data.""" profiles = session.query(CompanySocialMedia).filter( CompanySocialMedia.company_id == company_id, CompanySocialMedia.is_valid == True ).all() if not profiles: return None platforms = [] total_completeness = 0 for p in profiles: platform_data = { 'platform': p.platform, 'url': p.url, 'page_name': p.page_name, 'followers_count': p.followers_count, 'has_bio': p.has_bio, 'has_profile_photo': p.has_profile_photo, 'completeness_score': p.profile_completeness_score or 0, 'last_checked': p.last_checked_at.isoformat() if p.last_checked_at else None, } platforms.append(platform_data) total_completeness += (p.profile_completeness_score or 0) average = int(total_completeness / len(platforms)) if platforms else 0 return { 'platforms_found': len(platforms), 'platforms': platforms, 'average_completeness': average, 'missing_platforms': self._find_missing_platforms(profiles), } @staticmethod def _find_missing_platforms(profiles) -> List[str]: """Find platforms without profiles.""" all_platforms = {'facebook', 'instagram', 'linkedin', 'youtube', 'twitter', 'tiktok'} found = {p.platform for p in profiles} return sorted(all_platforms - found) def _get_gbp_data(self, session, company_id: int) -> Optional[Dict]: """Get GBP audit data.""" audit = session.query(GBPAudit).filter( GBPAudit.company_id == company_id ).order_by(GBPAudit.audit_date.desc()).first() if not audit: return None return { 'completeness_score': audit.completeness_score, 'score_category': audit.score_category, 'audit_date': audit.audit_date.isoformat() if audit.audit_date else None, 'fields_status': audit.fields_status, 'review_count': audit.review_count, 'average_rating': float(audit.average_rating) if audit.average_rating else None, 'photo_count': audit.photo_count, 'nap_consistent': audit.nap_consistent, 'review_response_rate': float(audit.review_response_rate) if audit.review_response_rate else None, 'review_sentiment': audit.review_sentiment, 'recommendations': audit.recommendations or [], } def _get_seo_data(self, session, company_id: int) -> Optional[Dict]: """Get SEO audit data.""" analysis = session.query(CompanyWebsiteAnalysis).filter( CompanyWebsiteAnalysis.company_id == company_id ).first() if not analysis: return None return { 'overall_score': analysis.seo_overall_score, 'pagespeed_seo': analysis.pagespeed_seo_score, 'pagespeed_performance': analysis.pagespeed_performance_score, 'pagespeed_accessibility': analysis.pagespeed_accessibility_score, 'meta_title': analysis.meta_title, 'has_structured_data': analysis.has_structured_data, 'has_sitemap': analysis.has_sitemap, 'has_robots_txt': analysis.has_robots_txt, 'is_mobile_friendly': analysis.is_mobile_friendly, 'local_seo_score': analysis.local_seo_score, 'has_local_business_schema': analysis.has_local_business_schema, 'citations_count': analysis.citations_count, 'content_freshness_score': analysis.content_freshness_score, '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, }, 'seo_issues': analysis.seo_issues, } def _get_competitor_data(self, session, company_id: int) -> Optional[Dict]: """Get competitor monitoring data.""" competitors = session.query(CompanyCompetitor).filter( CompanyCompetitor.company_id == company_id, CompanyCompetitor.is_active == True ).all() if not competitors: return None competitor_list = [] for comp in competitors: # Get latest snapshot latest = session.query(CompetitorSnapshot).filter( CompetitorSnapshot.competitor_id == comp.id ).order_by(CompetitorSnapshot.snapshot_date.desc()).first() competitor_list.append({ 'name': comp.competitor_name, 'rating': float(comp.competitor_rating) if comp.competitor_rating else None, 'review_count': comp.competitor_review_count, 'category': comp.competitor_category, 'latest_changes': latest.changes if latest else None, }) return { 'total_tracked': len(competitors), 'competitors': competitor_list, } def main(): parser = argparse.ArgumentParser(description='Generate Unified Audit Report') parser.add_argument('--company-id', type=int, help='Generate report for specific company') parser.add_argument('--all', action='store_true', help='Generate for all active companies') parser.add_argument('--type', choices=['full', 'social', 'gbp', 'seo'], default='full') parser.add_argument('--json', action='store_true', help='Output JSON to stdout') parser.add_argument('--verbose', '-v', action='store_true') args = parser.parse_args() if args.verbose: logging.getLogger().setLevel(logging.DEBUG) generator = AuditReportGenerator() if args.company_id: report = generator.generate_report(args.company_id, args.type) if args.json: print(json.dumps(report, default=str, indent=2, ensure_ascii=False)) else: print(f"\nReport #{report.get('report_id')} generated") print(f"Overall score: {report['scores'].get('overall', 0)}/100") for section, data in report.get('sections', {}).items(): print(f" {section}: included") elif args.all: engine = create_engine(DATABASE_URL) Session = sessionmaker(bind=engine) with Session() as session: companies = session.query(Company).filter(Company.status == 'active').all() company_ids = [c.id for c in companies] for cid in company_ids: try: report = generator.generate_report(cid, args.type) print(f"Company {cid}: score={report['scores'].get('overall', 0)}") except Exception as e: logger.error(f"Company {cid} failed: {e}") else: parser.print_help() sys.exit(1) if __name__ == '__main__': main()