# SPDX-License-Identifier: MIT # Copyright (c) 2026 Rug Munch Media LLC # # Part of Pry — https://git.rugmunch.io/RugMunchMedia/pryscraper # Licensed under MIT. See LICENSE. """Pry — Real data-driven auto-reports with PDF generation.""" import logging from datetime import UTC, datetime from html.parser import HTMLParser from pathlib import Path from typing import Any from paths import PRY_DATA_DIR logger = logging.getLogger(__name__) REPORTS_DIR = PRY_DATA_DIR / "reports_real" REPORTS_DIR.mkdir(parents=True, exist_ok=True) try: from reportlab.lib import colors from reportlab.lib.pagesizes import letter from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet from reportlab.lib.units import inch from reportlab.platypus import ( Paragraph, SimpleDocTemplate, Spacer, Table, ) _has_reportlab = True except ImportError: _has_reportlab = False class ReportGenerator: """Generate real data-driven reports with charts and PDF export.""" def __init__(self) -> None: self.styles = getSampleStyleSheet() if _has_reportlab else None def generate( self, report_type: str, data: dict[str, Any], output_format: str = "html", title: str = "Report", branding: dict | None = None, ) -> dict[str, Any]: """Generate a data-driven report from real data. Args: report_type: "competitive", "price_monitor", "seo_audit", "content_tracker", "anomaly_summary" data: Real data from database/cache output_format: "html" or "pdf" title: Report title branding: {"company": "...", "color": "#hex"} """ branding = branding or {} if report_type == "competitive": content = self._competitive_report(data, title, branding) elif report_type == "price_monitor": content = self._price_report(data, title, branding) elif report_type == "seo_audit": content = self._seo_report(data, title, branding) elif report_type == "anomaly_summary": content = self._anomaly_report(data, title, branding) else: return {"error": f"Unknown report type: {report_type}"} if output_format == "pdf" and _has_reportlab: pdf_path = self._to_pdf(content, title, branding) return {"success": True, "path": str(pdf_path), "format": "pdf"} timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S") html_path = REPORTS_DIR / f"{report_type}_{timestamp}.html" html_path.write_text(self._wrap_html(content, title, branding)) return {"success": True, "path": str(html_path), "format": "html"} def _competitive_report(self, data: dict, title: str, branding: dict) -> str: competitors = data.get("competitors", []) changes = data.get("changes", []) body = f"""

Competitive Analysis

Report generated: {datetime.now(UTC).strftime("%B %d, %Y")}

Competitors Tracked ({len(competitors)})

""" for c in competitors: change = c.get("price_change", 0) or 0 change_str = f"{change:+.1f}%" if change else "—" change_class = "up" if change > 0 else "down" if change < 0 else "stable" body += ( f"" f"" f"" f"" ) body += "
NamePriceChangeLast Updated
{c.get('name', '')}${c.get('price', 0):.2f}{change_str}{c.get('last_updated', '')}
" if changes: body += "

Recent Changes

" return body def _price_report(self, data: dict, title: str, branding: dict) -> str: products = data.get("products", []) drops = sum(1 for p in products if (p.get("price_change") or 0) < 0) rises = sum(1 for p in products if (p.get("price_change") or 0) > 0) body = f"""

Price Monitoring Report

Generated: {datetime.now(UTC).strftime("%B %d, %Y")}

{len(products)}

Products

{drops}

Price Drops

{rises}

Price Rises

""" for p in products[:50]: pc = p.get("price_change", 0) or 0 change_str = f"{pc:+.1f}%" if pc else "stable" change_class = "down" if pc < 0 else "up" if pc > 0 else "stable" body += ( f"" f"" f"" f"" ) body += "
ProductPreviousCurrentChange
{p.get('name', '')}${p.get('previous_price', 0):.2f}${p.get('current_price', 0):.2f}{change_str}
" return body def _seo_report(self, data: dict, title: str, branding: dict) -> str: seo = data.get("seo_data", {}) changes = data.get("changes", []) title_text = (seo.get("title") or "N/A")[:100] desc = (seo.get("meta_description") or "N/A")[:200] body = f"""

SEO Audit Report

Generated: {datetime.now(UTC).strftime("%B %d, %Y")}

Current SEO Status

Title{title_text}
Meta Description{desc}
Word Count{seo.get("word_count", 0):,}
Internal Links{seo.get("links_internal", 0)}
External Links{seo.get("links_external", 0)}
Schema Markup{"Yes" if seo.get("has_schema") else "No"}
""" if changes: body += "

Recent SEO Changes

" return body def _anomaly_report(self, data: dict, title: str, branding: dict) -> str: anomalies = data.get("anomalies", []) body = f"""

Anomaly Summary Report

Generated: {datetime.now(UTC).strftime("%B %d, %Y")}

Detected Anomalies ({len(anomalies)})

""" for a in anomalies: body += ( f"" f"" f"" f"" ) body += "
FieldTypeSeverityReason
{a.get('field', '')}{a.get('type', '')}{a.get('severity', '')}{a.get('reason', '')}
" return body def _to_pdf(self, html_content: str, title: str, branding: dict) -> Path: """Convert HTML report to PDF using reportlab.""" timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S") pdf_path = REPORTS_DIR / f"report_{timestamp}.pdf" doc = SimpleDocTemplate(str(pdf_path), pagesize=letter) story: list[Any] = [] assert self.styles is not None title_style = ParagraphStyle( "title", parent=self.styles["Heading1"], textColor=colors.HexColor(branding.get("color", "#f59e0b")), ) story.append(Paragraph(title, title_style)) story.append(Spacer(1, 0.3 * inch)) class SimpleHTMLToPDF(HTMLParser): def __init__(self, story: list[Any], styles: Any) -> None: super().__init__() self.story = story self.styles = styles self.current_row: list[str] = [] self.in_table = False def handle_starttag(self, tag: str, attrs: list) -> None: if tag in ("h2", "h3"): self.story.append(Paragraph(" ", self.styles["Normal"])) elif tag == "tr": self.current_row = [] def handle_endtag(self, tag: str) -> None: if tag == "tr" and self.current_row: import contextlib with contextlib.suppress(Exception): self.story.append( Table( [self.current_row], colWidths=[2 * inch] * len(self.current_row), ) ) elif tag == "table": self.story.append(Spacer(1, 0.1 * inch)) def handle_data(self, data: str) -> None: data = data.strip() if not data: return safe = data.replace("&", "&").replace("<", "<").replace(">", ">") if len(safe) > 100: safe = safe[:100] + "..." if self.in_table: self.current_row.append(safe) else: import contextlib with contextlib.suppress(Exception): self.story.append(Paragraph(safe, self.styles["Normal"])) SimpleHTMLToPDF(story, self.styles).feed(html_content) doc.build(story) return pdf_path def _wrap_html(self, body: str, title: str, branding: dict) -> str: """Wrap report body in full HTML with branding.""" company = branding.get("company", "Pry") color = branding.get("color", "#f59e0b") return f""" {title}

{title}

{body} """