Add 16 structured data detectors for JSON, YAML, TOML, INI, Markdown, Proto, SQL, Batch

Generic detectors (6): json_structure, yaml_structure, toml_structure,
ini_structure, markdown_structure, proto_structure — extract CONFIG_KEY
and CONFIG_FILE nodes from any file of their format.

Specialized detectors (10): package_json, tsconfig_json, openapi,
docker_compose, github_actions, kubernetes, pyproject_toml,
sql_structure, batch_structure, properties_detector — triggered by
filename patterns, extract domain-specific nodes (endpoints, entities,
infra resources, dependencies).

Infrastructure: add TOML/INI/Markdown/Proto parsers to analyzer.py,
register all 16 detectors in registry.py, add .env/.csv/.dockerfile
to include_extensions in config.py.

Benchmark results across 189K files in 8 projects:
- +155K new nodes (+14.3%), +201K new edges (+10.6%)
- 100% deterministic across 2 runs on 7/8 projects
- <5% performance overhead

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: aksOps

src/code_intelligence/analyzer.py

@@ -71,6 +71,33 @@ def _parse_structured(language: str, content: bytes, file_path: str) -> Any:
     elif language == "sql":
         from code_intelligence.parsing.structured.sql_parser import SqlParser
         return SqlParser().parse(content, file_path)
+    elif language == "toml":
+        try:
+            import tomllib
+        except ModuleNotFoundError:
+            import tomli as tomllib  # type: ignore[no-redef]
+        try:
+            text = content.decode("utf-8", errors="replace")
+            data = tomllib.loads(text)
+        except Exception as exc:
+            return {"error": "invalid_toml", "file": file_path, "detail": str(exc)}
+        return {"type": "toml", "file": file_path, "data": data}
+    elif language == "ini":
+        import configparser
+        try:
+            text = content.decode("utf-8", errors="replace")
+            parser = configparser.ConfigParser()
+            parser.read_string(text)
+            data = {section: dict(parser[section]) for section in parser.sections()}
+        except Exception as exc:
+            return {"error": "invalid_ini", "file": file_path, "detail": str(exc)}
+        return {"type": "ini", "file": file_path, "data": data}
+    elif language == "markdown":
+        # Return raw text for regex-based detection
+        return {"type": "markdown", "file": file_path, "data": content.decode("utf-8", errors="replace")}
+    elif language == "proto":
+        # Return raw text for regex-based detection
+        return {"type": "proto", "file": file_path, "data": content.decode("utf-8", errors="replace")}
     return None
 
 

src/code_intelligence/config.py

@@ -24,6 +24,7 @@ class DiscoveryConfig(BaseModel):
         ".rb", ".rs", ".kt", ".kts", ".scala", ".swift",
         ".r", ".R", ".pl", ".pm", ".lua", ".dart",
         ".toml", ".ini", ".cfg", ".conf",
+        ".env", ".csv", ".dockerfile",
     ])
     exclude_patterns: list[str] = Field(default_factory=lambda: [
         "**/node_modules/**",

src/code_intelligence/detectors/config/__init__.py

@@ -0,0 +1,127 @@
+"""Batch script (.bat/.cmd) structure detector for labels, calls, and variables."""
+
+from __future__ import annotations
+
+import re
+
+from code_intelligence.detectors.base import DetectorContext, DetectorResult
+from code_intelligence.models.graph import (
+    EdgeKind,
+    GraphEdge,
+    GraphNode,
+    NodeKind,
+    SourceLocation,
+)
+
+_LABEL_RE = re.compile(r'^:(\w+)', re.MULTILINE)
+_CALL_RE = re.compile(r'CALL\s+:?(\S+)', re.IGNORECASE)
+_SET_RE = re.compile(r'SET\s+(\w+)=', re.IGNORECASE)
+
+
+class BatchStructureDetector:
+    """Detects Batch script structures: labels, CALL commands, and SET variables."""
+
+    name: str = "batch_structure"
+    supported_languages: tuple[str, ...] = ("batch",)
+
+    def detect(self, ctx: DetectorContext) -> DetectorResult:
+        result = DetectorResult()
+
+        try:
+            text = ctx.content.decode("utf-8", errors="replace")
+        except Exception:
+            return result
+
+        filepath = ctx.file_path
+        lines = text.split("\n")
+        module_id = f"bat:{filepath}"
+
+        # MODULE node for the script
+        result.nodes.append(GraphNode(
+            id=module_id,
+            kind=NodeKind.MODULE,
+            label=filepath,
+            fqn=filepath,
+            module=ctx.module_name,
+            location=SourceLocation(
+                file_path=filepath,
+                line_start=1,
+            ),
+        ))
+
+        for i, line in enumerate(lines):
+            line_num = i + 1
+            stripped = line.strip()
+
+            # Skip comments and echo off
+            if not stripped:
+                continue
+            upper = stripped.upper()
+            if upper.startswith("@ECHO OFF"):
+                continue
+            if upper.startswith("REM ") or upper == "REM":
+                continue
+            if stripped.startswith("::"):
+                continue
+
+            # Labels
+            m = _LABEL_RE.match(stripped)
+            if m:
+                label_name = m.group(1)
+                result.nodes.append(GraphNode(
+                    id=f"bat:{filepath}:label:{label_name}",
+                    kind=NodeKind.METHOD,
+                    label=f":{label_name}",
+                    fqn=f"{filepath}:{label_name}",
+                    module=ctx.module_name,
+                    location=SourceLocation(
+                        file_path=filepath,
+                        line_start=line_num,
+                    ),
+                ))
+                result.edges.append(GraphEdge(
+                    source=module_id,
+                    target=f"bat:{filepath}:label:{label_name}",
+                    kind=EdgeKind.CONTAINS,
+                    label=f"{filepath} contains :{label_name}",
+                ))
+                continue
+
+            # CALL commands
+            m = _CALL_RE.search(stripped)
+            if m:
+                call_target = m.group(1)
+                # Determine if calling an internal label or external script
+                if call_target.startswith(":"):
+                    target_id = f"bat:{filepath}:label:{call_target[1:]}"
+                elif "." in call_target:
+                    # External script call
+                    target_id = call_target
+                else:
+                    target_id = f"bat:{filepath}:label:{call_target}"
+
+                result.edges.append(GraphEdge(
+                    source=module_id,
+                    target=target_id,
+                    kind=EdgeKind.CALLS,
+                    label=f"{filepath} calls {call_target}",
+                ))
+
+            # SET variables
+            m = _SET_RE.search(stripped)
+            if m:
+                var_name = m.group(1)
+                result.nodes.append(GraphNode(
+                    id=f"bat:{filepath}:set:{var_name}",
+                    kind=NodeKind.CONFIG_DEFINITION,
+                    label=f"SET {var_name}",
+                    fqn=f"{filepath}:{var_name}",
+                    module=ctx.module_name,
+                    location=SourceLocation(
+                        file_path=filepath,
+                        line_start=line_num,
+                    ),
+                    properties={"variable": var_name},
+                ))
+
+        return result

src/code_intelligence/detectors/config/batch_structure.py

@@ -0,0 +1,179 @@
+"""Docker Compose detector for container orchestration definitions."""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Any
+
+from code_intelligence.detectors.base import DetectorContext, DetectorResult
+from code_intelligence.models.graph import (
+    EdgeKind,
+    GraphEdge,
+    GraphNode,
+    NodeKind,
+    SourceLocation,
+)
+
+_COMPOSE_FILENAME_RE = re.compile(
+    r"^(docker-compose|compose).*\.(yml|yaml)$", re.IGNORECASE
+)
+
+
+def _is_compose_file(ctx: DetectorContext) -> bool:
+    """Check whether the file is a Docker Compose file."""
+    basename = os.path.basename(ctx.file_path)
+    if _COMPOSE_FILENAME_RE.match(basename):
+        return True
+    # Fallback: check parsed data for compose-like structure
+    if ctx.parsed_data and ctx.parsed_data.get("type") == "yaml":
+        data = ctx.parsed_data.get("data")
+        if isinstance(data, dict) and "services" in data:
+            return True
+    return False
+
+
+class DockerComposeDetector:
+    """Detects services, ports, volumes, networks, and dependencies from Docker Compose files."""
+
+    name: str = "docker_compose"
+    supported_languages: tuple[str, ...] = ("yaml",)
+
+    def detect(self, ctx: DetectorContext) -> DetectorResult:
+        result = DetectorResult()
+
+        if not _is_compose_file(ctx):
+            return result
+
+        if not ctx.parsed_data:
+            return result
+
+        data = ctx.parsed_data.get("data")
+        if not isinstance(data, dict):
+            return result
+
+        services = data.get("services")
+        if not isinstance(services, dict):
+            return result
+
+        fp = ctx.file_path
+
+        # Build a set of known service IDs for edge resolution
+        service_ids: dict[str, str] = {}
+        for svc_name in services:
+            service_ids[svc_name] = f"compose:{fp}:service:{svc_name}"
+
+        for svc_name, svc_def in services.items():
+            if not isinstance(svc_def, dict):
+                continue
+
+            svc_id = service_ids[svc_name]
+
+            # Properties for the service node
+            props: dict[str, Any] = {}
+            if "image" in svc_def:
+                props["image"] = str(svc_def["image"])
+            build = svc_def.get("build")
+            if isinstance(build, str):
+                props["build_context"] = build
+            elif isinstance(build, dict) and "context" in build:
+                props["build_context"] = str(build["context"])
+
+            # INFRA_RESOURCE node for the service
+            result.nodes.append(GraphNode(
+                id=svc_id,
+                kind=NodeKind.INFRA_RESOURCE,
+                label=svc_name,
+                fqn=f"compose:{svc_name}",
+                module=ctx.module_name,
+                location=SourceLocation(file_path=fp),
+                properties=props,
+            ))
+
+            # Ports
+            ports = svc_def.get("ports")
+            if isinstance(ports, list):
+                for port_entry in ports:
+                    port_str = str(port_entry)
+                    result.nodes.append(GraphNode(
+                        id=f"compose:{fp}:service:{svc_name}:port:{port_str}",
+                        kind=NodeKind.CONFIG_KEY,
+                        label=f"{svc_name} port {port_str}",
+                        module=ctx.module_name,
+                        location=SourceLocation(file_path=fp),
+                        properties={"port": port_str},
+                    ))
+
+            # depends_on
+            depends_on = svc_def.get("depends_on")
+            if isinstance(depends_on, list):
+                for dep in depends_on:
+                    dep_str = str(dep)
+                    if dep_str in service_ids:
+                        result.edges.append(GraphEdge(
+                            source=svc_id,
+                            target=service_ids[dep_str],
+                            kind=EdgeKind.DEPENDS_ON,
+                            label=f"{svc_name} depends on {dep_str}",
+                        ))
+            elif isinstance(depends_on, dict):
+                for dep_str in depends_on:
+                    if dep_str in service_ids:
+                        result.edges.append(GraphEdge(
+                            source=svc_id,
+                            target=service_ids[dep_str],
+                            kind=EdgeKind.DEPENDS_ON,
+                            label=f"{svc_name} depends on {dep_str}",
+                        ))
+
+            # links
+            links = svc_def.get("links")
+            if isinstance(links, list):
+                for link in links:
+                    link_name = str(link).split(":")[0]
+                    if link_name in service_ids:
+                        result.edges.append(GraphEdge(
+                            source=svc_id,
+                            target=service_ids[link_name],
+                            kind=EdgeKind.CONNECTS_TO,
+                            label=f"{svc_name} links to {link_name}",
+                        ))
+
+            # Volumes
+            volumes = svc_def.get("volumes")
+            if isinstance(volumes, list):
+                for vol_entry in volumes:
+                    vol_str = str(vol_entry) if not isinstance(vol_entry, dict) else vol_entry.get("source", str(vol_entry))
+                    result.nodes.append(GraphNode(
+                        id=f"compose:{fp}:service:{svc_name}:volume:{vol_str}",
+                        kind=NodeKind.CONFIG_KEY,
+                        label=f"{svc_name} volume {vol_str}",
+                        module=ctx.module_name,
+                        location=SourceLocation(file_path=fp),
+                        properties={"volume": vol_str},
+                    ))
+
+            # Networks
+            networks = svc_def.get("networks")
+            if isinstance(networks, list):
+                for net in networks:
+                    result.nodes.append(GraphNode(
+                        id=f"compose:{fp}:service:{svc_name}:network:{net}",
+                        kind=NodeKind.CONFIG_KEY,
+                        label=f"{svc_name} network {net}",
+                        module=ctx.module_name,
+                        location=SourceLocation(file_path=fp),
+                        properties={"network": str(net)},
+                    ))
+            elif isinstance(networks, dict):
+                for net_name in networks:
+                    result.nodes.append(GraphNode(
+                        id=f"compose:{fp}:service:{svc_name}:network:{net_name}",
+                        kind=NodeKind.CONFIG_KEY,
+                        label=f"{svc_name} network {net_name}",
+                        module=ctx.module_name,
+                        location=SourceLocation(file_path=fp),
+                        properties={"network": str(net_name)},
+                    ))
+
+        return result