feat: Phase 18 — PDF/HTML intelligence reports (110 tools, 344 tests)
This commit is contained in:
Marc Shade
+438
-105
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,735 @@
|
||||
"""PDF intelligence report generator.
|
||||
|
||||
Renders multi-domain intelligence summaries as styled PDF documents
|
||||
using WeasyPrint. Data is pulled from the same source modules and
|
||||
analysis engines used by the MCP server and dashboard.
|
||||
|
||||
Optional dependency: ``pip install -e ".[pdf]"`` (weasyprint>=62.0).
|
||||
Requires native pango/gobject libs (``brew install pango`` on macOS).
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .cache import Cache
|
||||
from .circuit_breaker import CircuitBreaker
|
||||
from .fetcher import Fetcher
|
||||
from .sources import (
|
||||
markets,
|
||||
seismology,
|
||||
military,
|
||||
infrastructure,
|
||||
intelligence, # noqa: F401
|
||||
wildfire,
|
||||
cyber,
|
||||
climate,
|
||||
conflict,
|
||||
health,
|
||||
shipping,
|
||||
nuclear,
|
||||
service_status,
|
||||
)
|
||||
from .analysis.alerts import fetch_alert_digest
|
||||
from .analysis.clustering import fetch_news_clusters # noqa: F401
|
||||
from .analysis.posture import fetch_strategic_posture
|
||||
from .analysis.world_brief import fetch_world_brief # noqa: F401
|
||||
|
||||
logger = logging.getLogger("world-intel-mcp.reports")
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HTML template for the PDF report
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_CSS = """\
|
||||
@page {
|
||||
size: A4;
|
||||
margin: 1.5cm 1.8cm;
|
||||
@bottom-center { content: "Page " counter(page) " of " counter(pages); font-size: 8pt; color: #888; }
|
||||
@top-right { content: "WORLD INTELLIGENCE REPORT"; font-size: 7pt; color: #aaa; letter-spacing: 1px; }
|
||||
}
|
||||
* { box-sizing: border-box; }
|
||||
body {
|
||||
font-family: -apple-system, 'Helvetica Neue', Helvetica, Arial, sans-serif;
|
||||
font-size: 9pt;
|
||||
line-height: 1.45;
|
||||
color: #1a1a2e;
|
||||
margin: 0;
|
||||
}
|
||||
h1 {
|
||||
font-size: 22pt;
|
||||
margin: 0 0 4pt;
|
||||
color: #0f0f23;
|
||||
letter-spacing: -0.5px;
|
||||
}
|
||||
.subtitle {
|
||||
font-size: 10pt;
|
||||
color: #555;
|
||||
margin-bottom: 14pt;
|
||||
border-bottom: 2px solid #0f0f23;
|
||||
padding-bottom: 8pt;
|
||||
}
|
||||
h2 {
|
||||
font-size: 13pt;
|
||||
color: #16213e;
|
||||
margin: 16pt 0 6pt;
|
||||
padding-bottom: 3pt;
|
||||
border-bottom: 1px solid #ddd;
|
||||
page-break-after: avoid;
|
||||
}
|
||||
h3 {
|
||||
font-size: 10pt;
|
||||
color: #1a1a2e;
|
||||
margin: 10pt 0 4pt;
|
||||
page-break-after: avoid;
|
||||
}
|
||||
table {
|
||||
width: 100%;
|
||||
border-collapse: collapse;
|
||||
margin: 6pt 0 10pt;
|
||||
font-size: 8.5pt;
|
||||
page-break-inside: avoid;
|
||||
}
|
||||
th {
|
||||
background: #16213e;
|
||||
color: white;
|
||||
padding: 4pt 6pt;
|
||||
text-align: left;
|
||||
font-weight: 600;
|
||||
font-size: 8pt;
|
||||
}
|
||||
td {
|
||||
padding: 3pt 6pt;
|
||||
border-bottom: 1px solid #eee;
|
||||
vertical-align: top;
|
||||
}
|
||||
tr:nth-child(even) td { background: #f8f9fa; }
|
||||
.alert-box {
|
||||
background: #fff3cd;
|
||||
border-left: 4px solid #ffc107;
|
||||
padding: 6pt 10pt;
|
||||
margin: 6pt 0;
|
||||
font-size: 8.5pt;
|
||||
page-break-inside: avoid;
|
||||
}
|
||||
.alert-box.critical {
|
||||
background: #f8d7da;
|
||||
border-left-color: #dc3545;
|
||||
}
|
||||
.metric {
|
||||
display: inline-block;
|
||||
background: #e8eaf6;
|
||||
border-radius: 3pt;
|
||||
padding: 2pt 8pt;
|
||||
margin: 2pt 4pt 2pt 0;
|
||||
font-size: 8pt;
|
||||
font-weight: 600;
|
||||
}
|
||||
.metric.green { background: #d4edda; color: #155724; }
|
||||
.metric.red { background: #f8d7da; color: #721c24; }
|
||||
.metric.amber { background: #fff3cd; color: #856404; }
|
||||
.section-grid {
|
||||
display: flex;
|
||||
flex-wrap: wrap;
|
||||
gap: 8pt;
|
||||
}
|
||||
.section-card {
|
||||
flex: 1 1 45%;
|
||||
border: 1px solid #dee2e6;
|
||||
border-radius: 4pt;
|
||||
padding: 6pt 8pt;
|
||||
page-break-inside: avoid;
|
||||
}
|
||||
.footer {
|
||||
margin-top: 20pt;
|
||||
padding-top: 8pt;
|
||||
border-top: 1px solid #ccc;
|
||||
font-size: 7pt;
|
||||
color: #999;
|
||||
text-align: center;
|
||||
}
|
||||
.no-data { color: #999; font-style: italic; font-size: 8pt; }
|
||||
"""
|
||||
|
||||
|
||||
def _esc(text: Any) -> str:
|
||||
"""Escape HTML special chars."""
|
||||
if text is None:
|
||||
return ""
|
||||
s = str(text)
|
||||
return (
|
||||
s.replace("&", "&")
|
||||
.replace("<", "<")
|
||||
.replace(">", ">")
|
||||
.replace('"', """)
|
||||
)
|
||||
|
||||
|
||||
def _fmt_num(val: Any, decimals: int = 2) -> str:
|
||||
"""Format a number with commas."""
|
||||
if val is None:
|
||||
return "N/A"
|
||||
try:
|
||||
f = float(val)
|
||||
if f == int(f) and decimals == 0:
|
||||
return f"{int(f):,}"
|
||||
return f"{f:,.{decimals}f}"
|
||||
except (ValueError, TypeError):
|
||||
return str(val)
|
||||
|
||||
|
||||
def _change_class(val: Any) -> str:
|
||||
"""Return CSS class based on +/- value."""
|
||||
try:
|
||||
v = float(val)
|
||||
if v > 0:
|
||||
return "green"
|
||||
elif v < 0:
|
||||
return "red"
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
return ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Section renderers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _render_markets(data: dict) -> str:
|
||||
"""Render market quotes section."""
|
||||
quotes = data.get("quotes") or data.get("indices") or []
|
||||
if not quotes:
|
||||
return '<p class="no-data">Market data unavailable</p>'
|
||||
|
||||
rows = []
|
||||
for q in quotes[:15]:
|
||||
name = _esc(q.get("shortName") or q.get("symbol", ""))
|
||||
price = _fmt_num(q.get("regularMarketPrice"))
|
||||
chg = q.get("regularMarketChangePercent")
|
||||
chg_str = f"{float(chg):+.2f}%" if chg is not None else "N/A"
|
||||
cls = _change_class(chg)
|
||||
rows.append(
|
||||
f'<tr><td>{name}</td><td>{price}</td><td><span class="metric {cls}">{chg_str}</span></td></tr>'
|
||||
)
|
||||
|
||||
return f"""
|
||||
<table>
|
||||
<tr><th>Index / Symbol</th><th>Price</th><th>Change</th></tr>
|
||||
{"".join(rows)}
|
||||
</table>"""
|
||||
|
||||
|
||||
def _render_earthquakes(data: dict) -> str:
|
||||
"""Render seismology section."""
|
||||
quakes = data.get("earthquakes", [])
|
||||
if not quakes:
|
||||
return '<p class="no-data">No significant seismic activity</p>'
|
||||
|
||||
rows = []
|
||||
for q in quakes[:10]:
|
||||
props = q.get("properties", {})
|
||||
mag = _fmt_num(props.get("mag"), 1)
|
||||
place = _esc(props.get("place", "Unknown"))
|
||||
t = props.get("time")
|
||||
time_str = (
|
||||
datetime.fromtimestamp(t / 1000, tz=timezone.utc).strftime(
|
||||
"%Y-%m-%d %H:%M UTC"
|
||||
)
|
||||
if t
|
||||
else ""
|
||||
)
|
||||
rows.append(f"<tr><td>{mag}</td><td>{place}</td><td>{time_str}</td></tr>")
|
||||
|
||||
return f"""
|
||||
<table>
|
||||
<tr><th>Mag</th><th>Location</th><th>Time</th></tr>
|
||||
{"".join(rows)}
|
||||
</table>"""
|
||||
|
||||
|
||||
def _render_conflicts(data: dict) -> str:
|
||||
"""Render conflict events section."""
|
||||
events = data.get("events", [])
|
||||
if not events:
|
||||
return '<p class="no-data">No recent conflict events</p>'
|
||||
|
||||
rows = []
|
||||
for e in events[:12]:
|
||||
etype = _esc(e.get("event_type", ""))
|
||||
country = _esc(e.get("country", ""))
|
||||
fatalities = e.get("fatalities", 0)
|
||||
date = _esc(e.get("event_date", ""))
|
||||
notes = _esc(str(e.get("notes", ""))[:120])
|
||||
rows.append(
|
||||
f"<tr><td>{etype}</td><td>{country}</td><td>{fatalities}</td><td>{date}</td><td>{notes}</td></tr>"
|
||||
)
|
||||
|
||||
return f"""
|
||||
<table>
|
||||
<tr><th>Type</th><th>Country</th><th>Fatal.</th><th>Date</th><th>Notes</th></tr>
|
||||
{"".join(rows)}
|
||||
</table>"""
|
||||
|
||||
|
||||
def _render_news_clusters(data: dict) -> str:
|
||||
"""Render top news clusters."""
|
||||
clusters = data.get("clusters", [])
|
||||
if not clusters:
|
||||
return '<p class="no-data">No news clusters available</p>'
|
||||
|
||||
items = []
|
||||
for c in clusters[:8]:
|
||||
title = _esc(c.get("label") or c.get("title", ""))
|
||||
count = c.get("article_count", c.get("count", ""))
|
||||
items.append(f"<li><strong>{title}</strong> ({count} articles)</li>")
|
||||
return f"<ul>{''.join(items)}</ul>"
|
||||
|
||||
|
||||
def _render_alerts(data: dict) -> str:
|
||||
"""Render alert digest."""
|
||||
alerts = data.get("alerts", [])
|
||||
if not alerts:
|
||||
return '<p class="no-data">No active alerts</p>'
|
||||
|
||||
boxes = []
|
||||
for a in alerts[:10]:
|
||||
severity = a.get("severity", "info")
|
||||
css = "critical" if severity in ("critical", "high") else ""
|
||||
title = _esc(a.get("title", a.get("type", "")))
|
||||
detail = _esc(str(a.get("detail", a.get("description", "")))[:200])
|
||||
boxes.append(
|
||||
f'<div class="alert-box {css}"><strong>{title}</strong><br>{detail}</div>'
|
||||
)
|
||||
return "".join(boxes)
|
||||
|
||||
|
||||
def _render_posture(data: dict) -> str:
|
||||
"""Render strategic posture summary."""
|
||||
assessment = data.get("overall_assessment") or data.get("summary", "")
|
||||
if not assessment:
|
||||
return '<p class="no-data">Posture data unavailable</p>'
|
||||
|
||||
level = _esc(data.get("threat_level", data.get("risk_level", "")))
|
||||
cls = (
|
||||
"red"
|
||||
if "high" in level.lower()
|
||||
else "amber"
|
||||
if "medium" in level.lower()
|
||||
else "green"
|
||||
)
|
||||
|
||||
html = f'<span class="metric {cls}">Threat Level: {level}</span>'
|
||||
html += f"<p>{_esc(str(assessment)[:500])}</p>"
|
||||
|
||||
regions = data.get("regional_assessments") or data.get("regions", {})
|
||||
if regions and isinstance(regions, dict):
|
||||
html += "<h3>Regional Breakdown</h3><ul>"
|
||||
for region, detail in list(regions.items())[:6]:
|
||||
summary = (
|
||||
detail
|
||||
if isinstance(detail, str)
|
||||
else detail.get("summary", str(detail))
|
||||
)
|
||||
html += (
|
||||
f"<li><strong>{_esc(region)}</strong>: {_esc(str(summary)[:150])}</li>"
|
||||
)
|
||||
html += "</ul>"
|
||||
return html
|
||||
|
||||
|
||||
def _render_infrastructure(data: dict) -> str:
|
||||
"""Render infrastructure status."""
|
||||
outages = data.get("outages", data.get("entries", []))
|
||||
if not outages:
|
||||
return '<p class="no-data">No infrastructure disruptions detected</p>'
|
||||
|
||||
rows = []
|
||||
for o in outages[:8]:
|
||||
name = _esc(o.get("entity") or o.get("name", ""))
|
||||
score = o.get("score") or o.get("severity", "")
|
||||
source = _esc(o.get("source", ""))
|
||||
rows.append(f"<tr><td>{name}</td><td>{score}</td><td>{source}</td></tr>")
|
||||
|
||||
return f"""
|
||||
<table>
|
||||
<tr><th>Entity</th><th>Score / Severity</th><th>Source</th></tr>
|
||||
{"".join(rows)}
|
||||
</table>"""
|
||||
|
||||
|
||||
def _render_cyber(data: dict) -> str:
|
||||
"""Render cyber threat intelligence."""
|
||||
threats = data.get("recent_threats") or data.get("threats", [])
|
||||
if not threats:
|
||||
return '<p class="no-data">No recent cyber threats</p>'
|
||||
|
||||
rows = []
|
||||
for t in threats[:8]:
|
||||
name = _esc(t.get("name") or t.get("tag", ""))
|
||||
ttype = _esc(t.get("type", ""))
|
||||
url = _esc(t.get("url", ""))
|
||||
rows.append(f"<tr><td>{name}</td><td>{ttype}</td><td>{url[:60]}</td></tr>")
|
||||
|
||||
return f"""
|
||||
<table>
|
||||
<tr><th>Threat</th><th>Type</th><th>Reference</th></tr>
|
||||
{"".join(rows)}
|
||||
</table>"""
|
||||
|
||||
|
||||
def _render_health(data: dict) -> str:
|
||||
"""Render health/disease outbreak data."""
|
||||
outbreaks = data.get("outbreaks") or data.get("events", [])
|
||||
if not outbreaks:
|
||||
return '<p class="no-data">No active disease outbreaks</p>'
|
||||
|
||||
rows = []
|
||||
for o in outbreaks[:8]:
|
||||
disease = _esc(o.get("disease") or o.get("title", ""))
|
||||
country = _esc(o.get("country", ""))
|
||||
date = _esc(o.get("date", ""))
|
||||
rows.append(f"<tr><td>{disease}</td><td>{country}</td><td>{date}</td></tr>")
|
||||
|
||||
return f"""
|
||||
<table>
|
||||
<tr><th>Disease/Event</th><th>Location</th><th>Date</th></tr>
|
||||
{"".join(rows)}
|
||||
</table>"""
|
||||
|
||||
|
||||
def _render_maritime(data: dict) -> str:
|
||||
"""Render maritime overview."""
|
||||
vessels = data.get("vessels") or data.get("snapshot", [])
|
||||
if not vessels:
|
||||
return '<p class="no-data">No maritime data</p>'
|
||||
|
||||
rows = []
|
||||
items = vessels if isinstance(vessels, list) else [vessels]
|
||||
for v in items[:8]:
|
||||
name = _esc(v.get("name") or v.get("vessel_name", ""))
|
||||
vtype = _esc(v.get("type") or v.get("ship_type", ""))
|
||||
flag = _esc(v.get("flag", ""))
|
||||
rows.append(f"<tr><td>{name}</td><td>{vtype}</td><td>{flag}</td></tr>")
|
||||
|
||||
return f"""
|
||||
<table>
|
||||
<tr><th>Vessel</th><th>Type</th><th>Flag</th></tr>
|
||||
{"".join(rows)}
|
||||
</table>"""
|
||||
|
||||
|
||||
def _render_situation_brief(data: dict) -> str:
|
||||
"""Render situation brief / world brief."""
|
||||
brief = data.get("brief") or data.get("summary", "")
|
||||
if not brief:
|
||||
return '<p class="no-data">Brief unavailable</p>'
|
||||
return f"<p>{_esc(str(brief)[:1000])}</p>"
|
||||
|
||||
|
||||
def _render_key_value(data: dict, keys: list[str] | None = None) -> str:
|
||||
"""Generic key-value renderer for simple dicts."""
|
||||
if not data:
|
||||
return '<p class="no-data">Data unavailable</p>'
|
||||
|
||||
items = []
|
||||
show_keys = keys or list(data.keys())[:20]
|
||||
for k in show_keys:
|
||||
v = data.get(k)
|
||||
if v is not None and k not in (
|
||||
"source",
|
||||
"cached",
|
||||
"cache_age_seconds",
|
||||
"fetched_at",
|
||||
):
|
||||
items.append(f"<li><strong>{_esc(k)}</strong>: {_esc(str(v)[:200])}</li>")
|
||||
return f"<ul>{''.join(items)}</ul>" if items else '<p class="no-data">No data</p>'
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Data collection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _collect_report_data(
|
||||
fetcher: Fetcher,
|
||||
sections: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Fetch data for all report sections in parallel.
|
||||
|
||||
Args:
|
||||
fetcher: Configured Fetcher instance.
|
||||
sections: Optional list of section names to include.
|
||||
Default: all sections.
|
||||
"""
|
||||
all_sections = {
|
||||
"world_brief": lambda: fetch_world_brief(fetcher),
|
||||
"strategic_posture": lambda: fetch_strategic_posture(fetcher),
|
||||
"alerts": lambda: fetch_alert_digest(fetcher),
|
||||
"markets": lambda: markets.fetch_market_quotes(fetcher),
|
||||
"economic": lambda: markets.fetch_macro_signals(fetcher),
|
||||
"earthquakes": lambda: seismology.fetch_earthquakes(
|
||||
fetcher, min_magnitude=4.5, hours=24
|
||||
),
|
||||
"wildfires": lambda: wildfire.fetch_wildfires(fetcher),
|
||||
"conflicts": lambda: conflict.fetch_acled_events(fetcher, limit=15),
|
||||
"military": lambda: military.fetch_military_flights(fetcher),
|
||||
"infrastructure": lambda: infrastructure.fetch_internet_outages(fetcher),
|
||||
"maritime": lambda: intelligence.fetch_vessel_snapshot(fetcher),
|
||||
"cyber": lambda: cyber.fetch_cyber_threats(fetcher),
|
||||
"health": lambda: health.fetch_disease_outbreaks(fetcher),
|
||||
"news": lambda: fetch_news_clusters(fetcher),
|
||||
"climate": lambda: climate.fetch_climate_anomalies(fetcher),
|
||||
"nuclear": lambda: nuclear.fetch_nuclear_monitor(fetcher),
|
||||
"shipping": lambda: shipping.fetch_shipping_index(fetcher),
|
||||
"service_status": lambda: service_status.fetch_service_status(fetcher),
|
||||
}
|
||||
|
||||
if sections:
|
||||
all_sections = {k: v for k, v in all_sections.items() if k in sections}
|
||||
|
||||
results: dict[str, Any] = {}
|
||||
tasks = {}
|
||||
for name, fn in all_sections.items():
|
||||
tasks[name] = asyncio.create_task(_safe_fetch(name, fn))
|
||||
|
||||
for name, task in tasks.items():
|
||||
results[name] = await task
|
||||
|
||||
return results
|
||||
|
||||
|
||||
async def _safe_fetch(name: str, fn) -> dict:
|
||||
"""Wrap a fetch call with error handling."""
|
||||
try:
|
||||
result = await fn()
|
||||
return result if isinstance(result, dict) else {"data": result}
|
||||
except Exception as exc:
|
||||
logger.warning("Report section '%s' failed: %s", name, exc)
|
||||
return {"error": str(exc)}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HTML assembly
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _build_html(data: dict[str, Any], title: str | None = None) -> str:
|
||||
"""Assemble the full HTML document from collected data."""
|
||||
now = datetime.now(timezone.utc)
|
||||
report_title = title or "World Intelligence Report"
|
||||
timestamp = now.strftime("%Y-%m-%d %H:%M UTC")
|
||||
|
||||
sections_html = []
|
||||
|
||||
# Executive summary (world brief)
|
||||
if "world_brief" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Executive Summary</h2>{_render_situation_brief(data['world_brief'])}"
|
||||
)
|
||||
|
||||
# Strategic posture
|
||||
if "strategic_posture" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Strategic Posture</h2>{_render_posture(data['strategic_posture'])}"
|
||||
)
|
||||
|
||||
# Alerts
|
||||
if "alerts" in data:
|
||||
sections_html.append(f"<h2>Active Alerts</h2>{_render_alerts(data['alerts'])}")
|
||||
|
||||
# Markets
|
||||
if "markets" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Financial Markets</h2>{_render_markets(data['markets'])}"
|
||||
)
|
||||
|
||||
# Economic
|
||||
if "economic" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Economic Indicators</h2>{_render_key_value(data['economic'])}"
|
||||
)
|
||||
|
||||
# Conflicts
|
||||
if "conflicts" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Conflict & Security</h2>{_render_conflicts(data['conflicts'])}"
|
||||
)
|
||||
|
||||
# Military
|
||||
if "military" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Military Activity</h2>{_render_key_value(data['military'])}"
|
||||
)
|
||||
|
||||
# Earthquakes
|
||||
if "earthquakes" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Seismology</h2>{_render_earthquakes(data['earthquakes'])}"
|
||||
)
|
||||
|
||||
# Infrastructure
|
||||
if "infrastructure" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Infrastructure</h2>{_render_infrastructure(data['infrastructure'])}"
|
||||
)
|
||||
|
||||
# Cyber
|
||||
if "cyber" in data:
|
||||
sections_html.append(f"<h2>Cyber Threats</h2>{_render_cyber(data['cyber'])}")
|
||||
|
||||
# Maritime
|
||||
if "maritime" in data:
|
||||
sections_html.append(f"<h2>Maritime</h2>{_render_maritime(data['maritime'])}")
|
||||
|
||||
# Health
|
||||
if "health" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Health & Disease</h2>{_render_health(data['health'])}"
|
||||
)
|
||||
|
||||
# Nuclear
|
||||
if "nuclear" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Nuclear Monitoring</h2>{_render_key_value(data['nuclear'])}"
|
||||
)
|
||||
|
||||
# Climate
|
||||
if "climate" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Climate & Environment</h2>{_render_key_value(data['climate'])}"
|
||||
)
|
||||
|
||||
# News
|
||||
if "news" in data:
|
||||
sections_html.append(
|
||||
f"<h2>News Clusters</h2>{_render_news_clusters(data['news'])}"
|
||||
)
|
||||
|
||||
# Shipping
|
||||
if "shipping" in data:
|
||||
sections_html.append(f"<h2>Shipping</h2>{_render_key_value(data['shipping'])}")
|
||||
|
||||
# Service status
|
||||
if "service_status" in data:
|
||||
sections_html.append(
|
||||
f"<h2>Cloud & Service Status</h2>{_render_key_value(data['service_status'])}"
|
||||
)
|
||||
|
||||
body = "\n".join(sections_html)
|
||||
|
||||
return f"""<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<style>{_CSS}</style>
|
||||
</head>
|
||||
<body>
|
||||
<h1>{_esc(report_title)}</h1>
|
||||
<div class="subtitle">Generated {timestamp} — World Intel MCP — 109 intelligence sources</div>
|
||||
{body}
|
||||
<div class="footer">
|
||||
World Intelligence MCP Server — github.com/marc-shade/world-intel-mcp<br>
|
||||
Report generated {timestamp}. Data sourced from public APIs.
|
||||
</div>
|
||||
</body>
|
||||
</html>"""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def pdf_dependencies_available() -> bool:
|
||||
"""Check if weasyprint is importable."""
|
||||
try:
|
||||
from importlib.util import find_spec
|
||||
|
||||
return find_spec("weasyprint") is not None
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
async def generate_report(
|
||||
fetcher: Fetcher,
|
||||
output_path: str | Path | None = None,
|
||||
title: str | None = None,
|
||||
sections: list[str] | None = None,
|
||||
fmt: str = "pdf",
|
||||
) -> dict[str, Any]:
|
||||
"""Generate an intelligence report.
|
||||
|
||||
Args:
|
||||
fetcher: Configured Fetcher instance.
|
||||
output_path: Where to write the file. Default: ``~/.cache/world-intel-mcp/report-<timestamp>.pdf``
|
||||
title: Report title.
|
||||
sections: List of section names to include (default: all).
|
||||
fmt: Output format — ``pdf`` or ``html``.
|
||||
|
||||
Returns:
|
||||
Dict with path, format, sections included, generation time.
|
||||
"""
|
||||
t0 = time.time()
|
||||
|
||||
# Collect data
|
||||
data = await _collect_report_data(fetcher, sections)
|
||||
|
||||
# Build HTML
|
||||
html = _build_html(data, title)
|
||||
|
||||
# Determine output path
|
||||
if output_path is None:
|
||||
cache_dir = Path.home() / ".cache" / "world-intel-mcp"
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
|
||||
ext = "pdf" if fmt == "pdf" else "html"
|
||||
output_path = cache_dir / f"report-{ts}.{ext}"
|
||||
else:
|
||||
output_path = Path(output_path)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if fmt == "pdf":
|
||||
if not pdf_dependencies_available():
|
||||
return {
|
||||
"error": 'WeasyPrint not installed. Install with `pip install -e ".[pdf]"`.',
|
||||
"fallback": "Use fmt='html' for HTML output without WeasyPrint.",
|
||||
}
|
||||
# Import inside the function to avoid import-time failures
|
||||
from weasyprint import HTML as WeasyHTML
|
||||
|
||||
pdf_bytes: bytes = await asyncio.to_thread(
|
||||
lambda: WeasyHTML(string=html).write_pdf() # type: ignore[return-value]
|
||||
)
|
||||
output_path.write_bytes(pdf_bytes)
|
||||
else:
|
||||
output_path.write_text(html, encoding="utf-8")
|
||||
|
||||
elapsed = time.time() - t0
|
||||
sections_included = [k for k, v in data.items() if "error" not in v]
|
||||
sections_failed = [k for k, v in data.items() if "error" in v]
|
||||
|
||||
return {
|
||||
"path": str(output_path),
|
||||
"format": fmt,
|
||||
"size_bytes": output_path.stat().st_size,
|
||||
"sections_included": sections_included,
|
||||
"sections_failed": sections_failed,
|
||||
"generation_seconds": round(elapsed, 2),
|
||||
}
|
||||
|
||||
|
||||
async def generate_report_standalone(
|
||||
output_path: str | Path | None = None,
|
||||
title: str | None = None,
|
||||
sections: list[str] | None = None,
|
||||
fmt: str = "pdf",
|
||||
) -> dict[str, Any]:
|
||||
"""Generate a report using a fresh Fetcher (for CLI / standalone use)."""
|
||||
cache = Cache()
|
||||
breaker = CircuitBreaker(failure_threshold=3, cooldown_seconds=300)
|
||||
fetcher = Fetcher(cache=cache, breaker=breaker)
|
||||
return await generate_report(fetcher, output_path, title, sections, fmt)
|
||||
@@ -30,6 +30,8 @@ Phase 16: Vector intelligence — semantic search, similar events, timeline, vec
|
||||
Collector daemon for 24/7 data accumulation. Enterprise-grade semantic retrieval.
|
||||
Phase 17: Cross-domain analytics — cross-domain correlation, domain summary, trend detection
|
||||
(+3 = 109 tools). Historical analysis and early warning from accumulated vector data.
|
||||
Phase 18: PDF/HTML intelligence reports (+1 = 110 tools). WeasyPrint-based multi-section
|
||||
report generation covering 18 intelligence domains in parallel.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
@@ -98,7 +100,9 @@ try:
|
||||
if vector_dependencies_available():
|
||||
_vector_store = VectorStore(enabled=True)
|
||||
else:
|
||||
logger.info("Vector store unavailable (qdrant_client / fastembed not installed)")
|
||||
logger.info(
|
||||
"Vector store unavailable (qdrant_client / fastembed not installed)"
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.info("Vector store unavailable: %s", exc)
|
||||
|
||||
@@ -1702,6 +1706,31 @@ TOOLS: list[Tool] = [
|
||||
},
|
||||
},
|
||||
),
|
||||
# --- Reports (1 tool) ---
|
||||
Tool(
|
||||
name="intel_generate_report",
|
||||
description="Generate a PDF or HTML intelligence report covering markets, conflicts, earthquakes, cyber threats, health, infrastructure, and more. Returns the file path. Optional: sections (list of section names), title (string), format ('pdf' or 'html').",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "Report title (default: 'World Intelligence Report')",
|
||||
},
|
||||
"sections": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "Sections to include: world_brief, strategic_posture, alerts, markets, economic, earthquakes, wildfires, conflicts, military, infrastructure, maritime, cyber, health, news, climate, nuclear, shipping, service_status. Default: all.",
|
||||
},
|
||||
"format": {
|
||||
"type": "string",
|
||||
"enum": ["pdf", "html"],
|
||||
"description": "Output format (default: pdf). Use html if weasyprint is not installed.",
|
||||
"default": "pdf",
|
||||
},
|
||||
},
|
||||
},
|
||||
),
|
||||
# --- System (1 tool) ---
|
||||
Tool(
|
||||
name="intel_status",
|
||||
@@ -2362,6 +2391,18 @@ async def _dispatch(name: str, arguments: dict[str, Any]) -> Any:
|
||||
baseline_hours=arguments.get("baseline_hours", 48.0),
|
||||
)
|
||||
|
||||
# System
|
||||
# Reports
|
||||
case "intel_generate_report":
|
||||
from .reports import generate_report
|
||||
|
||||
return await generate_report(
|
||||
fetcher,
|
||||
title=arguments.get("title"),
|
||||
sections=arguments.get("sections"),
|
||||
fmt=arguments.get("format", "pdf"),
|
||||
)
|
||||
|
||||
# System
|
||||
case "intel_status":
|
||||
vs_stats = {
|
||||
|
||||
@@ -0,0 +1,482 @@
|
||||
"""Tests for the PDF/HTML intelligence report generator."""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from world_intel_mcp.reports import (
|
||||
_build_html,
|
||||
_esc,
|
||||
_fmt_num,
|
||||
_change_class,
|
||||
_render_markets,
|
||||
_render_earthquakes,
|
||||
_render_conflicts,
|
||||
_render_news_clusters,
|
||||
_render_alerts,
|
||||
_render_posture,
|
||||
_render_infrastructure,
|
||||
_render_cyber,
|
||||
_render_health,
|
||||
_render_maritime,
|
||||
_render_situation_brief,
|
||||
_render_key_value,
|
||||
_collect_report_data,
|
||||
_safe_fetch,
|
||||
pdf_dependencies_available,
|
||||
generate_report,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Utility functions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEsc:
|
||||
def test_none(self):
|
||||
assert _esc(None) == ""
|
||||
|
||||
def test_html_chars(self):
|
||||
assert (
|
||||
_esc('<script>"alert&"</script>')
|
||||
== "<script>"alert&"</script>"
|
||||
)
|
||||
|
||||
def test_plain(self):
|
||||
assert _esc("hello world") == "hello world"
|
||||
|
||||
def test_number(self):
|
||||
assert _esc(42) == "42"
|
||||
|
||||
|
||||
class TestFmtNum:
|
||||
def test_none(self):
|
||||
assert _fmt_num(None) == "N/A"
|
||||
|
||||
def test_integer(self):
|
||||
assert _fmt_num(1234567, 0) == "1,234,567"
|
||||
|
||||
def test_float(self):
|
||||
assert _fmt_num(1234.567, 2) == "1,234.57"
|
||||
|
||||
def test_string(self):
|
||||
assert _fmt_num("not a number") == "not a number"
|
||||
|
||||
|
||||
class TestChangeClass:
|
||||
def test_positive(self):
|
||||
assert _change_class(1.5) == "green"
|
||||
|
||||
def test_negative(self):
|
||||
assert _change_class(-0.5) == "red"
|
||||
|
||||
def test_zero(self):
|
||||
assert _change_class(0) == ""
|
||||
|
||||
def test_none(self):
|
||||
assert _change_class(None) == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Section renderers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestRenderMarkets:
|
||||
def test_no_data(self):
|
||||
assert "unavailable" in _render_markets({})
|
||||
|
||||
def test_with_quotes(self):
|
||||
data = {
|
||||
"quotes": [
|
||||
{
|
||||
"shortName": "S&P 500",
|
||||
"regularMarketPrice": 4500.12,
|
||||
"regularMarketChangePercent": 1.23,
|
||||
},
|
||||
{
|
||||
"symbol": "DJI",
|
||||
"regularMarketPrice": 35000.0,
|
||||
"regularMarketChangePercent": -0.45,
|
||||
},
|
||||
]
|
||||
}
|
||||
html = _render_markets(data)
|
||||
assert "S&P 500" in html
|
||||
assert "4,500.12" in html
|
||||
assert "+1.23%" in html
|
||||
assert "green" in html
|
||||
assert "red" in html
|
||||
|
||||
|
||||
class TestRenderEarthquakes:
|
||||
def test_no_data(self):
|
||||
assert "No significant" in _render_earthquakes({})
|
||||
|
||||
def test_with_quakes(self):
|
||||
data = {
|
||||
"earthquakes": [
|
||||
{
|
||||
"properties": {
|
||||
"mag": 5.2,
|
||||
"place": "Near Tokyo",
|
||||
"time": 1700000000000,
|
||||
}
|
||||
},
|
||||
]
|
||||
}
|
||||
html = _render_earthquakes(data)
|
||||
assert "5.2" in html
|
||||
assert "Near Tokyo" in html
|
||||
|
||||
|
||||
class TestRenderConflicts:
|
||||
def test_no_data(self):
|
||||
assert "No recent" in _render_conflicts({})
|
||||
|
||||
def test_with_events(self):
|
||||
data = {
|
||||
"events": [
|
||||
{
|
||||
"event_type": "Battle",
|
||||
"country": "Ukraine",
|
||||
"fatalities": 5,
|
||||
"event_date": "2025-01-01",
|
||||
"notes": "Test event",
|
||||
},
|
||||
]
|
||||
}
|
||||
html = _render_conflicts(data)
|
||||
assert "Battle" in html
|
||||
assert "Ukraine" in html
|
||||
|
||||
|
||||
class TestRenderNewsClusters:
|
||||
def test_no_data(self):
|
||||
assert "No news clusters" in _render_news_clusters({})
|
||||
|
||||
def test_with_clusters(self):
|
||||
data = {"clusters": [{"label": "Climate Summit", "article_count": 42}]}
|
||||
html = _render_news_clusters(data)
|
||||
assert "Climate Summit" in html
|
||||
assert "42" in html
|
||||
|
||||
|
||||
class TestRenderAlerts:
|
||||
def test_no_data(self):
|
||||
assert "No active alerts" in _render_alerts({})
|
||||
|
||||
def test_with_alerts(self):
|
||||
data = {
|
||||
"alerts": [
|
||||
{
|
||||
"severity": "critical",
|
||||
"title": "Major Quake",
|
||||
"detail": "M7.2 event detected",
|
||||
},
|
||||
{
|
||||
"severity": "info",
|
||||
"title": "Minor Issue",
|
||||
"description": "Low impact",
|
||||
},
|
||||
]
|
||||
}
|
||||
html = _render_alerts(data)
|
||||
assert "critical" in html
|
||||
assert "Major Quake" in html
|
||||
|
||||
|
||||
class TestRenderPosture:
|
||||
def test_no_data(self):
|
||||
assert "unavailable" in _render_posture({})
|
||||
|
||||
def test_with_data(self):
|
||||
data = {
|
||||
"overall_assessment": "Tensions elevated in multiple regions.",
|
||||
"threat_level": "HIGH",
|
||||
"regions": {"Europe": "NATO exercises ongoing"},
|
||||
}
|
||||
html = _render_posture(data)
|
||||
assert "HIGH" in html
|
||||
assert "red" in html
|
||||
assert "Europe" in html
|
||||
|
||||
|
||||
class TestRenderInfrastructure:
|
||||
def test_no_data(self):
|
||||
assert "No infrastructure" in _render_infrastructure({})
|
||||
|
||||
def test_with_outages(self):
|
||||
data = {"outages": [{"entity": "AS12345", "score": 85, "source": "IODA"}]}
|
||||
html = _render_infrastructure(data)
|
||||
assert "AS12345" in html
|
||||
|
||||
|
||||
class TestRenderCyber:
|
||||
def test_no_data(self):
|
||||
assert "No recent cyber" in _render_cyber({})
|
||||
|
||||
def test_with_threats(self):
|
||||
data = {
|
||||
"recent_threats": [
|
||||
{"name": "Emotet", "type": "malware", "url": "https://example.com"}
|
||||
]
|
||||
}
|
||||
html = _render_cyber(data)
|
||||
assert "Emotet" in html
|
||||
|
||||
|
||||
class TestRenderHealth:
|
||||
def test_no_data(self):
|
||||
assert "No active disease" in _render_health({})
|
||||
|
||||
def test_with_outbreaks(self):
|
||||
data = {
|
||||
"outbreaks": [{"disease": "Mpox", "country": "DRC", "date": "2025-01-15"}]
|
||||
}
|
||||
html = _render_health(data)
|
||||
assert "Mpox" in html
|
||||
|
||||
|
||||
class TestRenderMaritime:
|
||||
def test_no_data(self):
|
||||
assert "No maritime" in _render_maritime({})
|
||||
|
||||
def test_with_vessels(self):
|
||||
data = {"vessels": [{"name": "USS Nimitz", "type": "CVN", "flag": "US"}]}
|
||||
html = _render_maritime(data)
|
||||
assert "USS Nimitz" in html
|
||||
|
||||
|
||||
class TestRenderSituationBrief:
|
||||
def test_no_data(self):
|
||||
assert "unavailable" in _render_situation_brief({})
|
||||
|
||||
def test_with_brief(self):
|
||||
data = {
|
||||
"brief": "Global tensions remain elevated with multiple hotspots active."
|
||||
}
|
||||
html = _render_situation_brief(data)
|
||||
assert "Global tensions" in html
|
||||
|
||||
|
||||
class TestRenderKeyValue:
|
||||
def test_no_data(self):
|
||||
assert "unavailable" in _render_key_value({})
|
||||
|
||||
def test_with_data(self):
|
||||
html = _render_key_value(
|
||||
{"metric1": 42, "metric2": "active", "source": "should_skip"}
|
||||
)
|
||||
assert "metric1" in html
|
||||
assert "42" in html
|
||||
assert "source" not in html # filtered out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HTML assembly
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestBuildHtml:
|
||||
def test_basic_structure(self):
|
||||
data = {"markets": {"quotes": []}, "world_brief": {"brief": "Test brief"}}
|
||||
html = _build_html(data)
|
||||
assert "<!DOCTYPE html>" in html
|
||||
assert "World Intelligence Report" in html
|
||||
assert "Executive Summary" in html
|
||||
assert "Financial Markets" in html
|
||||
|
||||
def test_custom_title(self):
|
||||
html = _build_html({}, title="Custom Report")
|
||||
assert "Custom Report" in html
|
||||
|
||||
def test_all_sections(self):
|
||||
data = {
|
||||
"world_brief": {"brief": "test"},
|
||||
"strategic_posture": {"overall_assessment": "test", "threat_level": "LOW"},
|
||||
"alerts": {"alerts": []},
|
||||
"markets": {"quotes": []},
|
||||
"economic": {"key": "val"},
|
||||
"conflicts": {"events": []},
|
||||
"military": {"data": "val"},
|
||||
"earthquakes": {"earthquakes": []},
|
||||
"infrastructure": {"outages": []},
|
||||
"cyber": {"recent_threats": []},
|
||||
"maritime": {"vessels": []},
|
||||
"health": {"outbreaks": []},
|
||||
"nuclear": {"sites": []},
|
||||
"climate": {"zones": {}},
|
||||
"news": {"clusters": []},
|
||||
"shipping": {"quotes": []},
|
||||
"service_status": {"services": []},
|
||||
}
|
||||
html = _build_html(data)
|
||||
assert "Executive Summary" in html
|
||||
assert "Strategic Posture" in html
|
||||
assert "Financial Markets" in html
|
||||
assert "Cyber Threats" in html
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Data collection
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSafeFetch:
|
||||
@pytest.mark.asyncio
|
||||
async def test_success(self):
|
||||
async def good_fn():
|
||||
return {"result": "ok"}
|
||||
|
||||
result = await _safe_fetch("test", good_fn)
|
||||
assert result == {"result": "ok"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failure(self):
|
||||
async def bad_fn():
|
||||
raise ValueError("boom")
|
||||
|
||||
result = await _safe_fetch("test", bad_fn)
|
||||
assert "error" in result
|
||||
assert "boom" in result["error"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_dict_result(self):
|
||||
async def list_fn():
|
||||
return [1, 2, 3]
|
||||
|
||||
result = await _safe_fetch("test", list_fn)
|
||||
assert result == {"data": [1, 2, 3]}
|
||||
|
||||
|
||||
class TestCollectReportData:
|
||||
@pytest.mark.asyncio
|
||||
async def test_section_filter(self):
|
||||
with patch("world_intel_mcp.reports.markets") as mock_markets:
|
||||
mock_markets.fetch_market_quotes = AsyncMock(return_value={"quotes": []})
|
||||
|
||||
data = await _collect_report_data(
|
||||
MagicMock(),
|
||||
sections=["markets"],
|
||||
)
|
||||
assert "markets" in data
|
||||
assert "earthquakes" not in data
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handles_failures(self):
|
||||
with patch("world_intel_mcp.reports.markets") as mock_markets:
|
||||
mock_markets.fetch_market_quotes = AsyncMock(
|
||||
side_effect=RuntimeError("api down")
|
||||
)
|
||||
|
||||
data = await _collect_report_data(MagicMock(), sections=["markets"])
|
||||
assert "error" in data["markets"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PDF dependency check
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPdfDependencies:
|
||||
def test_check(self):
|
||||
# Just verify it returns a bool without crashing
|
||||
result = pdf_dependencies_available()
|
||||
assert isinstance(result, bool)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Report generation (HTML mode — no weasyprint needed)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGenerateReport:
|
||||
@pytest.mark.asyncio
|
||||
async def test_html_output(self, tmp_path):
|
||||
output = tmp_path / "test_report.html"
|
||||
|
||||
with patch("world_intel_mcp.reports._collect_report_data") as mock_collect:
|
||||
mock_collect.return_value = {
|
||||
"markets": {
|
||||
"quotes": [
|
||||
{
|
||||
"shortName": "SPX",
|
||||
"regularMarketPrice": 4500,
|
||||
"regularMarketChangePercent": 0.5,
|
||||
}
|
||||
]
|
||||
},
|
||||
}
|
||||
|
||||
fetcher = MagicMock()
|
||||
result = await generate_report(fetcher, output_path=output, fmt="html")
|
||||
|
||||
assert result["format"] == "html"
|
||||
assert result["path"] == str(output)
|
||||
assert result["size_bytes"] > 0
|
||||
assert "markets" in result["sections_included"]
|
||||
assert output.exists()
|
||||
|
||||
content = output.read_text()
|
||||
assert "<!DOCTYPE html>" in content
|
||||
assert "SPX" in content
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_default_output_path(self, tmp_path):
|
||||
with patch("world_intel_mcp.reports._collect_report_data") as mock_collect:
|
||||
mock_collect.return_value = {"markets": {"quotes": []}}
|
||||
|
||||
with patch("world_intel_mcp.reports.Path.home", return_value=tmp_path):
|
||||
fetcher = MagicMock()
|
||||
result = await generate_report(fetcher, fmt="html")
|
||||
|
||||
assert result["format"] == "html"
|
||||
assert "report-" in result["path"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_sections_tracking(self, tmp_path):
|
||||
output = tmp_path / "test.html"
|
||||
|
||||
with patch("world_intel_mcp.reports._collect_report_data") as mock_collect:
|
||||
mock_collect.return_value = {
|
||||
"markets": {"quotes": []},
|
||||
"earthquakes": {"error": "api down"},
|
||||
}
|
||||
|
||||
result = await generate_report(MagicMock(), output_path=output, fmt="html")
|
||||
|
||||
assert "markets" in result["sections_included"]
|
||||
assert "earthquakes" in result["sections_failed"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_pdf_without_weasyprint(self, tmp_path):
|
||||
output = tmp_path / "test.pdf"
|
||||
|
||||
with patch(
|
||||
"world_intel_mcp.reports.pdf_dependencies_available", return_value=False
|
||||
):
|
||||
with patch("world_intel_mcp.reports._collect_report_data") as mock_collect:
|
||||
mock_collect.return_value = {"markets": {"quotes": []}}
|
||||
|
||||
result = await generate_report(
|
||||
MagicMock(), output_path=output, fmt="pdf"
|
||||
)
|
||||
|
||||
assert "error" in result
|
||||
assert "WeasyPrint" in result["error"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_custom_title(self, tmp_path):
|
||||
output = tmp_path / "custom.html"
|
||||
|
||||
with patch("world_intel_mcp.reports._collect_report_data") as mock_collect:
|
||||
mock_collect.return_value = {}
|
||||
|
||||
result = await generate_report(
|
||||
MagicMock(), output_path=output, title="Daily Brief", fmt="html"
|
||||
)
|
||||
|
||||
content = output.read_text()
|
||||
assert "Daily Brief" in content
|
||||
Reference in New Issue
Block a user