Advertisement

← refacto-feature-scout

scout/init.py

17.9 KB Download file
"""Setting up a project, for somebody who would rather not learn git.

Every question has a suggested answer already filled in, so the whole thing can be finished by
pressing return. Nothing here talks about branches, worktrees or repositories.
"""
from __future__ import annotations

import json
import re
import subprocess
from pathlib import Path
from typing import Any, Callable

import platform

from . import detect
from .claude_runner import ClaudeResult, run_claude
from .config import DEFAULT_CATEGORIES_DOC, MODEL_CHOICES, load_env, save_setting
from .llm import NoLLMAccess, require_llm_access
from .projects import DEFAULT_CATEGORIES, PROJECT_FILE, Project, from_dict, register, validate

Ask = Callable[[str, str], str]      # (question, suggested answer) -> answer
Say = Callable[[str], None]

DESCRIBE_PROMPT = """Read this project and describe it for a product researcher.

Reply with ONLY a JSON object:
{
  "name": "what this project is called, in a few words",
  "what_it_is": "2-3 sentences: what it does and what someone sees when they use it",
  "who_uses_it": "1-2 sentences: who it is for",
  "audience": "a short phrase naming those people, e.g. 'home cooks who plan their week'",
  "topic": "a short clause that could follow the name, starting with a comma or a space and \
an opening bracket, e.g. ', a meal planning app'. May be an empty string.",
  "communities": ["names of subreddits where these people talk, without the r/, best guesses"],
  "search_terms": ["2-4 phrases someone would search to find tools like this"]
}"""


def ask_terminal(question: str, suggested: str = "") -> str:
    shown = f" [{suggested}]" if suggested else ""
    got = input(f"{question}{shown}\n> ").strip()
    return got or suggested


def _run_tests_once(folder: Path, command: str, timeout_s: int = 600) -> tuple[bool, str]:
    try:
        proc = subprocess.run(command, shell=True, cwd=str(folder), capture_output=True,
                              text=True, timeout=timeout_s)
    except subprocess.TimeoutExpired:
        return False, "it was still going after ten minutes"
    except OSError as e:
        return False, str(e)
    return proc.returncode == 0, (proc.stdout + proc.stderr)[-600:]


def _describe_with_claude(folder: Path, runner: Callable[..., ClaudeResult]) -> dict[str, Any]:
    from .claude_runner import extract_json
    result = runner(DESCRIBE_PROMPT, cwd=folder, allowed_tools=["Read", "Glob", "Grep"],
                    max_turns=15, timeout_s=300)
    if not result.ok:
        return {}
    try:
        data = extract_json(result.text)
    except ValueError:
        return {}
    return data if isinstance(data, dict) else {}


def _reddit_feeds(names: list[str]) -> list[str]:
    """Turn what a person types — 'r/mealprep', 'mealprep', or a full address — into feeds."""
    out = []
    for raw in names:
        name = raw.strip().strip(",")
        if not name:
            continue
        if name.startswith("http"):
            out.append(name)
            continue
        name = name.removeprefix("r/").removeprefix("/r/").strip("/")
        out.append(f"https://www.reddit.com/r/{name}/new/.rss")
        out.append(f"https://www.reddit.com/r/{name}/top/.rss?t=week")
    return out


def _yes(answer: str, default: bool = True) -> bool:
    answer = (answer or "").strip().lower()
    if not answer:
        return default
    return answer[0] in "yt1"


def _reddit_names(feeds: list[str]) -> str:
    """The reverse of _reddit_feeds, for showing a person what they said last time."""
    names: list[str] = []
    for f in feeds:
        m = re.match(r"https://www\.reddit\.com/r/([^/]+)/", f)
        if m and m.group(1) not in names:
            names.append(m.group(1))
        elif not m:
            names.append(f)
    return ", ".join(names)


def _never_run_patterns(answer: str) -> list[str]:
    """'npm run verify, npx playwright' -> tool patterns the builder is refused."""
    out = []
    for raw in answer.split(","):
        cmd = raw.strip().strip("`")
        if not cmd:
            continue
        out.append(cmd if cmd.startswith("Bash(") else f"Bash({cmd}*)")
    return out


def _never_run_words(patterns: list[str]) -> str:
    return ", ".join(p[len("Bash("):-1].rstrip("*") if p.startswith("Bash(") else p for p in patterns)


def _time_of_day(answer: str, fallback: tuple[int, int] = (8, 15)) -> tuple[int, int]:
    m = re.match(r"^\s*(\d{1,2})[:.h]?(\d{2})?\s*(am|pm)?\s*$", (answer or "").lower())
    if not m:
        return fallback
    hour, minute = int(m.group(1)), int(m.group(2) or 0)
    if m.group(3) == "pm" and hour < 12:
        hour += 12
    if m.group(3) == "am" and hour == 12:
        hour = 0
    if not (0 <= hour <= 23 and 0 <= minute <= 59):
        return fallback
    return hour, minute


def _model_word(model: str) -> str:
    """The reverse of _model_choice: 'opus' -> 'best', '' -> 'default'."""
    for word, m in MODEL_CHOICES.items():
        if m == (model or ""):
            return word
    return model or "default"


def _model_choice(answer: str) -> str:
    word = (answer or "default").strip().lower()
    for k, v in MODEL_CHOICES.items():
        if word.startswith(k):
            return v
    return MODEL_CHOICES.get(word, word if word in ("opus", "sonnet", "haiku") else "")


def install_background_jobs(hour: int, minute: int, say: Say = print) -> bool:
    """On a Mac, install the two launch jobs (daily look for ideas, half-hourly build pass)."""
    script = Path(__file__).resolve().parent.parent / "scripts" / "install-launchd.sh"
    if platform.system() != "Darwin" or not script.exists():
        return False
    import os
    env = {**os.environ, "SCAN_HOUR": str(hour), "SCAN_MINUTE": str(minute)}
    proc = subprocess.run(["bash", str(script)], env=env, capture_output=True, text=True)
    if proc.returncode != 0:
        say("Couldn't install the background jobs: " + (proc.stderr or proc.stdout)[-300:])
        return False
    return True


def run_init(*, ask: Ask = ask_terminal, say: Say = print, folder: Path | None = None,
             runner: Callable[..., ClaudeResult] = run_claude,
             inspector: Callable[[Path], detect.Found] = detect.inspect,
             tester: Callable[[Path, str], tuple[bool, str]] = _run_tests_once,
             use_ai: bool = True, register_it: bool = True,
             installer: Callable[[int, int, Say], bool] = install_background_jobs,
             env_file: Path | None = None) -> Project:
    """The guided setup: the project, where ideas come from, how work is checked, which Claude
    does what, how you are told, and when it runs. Every question has a suggested answer, so
    pressing return all the way through gives a sensible setup. Run it again any time — your
    earlier answers are the suggestions."""
    folder = Path(folder or Path.cwd()).resolve()
    cur: dict[str, Any] = {}
    if (folder / PROJECT_FILE).exists():
        try:
            cur = json.loads((folder / PROJECT_FILE).read_text())
        except json.JSONDecodeError:
            cur = {}
    machine = load_env(env_file) if env_file else load_env()

    # 1. There is no point asking anything of someone who can't run anything.
    say("First, your Claude account — Feature Scout runs on yours, it has none of its own.")
    try:
        access = require_llm_access()
        say(f"  Good: {access.detail}\n")
    except NoLLMAccess as e:
        say(str(e))
        raise

    # 2. The folder.
    answer = ask("Which folder is your project in?", str(folder))
    folder = Path(answer).expanduser().resolve()
    if not folder.exists():
        raise FileNotFoundError(f"There's no folder at {folder}.")
    if not detect.is_git_project(folder):
        say("\nThis folder isn't saved with git yet. Feature Scout needs that so it can work on a\n"
            "copy of your project and put things back if something goes wrong.")
        if _yes(ask("Set that up now?", "yes")):
            subprocess.run(["git", "init", "-q"], cwd=str(folder), check=False)
            subprocess.run(["git", "add", "-A"], cwd=str(folder), check=False)
            subprocess.run(["git", "commit", "-qm", "Where things stood before Feature Scout"],
                           cwd=str(folder), check=False)
            say("Done — that's your starting point saved.")
        else:
            say("All right. Set that up before running a build, or nothing can be undone.")

    found = inspector(folder)

    # 3. What it is. Reading the project is much better than making someone write it cold.
    drafted: dict[str, Any] = {}
    if use_ai and _yes(ask("\nShall I read your project and write a description for you?", "yes")):
        say("Reading it now — this takes a minute and costs a few cents of your Claude usage.")
        drafted = _describe_with_claude(folder, runner)
        if not drafted:
            say("I couldn't make sense of it, so I'll just ask.")

    name = ask("\nWhat is this project called?", drafted.get("name") or cur.get("name") or folder.name)
    what_it_is = ask("In a sentence or two, what does it do?", drafted.get("what_it_is") or cur.get("what_it_is", ""))
    who_uses_it = ask("Who uses it?", drafted.get("who_uses_it") or cur.get("who_uses_it", ""))
    audience = ask("Short version of who to research for?",
                   drafted.get("audience") or cur.get("audience") or who_uses_it)
    topic = drafted.get("topic") or cur.get("topic", "")

    # 4. Where ideas come from.
    say("\nIdeas come from what people say in public. Your own users are the best source.")
    cur_sources = cur.get("sources") or {}
    pages = ask("Any page where your users leave feedback or requests? (blank for none)",
                ", ".join(cur_sources.get("urls") or []))
    communities = ask("Any Reddit communities where these people talk? (comma separated, blank for none)",
                      ", ".join(drafted.get("communities") or []) or _reddit_names(cur_sources.get("reddit") or []))
    search_terms = [t.strip() for t in
                    ask("Phrases someone would search to find tools like yours?",
                        ", ".join(drafted.get("search_terms") or cur_sources.get("search_terms") or [])).split(",") if t.strip()]

    # 5. How it checks its own work.
    say("\nBefore anything is added to your project, Feature Scout runs your tests.")
    test_command = ask("How do you run this project's tests? (blank if it has none yet)",
                       cur.get("test_command") or found.test_command)
    if test_command:
        say(f"Trying `{test_command}` now, to be sure it works...")
        ok, tail = tester(folder, test_command)
        if ok:
            say("It passes. That's what will be used to check any change.")
        else:
            say(f"That didn't pass right now:\n  {tail.strip().splitlines()[-1][:200] if tail.strip() else ''}")
            if not _yes(ask("Keep it anyway?", "yes")):
                test_command = ""
    if not test_command:
        say("No tests, then. Feature Scout will still build things, but it can't check them —\n"
            "you'll be told that every time, and asked to try each one yourself.")

    # 6. Deploying is off unless someone asks for it.
    deploy_command = None
    if found.deploy_command:
        say(f"\nThis project looks like it can be put live with `{found.deploy_command}`.")
        if _yes(ask("Should Feature Scout do that when you say 'put it live'? "
                    "(no means it only adds the change to your project)", "no"), default=False):
            deploy_command = found.deploy_command

    # 7. What the internal, hand-tuned setups always had and a newcomer had no way to say.
    say("\nA few more things that make the ideas better. All optional.")
    url = ask("Where does it live on the web? (an address, blank for none)", cur.get("url") or "")
    research_notes = ask("Where else should the researcher look, or what should it steer clear of? "
                         "(a sentence or two, blank for none)", cur.get("research_notes", ""))
    avoid = [a.strip() for a in ask("Anything you never want proposed? (comma separated, blank for none)",
                                    ", ".join(cur.get("avoid") or [])).split(",") if a.strip()]
    cats = ask(f"Kinds of ideas you want ({DEFAULT_CATEGORIES_DOC})? Press return to keep these, "
               "or type your own list", ", ".join(cur.get("categories") or DEFAULT_CATEGORIES))
    categories = [c.strip().lower() for c in cats.split(",") if c.strip()] or list(DEFAULT_CATEGORIES)
    plans = [x.strip() for x in ask("Does it have plans, like Free and Pro? Each idea will be placed in one. "
                                    "(comma separated, blank for none)", ", ".join(cur.get("plans") or [])).split(",") if x.strip()]
    house_rules = ask("Rules the builder must always follow? (for example 'never change the database "
                      "layout'; blank for none)", cur.get("house_rules", ""))
    never_run = _never_run_patterns(ask("Anything the builder must never run here — a browser test suite, "
                                        "a deploy script? (comma separated commands, blank for none)",
                                        _never_run_words(cur.get("never_run") or [])))
    ideas_raw = ask("How many ideas should each day's look turn up?", str(cur.get("ideas_per_scan") or 4))
    try:
        ideas_per_scan = max(1, int(ideas_raw))
    except ValueError:
        ideas_per_scan = 4

    # 8. Money. It is theirs.
    limit = ask("\nBuilding an idea costs a few dollars of your own Claude usage. Stop for the day "
                "after how many dollars? (0 for no limit)", f"{float(cur.get('spend_limit_usd', 20) or 20):g}")
    try:
        spend_limit = float(limit)
    except ValueError:
        spend_limit = 20.0

    # 9. Which Claude does what. Kept on this computer, not in the project.
    say("\nWhich Claude should do the work? 'default' uses whatever your Claude is set to; "
        "'best' is the strongest and slowest, 'balanced' the middle, 'fast' the cheapest.")
    scan_model = _model_choice(ask("For the research (default / best / balanced / fast)?",
                                   _model_word(machine.get("SCOUT_SCAN_MODEL", ""))))
    build_model = _model_choice(ask("For building (default / best / balanced / fast)?",
                                    _model_word(machine.get("SCOUT_BUILD_MODEL", ""))))

    # 10. When it runs. (New ideas, finished builds and problems all show on the private page
    # in your browser — there is nothing to set up for that.)
    hour, minute = _time_of_day(ask("\nWhat time each day should it look for ideas? (HH:MM, local time)",
                                    f"{int(machine.get('SCOUT_SCAN_HOUR', 8)):02d}:{int(machine.get('SCOUT_SCAN_MINUTE', 15)):02d}"),
                                fallback=(8, 15))
    background = False
    if platform.system() == "Darwin":
        background = _yes(ask("Keep it running in the background on this Mac, even when this window is closed?",
                              "no"), default=False)

    settings = {
        "version": 1,
        "key": _key_for(name, folder),
        "name": name,
        "what_it_is": what_it_is,
        "who_uses_it": who_uses_it,
        "audience": audience,
        "topic": topic,
        "context_files": found.context_files,
        "sources": {
            "reddit": _reddit_feeds(communities.split(",")),
            "urls": [p.strip() for p in pages.split(",") if p.strip()],
            "search_terms": search_terms,
        },
        "url": url or None,
        "test_command": test_command,
        "extra_checks": cur.get("extra_checks") or found.extra_checks,
        "install_hint": cur.get("install_hint") or found.install_hint,
        "deploy_command": deploy_command,
        "link_dirs": list(cur.get("link_dirs") or found.link_dirs),
        "spend_limit_usd": spend_limit,
        "research_notes": research_notes,
        "avoid": avoid,
        "categories": categories,
        "plans": plans,
        "house_rules": house_rules,
        "never_run": never_run,
        "ideas_per_scan": ideas_per_scan,
    }
    path = folder / PROJECT_FILE
    path.write_text(json.dumps(settings, indent=2) + "\n")
    say(f"\nSaved your answers in {path}. You can edit that file any time.")

    # This computer's own choices — never in the project file, which travels with the project.
    for key, value in (("SCOUT_SCAN_MODEL", scan_model), ("SCOUT_BUILD_MODEL", build_model),
                       ("SCOUT_SCAN_HOUR", str(hour)), ("SCOUT_SCAN_MINUTE", str(minute))):
        save_setting(key, value, env_file)
    say(f"Your Claude and timing choices are kept in {env_file or 'your settings file'} on this computer.")
    if background:
        if installer(hour, minute, say):
            say(f"Feature Scout now runs in the background on this Mac: a look for ideas at "
                f"{hour:02d}:{minute:02d} every day, and a build pass every half hour.")
        else:
            say("The background jobs couldn't be installed; `feature-scout watch` does the same while it runs.")

    project = from_dict(settings, folder)
    for problem in validate(project):
        say(f"  Worth knowing: {problem}")
    if register_it:
        register(project.key, folder)

    say(f"\nAll set. Try this:\n"
        f"  feature-scout scan      look for ideas now\n"
        f"  feature-scout review    see them and pick the ones you want\n"
        f"  feature-scout watch     leave it running and it does both for you\n"
        f"  feature-scout setup     change any of these answers later")
    return project


def _key_for(name: str, folder: Path) -> str:
    import re
    key = re.sub(r"[^a-z0-9]+", "-", (name or folder.name).lower()).strip("-")
    return key[:40] or "project"