# 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)})
| Name | Price | Change | Last Updated |
"""
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"| {c.get('name', '')} | "
f"${c.get('price', 0):.2f} | "
f"{change_str} | "
f"{c.get('last_updated', '')} |
"
)
body += "
"
if changes:
body += "Recent Changes
"
for c in changes[:20]:
body += f"- {c}
"
body += "
"
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")}
| Product | Previous | Current | Change |
"""
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"| {p.get('name', '')} | "
f"${p.get('previous_price', 0):.2f} | "
f"${p.get('current_price', 0):.2f} | "
f"{change_str} |
"
)
body += "
"
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
"
for c in changes[:20]:
body += f"- {c.get('field', '')}: {str(c.get('to', ''))[:80]}
"
body += "
"
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)})
| Field | Type | Severity | Reason |
"""
for a in anomalies:
body += (
f"| {a.get('field', '')} | "
f"{a.get('type', '')} | "
f"{a.get('severity', '')} | "
f"{a.get('reason', '')} |
"
)
body += "
"
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}
"""