diff --git a/pebble/export.py b/pebble/export.py index 084aa7e3..31333df9 100644 --- a/pebble/export.py +++ b/pebble/export.py @@ -2,23 +2,28 @@ from datetime import datetime, timezone +from .formatting import escape_md_table_cell, normalize_confidence, truncate_for_cell + +_CLAIM_TEXT_LIMIT = 240 + def render_profile_markdown(profile: dict, prospect_name: str, prospect_org: str) -> str: """Render a research profile as a Markdown document. Args: - profile: Profile dict with claims, summary, confidence_score. + profile: Profile dict with claims, summary, confidence_score, and + optionally `partial` (bool) and `partial_reason` (str). prospect_name: Display name of the prospect. - prospect_org: Organization name. + prospect_org: Organization name (may be blank). Returns: - Markdown string. + Markdown string. Stable structure across runs — pinned by tests in + `pebble/tests/test_export.py`. """ lines: list[str] = [] generated = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") - confidence = profile.get("confidence_score", "unknown") + confidence = normalize_confidence(profile.get("confidence_score")) - # Header lines.append(f"# Prospect Research: {prospect_name}") if prospect_org: lines.append(f"**Organization:** {prospect_org}") @@ -28,46 +33,99 @@ def render_profile_markdown(profile: dict, prospect_name: str, prospect_org: str lines.append("**Status:** Partial (some agents failed)") lines.append("") - # Summary - summary = profile.get("summary", "") + summary = (profile.get("summary") or "").strip() + claims = profile.get("claims") or [] + + if not summary and not claims: + lines.append("## No information gathered") + lines.append("") + reason = (profile.get("partial_reason") or "").strip() + if reason: + lines.append(f"This research run did not produce any claims. Reason: {reason}") + else: + lines.append( + "This research run did not produce any claims. The agents may have " + "timed out or found no public information for this prospect." + ) + lines.append("") + return "\n".join(lines) + if summary: lines.append("## Summary") lines.append("") lines.append(summary) lines.append("") - # Claims table - claims = profile.get("claims", []) + truncated_claims: list[tuple[int, str]] = [] if claims: lines.append(f"## Claims ({len(claims)})") lines.append("") lines.append("| # | Claim | Source | Confidence | Status |") lines.append("|---|-------|--------|------------|--------|") for i, claim in enumerate(claims, 1): - text = claim.get("text", "").replace("|", "\\|").replace("\n", " ") + raw_text = claim.get("text", "") + short, was_truncated = truncate_for_cell(raw_text, _CLAIM_TEXT_LIMIT) + cell_text = escape_md_table_cell(short) + if was_truncated: + cell_text = f"{cell_text} [^c{i}]" + truncated_claims.append((i, raw_text)) url = claim.get("source_url", "") source_link = f"[Link]({url})" if url else "-" - conf = claim.get("confidence", "medium") - temporal = claim.get("temporal_status", "") - if temporal and temporal != "unknown": - status = temporal - else: - status = "-" - lines.append(f"| {i} | {text} | {source_link} | {conf} | {status} |") + conf_cell = normalize_confidence(claim.get("confidence", "medium")) + temporal = (claim.get("temporal_status") or "").strip() + status = temporal if temporal and temporal != "unknown" else "-" + lines.append(f"| {i} | {cell_text} | {source_link} | {conf_cell} | {status} |") lines.append("") - # Sources list (unique URLs) - urls = sorted({c.get("source_url", "") for c in claims if c.get("source_url")}) - if urls: - lines.append("## Sources") - lines.append("") - for url in urls: - lines.append(f"- {url}") - lines.append("") + if truncated_claims: + lines.append("### Full claim text") + lines.append("") + for idx, full_text in truncated_claims: + normalized = full_text.replace("\r\n", "\n").strip() + lines.append(f"[^c{idx}]: {normalized}") + lines.append("") + + source_lines = _render_sources_section(claims) + if source_lines: + lines.extend(source_lines) return "\n".join(lines) +def _render_sources_section(claims: list[dict]) -> list[str]: + """Build the `## Sources` section ordered by first-appearance in claims. + + Each URL appears once; multi-claim URLs get a `(N claims)` suffix. + Uses `source_title` when present, falling back to the bare URL. + """ + counts: dict[str, int] = {} + titles: dict[str, str] = {} + order: list[str] = [] + for claim in claims: + url = (claim.get("source_url") or "").strip() + if not url: + continue + if url not in counts: + counts[url] = 0 + order.append(url) + counts[url] += 1 + title = (claim.get("source_title") or "").strip() + if title and url not in titles: + titles[url] = title + + if not order: + return [] + + out: list[str] = ["## Sources", ""] + for url in order: + title = titles.get(url) + label = f"[{title}]({url})" if title else url + suffix = f" ({counts[url]} claims)" if counts[url] > 1 else "" + out.append(f"- {label}{suffix}") + out.append("") + return out + + def render_profile_pdf(markdown_text: str) -> bytes: """Best-effort PDF rendering from markdown text. @@ -78,7 +136,6 @@ def render_profile_pdf(markdown_text: str) -> bytes: import markdown as md html = md.markdown(markdown_text, extensions=["tables"]) - # Wrap in minimal HTML for readability full_html = f"""