← refacto-feature-scout
"""Daily research: ask headless Claude Code to find what the community wants; store new ideas."""
from __future__ import annotations
import re
import urllib.request
from string import Template
from pathlib import Path
from typing import Any, Callable
from .claude_runner import ClaudeResult, extract_json, run_claude
from .config import IDEAS_PER_PLATFORM, LOG_DIR, SCAN_TIMEOUT_S, VENUES
from .projects import Project
from . import reddit
from .store import Store, clamp_score
from .text import html_to_text
PROMPT_PATH = Path(__file__).parent / "prompts" / "scan.md"
SCORE_PROMPT_PATH = Path(__file__).parent / "prompts" / "score.md"
CONTEXT_CHARS = 14000
SIGNAL_CHARS = 4000
VALID_EFFORT = {"S", "M", "L"}
VALID_CATEGORY = {"metric", "analysis", "screener", "alert", "visualisation", "data", "feature"}
def repo_context(project: Project, limit: int = CONTEXT_CHARS) -> str:
"""The product's own docs, so the researcher knows what already exists."""
chunks = []
budget = limit
for rel in project.context_files:
p = project.folder / rel
if not p.exists() or budget <= 0:
continue
text = p.read_text(errors="replace")
share = min(len(text), max(2000, budget // max(1, len(project.context_files))))
chunks.append(f"### {rel}\n{text[:share]}")
budget -= share
return "\n\n".join(chunks) or "(no docs found)"
def fetch_signals(project: Project, fetch: Callable[[str], str] | None = None,
reddit_posts: list[dict[str, Any]] | None = None) -> str:
fetch = fetch or _fetch_url
parts = []
if reddit_posts is None:
reddit_posts, _ = collect_reddit(project)
if project.reddit_feeds:
parts.append("### Recent Reddit posts\n" + reddit.render(reddit_posts))
elif not project.signal_urls:
parts.append("### No community feeds are set up for this project\n"
"There were no posts or pages to read today, so everything below has to come "
"from your own searching.")
for url in project.signal_urls:
try:
text = html_to_text(fetch(url))
except Exception as e: # noqa: BLE001 — a dead board must not stop the scan
parts.append(f"### {url}\n(could not fetch: {e})")
continue
text = re.sub(r"\n\s*\n+", "\n", text).strip()
parts.append(f"### {url}\n{text[:SIGNAL_CHARS]}")
return "\n\n".join(parts)
def collect_reddit(project: Project) -> tuple[list[dict[str, Any]], list[str]]:
"""Posts from the communities this project reads. No communities means no posts — which is
a fine way to run, and is said out loud rather than looking like a quiet day."""
if not project.reddit_feeds:
return [], []
return reddit.collect(project.reddit_feeds)
def _fetch_url(url: str) -> str:
req = urllib.request.Request(url, headers={"User-Agent": "FeatureScout/0.1"})
with urllib.request.urlopen(req, timeout=30) as r:
return r.read().decode("utf-8", "replace")
def prior_titles(store: Store, project: Project) -> str:
rows = store.all_titles(project.key)
if not rows:
return "(none yet)"
return "\n".join(f"- {t} [{s}]" for t, s in rows[-150:])
DEFAULT_WHERE_TO_LOOK = """- Public discussion of this kind of product: X/Twitter posts, Substack and blog posts, YouTube
video descriptions, Hacker News, forums, and Reddit snippets via search.
- Competing and adjacent tools: what they ship, and what their users praise or complain about.
- Official changelogs, API docs and blogs of anything this product relies on or connects to.
- News about this product's field that creates new needs."""
def where_to_look(project: Project) -> str:
notes = project.research_notes.strip() or DEFAULT_WHERE_TO_LOOK
if project.search_terms:
terms = ", ".join(f'"{t}"' for t in project.search_terms)
notes += f"\n- Search for: {terms}."
return notes
def constraints(project: Project) -> str:
return "".join(f"\n- Avoid {a}" for a in project.avoid)
def category_options(project: Project) -> str:
return " | ".join(f'"{c}"' for c in project.categories) or '"feature"'
def venue_field(project: Project) -> str:
"""Projects that cover more than one place tag each idea with which one it serves."""
if not project.areas:
return ""
options = " | ".join(f'"{a.lower()}"' for a in project.areas)
return f' "venue": {options} | "both" (which one this serves),\n'
def plan_field(project: Project) -> str:
"""Projects with more than one plan get each idea placed in one of them."""
if not project.plans:
return ""
options = " | ".join(f'"{a.lower()}"' for a in project.plans)
return (f' "plan": {options} (which plan this belongs in),\n'
' "plan_reason": "one sentence: why that plan — what makes it worth paying for, or why it must stay free",\n')
def build_prompt(project: Project, context: str, signals: str, prior: str,
count: int = IDEAS_PER_PLATFORM) -> str:
return Template(PROMPT_PATH.read_text()).substitute(
name=project.name, topic=project.topic, description=project.description,
audience=project.audience or "the people who use it",
where_to_look=where_to_look(project), constraints=constraints(project),
categories=category_options(project), venue_field=venue_field(project),
plan_field=plan_field(project),
context=context, signals=signals, prior=prior, count=count)
def parse_ideas(text: str) -> list[dict[str, Any]]:
data = extract_json(text)
if isinstance(data, dict):
data = data.get("ideas") or data.get("items") or [data]
if not isinstance(data, list):
raise ValueError("expected a JSON array of ideas")
ideas = []
for raw in data:
if not isinstance(raw, dict):
continue
title = str(raw.get("title") or "").strip()
summary = str(raw.get("summary") or "").strip()
if not title or not summary:
continue
evidence = []
for ev in raw.get("evidence") or []:
if isinstance(ev, dict) and str(ev.get("url", "")).startswith("http"):
evidence.append({"url": str(ev["url"]).strip(), "note": str(ev.get("note") or "").strip()})
elif isinstance(ev, str) and ev.startswith("http"):
evidence.append({"url": ev.strip(), "note": ""})
if not evidence:
continue # no evidence, no idea — the rule is real demand, not invention
effort = str(raw.get("effort") or "M").strip().upper()[:1]
category = str(raw.get("category") or "feature").strip().lower()
venue = str(raw.get("venue") or "both").strip().lower()
ideas.append({
"venue": venue if venue in VENUES else "both",
"title": title[:120],
"summary": summary,
"user_value": str(raw.get("user_value") or "").strip(),
"plan": str(raw.get("plan") or "").strip().lower()[:20],
"plan_reason": str(raw.get("plan_reason") or "").strip(),
"value_score": clamp_score(raw.get("value_score")),
"value_reason": str(raw.get("value_reason") or "").strip(),
"evidence": evidence,
"effort": effort if effort in VALID_EFFORT else "M",
"category": category if category in VALID_CATEGORY else "feature",
"spec": str(raw.get("spec") or "").strip(),
})
return ideas
_WORD = re.compile(r"[a-z0-9]+")
STOP = {"a", "an", "the", "of", "for", "and", "to", "in", "on", "with", "per", "by", "vs", "market",
"markets", "feature", "view", "page", "tool"}
def _tokens(s: str) -> set[str]:
return {w for w in _WORD.findall(s.lower()) if w not in STOP}
def is_duplicate(title: str, existing: list[str], threshold: float = 0.6) -> bool:
t = _tokens(title)
if not t:
return False
for other in existing:
o = _tokens(other)
if not o:
continue
overlap = len(t & o) / len(t | o)
if overlap >= threshold:
return True
return False
def run_scan(project: Project, store: Store, *, runner: Callable[..., ClaudeResult] = run_claude,
fetch: Callable[[str], str] | None = None,
reddit_posts: list[dict[str, Any]] | None = None) -> dict[str, Any]:
"""Research one project; returns counts so a run that examined nothing is visible."""
reddit_problems: list[str] = []
if reddit_posts is None:
reddit_posts, reddit_problems = collect_reddit(project)
prompt = build_prompt(project, repo_context(project), fetch_signals(project, fetch, reddit_posts),
prior_titles(store, project), count=project.ideas_per_scan)
from .config import scan_model
result = runner(prompt, cwd=project.folder, allowed_tools=["WebSearch", "WebFetch"],
max_turns=60, timeout_s=SCAN_TIMEOUT_S, model=scan_model(),
log_path=LOG_DIR / f"scan-{project.key}-latest.json")
if not result.ok:
return {"project": project.key, "ok": False, "error": result.error, "found": 0, "added": 0,
"duplicates": 0, "cost_usd": result.cost_usd}
try:
ideas = parse_ideas(result.text)
except ValueError as e:
return {"project": project.key, "ok": False, "error": f"{e}: {result.text[:300]}", "found": 0,
"added": 0, "duplicates": 0, "cost_usd": result.cost_usd}
existing = [t for t, _ in store.all_titles(project.key)]
added, dupes = [], 0
for idea in ideas[:project.ideas_per_scan]:
if is_duplicate(idea["title"], existing):
dupes += 1
continue
idea_id = store.add_idea(project.key, idea["title"], idea["summary"], user_value=idea["user_value"],
evidence=idea["evidence"], effort=idea["effort"], category=idea["category"],
spec=idea["spec"], source="scan", venue=idea["venue"],
value_score=idea["value_score"], value_reason=idea["value_reason"],
plan=idea["plan"], plan_reason=idea["plan_reason"])
existing.append(idea["title"])
added.append(idea_id)
return {"project": project.key, "ok": True, "found": len(ideas), "added": len(added), "ids": added,
"duplicates": dupes, "cost_usd": result.cost_usd, "reddit_posts": len(reddit_posts),
"reddit_problems": reddit_problems}
# ---- value scores for ideas that arrived before scoring existed -----------------------
def build_score_prompt(project: Project, ideas: list[dict[str, Any]]) -> str:
lines = []
for i in ideas:
ev = "; ".join(f"{e.get('note') or e.get('url')}" for e in (i.get("evidence") or [])[:3])
lines.append(f"- id {i['id']}: {i['title']}\n What: {i['summary']}\n"
f" Why users want it: {i.get('user_value') or '(not given)'}\n Evidence: {ev or '(none)'}")
tpl = Template(SCORE_PROMPT_PATH.read_text())
return tpl.substitute(name=project.name, topic=project.topic, description=project.description,
audience=project.audience or "the people who use it", ideas="\n".join(lines))
def parse_scores(text: str) -> dict[int, tuple[int, str]]:
"""{idea id: (score, reason)} for every well-formed entry in the answer."""
data = extract_json(text)
if isinstance(data, dict):
data = data.get("scores") or data.get("ideas") or data.get("items") or [data]
if not isinstance(data, list):
raise ValueError("expected a JSON array of scores")
out: dict[int, tuple[int, str]] = {}
for raw in data:
if not isinstance(raw, dict):
continue
try:
iid = int(raw.get("id"))
except (TypeError, ValueError):
continue
score = clamp_score(raw.get("value_score"))
if score is None:
continue
out[iid] = (score, str(raw.get("value_reason") or "").strip())
return out
def score_ideas(project: Project, store: Store, *, runner: Callable[..., ClaudeResult] = run_claude,
rescore: bool = False) -> dict[str, Any]:
"""Give every unscored (or, with ``rescore``, every open) idea a value score."""
open_statuses = ("proposed", "approved", "building", "built", "blocked", "failed")
ideas = [i for i in store.ideas(platform=project.key, statuses=open_statuses)
if rescore or not i.get("value_score")]
if not ideas:
return {"project": project.key, "ok": True, "asked": 0, "scored": 0, "cost_usd": 0.0}
prompt = build_score_prompt(project, ideas)
result = runner(prompt, cwd=project.folder, allowed_tools=[], max_turns=3, timeout_s=600,
log_path=LOG_DIR / f"score-{project.key}-latest.json")
if not result.ok:
return {"project": project.key, "ok": False, "error": result.error, "asked": len(ideas),
"scored": 0, "cost_usd": result.cost_usd}
try:
scores = parse_scores(result.text)
except ValueError as e:
return {"project": project.key, "ok": False, "error": f"{e}: {result.text[:300]}",
"asked": len(ideas), "scored": 0, "cost_usd": result.cost_usd}
wanted = {i["id"] for i in ideas}
scored = 0
for iid, (score, reason) in scores.items():
if iid in wanted:
store.set_value(iid, score, reason)
scored += 1
missing = sorted(wanted - set(scores))
return {"project": project.key, "ok": True, "asked": len(ideas), "scored": scored,
"unscored_ids": missing, "cost_usd": result.cost_usd}