文件内容
src/founder_signal/run_review.py
"""Public-safe run review rendering for Founder Signal."""
from __future__ import annotations
from pathlib import Path
import re
from typing import Any, Mapping
PUBLIC_RUN_REVIEW_FILENAME = "public-run-review.md"
_MAX_PREVIEW_SENTENCES = 2
_MAX_PREVIEW_WORDS = 40
def write_public_run_review(
*,
run_dir: Path,
artifact: Mapping[str, Any],
candidates: list[dict[str, Any]],
) -> Path:
output_path = run_dir / PUBLIC_RUN_REVIEW_FILENAME
output_path.write_text(
render_public_run_review(
run_dir=run_dir,
artifact=artifact,
candidates=candidates,
),
encoding="utf-8",
)
return output_path
def render_public_run_review(
*,
run_dir: Path,
artifact: Mapping[str, Any],
candidates: list[dict[str, Any]],
) -> str:
profile_id = str(artifact.get("profile_id") or "").strip()
product_name = str(artifact.get("product_name") or "").strip()
title_suffix = f" - {product_name}" if product_name else ""
lines = [
f"# Founder Signal Run Review{title_suffix}",
"",
"## Summary",
"",
f"- Run ID: {artifact.get('run_id') or run_dir.name}",
f"- Profile ID: {profile_id or 'unknown'}",
]
if product_name:
lines.append(f"- Product: {product_name}")
lines.extend(
[
f"- Status: {_report_status(artifact)}",
f"- Candidates found: {int(artifact.get('candidates_found') or 0)}",
f"- Candidates verified: {int(artifact.get('candidates_verified') or 0)}",
f"- Action card generated: {_yes_no(artifact.get('action_card_generated'))}",
f"- Selected candidate: {_selected_candidate_label(artifact)}",
"",
"## Summary Snapshot",
"",
"```text",
*_summary_snapshot_lines(artifact),
"```",
"",
"## Discovery Metrics",
"",
]
)
lines.extend(_discovery_metric_lines(artifact))
lines.extend(["", "## Candidate Review Table", ""])
lines.extend(_candidate_review_table(candidates))
daily_review_path = run_dir / "daily-review.md"
if daily_review_path.exists():
lines.extend(
[
"",
"## Embedded Daily Review",
"",
"The selected-candidate action card is embedded below for founder review.",
"",
daily_review_path.read_text(encoding="utf-8").rstrip(),
]
)
return "\n".join(lines).rstrip() + "\n"
def _report_status(artifact: Mapping[str, Any]) -> str:
status = str(artifact.get("status") or "").strip().lower()
return "failed" if "fail" in status or status == "error" else "success"
def _yes_no(value: Any) -> str:
return "yes" if bool(value) else "no"
def _selected_candidate_label(artifact: Mapping[str, Any]) -> str:
selected = artifact.get("selected_candidate")
if not isinstance(selected, Mapping):
return "None"
candidate_id = str(selected.get("candidate_id") or "").strip()
source_url = str(selected.get("source_url") or selected.get("reddit_url") or "").strip()
if candidate_id and source_url:
return f"{candidate_id} ({source_url})"
return candidate_id or source_url or "None"
def _discovery_metric_lines(artifact: Mapping[str, Any]) -> list[str]:
metrics = artifact.get("discovery_metrics")
if not isinstance(metrics, Mapping):
return ["- None"]
return [
f"- Searched pages: {int(metrics.get('searched_pages') or 0)}",
f"- Fresh candidates found: {int(metrics.get('fresh_candidates_found') or 0)}",
f"- Excluded by history: {int(metrics.get('excluded_by_history') or 0)}",
f"- Excluded by profile: {int(metrics.get('excluded_by_profile') or 0)}",
f"- Total excluded: {int(metrics.get('total_excluded') or 0)}",
f"- Discovery budget pages: {int(metrics.get('discovery_budget_pages') or 0)}",
f"- Discovery exhausted: {_yes_no(metrics.get('discovery_exhausted'))}",
]
def _candidate_review_table(candidates: list[dict[str, Any]]) -> list[str]:
if not candidates:
return ["No candidates were available for this run."]
lines = [
"| Post preview | Platform | Source | Evidence | Status | Score | Agent | Eligible | Source type | Discovery method | Gate | Rejection signals |",
"| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |",
]
for candidate in candidates:
source_url = str(candidate.get("source_url") or candidate.get("reddit_url") or "").strip()
evidence_url = str(candidate.get("evidence_url") or "").strip()
lines.append(
"| "
+ " | ".join(
[
_cell(_candidate_preview(candidate)),
_cell(candidate.get("source_platform") or candidate.get("platform") or "reddit"),
_link_cell(source_url, label="source"),
_link_cell(evidence_url, label="evidence"),
_cell(candidate.get("read_status") or candidate.get("status")),
_cell(candidate.get("score_total")),
_cell(candidate.get("agent_review_score_total")),
_cell(_eligible_label(candidate.get("selection_eligible"))),
_cell(candidate.get("source_type")),
_cell(candidate.get("discovery_method")),
_cell(candidate.get("selection_gate_reason") or "passed"),
_cell(", ".join(_string_list(candidate.get("rejection_signals"))) or "none"),
]
)
+ " |"
)
return lines
def _candidate_preview(candidate: Mapping[str, Any]) -> str:
structured = candidate.get("structured_evidence")
if isinstance(structured, Mapping):
title = _normalize_text(structured.get("post_title"))
body = _normalize_text(structured.get("post_body"))
comments = _normalize_text(structured.get("comments_excerpt"))
preview = _join_preview_parts(title, body or comments)
if preview:
return preview
title = _normalize_text(candidate.get("post_title"))
body = _normalize_text(candidate.get("post_body"))
text_snapshot = _normalize_text(candidate.get("text_snapshot"))
preview = _join_preview_parts(title, body or text_snapshot)
if preview:
return preview
snapshot_path = str(candidate.get("evidence_snapshot_path") or "").strip()
if snapshot_path:
snapshot_title, snapshot_body = _snapshot_title_and_body(Path(snapshot_path))
preview = _join_preview_parts(snapshot_title, snapshot_body)
if preview:
return preview
return str(candidate.get("candidate_id") or "Untitled candidate")
def _join_preview_parts(title: str, body: str) -> str:
title = title.strip()
body = body.strip()
if title and body:
return f"{title} — {_body_excerpt(body)}"
if title:
return _clip_words(title, _MAX_PREVIEW_WORDS)
if body:
return _body_excerpt(body)
return ""
def _summary_snapshot_lines(artifact: Mapping[str, Any]) -> list[str]:
selected = artifact.get("selected_candidate")
selected_label = _selected_candidate_label(artifact)
selected_score = "None"
if isinstance(selected, Mapping):
raw_score = selected.get("score_total")
if raw_score not in (None, ""):
selected_score = str(raw_score)
return [
"+--------------------------------------------------------------+",
f"| status: {_report_status(artifact):<52}|",
f"| candidates found: {int(artifact.get('candidates_found') or 0):<42}|",
f"| candidates verified: {int(artifact.get('candidates_verified') or 0):<39}|",
f"| action card generated: {_yes_no(artifact.get('action_card_generated')):<36}|",
f"| selected score: {selected_score:<43}|",
"+--------------------------------------------------------------+",
f"| selected: {_clip_snapshot_text(selected_label, 50):<50}|",
"+--------------------------------------------------------------+",
]
def _eligible_label(value: Any) -> str:
if value is True:
return "yes"
if value is False:
return "no"
return "n/a"
def _string_list(value: Any) -> list[str]:
if not isinstance(value, list):
return []
return [str(item).strip() for item in value if str(item).strip()]
def _cell(value: Any) -> str:
text = str(value or "").strip() or "None"
return text.replace("|", "\\|").replace("\n", "<br>")
def _link_cell(url: str, *, label: str) -> str:
if not url:
return "None"
safe_url = url.replace(")", "%29")
return f"[{label}]({safe_url})"
def _clip_snapshot_text(text: str, limit: int) -> str:
compact = " ".join(str(text or "").split())
if len(compact) <= limit:
return compact
return compact[: max(limit - 3, 0)].rstrip() + "..."
def _body_excerpt(text: str) -> str:
sentences = _sentences(text)
excerpt = " ".join(sentences[:_MAX_PREVIEW_SENTENCES]).strip()
if not excerpt:
excerpt = _normalize_text(text)
return _clip_words(excerpt, _MAX_PREVIEW_WORDS)
def _sentences(text: str) -> list[str]:
compact = " ".join(_normalize_text(text).split())
if not compact:
return []
parts = re.split(r"(?<=[.!?])\s+", compact)
return [part.strip() for part in parts if part.strip()]
def _clip_words(text: str, limit: int) -> str:
words = text.split()
if len(words) <= limit:
return text
return " ".join(words[:limit]).rstrip(".,;:") + "..."
def _normalize_text(value: Any) -> str:
return " ".join(str(value or "").split())
def _snapshot_title_and_body(snapshot_path: Path) -> tuple[str, str]:
try:
snapshot = snapshot_path.read_text(encoding="utf-8")
except OSError:
return "", ""
lines = [line.strip() for line in snapshot.splitlines() if line.strip()]
if not lines:
return "", ""
title = _clip_words(lines[0], 18)
body = " ".join(lines[1:]).strip() or lines[0]
return title, body