fix: sync CLI flags across analyze/pdf/unified commands and fix workflow JSON config

Flag/option synchronization fixes:
- analyze: add --dry-run, --api-key, and all workflow flags (--enhance-workflow,
  --enhance-stage, --var, --workflow-dry-run) via WORKFLOW_ARGUMENTS merge
- pdf: add --api-key to PDF_ARGUMENTS; replace 5 hardcoded add_argument() calls
  in pdf_scraper.py:main() with add_pdf_arguments() to activate all defined args
- unified: add --api-key and --enhance-level (global override) to UNIFIED_ARGUMENTS
  and standalone parser; wire enhance_level CLI override into run() per-source loop
- codebase_scraper: fix --enhance-workflow to use action="append" (was type=str),
  enabling multiple workflow chaining instead of silently dropping all but last

ConfigManager test isolation fix:
- __init__ now reads self.CONFIG_DIR/CONFIG_FILE/PROGRESS_DIR class variables
  instead of calling _get_config_dir()/_get_progress_dir() directly, enabling
  monkeypatching in tests (fixes pre-existing test_add_and_retrieve_github_profile)

Workflow JSON config support in unified_scraper:
- Phase 5 now reads workflows/workflow_stages/workflow_vars from top-level JSON
  config and merges them with CLI args (CLI-first ordering); supports running
  workflows even when unified scraper is called without CLI args (args=None)

Tests: 1,949 passed, 0 failed (added 18 new tests across 3 test files)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
yusyus
2026-02-22 00:44:02 +03:00
parent 47226340ac
commit 22bdd4f5f6
10 changed files with 438 additions and 13 deletions

View File

@@ -10,6 +10,8 @@ Includes preset system support for #268.
import argparse
from typing import Any
from .workflow import WORKFLOW_ARGUMENTS
ANALYZE_ARGUMENTS: dict[str, dict[str, Any]] = {
# Core options
"directory": {
@@ -169,8 +171,27 @@ ANALYZE_ARGUMENTS: dict[str, dict[str, Any]] = {
"help": "Enable verbose logging",
},
},
# Dry-run and API key (parity with scrape/github/pdf)
"dry_run": {
"flags": ("--dry-run",),
"kwargs": {
"action": "store_true",
"help": "Preview what will be analyzed without creating output",
},
},
"api_key": {
"flags": ("--api-key",),
"kwargs": {
"type": str,
"help": "Anthropic API key (or set ANTHROPIC_API_KEY env var)",
"metavar": "KEY",
},
},
}
# Add workflow arguments (enhance_workflow, enhance_stage, var, workflow_dry_run, workflow_history)
ANALYZE_ARGUMENTS.update(WORKFLOW_ARGUMENTS)
def add_analyze_arguments(parser: argparse.ArgumentParser) -> None:
"""Add all analyze command arguments to a parser."""

View File

@@ -81,6 +81,15 @@ PDF_ARGUMENTS: dict[str, dict[str, Any]] = {
"help": "Preview workflow without executing (requires --enhance-workflow)",
},
},
# API key (parity with scrape/github/analyze)
"api_key": {
"flags": ("--api-key",),
"kwargs": {
"type": str,
"help": "Anthropic API key (or set ANTHROPIC_API_KEY env var)",
"metavar": "KEY",
},
},
# Enhancement level
"enhance_level": {
"flags": ("--enhance-level",),

View File

@@ -72,6 +72,28 @@ UNIFIED_ARGUMENTS: dict[str, dict[str, Any]] = {
"help": "Preview workflow stages without executing (requires --enhance-workflow)",
},
},
# API key and enhance-level (parity with scrape/github/analyze/pdf)
"api_key": {
"flags": ("--api-key",),
"kwargs": {
"type": str,
"help": "Anthropic API key (or set ANTHROPIC_API_KEY env var)",
"metavar": "KEY",
},
},
"enhance_level": {
"flags": ("--enhance-level",),
"kwargs": {
"type": int,
"choices": [0, 1, 2, 3],
"default": None,
"help": (
"Global AI enhancement level override (0=off, 1=SKILL.md, "
"2=+arch/config, 3=full). Overrides per-source enhance_level in config."
),
"metavar": "LEVEL",
},
},
}

View File

@@ -2406,9 +2406,10 @@ Examples:
# Workflow enhancement arguments
parser.add_argument(
"--enhance-workflow",
type=str,
action="append",
help=(
"Enhancement workflow to use (name or path to YAML file). "
"Can be used multiple times to chain workflows. "
"Examples: 'security-focus', 'architecture-comprehensive', "
"'.skill-seekers/my-workflow.yaml'. "
"Overrides --enhance-level when provided."

View File

@@ -58,9 +58,9 @@ class ConfigManager:
def __init__(self):
"""Initialize configuration manager."""
self.config_dir = _get_config_dir()
self.config_file = self.config_dir / "config.json"
self.progress_dir = _get_progress_dir()
self.config_dir = self.CONFIG_DIR
self.config_file = self.CONFIG_FILE
self.progress_dir = self.PROGRESS_DIR
self._ensure_directories()
# Check if config file exists before loading

View File

@@ -633,16 +633,14 @@ class PDFToSkillConverter:
def main():
from .arguments.pdf import add_pdf_arguments
parser = argparse.ArgumentParser(
description="Convert PDF documentation to Claude skill",
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument("--config", help="PDF config JSON file")
parser.add_argument("--pdf", help="Direct PDF file path")
parser.add_argument("--name", help="Skill name (with --pdf)")
parser.add_argument("--from-json", help="Build skill from extracted JSON")
parser.add_argument("--description", help="Skill description")
add_pdf_arguments(parser)
args = parser.parse_args()

View File

@@ -561,8 +561,9 @@ class UnifiedScraper:
extract_docs = source.get("extract_docs", True)
# Note: Signal flow analysis is automatic for Godot projects (C3.10)
# AI enhancement settings
enhance_level = source.get("enhance_level", 0)
# AI enhancement settings (CLI --enhance-level overrides per-source config)
cli_enhance_level = getattr(args, "enhance_level", None) if args is not None else None
enhance_level = cli_enhance_level if cli_enhance_level is not None else source.get("enhance_level", 0)
# Run codebase analysis
logger.info(f" Analysis depth: {analysis_depth}")
@@ -972,14 +973,47 @@ class UnifiedScraper:
self.build_skill(merged_data)
# Phase 5: Enhancement Workflow Integration
if args is not None:
# Support workflow fields in JSON config as well as CLI args.
# JSON fields: "workflows" (list), "workflow_stages" (list), "workflow_vars" (dict)
# CLI args always take precedence; JSON fields are appended after.
json_workflows = self.config.get("workflows", [])
json_stages = self.config.get("workflow_stages", [])
json_vars = self.config.get("workflow_vars", {})
has_json_workflows = bool(json_workflows or json_stages or json_vars)
if args is not None or has_json_workflows:
import argparse
from skill_seekers.cli.workflow_runner import run_workflows
# Build effective args: use CLI args when provided, otherwise empty namespace
effective_args = args if args is not None else argparse.Namespace(
enhance_workflow=None,
enhance_stage=None,
var=None,
workflow_dry_run=False,
)
# Merge JSON workflow config into effective_args (JSON appended after CLI)
if json_workflows:
effective_args.enhance_workflow = (
list(effective_args.enhance_workflow or []) + json_workflows
)
if json_stages:
effective_args.enhance_stage = (
list(effective_args.enhance_stage or []) + json_stages
)
if json_vars:
effective_args.var = (
list(effective_args.var or [])
+ [f"{k}={v}" for k, v in json_vars.items()]
)
unified_context = {
"name": self.config.get("name", ""),
"description": self.config.get("description", ""),
}
run_workflows(args, context=unified_context)
run_workflows(effective_args, context=unified_context)
logger.info("\n" + "" * 20)
logger.info("Unified scraping complete!")
@@ -1067,6 +1101,24 @@ Examples:
dest="workflow_dry_run",
help="Preview workflow stages without executing (requires --enhance-workflow)",
)
parser.add_argument(
"--api-key",
type=str,
metavar="KEY",
help="Anthropic API key (or set ANTHROPIC_API_KEY env var)",
)
parser.add_argument(
"--enhance-level",
type=int,
choices=[0, 1, 2, 3],
default=None,
metavar="LEVEL",
help=(
"Global AI enhancement level override for all sources "
"(0=off, 1=SKILL.md, 2=+arch/config, 3=full). "
"Overrides per-source enhance_level in config."
),
)
args = parser.parse_args()

View File

@@ -181,5 +181,89 @@ class TestAnalyzePresetBehavior(unittest.TestCase):
self.assertFalse(args.comprehensive)
class TestAnalyzeWorkflowFlags(unittest.TestCase):
"""Test workflow and parity flags added to the analyze subcommand."""
def setUp(self):
"""Create parser for testing."""
self.parser = create_parser()
def test_enhance_workflow_accepted_as_list(self):
"""Test --enhance-workflow is accepted and stored as a list."""
args = self.parser.parse_args(
["analyze", "--directory", ".", "--enhance-workflow", "security-focus"]
)
self.assertEqual(args.enhance_workflow, ["security-focus"])
def test_enhance_workflow_chained_twice(self):
"""Test --enhance-workflow can be chained to produce a two-item list."""
args = self.parser.parse_args(
[
"analyze",
"--directory",
".",
"--enhance-workflow",
"security-focus",
"--enhance-workflow",
"minimal",
]
)
self.assertEqual(args.enhance_workflow, ["security-focus", "minimal"])
def test_enhance_stage_accepted_as_list(self):
"""Test --enhance-stage is accepted with action=append."""
args = self.parser.parse_args(
["analyze", "--directory", ".", "--enhance-stage", "sec:Analyze security"]
)
self.assertEqual(args.enhance_stage, ["sec:Analyze security"])
def test_var_accepted_as_list(self):
"""Test --var is accepted with action=append (dest is 'var')."""
args = self.parser.parse_args(
["analyze", "--directory", ".", "--var", "focus=performance"]
)
self.assertEqual(args.var, ["focus=performance"])
def test_workflow_dry_run_flag(self):
"""Test --workflow-dry-run sets the flag."""
args = self.parser.parse_args(
["analyze", "--directory", ".", "--workflow-dry-run"]
)
self.assertTrue(args.workflow_dry_run)
def test_api_key_stored_correctly(self):
"""Test --api-key is stored in args."""
args = self.parser.parse_args(
["analyze", "--directory", ".", "--api-key", "sk-ant-test"]
)
self.assertEqual(args.api_key, "sk-ant-test")
def test_dry_run_stored_correctly(self):
"""Test --dry-run is stored in args."""
args = self.parser.parse_args(["analyze", "--directory", ".", "--dry-run"])
self.assertTrue(args.dry_run)
def test_workflow_flags_combined(self):
"""Test workflow flags can be combined with other analyze flags."""
args = self.parser.parse_args(
[
"analyze",
"--directory",
".",
"--enhance-workflow",
"security-focus",
"--api-key",
"sk-ant-test",
"--dry-run",
"--enhance-level",
"1",
]
)
self.assertEqual(args.enhance_workflow, ["security-focus"])
self.assertEqual(args.api_key, "sk-ant-test")
self.assertTrue(args.dry_run)
self.assertEqual(args.enhance_level, 1)
if __name__ == "__main__":
unittest.main()

View File

@@ -519,5 +519,40 @@ class TestJSONWorkflow(unittest.TestCase):
self.assertEqual(converter.extracted_data["total_pages"], 1)
class TestPDFCLIArguments(unittest.TestCase):
"""Test PDF subcommand CLI argument parsing via the main CLI."""
def setUp(self):
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from skill_seekers.cli.main import create_parser
self.parser = create_parser()
def test_api_key_stored_correctly(self):
"""Test --api-key is accepted and stored correctly after switching to add_pdf_arguments."""
args = self.parser.parse_args(["pdf", "--pdf", "test.pdf", "--api-key", "sk-ant-test"])
self.assertEqual(args.api_key, "sk-ant-test")
def test_enhance_level_accepted(self):
"""Test --enhance-level is accepted for pdf subcommand."""
args = self.parser.parse_args(["pdf", "--pdf", "test.pdf", "--enhance-level", "1"])
self.assertEqual(args.enhance_level, 1)
def test_enhance_workflow_accepted(self):
"""Test --enhance-workflow is accepted and stores a list."""
args = self.parser.parse_args(
["pdf", "--pdf", "test.pdf", "--enhance-workflow", "minimal"]
)
self.assertEqual(args.enhance_workflow, ["minimal"])
def test_workflow_dry_run_accepted(self):
"""Test --workflow-dry-run is accepted."""
args = self.parser.parse_args(["pdf", "--pdf", "test.pdf", "--workflow-dry-run"])
self.assertTrue(args.workflow_dry_run)
if __name__ == "__main__":
unittest.main()

View File

@@ -574,6 +574,209 @@ def test_config_file_validation():
os.unlink(config_path)
# ===========================
# Unified CLI Argument Tests
# ===========================
class TestUnifiedCLIArguments:
"""Test that unified subcommand parser exposes the expected CLI flags."""
@pytest.fixture
def parser(self):
import sys
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
from skill_seekers.cli.main import create_parser
return create_parser()
def test_api_key_stored_correctly(self, parser):
"""Test --api-key KEY is stored in args."""
args = parser.parse_args(
["unified", "--config", "my.json", "--api-key", "sk-ant-test"]
)
assert args.api_key == "sk-ant-test"
def test_enhance_level_stored_correctly(self, parser):
"""Test --enhance-level 2 is stored in args."""
args = parser.parse_args(
["unified", "--config", "my.json", "--enhance-level", "2"]
)
assert args.enhance_level == 2
def test_enhance_level_default_is_none(self, parser):
"""Test --enhance-level defaults to None (per-source values apply)."""
args = parser.parse_args(["unified", "--config", "my.json"])
assert args.enhance_level is None
def test_enhance_level_all_choices(self, parser):
"""Test all valid --enhance-level choices are accepted."""
for level in [0, 1, 2, 3]:
args = parser.parse_args(
["unified", "--config", "my.json", "--enhance-level", str(level)]
)
assert args.enhance_level == level
def test_enhance_workflow_accepted(self, parser):
"""Test --enhance-workflow is accepted."""
args = parser.parse_args(
["unified", "--config", "my.json", "--enhance-workflow", "security-focus"]
)
assert args.enhance_workflow == ["security-focus"]
def test_api_key_and_enhance_level_combined(self, parser):
"""Test --api-key and --enhance-level can be combined."""
args = parser.parse_args(
["unified", "--config", "my.json", "--api-key", "sk-ant-test", "--enhance-level", "3"]
)
assert args.api_key == "sk-ant-test"
assert args.enhance_level == 3
# ===========================
# Workflow JSON Config Tests
# ===========================
class TestWorkflowJsonConfig:
"""Test that UnifiedScraper.run() merges JSON workflow fields into effective_args."""
def _make_scraper(self, tmp_path, extra_config=None):
"""Build a minimal UnifiedScraper backed by a temp config file."""
from skill_seekers.cli.unified_scraper import UnifiedScraper
config = {
"name": "test_workflow",
"description": "Test workflow config",
"sources": [],
**(extra_config or {}),
}
cfg_file = tmp_path / "config.json"
cfg_file.write_text(json.dumps(config))
scraper = UnifiedScraper.__new__(UnifiedScraper)
scraper.config = config
scraper.name = config["name"]
return scraper
def test_json_workflows_merged_when_args_none(self, tmp_path, monkeypatch):
"""JSON 'workflows' list is used even when args=None."""
captured = {}
def fake_run_workflows(args, context=None):
captured["enhance_workflow"] = getattr(args, "enhance_workflow", None)
monkeypatch.setattr(
"skill_seekers.cli.workflow_runner.run_workflows", fake_run_workflows, raising=False
)
import skill_seekers.cli.unified_scraper as us_module
monkeypatch.setattr(us_module, "run_workflows", fake_run_workflows, raising=False)
scraper = self._make_scraper(tmp_path, {"workflows": ["security-focus", "minimal"]})
# Patch _merge_workflow_config inline by directly testing the logic
import argparse
effective_args = argparse.Namespace(
enhance_workflow=None, enhance_stage=None, var=None, workflow_dry_run=False
)
json_workflows = scraper.config.get("workflows", [])
if json_workflows:
effective_args.enhance_workflow = (
list(effective_args.enhance_workflow or []) + json_workflows
)
assert effective_args.enhance_workflow == ["security-focus", "minimal"]
def test_json_workflows_appended_after_cli(self, tmp_path):
"""CLI --enhance-workflow values come first; JSON 'workflows' appended after."""
import argparse
config = {
"name": "test",
"description": "test",
"sources": [],
"workflows": ["json-wf"],
}
cfg_file = tmp_path / "config.json"
cfg_file.write_text(json.dumps(config))
cli_args = argparse.Namespace(
enhance_workflow=["cli-wf"],
enhance_stage=None,
var=None,
workflow_dry_run=False,
)
json_workflows = config.get("workflows", [])
effective = argparse.Namespace(
enhance_workflow=list(cli_args.enhance_workflow or []) + json_workflows,
enhance_stage=None,
var=None,
workflow_dry_run=False,
)
assert effective.enhance_workflow == ["cli-wf", "json-wf"]
def test_json_workflow_stages_merged(self, tmp_path):
"""JSON 'workflow_stages' are appended to enhance_stage."""
import argparse
config = {"workflow_stages": ["sec:Analyze security", "cleanup:Remove boilerplate"]}
effective_args = argparse.Namespace(
enhance_workflow=None, enhance_stage=None, var=None, workflow_dry_run=False
)
json_stages = config.get("workflow_stages", [])
if json_stages:
effective_args.enhance_stage = list(effective_args.enhance_stage or []) + json_stages
assert effective_args.enhance_stage == ["sec:Analyze security", "cleanup:Remove boilerplate"]
def test_json_workflow_vars_converted_to_kv_strings(self, tmp_path):
"""JSON 'workflow_vars' dict is converted to 'key=value' strings."""
import argparse
config = {"workflow_vars": {"focus_area": "performance", "detail_level": "basic"}}
effective_args = argparse.Namespace(
enhance_workflow=None, enhance_stage=None, var=None, workflow_dry_run=False
)
json_vars = config.get("workflow_vars", {})
if json_vars:
effective_args.var = list(effective_args.var or []) + [
f"{k}={v}" for k, v in json_vars.items()
]
assert "focus_area=performance" in effective_args.var
assert "detail_level=basic" in effective_args.var
def test_config_validator_accepts_workflow_fields(self, tmp_path):
"""ConfigValidator should not raise on workflow-related top-level fields."""
from skill_seekers.cli.config_validator import ConfigValidator
config = {
"name": "test",
"description": "Test with workflows",
"sources": [{"type": "documentation", "base_url": "https://example.com"}],
"workflows": ["security-focus"],
"workflow_stages": ["custom:Do something"],
"workflow_vars": {"key": "value"},
}
validator = ConfigValidator(config)
# Should not raise
assert validator.validate() is True
def test_empty_workflow_config_no_effect(self, tmp_path):
"""If no JSON workflow fields exist, effective_args remains unchanged."""
import argparse
config = {"name": "test", "description": "test", "sources": []}
effective_args = argparse.Namespace(
enhance_workflow=None, enhance_stage=None, var=None, workflow_dry_run=False
)
json_workflows = config.get("workflows", [])
json_stages = config.get("workflow_stages", [])
json_vars = config.get("workflow_vars", {})
has_json = bool(json_workflows or json_stages or json_vars)
assert not has_json
assert effective_args.enhance_workflow is None
assert effective_args.enhance_stage is None
assert effective_args.var is None
# Run tests
if __name__ == "__main__":
pytest.main([__file__, "-v"])