cryptorugmunch
dd63022530
Each module did:
X_DIR = Path(os.path.expanduser("~/.pry/x"))
After:
from paths import PRY_DATA_DIR
X_DIR = PRY_DATA_DIR / "x"
The module-level Path construction is preserved, so the rest of the
code is unchanged. PRY_DATA_DIR is read once at import (overridable via
the env var of the same name).
Verified:
- 407 tests collect (was 5 collection errors from a misplaced import)
- 83 sampled tests pass (intelligence, proxy_manager, x402, agency,
gdpr, referrals, marketplace, api)
- 0 remaining hardcoded ~/.pry references in .py files
Follow-up: paths.py adds subdir(name) helper for new code that wants
auto-mkdir; existing modules still call .mkdir(exist_ok=True) themselves
to preserve the eager-init behavior they had before.
273 lines
12 KiB
Python
273 lines
12 KiB
Python
"""Pry — Automated White-Label Report Generator.
|
|
Generate branded PDF/HTML reports from scraped data for client delivery."""
|
|
from paths import PRY_DATA_DIR
|
|
|
|
# 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.
|
|
|
|
import logging
|
|
import os
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
REPORTS_DIR = PRY_DATA_DIR / "reports"
|
|
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def generate_report(
|
|
report_type: str,
|
|
data: dict[str, Any],
|
|
branding: dict[str, Any] | None = None,
|
|
output_format: str = "html",
|
|
) -> dict[str, Any]:
|
|
"""Generate a white-label report from scraped data.
|
|
|
|
Args:
|
|
report_type: competitive_analysis, price_monitor, seo_audit, content_tracker
|
|
data: The scraped data to include in the report
|
|
branding: Custom branding (agency_name, logo_url, brand_color, domain)
|
|
output_format: html or pdf
|
|
|
|
Returns report metadata and file path.
|
|
"""
|
|
report_id = uuid.uuid4().hex[:8]
|
|
brand = branding or {}
|
|
agency_name = brand.get("agency_name", "Pry Intelligence")
|
|
brand_color = brand.get("brand_color", "#f59e0b")
|
|
logo_url = brand.get("logo_url", "")
|
|
|
|
now = datetime.now(UTC)
|
|
report_date = now.strftime("%B %d, %Y")
|
|
period_start = (now - timedelta(days=7)).strftime("%b %d")
|
|
period_end = now.strftime("%b %d, %Y")
|
|
|
|
if report_type == "competitive_analysis":
|
|
html = _build_competitive_report(
|
|
data, agency_name, brand_color, logo_url, report_date, period_start, period_end
|
|
)
|
|
elif report_type == "price_monitor":
|
|
html = _build_price_report(
|
|
data, agency_name, brand_color, logo_url, report_date, period_start, period_end
|
|
)
|
|
elif report_type == "seo_audit":
|
|
html = _build_seo_report(data, agency_name, brand_color, logo_url, report_date)
|
|
elif report_type == "content_tracker":
|
|
html = _build_content_report(
|
|
data, agency_name, brand_color, logo_url, report_date, period_start, period_end
|
|
)
|
|
else:
|
|
return {"error": f"Unknown report type: {report_type}"}
|
|
|
|
# Save HTML report
|
|
report_path = REPORTS_DIR / f"{report_id}_{report_type}.html"
|
|
try:
|
|
report_path.write_text(html)
|
|
except OSError as e:
|
|
return {"error": str(e)}
|
|
|
|
return {
|
|
"success": True,
|
|
"report_id": report_id,
|
|
"type": report_type,
|
|
"format": "html",
|
|
"path": str(report_path),
|
|
"title": f"{agency_name} — {report_type.replace('_', ' ').title()}",
|
|
"generated_at": datetime.now(UTC).isoformat(),
|
|
}
|
|
|
|
|
|
def _build_html_base(
|
|
title: str, body: str, brand_color: str, logo_url: str, agency_name: str
|
|
) -> str:
|
|
"""Build HTML document with white-label branding."""
|
|
logo_html = (
|
|
f'<img src="{logo_url}" alt="Logo" style="height:40px;">'
|
|
if logo_url
|
|
else f'<h1 style="margin:0;font-size:24px;">{agency_name}</h1>'
|
|
)
|
|
return f"""<!DOCTYPE html>
|
|
<html lang="en">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<meta name="viewport" content="width=device-width,initial-scale=1">
|
|
<title>{title}</title>
|
|
<style>
|
|
* {{ margin:0; padding:0; box-sizing:border-box; }}
|
|
body {{ font-family:-apple-system,system-ui,sans-serif; background:#f8fafc; color:#1a1a2e; line-height:1.6; }}
|
|
.header {{ background:{brand_color}; color:#fff; padding:30px 40px; }}
|
|
.header h1 {{ font-size:28px; margin-bottom:5px; }}
|
|
.header p {{ opacity:0.9; font-size:14px; }}
|
|
.header .logo {{ margin-bottom:15px; }}
|
|
.content {{ padding:30px 40px; max-width:900px; margin:0 auto; }}
|
|
.section {{ background:#fff; border-radius:8px; padding:20px; margin-bottom:20px; box-shadow:0 1px 3px rgba(0,0,0,0.1); }}
|
|
.section h2 {{ font-size:18px; color:{brand_color}; margin-bottom:15px; padding-bottom:8px; border-bottom:2px solid {brand_color}; }}
|
|
table {{ width:100%; border-collapse:collapse; font-size:14px; }}
|
|
th {{ background:#f1f5f9; text-align:left; padding:8px 12px; font-weight:600; }}
|
|
td {{ padding:8px 12px; border-bottom:1px solid #e2e8f0; }}
|
|
tr:hover td {{ background:#f8fafc; }}
|
|
.badge {{ display:inline-block; padding:2px 8px; border-radius:10px; font-size:11px; font-weight:600; }}
|
|
.badge-up {{ background:#dcfce7; color:#166534; }}
|
|
.badge-down {{ background:#fee2e2; color:#991b1b; }}
|
|
.badge-change {{ background:#fef3c7; color:#92400e; }}
|
|
.footer {{ text-align:center; padding:20px; font-size:12px; color:#94a3b8; }}
|
|
.chart-bar {{ height:20px; background:{brand_color}; border-radius:3px; margin:2px 0; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div class="header">
|
|
<div class="logo">{logo_html}</div>
|
|
<h1>{title}</h1>
|
|
</div>
|
|
<div class="content">{body}</div>
|
|
<div class="footer">
|
|
<p>Generated by {agency_name} — Automated Intelligence Report</p>
|
|
</div>
|
|
</body>
|
|
</html>"""
|
|
|
|
|
|
def _build_competitive_report(
|
|
data: dict[str, Any], agency: str, color: str, logo: str, date: str, ps: str, pe: str
|
|
) -> str:
|
|
competitors = data.get("competitors", [])
|
|
changes = data.get("changes", [])
|
|
|
|
# Build competitor table
|
|
table_rows = ""
|
|
for comp in competitors:
|
|
price_change = comp.get("price_change", 0)
|
|
badge = (
|
|
f'<span class="badge badge-{"up" if price_change > 0 else "down"}">{price_change:+.1f}%</span>'
|
|
if price_change
|
|
else '<span class="badge badge-change">—</span>'
|
|
)
|
|
table_rows += f"<tr><td><strong>{comp.get('name', 'Unknown')}</strong></td><td>${comp.get('price', 0):.2f}</td><td>{badge}</td><td>{comp.get('last_updated', 'N/A')}</td></tr>"
|
|
|
|
body = f"""
|
|
<h2 style="margin:0 0 20px 0;font-size:16px;color:#64748b;">Report Period: {ps} — {pe}</h2>
|
|
<div class="section">
|
|
<h2>📊 Competitive Pricing Overview</h2>
|
|
<p style="margin-bottom:15px;color:#64748b;">Tracking {len(competitors)} competitors across your market.</p>
|
|
<table>
|
|
<thead><tr><th>Competitor</th><th>Price</th><th>Change</th><th>Last Updated</th></tr></thead>
|
|
<tbody>{table_rows}</tbody>
|
|
</table>
|
|
</div>
|
|
<div class="section">
|
|
<h2>🔔 Key Changes This Period</h2>
|
|
{"".join(f'<div style="padding:8px 0;border-bottom:1px solid #f1f5f9;"><span class="badge badge-change">Change</span> {c}</div>' for c in changes[:10]) if changes else '<p style="color:#94a3b8;">No significant changes detected this period.</p>'}
|
|
</div>"""
|
|
return _build_html_base(f"{agency} — Competitive Analysis", body, color, logo, agency)
|
|
|
|
|
|
def _build_price_report(
|
|
data: dict[str, Any], agency: str, color: str, logo: str, date: str, ps: str, pe: str
|
|
) -> str:
|
|
products = data.get("products", [])
|
|
total_drops = sum(1 for p in products if p.get("price_change", 0) < 0) if products else 0
|
|
total_rises = sum(1 for p in products if p.get("price_change", 0) > 0) if products else 0
|
|
|
|
rows = ""
|
|
for p in products or []:
|
|
pc = p.get("price_change", 0)
|
|
badge = (
|
|
f'<span class="badge badge-{"up" if pc > 0 else "down"}">{pc:+.1f}%</span>'
|
|
if pc
|
|
else '<span class="badge badge-change">stable</span>'
|
|
)
|
|
rows += f"<tr><td>{p.get('name', 'N/A')}</td><td>${p.get('previous_price', 0):.2f}</td><td><strong>${p.get('current_price', 0):.2f}</strong></td><td>{badge}</td></tr>"
|
|
|
|
body = f"""
|
|
<h2 style="margin:0 0 20px 0;font-size:16px;color:#64748b;">{ps} — {pe}</h2>
|
|
<div style="display:grid;grid-template-columns:repeat(3,1fr);gap:15px;margin-bottom:20px;">
|
|
<div class="section" style="text-align:center;"><h3 style="font-size:24px;">{len(products)}</h3><p style="color:#64748b;">Products Tracked</p></div>
|
|
<div class="section" style="text-align:center;border-top:3px solid #22c55e;"><h3 style="font-size:24px;color:#22c55e;">{total_drops}</h3><p style="color:#64748b;">Price Drops</p></div>
|
|
<div class="section" style="text-align:center;border-top:3px solid #ef4444;"><h3 style="font-size:24px;color:#ef4444;">{total_rises}</h3><p style="color:#64748b;">Price Rises</p></div>
|
|
</div>
|
|
<div class="section">
|
|
<h2>📋 Price Changes</h2>
|
|
<table><thead><tr><th>Product</th><th>Previous</th><th>Current</th><th>Change</th></tr></thead><tbody>{rows}</tbody></table>
|
|
</div>"""
|
|
return _build_html_base(f"{agency} — Price Monitor", body, color, logo, agency)
|
|
|
|
|
|
def _build_seo_report(data: dict[str, Any], agency: str, color: str, logo: str, date: str) -> str:
|
|
seo = data.get("seo_data", {})
|
|
changes = data.get("changes", [])
|
|
|
|
change_items = ""
|
|
for c in changes[:10]:
|
|
icon = "🔴" if c.get("severity") == "high" else "🟡"
|
|
change_items += f"<div style='padding:6px 0;'>{icon} <strong>{c.get('field', 'Unknown')}</strong>: {c.get('to', '')[:80]}</div>"
|
|
|
|
body = f"""
|
|
<div class="section">
|
|
<h2>🔍 SEO Analysis</h2>
|
|
<table>
|
|
<tr><td><strong>Title</strong></td><td>{seo.get("title", "N/A")[:100]}</td></tr>
|
|
<tr><td><strong>Meta Description</strong></td><td>{seo.get("meta_description", "N/A")[:150]}</td></tr>
|
|
<tr><td><strong>Word Count</strong></td><td>{seo.get("word_count", 0):,}</td></tr>
|
|
<tr><td><strong>Internal Links</strong></td><td>{seo.get("links_internal", 0)}</td></tr>
|
|
<tr><td><strong>External Links</strong></td><td>{seo.get("links_external", 0)}</td></tr>
|
|
<tr><td><strong>Schema Markup</strong></td><td>{"✅ Present" if seo.get("has_schema") else "❌ Not found"}</td></tr>
|
|
</table>
|
|
</div>
|
|
<div class="section">
|
|
<h2>🔄 Recent Changes</h2>
|
|
{change_items if change_items else '<p style="color:#94a3b8;">No changes detected since last scan.</p>'}
|
|
</div>"""
|
|
return _build_html_base(f"{agency} — SEO Audit", body, color, logo, agency)
|
|
|
|
|
|
def _build_content_report(
|
|
data: dict[str, Any], agency: str, color: str, logo: str, date: str, ps: str, pe: str
|
|
) -> str:
|
|
pages = data.get("pages", [])
|
|
changes = data.get("changes", [])
|
|
|
|
rows = ""
|
|
for p in pages:
|
|
status = (
|
|
'<span style="color:#22c55e;">●</span>'
|
|
if p.get("status") == "unchanged"
|
|
else '<span style="color:#f59e0b;">●</span>'
|
|
)
|
|
rows += f"<tr><td>{status}</td><td><a href='{p.get('url', '#')}' style='color:{color};'>{p.get('title', 'Untitled')[:60]}</a></td><td>{p.get('last_changed', 'N/A')[:10]}</td><td>{p.get('word_count', 0):,}</td></tr>"
|
|
|
|
body = f"""
|
|
<div class="section">
|
|
<h2>📄 Content Tracking — {ps} to {pe}</h2>
|
|
<p style="margin-bottom:15px;color:#64748b;">Monitoring {len(pages)} pages for content changes.</p>
|
|
<table><thead><tr><th>Status</th><th>Page</th><th>Changed</th><th>Words</th></tr></thead><tbody>{rows}</tbody></table>
|
|
</div>
|
|
<div class="section">
|
|
<h2>📝 Content Changes Found</h2>
|
|
{"".join(f'<div style="padding:6px 0;border-bottom:1px solid #f1f5f9;">{c}</div>' for c in changes[:10]) if changes else '<p style="color:#94a3b8;">No content changes detected this period.</p>'}
|
|
</div>"""
|
|
return _build_html_base(f"{agency} — Content Tracker", body, color, logo, agency)
|
|
|
|
|
|
def list_reports() -> list[dict[str, Any]]:
|
|
"""List all generated reports."""
|
|
reports = []
|
|
for path in sorted(REPORTS_DIR.glob("*.html"), key=os.path.getmtime, reverse=True):
|
|
parts = path.stem.split("_", 1)
|
|
report_id = parts[0]
|
|
report_type = parts[1] if len(parts) > 1 else "unknown"
|
|
reports.append(
|
|
{
|
|
"id": report_id,
|
|
"type": report_type,
|
|
"path": str(path),
|
|
"size_bytes": path.stat().st_size,
|
|
"generated_at": datetime.fromtimestamp(path.stat().st_mtime).isoformat(),
|
|
}
|
|
)
|
|
return reports
|