qt6-migration-pipeline/app/migration_pipeline/chunker.py

"""Semantic C++ source-file chunker.

Splits large C++ files at meaningful boundaries (class, struct, namespace,
free function definitions) so that each chunk can be migrated independently
by the LLM without losing context at arbitrary line positions.

Secondary fallback: nearest blank line below the 500-line mark.
"""

from __future__ import annotations

import re
from typing import List

from migration_pipeline.models import CodeChunk


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def split_semantic(source: str, max_lines: int = 500) -> List[CodeChunk]:
    """Split *source* into chunks of at most *max_lines* lines.

    Returns a list with a single CodeChunk (boundary_type="complete") when
    the file is small enough to fit in one shot.
    """
    lines = source.splitlines(keepends=True)
    n = len(lines)

    if n <= max_lines:
        return [CodeChunk(0, 0, n, source, "complete")]

    boundaries = _find_top_level_boundaries(lines)
    return _partition(lines, boundaries, max_lines)


# ---------------------------------------------------------------------------
# Boundary detection
# ---------------------------------------------------------------------------

# Matches the opening of a top-level class/struct/namespace/enum-class.
_COMPOSITE_RE = re.compile(
    r"""^\s*
    (?:template\s*<[^>]*>\s*)?          # optional template<...>
    (?:class|struct|namespace|enum\s+class)\s+\w
    """,
    re.VERBOSE,
)

# Matches a free-function or method *definition* opening (has `{` at end of
# logical line, not just a forward declaration ending with `;`).
_FUNC_DEF_RE = re.compile(
    r"""^                               # start of line
    [\w:<>\*&~\s]+                      # return type / qualifier
    \s+\w[\w:<>]*                       # function name
    \s*\(                               # opening paren
    """,
    re.VERBOSE,
)


def _find_top_level_boundaries(lines: List[str]) -> List[int]:
    """Return line indices (0-based) that start a new top-level declaration."""
    boundaries: List[int] = [0]
    brace_depth = 0
    in_block_comment = False
    i = 0

    while i < len(lines):
        raw = lines[i]
        line = raw.strip()

        # ---- Block-comment tracking ----------------------------------------
        if in_block_comment:
            if "*/" in raw:
                in_block_comment = False
            i += 1
            continue

        if "/*" in raw and "*/" not in raw:
            in_block_comment = True

        # ---- Skip single-line comments, empty lines, preprocessor ----------
        if not line or line.startswith("//") or line.startswith("#"):
            i += 1
            continue

        # ---- At top level: look for new declarations -----------------------
        if brace_depth == 0:
            if _COMPOSITE_RE.match(raw) or _looks_like_func_def(lines, i):
                if i > 0 and i not in boundaries:
                    boundaries.append(i)

        # ---- Track brace depth (naïve, ignores strings/chars) -------------
        # Remove string/char literals to avoid counting braces inside them.
        cleaned = _strip_literals(raw)
        brace_depth += cleaned.count("{") - cleaned.count("}")
        brace_depth = max(0, brace_depth)

        i += 1

    return sorted(boundaries)


def _looks_like_func_def(lines: List[str], start: int) -> bool:
    """Check whether lines[start:] begins a function *definition* (has `{`)."""
    # Collect up to 5 continuation lines for multi-line signatures.
    snippet = ""
    for j in range(start, min(start + 5, len(lines))):
        snippet += lines[j]
        if "{" in snippet:
            return bool(_FUNC_DEF_RE.match(snippet.lstrip()))
        if ";" in snippet:
            return False  # forward declaration
    return False


_STRING_RE = re.compile(r'"[^"\\]*(?:\\.[^"\\]*)*"')
_CHAR_RE = re.compile(r"'[^'\\]*(?:\\.[^'\\]*)*'")


def _strip_literals(s: str) -> str:
    s = _STRING_RE.sub('""', s)
    s = _CHAR_RE.sub("''", s)
    return s


# ---------------------------------------------------------------------------
# Partitioning
# ---------------------------------------------------------------------------

def _partition(
    lines: List[str],
    boundaries: List[int],
    max_lines: int,
) -> List[CodeChunk]:
    """Greedily group lines into chunks that respect *boundaries* and *max_lines*."""
    chunks: List[CodeChunk] = []
    chunk_idx = 0
    i = 0
    n = len(lines)
    boundary_set = set(boundaries)

    while i < n:
        target_end = min(i + max_lines, n)

        # Prefer ending at the latest semantic boundary before target_end.
        best_end = _best_boundary_before(boundaries, i + 1, target_end)

        if best_end is None or best_end <= i:
            # No semantic boundary: fall back to the nearest blank line.
            best_end = _nearest_blank_before(lines, target_end, i)

        content = "".join(lines[i:best_end])
        boundary_type = "semantic" if best_end in boundary_set else "fallback"
        if i == 0 and best_end == n:
            boundary_type = "complete"

        chunks.append(CodeChunk(chunk_idx, i, best_end, content, boundary_type))
        chunk_idx += 1
        i = best_end

    return chunks


def _best_boundary_before(
    boundaries: List[int], lo: int, hi: int
) -> int | None:
    """Return the largest boundary in (lo, hi], or None."""
    best = None
    for b in boundaries:
        if lo < b <= hi:
            best = b
    return best


def _nearest_blank_before(lines: List[str], target: int, floor: int) -> int:
    """Search backwards from *target* for a blank line; return its index+1."""
    for j in range(min(target, len(lines)) - 1, max(floor, target - 60), -1):
        if not lines[j].strip():
            return j + 1
    return target  # no blank found → hard cut