🧹 refactor: Extract section validation logic to shared utility

services/analysis-engine/src/bandscope_analysis/chords/analyzer.py

@@ -5,6 +5,7 @@
 import logging
 from typing import Any, Literal
 
+from ..sections.utils import validate_section
 from .model import ChordAnalysisResult, ChordLabel, SectionChordSummary
 
 logger = logging.getLogger(__name__)
@@ -48,15 +49,7 @@ def analyze(
         summaries: list[SectionChordSummary] = []
 
         for i, section in enumerate(sections):
-            if not isinstance(section, dict):
-                logger.warning(
-                    "Invalid section format at index %d; expected dict, got %s",
-                    i,
-                    type(section).__name__,
-                )
-                section_id = f"section-{i}"
-            else:
-                section_id = section.get("id", f"section-{i}")
+            section_id = validate_section(section, i, logger)
 
             chords: list[ChordLabel] = []
             key_center = _DEFAULT_KEY_CENTER

services/analysis-engine/src/bandscope_analysis/ranges/analyzer.py

@@ -5,6 +5,7 @@
 import logging
 from typing import Any, Literal
 
+from ..sections.utils import validate_section
 from .model import (
     RangeAnalysisResult,
     RangeInfo,
@@ -182,15 +183,7 @@ def analyze(
         summaries: list[SectionRangeSummary] = []
 
         for i, section in enumerate(sections):
-            if not isinstance(section, dict):
-                logger.warning(
-                    "Invalid section format at index %d; expected dict, got %s",
-                    i,
-                    type(section).__name__,
-                )
-                section_id = f"section-{i}"
-            else:
-                section_id = section.get("id", f"section-{i}")
+            section_id = validate_section(section, i, logger)
 
             section_roles = (roles_by_section or {}).get(section_id, [])
             ranges: list[RangeInfo] = []

services/analysis-engine/src/bandscope_analysis/roles/extractor.py

@@ -5,6 +5,7 @@
 import logging
 from typing import Any
 
+from ..sections.utils import validate_section
 from .model import (
     CueAnchorKind,
     PartGraphNode,
@@ -53,15 +54,7 @@ def extract(
 
         # Simple mock implementation for testing/demonstration purposes
         for i, section in enumerate(sections):
-            if not isinstance(section, dict):
-                logger.warning(
-                    "Invalid section format at index %d; expected dict, got %s",
-                    i,
-                    type(section).__name__,
-                )
-                section_id = f"section-{i}"
-            else:
-                section_id = section.get("id", f"section-{i}")
+            section_id = validate_section(section, i, logger)
 
             topology = self._build_topology(section_id, i == 0, roles)
             topologies.append(topology)

services/analysis-engine/src/bandscope_analysis/sections/__init__.py

@@ -14,6 +14,7 @@
     SectionExtractionResult,
     SectionLabel,
 )
+from .utils import validate_section
 
 __all__ = [
     "CueAnchor",
@@ -25,4 +26,5 @@
     "count_based_anchor",
     "lyric_phrase_anchor",
     "extract_sections",
+    "validate_section",
 ]

services/analysis-engine/src/bandscope_analysis/sections/utils.py

@@ -0,0 +1,18 @@
+"""Utility functions for section processing."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+
+def validate_section(section: Any, index: int, logger: logging.Logger) -> str:
+    """Return a stable section id, warning when section data is malformed."""
+    if not isinstance(section, dict):
+        logger.warning(
+            "Invalid section format at index %d; expected dict, got %s",
+            index,
+            type(section).__name__,
+        )
+        return f"section-{index}"
+    return str(section.get("id", f"section-{index}"))