diff --git a/scripts/python/common.py b/scripts/python/common.py index 77c13eefbb..fb4c46cacf 100644 --- a/scripts/python/common.py +++ b/scripts/python/common.py @@ -182,6 +182,78 @@ def get_feature_paths( ) +def _sorted_preset_ids(presets_dir: Path) -> list[str]: + registry = presets_dir / ".registry" + if registry.is_file(): + # Mirrors bash: any failure while reading or sorting the registry + # (invalid JSON, non-dict shapes, unorderable priority values) falls + # back to the directory scan below. + try: + data = json.loads(registry.read_text(encoding="utf-8")) + presets = data.get("presets", {}) + return [ + pid + for pid, meta in sorted( + presets.items(), + key=lambda kv: kv[1].get("priority", 10) + if isinstance(kv[1], dict) + else 10, + ) + if isinstance(meta, dict) and meta.get("enabled", True) is not False + ] + except Exception: + pass + try: + return sorted( + p.name + for p in presets_dir.iterdir() + if p.is_dir() and not p.name.startswith(".") + ) + except OSError: + return [] + + +def resolve_template(template_name: str, repo_root: Path) -> Path | None: + """Resolve a template name to a file path using the priority stack. + + Order (mirrors resolve_template in scripts/bash/common.sh): + 1. .specify/templates/overrides/ + 2. .specify/presets//templates/ (sorted by .registry priority) + 3. .specify/extensions//templates/ (hidden directories skipped) + 4. .specify/templates/ (core) + """ + base = repo_root / ".specify" / "templates" + + override = base / "overrides" / f"{template_name}.md" + if override.is_file(): + return override + + presets_dir = repo_root / ".specify" / "presets" + if presets_dir.is_dir(): + for preset_id in _sorted_preset_ids(presets_dir): + candidate = presets_dir / preset_id / "templates" / f"{template_name}.md" + if candidate.is_file(): + return candidate + + ext_dir = repo_root / ".specify" / "extensions" + if ext_dir.is_dir(): + try: + extensions = sorted(p for p in ext_dir.iterdir() if p.is_dir()) + except OSError: + extensions = [] + for ext in extensions: + if ext.name.startswith("."): + continue + candidate = ext / "templates" / f"{template_name}.md" + if candidate.is_file(): + return candidate + + core = base / f"{template_name}.md" + if core.is_file(): + return core + return None + + def get_invoke_separator(repo_root: Path) -> str: integration_json = repo_root / ".specify" / "integration.json" if not integration_json.is_file(): diff --git a/scripts/python/create_new_feature.py b/scripts/python/create_new_feature.py new file mode 100644 index 0000000000..c5f349a8fe --- /dev/null +++ b/scripts/python/create_new_feature.py @@ -0,0 +1,311 @@ +#!/usr/bin/env python3 +"""Create a new feature directory and spec file.""" + +from __future__ import annotations + +import datetime +import json +import re +import shlex +import shutil +import sys +from dataclasses import dataclass +from pathlib import Path + +try: + from common import get_repo_root, persist_feature_json, resolve_template +except ImportError: # pragma: no cover - direct execution from unusual cwd + sys.path.insert(0, str(Path(__file__).resolve().parent)) + from common import get_repo_root, persist_feature_json, resolve_template + + +def _json_line(payload: object) -> str: + return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n" + + +_STOP_WORDS = frozenset( + """ + i a an the to for of in on at by with from is are was were be been being + have has had do does did will would should could can may might must shall + this that these those my your our their want need add get set + """.split() +) + +_MAX_BRANCH_LENGTH = 244 + + +def _usage(argv0: str) -> str: + return ( + f"Usage: {argv0} [--json] [--dry-run] [--allow-existing-branch] " + "[--short-name ] [--number N] [--timestamp] " + ) + + +def _help_text(argv0: str) -> str: + return f"""{_usage(argv0)} + +Options: + --json Output in JSON format + --dry-run Compute feature name and paths without creating directories or files + --allow-existing-branch Reuse an existing feature directory if it already exists + --short-name Provide a custom short name (2-4 words) for the feature + --number N Specify branch number manually (overrides auto-detection) + --timestamp Use timestamp prefix (YYYYMMDD-HHMMSS) instead of sequential numbering + --help, -h Show this help message + +Examples: + {argv0} 'Add user authentication system' --short-name 'user-auth' + {argv0} 'Implement OAuth2 integration for API' --number 5 + {argv0} --timestamp --short-name 'user-auth' 'Add user authentication' +""" + + +@dataclass(frozen=True) +class Args: + json_mode: bool = False + dry_run: bool = False + allow_existing: bool = False + short_name: str = "" + branch_number: str = "" + use_timestamp: bool = False + description: str = "" + + +def _parse_args(argv: list[str], argv0: str) -> Args: + json_mode = False + dry_run = False + allow_existing = False + short_name = "" + branch_number = "" + use_timestamp = False + rest: list[str] = [] + + i = 0 + while i < len(argv): + arg = argv[i] + if arg == "--json": + json_mode = True + elif arg == "--dry-run": + dry_run = True + elif arg == "--allow-existing-branch": + allow_existing = True + elif arg in {"--short-name", "--number"}: + if i + 1 >= len(argv) or argv[i + 1].startswith("--"): + print(f"Error: {arg} requires a value", file=sys.stderr) + raise SystemExit(1) + i += 1 + if arg == "--short-name": + short_name = argv[i] + else: + branch_number = argv[i] + elif arg == "--timestamp": + use_timestamp = True + elif arg in {"--help", "-h"}: + sys.stdout.write(_help_text(argv0)) + raise SystemExit(0) + else: + rest.append(arg) + i += 1 + + description = " ".join(rest).strip() + if not description: + if rest: + print( + "Error: Feature description cannot be empty or contain only whitespace", + file=sys.stderr, + ) + else: + print(_usage(argv0), file=sys.stderr) + raise SystemExit(1) + + return Args( + json_mode=json_mode, + dry_run=dry_run, + allow_existing=allow_existing, + short_name=short_name, + branch_number=branch_number, + use_timestamp=use_timestamp, + description=description, + ) + + +def _clean_branch_name(name: str) -> str: + cleaned = re.sub(r"[^a-z0-9]", "-", name.lower()) + cleaned = re.sub(r"-+", "-", cleaned) + return cleaned.strip("-") + + +def _generate_branch_name(description: str) -> str: + clean = re.sub(r"[^a-z0-9]", " ", description.lower()) + meaningful: list[str] = [] + for word in clean.split(): + if word in _STOP_WORDS: + continue + if len(word) >= 3: + meaningful.append(word) + # Keep short words that appear as an uppercase acronym in the original, + # mirroring the bash twin's case-sensitive `grep -qw` check. + elif re.search( + rf"(? int: + highest = 0 + if not specs_dir.is_dir(): + return highest + for entry in specs_dir.iterdir(): + if not entry.is_dir(): + continue + name = entry.name + # Match sequential prefixes (>=3 digits), but skip timestamp dirs. + if re.match(r"^[0-9]{3,}-", name) and not re.match( + r"^[0-9]{8}-[0-9]{6}-", name + ): + number = int(re.match(r"^[0-9]+", name).group(), 10) + highest = max(highest, number) + return highest + + +def main(argv: list[str] | None = None) -> int: + argv0 = sys.argv[0] + args = _parse_args(list(argv if argv is not None else sys.argv[1:]), argv0) + + repo_root = get_repo_root(Path(__file__)) + specs_dir = repo_root / "specs" + if not args.dry_run: + specs_dir.mkdir(parents=True, exist_ok=True) + + if args.short_name: + branch_suffix = _clean_branch_name(args.short_name) + else: + branch_suffix = _generate_branch_name(args.description) + + branch_number = args.branch_number + if args.use_timestamp and branch_number: + print( + "[specify] Warning: --number is ignored when --timestamp is used", + file=sys.stderr, + ) + branch_number = "" + + if args.use_timestamp: + feature_num = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") + else: + if branch_number: + try: + number = int(branch_number, 10) + except ValueError: + print( + f"Error: --number must be an integer, got '{branch_number}'", + file=sys.stderr, + ) + return 1 + else: + number = _get_highest_from_specs(specs_dir) + 1 + feature_num = f"{number:03d}" + branch_name = f"{feature_num}-{branch_suffix}" + + # GitHub enforces a 244-byte limit on branch names. + if len(branch_name) > _MAX_BRANCH_LENGTH: + max_suffix_length = _MAX_BRANCH_LENGTH - (len(feature_num) + 1) + truncated_suffix = re.sub(r"-$", "", branch_suffix[:max_suffix_length]) + original_branch_name = branch_name + branch_name = f"{feature_num}-{truncated_suffix}" + print( + "[specify] Warning: Branch name exceeded GitHub's 244-byte limit", + file=sys.stderr, + ) + print( + f"[specify] Original: {original_branch_name} " + f"({len(original_branch_name)} bytes)", + file=sys.stderr, + ) + print( + f"[specify] Truncated to: {branch_name} ({len(branch_name)} bytes)", + file=sys.stderr, + ) + + feature_dir = specs_dir / branch_name + spec_file = feature_dir / "spec.md" + + if not args.dry_run: + if feature_dir.is_dir() and not args.allow_existing: + if args.use_timestamp: + print( + f"Error: Feature directory '{feature_dir}' already exists. " + "Rerun to get a new timestamp or use a different --short-name.", + file=sys.stderr, + ) + else: + print( + f"Error: Feature directory '{feature_dir}' already exists. " + "Please use a different feature name or specify a different " + "number with --number.", + file=sys.stderr, + ) + return 1 + + feature_dir.mkdir(parents=True, exist_ok=True) + + if not spec_file.is_file(): + template = resolve_template("spec-template", repo_root) + if template is not None and template.is_file(): + shutil.copy(template, spec_file) + else: + print( + "Warning: Spec template not found; created empty spec file", + file=sys.stderr, + ) + spec_file.touch() + + # Persist to .specify/feature.json so downstream commands can find the feature. + persist_feature_json(repo_root, str(feature_dir)) + + # Inform the user how to set feature state in their own shell. + print( + f"# To persist: export SPECIFY_FEATURE={shlex.quote(branch_name)}", + file=sys.stderr, + ) + print( + "# export " + f"SPECIFY_FEATURE_DIRECTORY={shlex.quote(str(feature_dir))}", + file=sys.stderr, + ) + + if args.json_mode: + payload: dict[str, object] = { + "BRANCH_NAME": branch_name, + "SPEC_FILE": str(spec_file), + "FEATURE_NUM": feature_num, + } + if args.dry_run: + payload["DRY_RUN"] = True + sys.stdout.write(_json_line(payload)) + else: + print(f"BRANCH_NAME: {branch_name}") + print(f"SPEC_FILE: {spec_file}") + print(f"FEATURE_NUM: {feature_num}") + if not args.dry_run: + print( + "# To persist in your shell: export " + f"SPECIFY_FEATURE={shlex.quote(branch_name)}" + ) + print( + "# export " + f"SPECIFY_FEATURE_DIRECTORY={shlex.quote(str(feature_dir))}" + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/python/setup_plan.py b/scripts/python/setup_plan.py new file mode 100644 index 0000000000..7b8e77ce5a --- /dev/null +++ b/scripts/python/setup_plan.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +"""Setup implementation plan for a feature.""" + +from __future__ import annotations + +import json +import shutil +import sys +from pathlib import Path + +try: + from common import get_feature_paths, resolve_template +except ImportError: # pragma: no cover - direct execution from unusual cwd + sys.path.insert(0, str(Path(__file__).resolve().parent)) + from common import get_feature_paths, resolve_template + + +def _json_line(payload: object) -> str: + return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n" + + +def _help_text(argv0: str) -> str: + return f"""Usage: {argv0} [--json] + --json Output results in JSON format + --help Show this help message +""" + + +def main(argv: list[str] | None = None) -> int: + args = list(argv if argv is not None else sys.argv[1:]) + json_mode = False + for arg in args: + if arg == "--json": + json_mode = True + elif arg in {"--help", "-h"}: + sys.stdout.write(_help_text(sys.argv[0])) + return 0 + # Other arguments are accepted and silently ignored, matching setup-plan.sh. + + try: + paths = get_feature_paths(script_file=Path(__file__)) + except SystemExit as exc: + if exc.code == 0: + return 0 + print("ERROR: Failed to resolve feature paths", file=sys.stderr) + return int(exc.code) if isinstance(exc.code, int) else 1 + + paths.feature_dir.mkdir(parents=True, exist_ok=True) + + # Status messages go to stderr in JSON mode so stdout stays pure JSON. + status_stream = sys.stderr if json_mode else sys.stdout + if paths.impl_plan.is_file(): + print( + f"Plan already exists at {paths.impl_plan}, skipping template copy", + file=status_stream, + ) + else: + template = resolve_template("plan-template", paths.repo_root) + if template is not None and template.is_file(): + shutil.copy(template, paths.impl_plan) + print(f"Copied plan template to {paths.impl_plan}", file=status_stream) + else: + print("Warning: Plan template not found", file=status_stream) + paths.impl_plan.touch() + + if json_mode: + sys.stdout.write( + _json_line( + { + "FEATURE_SPEC": str(paths.feature_spec), + "IMPL_PLAN": str(paths.impl_plan), + "SPECS_DIR": str(paths.feature_dir), + "BRANCH": paths.current_branch, + } + ) + ) + else: + print(f"FEATURE_SPEC: {paths.feature_spec}") + print(f"IMPL_PLAN: {paths.impl_plan}") + print(f"SPECS_DIR: {paths.feature_dir}") + print(f"BRANCH: {paths.current_branch}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/python/setup_tasks.py b/scripts/python/setup_tasks.py new file mode 100644 index 0000000000..b3abb6dc1a --- /dev/null +++ b/scripts/python/setup_tasks.py @@ -0,0 +1,145 @@ +#!/usr/bin/env python3 +"""Check tasks prerequisites and resolve the tasks template.""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +try: + from common import ( + FeaturePaths, + format_speckit_command, + get_feature_paths, + resolve_template, + ) +except ImportError: # pragma: no cover - direct execution from unusual cwd + sys.path.insert(0, str(Path(__file__).resolve().parent)) + from common import ( + FeaturePaths, + format_speckit_command, + get_feature_paths, + resolve_template, + ) + + +def _json_line(payload: object) -> str: + return json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n" + + +def _help_text(argv0: str) -> str: + return f"""Usage: {argv0} [--json] + --json Output results in JSON format + --help Show this help message +""" + + +def _dir_has_entries(path: Path) -> bool: + try: + return path.is_dir() and any(path.iterdir()) + except OSError: + return False + + +def _available_docs(paths: FeaturePaths) -> list[str]: + docs: list[str] = [] + if paths.research.is_file(): + docs.append("research.md") + if paths.data_model.is_file(): + docs.append("data-model.md") + if _dir_has_entries(paths.contracts_dir): + docs.append("contracts/") + if paths.quickstart.is_file(): + docs.append("quickstart.md") + return docs + + +def _check_file(path: Path, description: str) -> None: + marker = "✓" if path.is_file() else "✗" + print(f" {marker} {description}") + + +def _check_dir(path: Path, description: str) -> None: + marker = "✓" if _dir_has_entries(path) else "✗" + print(f" {marker} {description}") + + +def main(argv: list[str] | None = None) -> int: + json_mode = False + for arg in list(argv if argv is not None else sys.argv[1:]): + if arg == "--json": + json_mode = True + elif arg in {"--help", "-h"}: + sys.stdout.write(_help_text(sys.argv[0])) + return 0 + else: + print(f"ERROR: Unknown option '{arg}'", file=sys.stderr) + return 1 + + try: + paths = get_feature_paths(script_file=Path(__file__)) + except SystemExit as exc: + if exc.code == 0: + return 0 + print("ERROR: Failed to resolve feature paths", file=sys.stderr) + return int(exc.code) if isinstance(exc.code, int) else 1 + + if not paths.impl_plan.is_file(): + print(f"ERROR: plan.md not found in {paths.feature_dir}", file=sys.stderr) + print( + f"Run {format_speckit_command('plan', paths.repo_root)} first to create the implementation plan.", + file=sys.stderr, + ) + return 1 + + if not paths.feature_spec.is_file(): + print(f"ERROR: spec.md not found in {paths.feature_dir}", file=sys.stderr) + print( + f"Run {format_speckit_command('specify', paths.repo_root)} first to create the feature structure.", + file=sys.stderr, + ) + return 1 + + docs = _available_docs(paths) + + tasks_template = resolve_template("tasks-template", paths.repo_root) + if tasks_template is None or not tasks_template.is_file(): + print( + "ERROR: Could not resolve required tasks-template from the template " + f"override stack for {paths.repo_root}", + file=sys.stderr, + ) + print( + "Template 'tasks-template' was not found in any supported location " + "(overrides, presets, extensions, or shared core). Add an override at " + ".specify/templates/overrides/tasks-template.md, or run 'specify init' " + "/ reinstall shared infra to restore the core " + ".specify/templates/tasks-template.md template.", + file=sys.stderr, + ) + return 1 + + if json_mode: + sys.stdout.write( + _json_line( + { + "FEATURE_DIR": str(paths.feature_dir), + "AVAILABLE_DOCS": docs, + "TASKS_TEMPLATE": str(tasks_template), + } + ) + ) + else: + print(f"FEATURE_DIR: {paths.feature_dir}") + print(f"TASKS_TEMPLATE: {tasks_template}") + print("AVAILABLE_DOCS:") + _check_file(paths.research, "research.md") + _check_file(paths.data_model, "data-model.md") + _check_dir(paths.contracts_dir, "contracts/") + _check_file(paths.quickstart, "quickstart.md") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/specify_cli/agents.py b/src/specify_cli/agents.py index 7864260a99..3458867d01 100644 --- a/src/specify_cli/agents.py +++ b/src/specify_cli/agents.py @@ -408,7 +408,7 @@ def resolve_skill_placeholders( init_opts = {} script_variant = init_opts.get("script") - if script_variant not in {"sh", "ps"}: + if script_variant not in {"sh", "ps", "py"}: fallback_order = [] default_variant = ( "ps" if platform.system().lower().startswith("win") else "sh" @@ -428,6 +428,16 @@ def resolve_skill_placeholders( script_command = scripts.get(script_variant) if script_variant else None if script_command: + if script_variant == "py": + # Same portability handling as process_template: .py files + # are not directly executable on Windows, so prefix the + # resolved interpreter (quoted when it contains whitespace). + from specify_cli.integrations.base import IntegrationBase + + interpreter = IntegrationBase.resolve_python_interpreter(project_root) + if any(ch.isspace() for ch in interpreter): + interpreter = f'"{interpreter}"' + script_command = f"{interpreter} {script_command}" script_command = script_command.replace("{ARGS}", "$ARGUMENTS") body = body.replace("{SCRIPT}", script_command) diff --git a/templates/commands/plan.md b/templates/commands/plan.md index e82bd4b303..3a56c671ca 100644 --- a/templates/commands/plan.md +++ b/templates/commands/plan.md @@ -11,6 +11,7 @@ handoffs: scripts: sh: scripts/bash/setup-plan.sh --json ps: scripts/powershell/setup-plan.ps1 -Json + py: scripts/python/setup_plan.py --json --- ## User Input diff --git a/templates/commands/tasks.md b/templates/commands/tasks.md index 4d3e45a7c4..454d7e3b32 100644 --- a/templates/commands/tasks.md +++ b/templates/commands/tasks.md @@ -12,6 +12,7 @@ handoffs: scripts: sh: scripts/bash/setup-tasks.sh --json ps: scripts/powershell/setup-tasks.ps1 -Json + py: scripts/python/setup_tasks.py --json --- ## User Input diff --git a/tests/parity_helpers.py b/tests/parity_helpers.py new file mode 100644 index 0000000000..ea83db8247 --- /dev/null +++ b/tests/parity_helpers.py @@ -0,0 +1,130 @@ +"""Shared helpers for the core-script Python parity tests.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +BASH_DIR = PROJECT_ROOT / "scripts" / "bash" +PS_DIR = PROJECT_ROOT / "scripts" / "powershell" +PY_DIR = PROJECT_ROOT / "scripts" / "python" + +HAS_PWSH = shutil.which("pwsh") is not None +WINDOWS_POWERSHELL = ( + (shutil.which("powershell.exe") or shutil.which("powershell")) + if os.name == "nt" + else None +) +POWERSHELL_EXE = "pwsh" if HAS_PWSH else WINDOWS_POWERSHELL +HAS_POWERSHELL = POWERSHELL_EXE is not None + + +def make_repo(tmp_path: Path, name: str = "proj") -> Path: + repo = tmp_path / name + (repo / ".specify").mkdir(parents=True) + return repo + + +def install_scripts(repo: Path, script: str) -> None: + """Install the bash/powershell/python twins of a kebab-case script name.""" + py_name = script.replace("-", "_") + + bash_dir = repo / ".specify" / "scripts" / "bash" + bash_dir.mkdir(parents=True, exist_ok=True) + shutil.copy(BASH_DIR / "common.sh", bash_dir / "common.sh") + shutil.copy(BASH_DIR / f"{script}.sh", bash_dir / f"{script}.sh") + + ps_dir = repo / ".specify" / "scripts" / "powershell" + ps_dir.mkdir(parents=True, exist_ok=True) + shutil.copy(PS_DIR / "common.ps1", ps_dir / "common.ps1") + shutil.copy(PS_DIR / f"{script}.ps1", ps_dir / f"{script}.ps1") + + py_dir = repo / ".specify" / "scripts" / "python" + py_dir.mkdir(parents=True, exist_ok=True) + shutil.copy(PY_DIR / "common.py", py_dir / "common.py") + shutil.copy(PY_DIR / f"{py_name}.py", py_dir / f"{py_name}.py") + + +def bash_cmd(repo: Path, script: str, *args: str) -> list[str]: + return ["bash", str(repo / ".specify" / "scripts" / "bash" / f"{script}.sh"), *args] + + +def py_cmd(repo: Path, script: str, *args: str) -> list[str]: + py_name = script.replace("-", "_") + return [ + sys.executable, + str(repo / ".specify" / "scripts" / "python" / f"{py_name}.py"), + *args, + ] + + +def ps_cmd(repo: Path, script: str, *args: str) -> list[str]: + assert POWERSHELL_EXE, "no PowerShell available; guard the test with HAS_POWERSHELL" + return [ + POWERSHELL_EXE, + "-NoProfile", + "-File", + str(repo / ".specify" / "scripts" / "powershell" / f"{script}.ps1"), + *args, + ] + + +def clean_env() -> dict[str, str]: + env = os.environ.copy() + for key in list(env): + if key.startswith("SPECIFY_"): + env.pop(key) + return env + + +def run( + cmd: list[str], repo: Path, env: dict[str, str] | None = None +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + cmd, + cwd=repo, + capture_output=True, + text=True, + check=False, + env=env if env is not None else clean_env(), + ) + + +def json_stdout(result: subprocess.CompletedProcess[str]) -> object: + return json.loads(result.stdout) + + +def write_feature_json( + repo: Path, feature_directory: str = "specs/001-my-feature" +) -> None: + (repo / ".specify" / "feature.json").write_text( + json.dumps({"feature_directory": feature_directory}, separators=(",", ":")) + + "\n", + encoding="utf-8", + ) + + +def normalize_repo_paths(text: str, repo: Path) -> str: + """Replace the repo path with a placeholder so two-repo runs compare equal.""" + return text.replace(str(repo), "").replace("\r\n", "\n") + + +def normalize_script_names(text: str, repo: Path, script: str) -> str: + """Replace per-runtime script paths (argv[0] in usage/help output).""" + py_name = script.replace("-", "_") + bash_script = str(repo / ".specify" / "scripts" / "bash" / f"{script}.sh") + py_script = str(repo / ".specify" / "scripts" / "python" / f"{py_name}.py") + return text.replace(bash_script, "