Analyzers¶

@abstractmethod
def detect(self, context: AnalysisContext, config: ConfigT) -> list[Violation]:
    """Run this detector's algorithm and return any violations found.

    Implementations should read only the fields they need from
    *context* (e.g. ``cyclomatic_summary`` for a complexity check) and
    compare against thresholds stored in *config*. Use
    [`build_violation`][mcp_zen_of_languages.analyzers.base.ViolationDetector.build_violation]
    to construct ``Violation`` instances with correct principle IDs and
    severity.

    Args:
        context (AnalysisContext): Pipeline state carrying parsed AST, metrics, and
            source text populated by earlier analysis stages.
        config (AnalyzerConfig): Typed detector configuration holding thresholds,
            severity, and violation-message templates for this
            detector.

    Returns:
        list[Violation]: Zero or more violations discovered by this strategy. An empty
        list signals clean code for this detector's concern.
    """

def build_violation(  # noqa: PLR0913
    self,
    config: ConfigT,
    *,
    rule_id: str | None = None,
    message: str | None = None,
    contains: str | None = None,
    index: int = 0,
    severity: int | None = None,
    location: Location | None = None,
    suggestion: str | None = None,
    files: list[str] | None = None,
) -> Violation:
    """Construct a ``Violation`` from detector config with optional overrides.

    This convenience factory saves every detector from repeating the
    same principle-ID look-up, severity resolution, and message
    selection logic. The algorithm is:

    1. Resolve ``principle`` from ``config.principle``,
       ``config.principle_id``, or ``config.type`` (first non-``None``
       wins).
    2. If *message* is not given explicitly, delegate to
       ``config.select_violation_message()`` using *contains* or
       *index* to pick the right template.
    3. If *severity* is not given, fall back to ``config.severity``
       or a default of ``5``.

    Args:
        config (DetectorConfig): Detector configuration carrying principle metadata,
            severity, and violation-message templates.
        rule_id (str | None, optional): Specific rule identifier for
            composite detectors. When provided, principle text, severity,
            and default message selection are resolved from that rule's
            preserved context. Default to None.
        message (str | None, optional): Explicit violation message. When ``None``, the
            message is auto-selected from the config's template list. Default to None.
        contains (str | None, optional): Substring filter passed to
            ``select_violation_message`` to pick a matching template. Default to None.
        index (int | None, optional): Zero-based position selecting one template from the
            config's ``violation_messages`` list (default ``0``). Default to 0.
        severity (int | None, optional): Override severity score (1-10). Falls back to the
            config-level severity when omitted. Default to None.
        location (Location | None, optional): Source location to attach to the violation, typically
            produced by
            [`LocationHelperMixin`][mcp_zen_of_languages.analyzers.base.LocationHelperMixin]. Default to None.
        suggestion (str | None, optional): Remediation hint shown alongside the violation in
            reports and IDE integrations. Default to None.
        files (list[str] | None, optional): Related file paths included for cross-file violations
            such as duplicate-code detection. Default to None.

    Returns:
        Violation: Fully populated ``Violation`` ready for collection by the
        [`DetectionPipeline`][mcp_zen_of_languages.analyzers.base.DetectionPipeline].
    """
    principle_resolver = getattr(config, "principle_for_rule", None)
    if callable(principle_resolver):
        principle = principle_resolver(rule_id)
    else:
        principle = (
            getattr(config, "principle", None)
            or getattr(config, "principle_id", None)
            or getattr(config, "type", "violation")
        )
    if message is None:
        selector = getattr(config, "select_violation_message", None)
        if callable(selector):
            message = selector(contains=contains, index=index, rule_id=rule_id)
        else:
            message = principle
    resolved_severity = severity
    if resolved_severity is None:
        severity_resolver = getattr(config, "severity_for_rule", None)
        if callable(severity_resolver):
            resolved_severity = severity_resolver(rule_id, 5)
        else:
            resolved_severity = getattr(config, "severity", None) or 5
    return Violation(
        principle=principle,
        severity=resolved_severity,
        message=message,
        location=location,
        suggestion=suggestion,
        files=files,
        rule_id=rule_id or getattr(config, "principle_id", None),
        detector_id=getattr(config, "type", None),
    )

def run(
    self,
    context: AnalysisContext,
    config: AnalyzerConfig | DetectorConfig,
) -> list[Violation]:
    """Execute every detector against the shared context and merge results.

    For each detector the pipeline resolves which configuration to
    use: if the detector carries its own ``config`` (injected during
    pipeline construction), that takes precedence; otherwise the
    analyzer-level *config* is used as a fallback.

    If a detector raises an exception, the error is printed and the
    pipeline continues with the next detector — no violations from
    healthy detectors are lost.

    Args:
        context (AnalysisContext): Shared analysis state populated by
            [`BaseAnalyzer.analyze`][mcp_zen_of_languages.analyzers.base.BaseAnalyzer.analyze]
            before the pipeline starts.
        config (AnalyzerConfig): Fallback configuration used when a detector does not
            carry its own per-detector config.

    Returns:
        list[Violation]: Flat list of violations aggregated from all detectors that
        executed successfully.
    """
    all_violations: list[Violation] = []

    for detector in self.detectors:
        try:
            detector_config = detector.config or config
            detector_name = self._detector_name(detector)
            with analysis_span(
                "detector.run",
                {"language": context.language, "detector": detector_name},
            ):
                violations = detector.detect(context, detector_config)
            all_violations.extend(
                [
                    self._enrich_violation(
                        context,
                        detector,
                        detector_config,
                        detector_name,
                        violation,
                    )
                    for violation in violations
                ],
            )
        except Exception:
            # Log error but continue with other detectors
            detector_name = self._detector_name(detector)
            logger.exception("Error in detector %s", detector_name)

    return all_violations

@abstractmethod
def default_config(self) -> AnalyzerConfig:
    """Provide the default configuration for this language.

    Called by ``__init__`` when the caller does not pass an explicit
    config. Language subclasses return their own typed config (e.g.
    [`PythonAnalyzerConfig`][mcp_zen_of_languages.analyzers.base.PythonAnalyzerConfig])
    pre-populated with sensible defaults.

    Returns:
        AnalyzerConfig: Language-appropriate configuration with default thresholds.
    """

@abstractmethod
def language(self) -> str:
    """Return the language identifier this analyzer handles.

    The string must match the keys used in ``zen-config.yaml`` and
    the language registry (e.g. ``"python"``, ``"typescript"``,
    ``"rust"``).

    Returns:
        str: Lowercase language name used for rule lookup and result
        tagging.
    """

@abstractmethod
def parse_code(self, code: str) -> ParserResult | None:
    """Parse raw source text into a language-specific syntax tree.

    This is the first Template Method hook called by ``analyze()``.
    Python subclasses typically delegate to the ``ast`` module; other
    languages may use tree-sitter or custom parsers.

    Args:
        code (str): Complete source text of the file being analyzed.

    Returns:
        ParserResult | None: Wrapped parse result, or ``None`` when the source cannot be
        parsed (e.g. syntax errors). A ``None`` return does not abort
        analysis — metric computation and detectors will proceed
        with whatever data is available.
    """

@abstractmethod
def compute_metrics(
    self,
    code: str,
    ast_tree: ParserResult | None,
) -> tuple[CyclomaticSummary | None, float | None, int]:
    """Compute quantitative code-quality metrics for the given source.

    This is the second Template Method hook. Implementations should
    calculate at least cyclomatic complexity, a maintainability index,
    and a physical line count. The returned tuple is unpacked by
    ``analyze()`` and stored on the
    [`AnalysisContext`][mcp_zen_of_languages.analyzers.base.AnalysisContext]
    for downstream detectors.

    Args:
        code (str): Source text to measure.
        ast_tree (ParserResult | None): Previously parsed syntax tree (may be ``None`` if
            parsing failed), useful for AST-driven metric tools.

    Returns:
        tuple[CyclomaticSummary | None, float | None, int]: Three-element tuple of ``(cyclomatic_summary,
        maintainability_index, lines_of_code)``. Any element may be
        ``None`` when the corresponding metric is unavailable.
    """

def capabilities(self) -> AnalyzerCapabilities:
    """Declare language features supported by this analyzer implementation.

    The default implementation marks all advanced capabilities as
    unsupported. Language analyzers with concrete parser or dependency
    support override this method to advertise availability.

    Returns:
        AnalyzerCapabilities: Capability flags consumed by
            ``AnalysisContext`` and downstream detectors.
    """
    return AnalyzerCapabilities()

def analyze(  # noqa: PLR0913
    self,
    code: str,
    path: str | None = None,
    other_files: dict[str, str] | None = None,
    repository_imports: dict[str, list[str]] | None = None,
    *,
    enable_external_tools: bool = False,
    allow_temporary_tools: bool = False,
) -> AnalysisResult:
    """Run the full analysis workflow against a single source file.

    This is the **Template Method**: it defines the invariant
    seven-step algorithm that every language analyzer follows, calling
    abstract hooks (``parse_code``, ``compute_metrics``) and the
    pre-built [`DetectionPipeline`][mcp_zen_of_languages.analyzers.base.DetectionPipeline]
    at the appropriate moments.

    **Workflow steps:**

    1. Create an [`AnalysisContext`][mcp_zen_of_languages.analyzers.base.AnalysisContext]
       from the inputs.
    2. Parse source via ``parse_code()`` → ``context.ast_tree``.
    3. Compute metrics via ``compute_metrics()`` → cyclomatic,
       maintainability, LOC.
    4. (Optional) Build dependency graph →
       ``context.dependency_analysis``.
    5. Run the detector pipeline → initial violation list.
    6. Merge with ``RulesAdapter`` violations and attach
       ``rules_summary`` (gracefully skipped if the adapter is
       unavailable).
    7. Assemble and return the final ``AnalysisResult``.

    Args:
        code (str): Complete source text to analyze.
        path (str | None, optional): Filesystem path of the source file, used for
            cross-file detectors and result metadata. Default to None.
        other_files (dict[str, str] | None, optional): Map of sibling file paths to
            their contents, enabling detectors like duplicate-code that
            compare across files. Default to None.
        repository_imports (dict[str, list[str]] | None, optional): Per-file import
            lists from the wider repository, enabling coupling and
            dependency-fan detectors. Default to None.
        enable_external_tools (bool, optional): Run allow-listed external
            linters/tools in best-effort mode for additional diagnostics. Default to False.
        allow_temporary_tools (bool, optional): Allow temporary-runner strategies
            (for example ``npx``/``uvx``) when direct/no-install Default to False.
            resolution is unavailable.

    Returns:
        AnalysisResult: Fully populated analysis result containing metrics,
        violations, an overall quality score, and (when available)
        a ``rules_summary``.
    """
    with analysis_span(
        "analyzer.analyze",
        {"language": self.language(), "path": path or "<snippet>"},
    ):
        context = self._create_context(
            code=code,
            path=path,
            other_files=other_files,
            repository_imports=repository_imports,
        )

        with analysis_span("analyzer.parse", {"language": self.language()}):
            context.ast_tree = self.parse_code(code)
            if context.capabilities.supports_ast:
                context.ast_status = (
                    AstStatus.parsed
                    if context.ast_tree is not None
                    else AstStatus.parse_failed
                )

        with analysis_span("analyzer.metrics", {"language": self.language()}):
            cc, mi, loc = self.compute_metrics(code, context.ast_tree)
            context.cyclomatic_summary = cc
            context.maintainability_index = mi
            context.lines_of_code = loc

        with analysis_span("analyzer.dependencies", {"language": self.language()}):
            if self.config.enable_dependency_analysis:
                context.dependency_analysis = self._build_dependency_analysis(
                    context
                )
            context.external_analysis = self._build_external_analysis(
                context,
                enable_external_tools=enable_external_tools,
                allow_temporary_tools=allow_temporary_tools,
            )

        with analysis_span(
            "analyzer.pipeline",
            {
                "language": self.language(),
                "detector_count": len(self.pipeline.detectors),
            },
        ):
            violations = self.pipeline.run(context, self.config)

        result = self._build_result(context, violations)

        try:
            from mcp_zen_of_languages.adapters.rules_adapter import RulesAdapter
            from mcp_zen_of_languages.adapters.rules_adapter import (
                RulesAdapterConfig,
            )
            from mcp_zen_of_languages.models import DependencyAnalysis

            adapter_config = RulesAdapterConfig(
                max_nesting_depth=self.config.max_nesting_depth,
                max_cyclomatic_complexity=self.config.max_cyclomatic_complexity,
                min_maintainability_index=None,
            )

            adapter = RulesAdapter(language=self.language(), config=adapter_config)

            dep_analysis: DependencyAnalysis | None = None
            raw_dep = context.dependency_analysis
            if isinstance(raw_dep, DependencyAnalysis):
                dep_analysis = raw_dep
            elif isinstance(raw_dep, dict):
                try:
                    dep_analysis = DependencyAnalysis.model_validate(raw_dep)
                except (ValueError, TypeError):
                    dep_analysis = None

            with analysis_span(
                "analyzer.rules_adapter",
                {"language": self.language()},
            ):
                rules_violations = (
                    adapter.find_violations(
                        code=code,
                        cyclomatic_summary=context.cyclomatic_summary,
                        maintainability_index=context.maintainability_index,
                        dependency_analysis=dep_analysis,
                    )
                    if self.config.enable_pattern_detection
                    else []
                )
            all_violations = violations + rules_violations
            result.rules_summary = RulesSummary(
                **adapter.summarize_violations(all_violations),
            )
            result.violations = all_violations
            from mcp_zen_of_languages.dogmas.interface import attach_dogma_analysis

            result = attach_dogma_analysis(
                result.model_copy(update={"dogma_analysis": None})
            )
        except Exception as exc:
            logger.debug(
                "RulesAdapter integration failed; continuing", exc_info=exc
            )

        return result

def build_pipeline(self) -> DetectionPipeline:
    """Assemble the detector pipeline from zen rules and config overrides.

    The default implementation follows a four-stage process:

    1. Load canonical zen rules for ``self.language()`` via
       ``get_language_zen()``.
    2. Project those rules into
       [`DetectorConfig`][mcp_zen_of_languages.languages.configs.DetectorConfig]
       instances using the detector registry.
    3. Merge any overrides from ``zen-config.yaml`` (matched by
       detector ``type``).
    4. Instantiate each registered detector, inject its config and
       rule IDs, and wrap them in a
       [`DetectionPipeline`][mcp_zen_of_languages.analyzers.base.DetectionPipeline].

    Language subclasses may override this method entirely to build a
    hand-crafted pipeline, but the rule-driven default covers most
    use-cases.

    Returns:
        DetectionPipeline: Ready-to-run pipeline containing all detectors registered
        for this language.

    Raises:
        ValueError: If no zen rules are found for the language.
    """
    from mcp_zen_of_languages.analyzers import registry_bootstrap  # noqa: F401
    from mcp_zen_of_languages.analyzers.pipeline import PipelineConfig
    from mcp_zen_of_languages.analyzers.registry import REGISTRY
    from mcp_zen_of_languages.rules import get_language_zen

    lang_zen = get_language_zen(self.language())
    if lang_zen is None:
        msg = f"No zen rules for language: {self.language()}"
        raise ValueError(msg)
    if self._pipeline_config is None:
        self._pipeline_config = PipelineConfig(
            language=self.language(),
            detectors=[],
        )

    base_config = PipelineConfig(
        language=self.language(),
        detectors=REGISTRY.configs_from_rules(lang_zen),
    )
    if self._pipeline_config:
        merged = REGISTRY.merge_configs(
            base_config.detectors,
            self._pipeline_config.detectors,
        )
        pipeline_config = PipelineConfig(
            language=base_config.language,
            detectors=merged,
        )
    else:
        pipeline_config = base_config

    analyzer_defaults = AnalyzerConfig()
    detectors: list[ViolationDetector] = []
    for detector_config in pipeline_config.detectors:
        if detector_config.type == "analyzer_defaults":
            if isinstance(detector_config, AnalyzerConfig):
                analyzer_defaults = detector_config
            continue
        meta = REGISTRY.get(detector_config.type)
        detector = meta.detector_class()
        detector.config = detector_config
        detector.rule_ids = list(meta.rule_ids)
        detectors.append(detector)

    self.config = analyzer_defaults
    return DetectionPipeline(detectors)

def find_location_by_substring(self, code: str, substring: str) -> Location:
    """Locate the first occurrence of *substring* in the source text.

    Scans *code* line-by-line and returns a one-based ``Location``
    pointing to the first character of the match. When the substring
    is not found the method returns ``Location(line=1, column=1)``
    as a safe fallback rather than raising.

    Args:
        code (str): Full source text to search, potentially multi-line.
        substring (str): Token or identifier to locate (exact,
            case-sensitive match).

    Returns:
        Location: One-based ``Location`` of the first match, or ``(1, 1)``
        when the substring does not appear in *code*.
    """
    lines = code.splitlines()
    for i, line in enumerate(lines, start=1):
        col = line.find(substring)
        if col >= 0:
            return Location(line=i, column=col + 1)
    return Location(line=1, column=1)

def ast_node_to_location(
    self,
    _ast_tree: ParserResult | None,
    node: object | None,
) -> Location | None:
    """Extract a ``Location`` from a Python-style AST node.

    Reads ``lineno`` and ``col_offset`` attributes via ``getattr``
    so this helper works with any AST node that exposes those fields
    (stdlib ``ast``, tree-sitter adapters, etc.). Column offsets are
    converted from zero-based to one-based to match the ``Location``
    convention.

    Args:
        _ast_tree (ParserResult | None): Parsed tree wrapper (currently unused but reserved
            for future tree-sitter adapters that need the root).
        node (object): AST node expected to carry ``lineno`` (int) and
            ``col_offset`` (int) attributes.

    Returns:
        Location: One-based ``Location`` when both attributes are present,
        otherwise ``None``.
    """
    if node is None:
        return None

    try:
        # Extract tree if ParserResult wrapper
        # Try to get line/column from node
        lineno = getattr(node, "lineno", None)
        col_offset = getattr(node, "col_offset", None)

        if lineno is not None and col_offset is not None:
            return Location(line=int(lineno), column=int(col_offset) + 1)
    except (TypeError, ValueError):
        pass

    return None

@classmethod
def from_rules(cls, language: str) -> PipelineConfig:
    """Build a complete pipeline by projecting a language's zen principles.

    Loads the [`LanguageZenPrinciples`][mcp_zen_of_languages.rules.base_models.LanguageZenPrinciples]
    for *language*, then delegates to
    ``configs_from_rules``
    to project each principle's metrics onto the matching detector configs.

    Args:
        language (str): Language key recognised by [`get_language_zen`][mcp_zen_of_languages.rules.get_language_zen]
            (e.g. ``"python"``).

    Returns:
        PipelineConfig: A fully populated ``PipelineConfig`` whose detector list reflects
        all zen principles defined for the language.

    Raises:
        ValueError: If no zen rules exist for *language*.

    Examples:
        >>> cfg = PipelineConfig.from_rules("python")
        >>> cfg.language
        'python'
    """
    lang_zen = get_language_zen(language)
    if not lang_zen:
        msg = f"No zen rules for language: {language}"
        raise ValueError(msg)
    from mcp_zen_of_languages.analyzers.registry import REGISTRY

    detectors = REGISTRY.configs_from_rules(lang_zen)
    return cls(language=language, detectors=detectors)