"""Section and lyric-variant parsing for Carnatic compositions. Ports deterministic header/regex heuristics from the Kotlin KrithiStructureParser into the Python extraction service. Produces a frozen parse contract with: - canonical sections - lyric variants (language/script scoped) - metadata boundaries (meaning/notes/etc.) """ from __future__ import annotations import logging import re from dataclasses import dataclass, field from .schema import ( CanonicalLyricSection, CanonicalLyricVariant, CanonicalMetadataBoundary, CanonicalSection, SectionType, ) logger = logging.getLogger(__name__) @dataclass class DetectedSection: """A lyric section detected in extracted text.""" section_type: SectionType order: int label: str text: str start_pos: int end_pos: int @dataclass class DetectedLyricVariant: """A full lyric variant in one language/script.""" language: str script: str sections: list[DetectedSection] = field(default_factory=list) @dataclass class MetadataBoundary: """Start/end offsets for non-lyric metadata blocks in source text.""" label: str start_pos: int end_pos: int @dataclass class RagamalikaSubsection: """A detected raga subsection within a ragamalika composition.""" raga_name: str parent_section_type: SectionType is_viloma: bool order: int @dataclass class StructureParseResult: """Frozen parser contract emitted by the extraction layer.""" sections: list[DetectedSection] = field(default_factory=list) lyric_variants: list[DetectedLyricVariant] = field(default_factory=list) metadata_boundaries: list[MetadataBoundary] = field(default_factory=list) ragamalika_subsections: list[RagamalikaSubsection] = field(default_factory=list) @dataclass class _LineToken: text: str start_pos: int end_pos: int preceded_by_blank: bool = False @dataclass class _TextBlock: label: str lines: list[_LineToken] start_pos: int end_pos: int raga_name: str | None = None is_viloma: bool = False @dataclass class _HeaderMatch: label: str remainder: str raga_name: str | None = None is_viloma: bool = False LANGUAGE_LABELS = { "DEVANAGARI", "TAMIL", "TELUGU", "KANNADA", "MALAYALAM", "ENGLISH", "LATIN", "SANSKRIT", "HINDI", "WORD_DIVISION", "MEANING", "GIST", "NOTES", "VARIATIONS", } METADATA_LABELS = {"WORD_DIVISION", "MEANING", "GIST", "NOTES", "VARIATIONS"} LANGUAGE_HEADER_CANDIDATES: list[tuple[str, str]] = [ # TRACK-102: Compound "Script - Word Division" headers must appear before simple script entries ("english - word division", "WORD_DIVISION"), ("devanagari - word division", "WORD_DIVISION"), ("tamil - word division", "WORD_DIVISION"), ("telugu - word division", "WORD_DIVISION"), ("kannada - word division", "WORD_DIVISION"), ("malayalam - word division", "WORD_DIVISION"), ("sanskrit - word division", "WORD_DIVISION"), # Simple script entries ("devanagari", "DEVANAGARI"), ("tamil", "TAMIL"), ("telugu", "TELUGU"), ("kannada", "KANNADA"), ("malayalam", "MALAYALAM"), ("english", "ENGLISH"), ("roman", "LATIN"), ("latin", "LATIN"), ("sanskrit", "SANSKRIT"), ("hindi", "HINDI"), ("word division", "WORD_DIVISION"), ("meaning", "MEANING"), ("gist", "GIST"), ("notes", "NOTES"), ("variations", "VARIATIONS"), ] SECTION_LABEL_TO_TYPE: dict[str, SectionType] = { "PALLAVI": SectionType.PALLAVI, "ANUPALLAVI": SectionType.ANUPALLAVI, "CHARANAM": SectionType.CHARANAM, "SAMASHTI_CHARANAM": SectionType.SAMASHTI_CHARANAM, "CHITTASWARAM": SectionType.CHITTASWARAM, "SWARA_SAHITYA": SectionType.SWARA_SAHITYA, "MADHYAMAKALA": SectionType.MADHYAMA_KALA, "MADHYAMA_KALA": SectionType.MADHYAMA_KALA, # TRACK-133 (WORK ITEM 2): a per-raga stanza boundary in a "pure" ragamalika # (each avatara stanza headed only by " rAgaM", no P/A/C). Section-type # assignment per stanza is a lakshana call owned by the curator/musicologist, # so the splitter emits OTHER and leaves the typing decision downstream. "RAGA_SEGMENT": SectionType.OTHER, } # Kotlin parity: matches "1. SrI rAgaM" or "Arabhi rAgaM" RAGA_SUBSECTION_PATTERN = re.compile(r"^(?:\d+\.\s*)?(.+?)\s+rAgaM\s*$", re.IGNORECASE) # Kotlin parity: matches "vilOma - mOhana rAgaM" VILOMA_SUBSECTION_PATTERN = re.compile(r"^vilOma\s*-\s*(.+?)\s+rAgaM\s*$", re.IGNORECASE) # TRACK-133 follow-up: the Latin patterns above only match the transliterated # " rAgaM" header. In multi-script sources (e.g. the Dashavatara ragamalika # mAdhavO mAM pAtu) each script variant carries the SAME raga header in its own # script — Devanagari "नाट रागं", Tamil "நாட ராகம்", Telugu "నాట రాగం", # Kannada "ನಾಟ ರಾಗಂ", Malayalam "നാട രാഗം". Without an Indic-script form the # raga boundary is invisible in the transliterated variants, so the whole stanza # sequence collapses into one section on the variant side while the (Latin-built) # canonical structure segments correctly. This mirrors RAGA_SUBSECTION_PATTERN for # the five Indic scripts; the raga word is "rāga" + an anusvāra / virāma-m ending. _INDIC_RAGA_WORD = r"(?:राग[ंम]्?|ராக[ம]்?|రాగ[ంమ]్?|ರಾಗ[ಂಮ]್?|രാഗ[ംമ]്?)" RAGA_SUBSECTION_INDIC_PATTERN = re.compile(rf"^(?:\d+[.।]?\s*)?(.+?)\s+{_INDIC_RAGA_WORD}\s*$") # Honorific "SrI " before a raga name (e.g. "SrI gauLa rAgaM") is a blog artifact, # not the raga Sri. Bare "SrI rAgaM" stays "SrI". _SRI_HONORIFIC_PREFIX = re.compile(r"^SrI\s+", re.IGNORECASE) # Blog line-wrap of a hyphenated word ("kani(y)-" / "A rAN-muni") must not be read # as an inline P/A/C header. TRACK-139 / Alakalallalaadaga two-line pallavi. _INLINE_PAC_LABELS = frozenset({"PALLAVI", "ANUPALLAVI", "CHARANAM"}) # Only the Dashavatara blog spelling 'SrI gauLa' is an honorific+raga, not Sri ranjani. _SRI_HONORIFIC_REMAINDERS = frozenset({"gaula"}) def _strip_sri_honorific(name: str) -> str: """Drop a leading honorific 'SrI ' only for the adjudicated blog spelling. 'SrI gauLa' → 'gauLa'. Bare 'SrI' (the raga) and 'SrI ranjani' stay intact (Śrīranjani ≠ Ranjani; TRACK-136 / normalize_for_matching). """ match = _SRI_HONORIFIC_PREFIX.match(name) if match is None: return name remainder = name[match.end() :].strip() if remainder.casefold() in _SRI_HONORIFIC_REMAINDERS: return remainder return name def _raga_name_from_segment_header(line: str) -> str | None: match = RAGA_SUBSECTION_PATTERN.search(line) or RAGA_SUBSECTION_INDIC_PATTERN.search(line) if match is None: return None return _strip_sri_honorific(match.group(1).strip()) SECTION_HEADER_PATTERNS: list[tuple[re.Pattern[str], str]] = [ (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*pallavi(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "PALLAVI"), (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*anupallavi(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "ANUPALLAVI"), ( re.compile( r"^\s*[\-–—•*()=\[\]]*\s*(?:(?:ch|c)ara?n(?:\.\s*am|am)|caraṇam)(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE, ), "CHARANAM", ), ( re.compile(r"^\s*[\-–—•*()=\[\]]*\s*samashti\s+(?:ch|c)ara?nam(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "SAMASHTI_CHARANAM", ), ( re.compile(r"^\s*[\-–—•*()=\[\]]*\s*samash?ti\s+(?:ch|c)ara?nam(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "SAMASHTI_CHARANAM", ), (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*chittaswaram(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "CHITTASWARAM"), ( re.compile(r"^\s*[\-–—•*()=\[\]]*\s*madhyama\s+kAla(?:\s+sAhityam)?(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "MADHYAMAKALA", ), ( re.compile(r"^\s*[\-–—•*()=\[\]]*\s*madhyama\s+kala(?:\s+sahityam)?(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "MADHYAMAKALA", ), ( re.compile(r"^\s*[\-–—•*()=\[\]]*\s*madhyamakala(?:\s+sahityam)?(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "MADHYAMAKALA", ), ( re.compile(r"^\s*[\-–—•*()=\[\]]*\s*m\.\s*k(?:\s+sahityam)?(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "MADHYAMAKALA", ), (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*svara\s+sahitya(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "SWARA_SAHITYA"), (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*swarasahitya(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "SWARA_SAHITYA"), # Indic-script "svara sAhitya N" headers (CAT-B: Syama Sastri kritis whose # swara-sahitya sections are labelled inline in each variant's own script). # Without these, the whole Indic variant collapses into one PALLAVI blob and # never matches the multi-section template. (re.compile(r"^\s*स्वर\s+साहित्य(?:\s|:|\-|\.|\)|]|\d|$)"), "SWARA_SAHITYA"), # Devanagari (re.compile(r"^\s*ஸ்வர\s+ஸாஹித்ய(?:\s|:|\-|\.|\)|]|\d|$)"), "SWARA_SAHITYA"), # Tamil (re.compile(r"^\s*స్వర\s+సాహిత్య(?:\s|:|\-|\.|\)|]|\d|$)"), "SWARA_SAHITYA"), # Telugu (re.compile(r"^\s*ಸ್ವರ\s+ಸಾಹಿತ್ಯ(?:\s|:|\-|\.|\)|]|\d|$)"), "SWARA_SAHITYA"), # Kannada (re.compile(r"^\s*സ്വര\s+സാഹിത്യ(?:\s|:|\-|\.|\)|]|\d|$)"), "SWARA_SAHITYA"), # Malayalam # Inline C + digit(s) (thyagaraja-vaibhavam blog format): "C1 venuka tIka", # "C12 rAjillu SrI tyAgarAja". Uppercase only, unambiguous — no lyric line # starts with "C" + digit + space. (re.compile(r"^\s*C\s?\d{1,2}\s+(?=\S)"), "CHARANAM"), # Latin single-letter abbreviations. # Must be the ONLY content on the line to avoid false positives on lyric text. (re.compile(r"^\s*P[\.:\-\s]*$", re.IGNORECASE), "PALLAVI"), (re.compile(r"^\s*A[\.:\-\s]*$", re.IGNORECASE), "ANUPALLAVI"), (re.compile(r"^\s*C[\.:\-\s]*$", re.IGNORECASE), "CHARANAM"), (re.compile(r"^\s*Ch[\.:\-\s]*$", re.IGNORECASE), "CHARANAM"), # Indic single-letter abbreviations. # Must be the ONLY content on the line (letter + optional punctuation/whitespace) # to avoid false positives on lyric lines like "ப 4 ரதாக்..." being detected as headers. (re.compile(r"^\s*प[\.:\-\s]*$"), "PALLAVI"), (re.compile(r"^\s*अ[\.:\-\s]*$"), "ANUPALLAVI"), (re.compile(r"^\s*च[\.:\-\s]*$"), "CHARANAM"), (re.compile(r"^\s*ப[\.:\-\s]*$"), "PALLAVI"), (re.compile(r"^\s*அ[\.:\-\s]*$"), "ANUPALLAVI"), (re.compile(r"^\s*ச[\.:\-\s]*$"), "CHARANAM"), (re.compile(r"^\s*ప[\.:\-\s]*$"), "PALLAVI"), (re.compile(r"^\s*అ[\.:\-\s]*$"), "ANUPALLAVI"), (re.compile(r"^\s*చ[\.:\-\s]*$"), "CHARANAM"), (re.compile(r"^\s*ಪ[\.:\-\s]*$"), "PALLAVI"), (re.compile(r"^\s*ಅ[\.:\-\s]*$"), "ANUPALLAVI"), (re.compile(r"^\s*ಚ[\.:\-\s]*$"), "CHARANAM"), (re.compile(r"^\s*പ[\.:\-\s]*$"), "PALLAVI"), (re.compile(r"^\s*അ[\.:\-\s]*$"), "ANUPALLAVI"), (re.compile(r"^\s*ച[\.:\-\s]*$"), "CHARANAM"), # Devanagari full headers. (re.compile(r"^\s*पल्लवि(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "PALLAVI"), (re.compile(r"^\s*अनुपल्लवि(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "ANUPALLAVI"), # TRACK-133 (WORK ITEM 1, form 2): accept the dental-nasal spelling variant # चरन/చరన/சரந/ಚರನ/ചരന ("caraNam" written with dental na) alongside the # retroflex चरण form. Some Indic variants (e.g. ennEramum un pAda) label the # inline charanam with the dental spelling, which the retroflex-only patterns # missed, so every charanam collapsed into the preceding anupallavi block. (re.compile(r"^\s*चर[णन][म्ंम]+(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "CHARANAM"), (re.compile(r"^\s*समष्टि\s+चरण[म्ंम]+(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "SAMASHTI_CHARANAM"), (re.compile(r"^\s*[(]?मध्यम\s+काल\s+साहित्य[म्ंम]*[)]?(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "MADHYAMAKALA"), # Tamil full headers. (re.compile(r"^\s*பல்லவி(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "PALLAVI"), (re.compile(r"^\s*அனுபல்லவி(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "ANUPALLAVI"), (re.compile(r"^\s*சர[ணநன]ம்(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "CHARANAM"), (re.compile(r"^\s*ஸமஷ்டி\s+சரணம்(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "SAMASHTI_CHARANAM"), # Telugu full headers (both anusvara చరణం and explicit చరణమ్ forms). (re.compile(r"^\s*పల్లవి(?:\s|:|\-|\)|]|$)", re.IGNORECASE), "PALLAVI"), (re.compile(r"^\s*అనుపల్లవి(?:\s|:|\-|\)|]|$)", re.IGNORECASE), "ANUPALLAVI"), (re.compile(r"^\s*చర[ణన]ం(?:\s|:|\-|\)|]|$)", re.IGNORECASE), "CHARANAM"), (re.compile(r"^\s*చర[ణన]మ్(?:\s|:|\-|\)|]|$)", re.IGNORECASE), "CHARANAM"), (re.compile(r"^\s*సమష్టి\s+చరణ[ంమ్]+(?:\s|:|\-|\)|]|$)", re.IGNORECASE), "SAMASHTI_CHARANAM"), # Kannada full headers (both short ಚರಣ and full ಚರಣಮ್/ಚರಣಂ forms). (re.compile(r"^\s*ಪಲ್ಲವಿ(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "PALLAVI"), (re.compile(r"^\s*ಅನುಪಲ್ಲವಿ(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "ANUPALLAVI"), (re.compile(r"^\s*ಚರ[ಣನ][ಮ್ಂ]*(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "CHARANAM"), (re.compile(r"^\s*ಸಮಷ್ಟಿ\s+ಚರಣ[ಮ್ಂ]*(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "SAMASHTI_CHARANAM"), # Malayalam full headers (both ചരണം and ചരണമ് forms). (re.compile(r"^\s*പല്ലവി(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "PALLAVI"), (re.compile(r"^\s*അനുപല്ലവി(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "ANUPALLAVI"), (re.compile(r"^\s*ചര[ണന][ംമ്]+(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "CHARANAM"), (re.compile(r"^\s*സമഷ്ടി\s+ചരണ[ംമ്]+(?:\s|:|\-|\.|\)|]|$)", re.IGNORECASE), "SAMASHTI_CHARANAM"), # Telugu parenthesized MKS: (మధ్యమ కాల సాహిత్యం) (re.compile(r"^\s*[(]మధ్యమ\s+కాల\s+సాహిత్య[ంమ్]+[)]\s*$", re.IGNORECASE), "MADHYAMAKALA"), # Kannada parenthesized MKS: (ಮಧ್ಯಮ ಕಾಲ ಸಾಹಿತ್ಯಂ) (re.compile(r"^\s*[(]ಮಧ್ಯಮ\s+ಕಾಲ\s+ಸಾಹಿತ್ಯ[ಂಮ್]*[)]\s*$", re.IGNORECASE), "MADHYAMAKALA"), # Malayalam parenthesized MKS: (മധ്യമ കാല സാഹിത്യം) (re.compile(r"^\s*[(]മധ്യമ\s+കാല\s+സാഹിത്യ[ംമ്]+[)]\s*$", re.IGNORECASE), "MADHYAMAKALA"), # Tamil parenthesized MKS: (மத்யம கால ஸாஹித்யம்) and the grantha-numeral form # (மத் 4 யம கால ஸாஹித்யம்), where the aspirated dha is written "த் 4" with the # superscript number spaced out (as emitted by the guruguha Tamil transliteration). (re.compile(r"^\s*[(]மத்\s*[0-9]*\s*யம\s+கால\s+ஸாஹித்ய[ம்]+[)]\s*$", re.IGNORECASE), "MADHYAMAKALA"), # Bracket-format headers with underscores (from stored lyrics). (re.compile(r"^\s*\[PALLAVI\]\s*$", re.IGNORECASE), "PALLAVI"), (re.compile(r"^\s*\[ANUPALLAVI\]\s*$", re.IGNORECASE), "ANUPALLAVI"), (re.compile(r"^\s*\[CHARANAM\]\s*$", re.IGNORECASE), "CHARANAM"), (re.compile(r"^\s*\[SAMASHTI_CHARANAM\]\s*$", re.IGNORECASE), "SAMASHTI_CHARANAM"), (re.compile(r"^\s*\[CHITTASWARAM\]\s*$", re.IGNORECASE), "CHITTASWARAM"), (re.compile(r"^\s*\[MADHYAMA_KALA\]\s*$", re.IGNORECASE), "MADHYAMAKALA"), (re.compile(r"^\s*\[SWARA_SAHITYA\]\s*$", re.IGNORECASE), "SWARA_SAHITYA"), (re.compile(r"^\s*\[SOLKATTU_SWARA\]\s*$", re.IGNORECASE), "SOLKATTU_SWARA"), # Non-lyric metadata headers. (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*meaning(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "MEANING"), (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*notes?(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "NOTES"), (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*gist(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "GIST"), (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*word\s+division(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "WORD_DIVISION"), (re.compile(r"^\s*[\-–—•*()=\[\]]*\s*variations?(?:\b|:|\.|\-|\)|]|=|$)", re.IGNORECASE), "VARIATIONS"), ] # TRACK-097: Metadata boundary patterns use negative lookaheads to avoid false # positives from navigation links like "Meaning of Kriti-1" or "Notes on the composition". # A genuine metadata header is a standalone label, optionally followed by `:`, `-`, or newline. METADATA_BOUNDARY_PATTERNS: list[tuple[str, re.Pattern[str]]] = [ ( "MEANING", re.compile( r"^\s*(?:meaning(?!\s+of\b)|artha|artham|भावार्थ)\b", re.IGNORECASE | re.MULTILINE, ), ), ("GIST", re.compile(r"^\s*(?:gist|summary)\b", re.IGNORECASE | re.MULTILINE)), ( "NOTES", re.compile( r"^\s*(?:notes?(?!\s+(?:on|about|from|by|for)\b)|tippani)\b", re.IGNORECASE | re.MULTILINE, ), ), ("WORD_DIVISION", re.compile(r"^\s*(?:word\s*division|pada\s*ccheda)\b", re.IGNORECASE | re.MULTILINE)), ("VARIATIONS", re.compile(r"^\s*(?:variations?|alternate\s*reading)\b", re.IGNORECASE | re.MULTILINE)), ] # Inline P/A patterns are context-dependent: only activated when the document # also contains inline C/C\d+ labels (thyagaraja-vaibhavam blog format). # Without this guard, "A jagadamba" in a document using full-word headers # would be falsely split into a new ANUPALLAVI section. # The lookahead allows a digit so a leading footnote reference is tolerated: # "P 1giripai" (rendered from P1giripai) must still be # detected as the pallavi marker. The footnote digit is then stripped from the # remainder. Without this, the Latin pallavi is head-captured and the whole # template under-counts (see giripai nelakonna). INLINE_PAC_PATTERNS: list[tuple[re.Pattern[str], str]] = [ (re.compile(r"^\s*P (?=[a-zA-Z\d])"), "PALLAVI"), (re.compile(r"^\s*A (?=[a-zA-Z\d])"), "ANUPALLAVI"), (re.compile(r"^\s*C (?=[a-zA-Z\d])"), "CHARANAM"), ] _INLINE_CHARANAM_PROBE = re.compile(r"(?m)^\s*C(?:\d{1,2})? (?=[a-zA-Z\d])") # Indic-script inline abbreviations "प." / "अ." / "च1." (thyagaraja-vaibhavam blog # format, Indic variants). Like INLINE_PAC_PATTERNS these are context-dependent: # only activated when the document actually contains such abbreviation markers # (a line starting with a P/A/C letter + optional digit + period), so documents # that use full-word Indic headers are unaffected. The trailing period is the # disambiguator — natural lyric lines do not begin ".". INLINE_INDIC_PAC_PATTERNS: list[tuple[re.Pattern[str], str]] = [ (re.compile(r"^\s*प\d*\s*\.\s*(?=\S)"), "PALLAVI"), # Devanagari (re.compile(r"^\s*अ\d*\s*\.\s*(?=\S)"), "ANUPALLAVI"), (re.compile(r"^\s*च\d*\s*\.\s*(?=\S)"), "CHARANAM"), (re.compile(r"^\s*ప\d*\s*\.\s*(?=\S)"), "PALLAVI"), # Telugu (re.compile(r"^\s*అ\d*\s*\.\s*(?=\S)"), "ANUPALLAVI"), (re.compile(r"^\s*చ\d*\s*\.\s*(?=\S)"), "CHARANAM"), (re.compile(r"^\s*ಪ\d*\s*\.\s*(?=\S)"), "PALLAVI"), # Kannada (re.compile(r"^\s*ಅ\d*\s*\.\s*(?=\S)"), "ANUPALLAVI"), (re.compile(r"^\s*ಚ\d*\s*\.\s*(?=\S)"), "CHARANAM"), (re.compile(r"^\s*പ\d*\s*\.\s*(?=\S)"), "PALLAVI"), # Malayalam (re.compile(r"^\s*അ\d*\s*\.\s*(?=\S)"), "ANUPALLAVI"), (re.compile(r"^\s*ച\d*\s*\.\s*(?=\S)"), "CHARANAM"), (re.compile(r"^\s*ப\d*\s*\.\s*(?=\S)"), "PALLAVI"), # Tamil (re.compile(r"^\s*அ\d*\s*\.\s*(?=\S)"), "ANUPALLAVI"), (re.compile(r"^\s*ச\d*\s*\.\s*(?=\S)"), "CHARANAM"), ] _INLINE_INDIC_PAC_PROBE = re.compile(r"(?m)^\s*(?:प|अ|च|ప|అ|చ|ಪ|ಅ|ಚ|പ|അ|ച|ப|அ|ச)\d*\s*\.\s*(?=\S)") # Indic-script inline swara-sahitya ordinal markers "स्व1." / "ஸ்வ4(A)." (TRACK-133, # thyagaraja-vaibhavam blog format, Indic variants). The English/IAST variant labels # each swara-sahitya sub-block with the full words "svara sAhitya N"; the Indic variants # use the abbreviated cluster "sva" + ordinal + period instead ("स्व1.", "ஸ்வ4(A)."), # which the full-word CAT-B header patterns above do NOT match. Without these, only the # lone bare "स्वर साहित्य" group title is detected and every "sva N" sub-block collapses # into that single SWARA_SAHITYA section, under-segmenting the Indic variant relative to # the English canonical skeleton (e.g. sAdhincenE: 11 canon vs 5 Indic). # # Like INLINE_INDIC_PAC_PATTERNS these are context-gated (only activated when the block # actually contains such markers, via _INLINE_INDIC_SWARA_PROBE), so blocks that use only # the full-word header are unaffected. The bare "स्वर साहित्य" group-title line that # precedes "sva 1." then becomes a header with no body and is dropped by the empty-block # guard in _extract_sections, so the count matches the English side (N sub-blocks, not N+1). # The trailing period is the disambiguator — natural lyric lines do not begin "स्व.". # The optional "(A)" suffix mirrors the English "svara sAhitya 4A" continuation marker. _SVA_CLUSTERS = "स्व|ஸ்வ|స్వ|ಸ್ವ|സ്വ" # Devanagari, Tamil, Telugu, Kannada, Malayalam INLINE_INDIC_SWARA_PATTERNS: list[tuple[re.Pattern[str], str]] = [ (re.compile(rf"^\s*(?:{_SVA_CLUSTERS})\d+(?:\([A-Za-z]\))?\s*\.\s*(?=\S)"), "SWARA_SAHITYA"), ] _INLINE_INDIC_SWARA_PROBE = re.compile(rf"(?m)^\s*(?:{_SVA_CLUSTERS})\d+(?:\([A-Za-z]\))?\s*\.\s*(?=\S)") # --- TRACK-133 WORK ITEM 1: second inline-Indic charanam-marker gap ------------ # # A distinct class of undetected inline Indic P/A/C markers, the analogue of the # INLINE_INDIC_PAC (period) and INLINE_INDIC_SWARA (sva-ordinal) seams. Three forms, # each context-gated so full-word-header documents stay unaffected: # # form 3 — DIGIT WITHOUT TRAILING PERIOD: "च4 सुर तारक …" (vs the "च4." form that # INLINE_INDIC_PAC already handles). Seen in rAma Eva daivataM (sa), where # the no-period charanam 4 marker leaks and charanam 4 merges into 3. # form 1 — BARE AKSHARA, no digit and no period: "च अल नाडु सौमित्रि". Seen in # kaNTa jUDumi / enta bhAgyamu, where the charanam collapses into the # preceding block. Real source has UNMARKED pallavi/anupallavi, so form 1 # is self-gated on the bare-charanam-token line itself (see below), not on # a bare P/A twin. The digit (form 3) and the bare token followed by # whitespace (form 1) are the disambiguators against ordinary sahitya. _INDIC_PALLAVI_AKSHARAS = "प|ప|ಪ|പ|ப" _INDIC_ANUPALLAVI_AKSHARAS = "अ|అ|ಅ|അ|அ" _INDIC_CHARANAM_AKSHARAS = "च|చ|ಚ|ച|ச" INLINE_INDIC_DIGIT_PATTERNS: list[tuple[re.Pattern[str], str]] = [ (re.compile(rf"^\s*(?:{_INDIC_PALLAVI_AKSHARAS})\d+\s+(?=\S)"), "PALLAVI"), (re.compile(rf"^\s*(?:{_INDIC_ANUPALLAVI_AKSHARAS})\d+\s+(?=\S)"), "ANUPALLAVI"), (re.compile(rf"^\s*(?:{_INDIC_CHARANAM_AKSHARAS})\d+\s+(?=\S)"), "CHARANAM"), ] _INLINE_INDIC_DIGIT_PROBE = re.compile( rf"(?m)^\s*(?:{_INDIC_PALLAVI_AKSHARAS}|{_INDIC_ANUPALLAVI_AKSHARAS}|{_INDIC_CHARANAM_AKSHARAS})\d+\s+(?=\S)" ) # Form 1 - bare charanam akshara as its own token. Charanam-only: the real source # marks only the charanam this way (pallavi/anupallavi are unmarked), so this is NOT # gated on a bare P/A twin. The trailing "\s+" is the discriminator - a bare akshara # followed by whitespace ("च ") is a marker; an akshara bound to a vowel matra with no # space ("चॆन्त"/"चिन्त"/"செந்த"/"சிந்த") is the first consonant of a lyric word. INLINE_INDIC_BARE_PATTERNS: list[tuple[re.Pattern[str], str]] = [ (re.compile(rf"^\s*(?:{_INDIC_CHARANAM_AKSHARAS})\s+(?=\S)"), "CHARANAM"), ] # Self-gate: enable only when a bare charanam-akshara-as-token line actually exists. _INLINE_INDIC_BARE_CA_PROBE = re.compile(rf"(?m)^\s*(?:{_INDIC_CHARANAM_AKSHARAS})\s+(?=\S)") # --- TRACK-133 WORK ITEM 1: "caraNam"-in-sahitya false-positive guard ---------- # # A real spelled charanam header sits alone on its line (optionally with an ordinal # like "caraNam 2" / "caraNam 4A"). When the word "caraNam" instead heads a line but # is FOLLOWED by more sahitya on the same line, it is lyric — a hyphenation # continuation such as "…SaraNAgata tvac-\ncaraNam bhava tAraNambu cEsunu" # (ramA ramaNa rArA, over-split in en/ta). This guard keeps the trailing-lyric case # from faking a charanam header, so re-extract cannot reintroduce the spurious split. _CARANAM_WORD_AT_START = re.compile( r"^\s*[\-–—•*()=\[\]]*\s*" r"(?:(?:ch|c)ara?nam|caraṇam|चर[णन][म्ंम]+|சர[ணநன]ம்|చర[ணన][ంమ్]+|ಚರ[ಣನ][ಮ್ಂ]*|ചര[ണന][ംമ്]+)", re.IGNORECASE, ) # Allowed tail after the header word: ordinal digits, a "(A)" or bare "A" # continuation, or punctuation. Anything with letters remaining is lyric. _CARANAM_ORDINAL_TAIL = re.compile(r"^[\s:.\-)\]]*\d*\s*(?:\([A-Za-z]\)|[A-Za-z])?\s*") def _is_caranam_lyric_line(line: str) -> bool: """True when a line opens with the word ``caraNam`` but continues with lyric. Distinguishes a genuine standalone charanam header (``caraNam`` / ``caraNam 4A``) from a sahitya line that merely begins with the word after a hyphenation break (``tvac-caraNam bhava tAraNambu``). Only the latter must be suppressed. """ match = _CARANAM_WORD_AT_START.match(line) if match is None: return False tail = line[match.end() :] tail = _CARANAM_ORDINAL_TAIL.sub("", tail, count=1) return bool(re.search(r"[^\W\d_]", tail)) # --- TRACK-133 WORK ITEM 2: pure-ragamalika raga-header segmentation ----------- # # A "pure" ragamalika (e.g. mAdhavO mAM pAtu, the Dashavatara Ragamalika) has no # pallavi/anupallavi/charanam headers at all — each avatara stanza is headed only by # " rAgaM" and carries a Madhyamakala Sahitya. Without treating the raga header # as a section boundary the whole lyric collapses into one unsegmented block. This is # gated so it fires ONLY when the document has multiple raga headers AND no ordinary # P/A/C structure — an ordinary ragamalika (raga markers nested inside P/A/C, e.g. # SrI viSva nAthaM) keeps its P/A/C segmentation with ragas as metadata subsections. _RAGA_SEGMENT_HEADER_PROBE = re.compile(rf"(?mi)^\s*(?:\d+[.।]?\s*)?\S[^\n]*?\s+(?:rAgaM|{_INDIC_RAGA_WORD})\s*$") _STANDARD_SECTION_PROBE = re.compile( r"(?mi)^\s*[\-–—•*()=\[\]]*\s*(?:" r"pallavi|anupallavi|(?:ch|c)ara?n(?:\.\s*am|am)|caraṇam|samash?ti|chittaswaram|" r"पल्लवि|अनुपल्लवि|चर[णन]|समष्टि|" r"பல்லவி|அனுபல்லவி|சர[ணநன]|" r"పల్లవి|అనుపల్లవి|చర[ణన]|" r"ಪಲ್ಲವಿ|ಅನುಪಲ್ಲವಿ|ಚರ[ಣನ]|" r"പല്ലവി|അനുപല്ലവി|ചര[ണന]" r")" ) METADATA_KEYWORDS = ( "title", "raga", "ragam", "raagam", "tala", "talam", "taala", "composer", "composed by", "deity", "temple", "kshetra", "kshetram", "kshethra", "kriti", "krithi", "language", ) # Inline pronunciation-disambiguation digit (Modified HK): a digit glued to an Indic # consonant, e.g. "க3" (ga), "பா4" (bha), "விது4லு". Transliterated Carnatic lyric is # dense with these; a natural-language (Tamil) translation trailer has none. That # contrast is what locates the lyric→trailer boundary. _PRONUNCIATION_DIGIT = re.compile(r"[ऀ-෿]\d") _MIN_TRAILER_CHARS = 40 _MIN_TRAILER_LINES = 2 def strip_refrain_trailer(sections: list[DetectedSection]) -> list[DetectedSection]: """Trim an inline translation trailer from a variant's last section. Govindan-blog Tamil variants append a word-by-word Tamil *meaning* after the lyric; marker splitting leaves it glued to the final charanam, blowing that section up to many times its siblings. The lyric is transliterated (every line carries HK pronunciation digits); the appended meaning is natural Tamil prose with none. We cut everything after the **last line that still carries a pronunciation digit**, provided what follows is a multi-line prose block. Conservative by construction — the strip fires only when: * the last section has ≥1 transliterated (digit-bearing) line, AND * ≥2 trailing lines with no pronunciation digit follow it, totalling ≥40 chars. Otherwise the section is returned unchanged (deferred, never guessed). This never touches Latin variants (no Indic digits) and leaves clean lyric — whose lines all carry digits — untouched. """ if len(sections) < 2: return sections last = sections[-1] lines = last.text.split("\n") last_lyric = -1 for i, line in enumerate(lines): if _PRONUNCIATION_DIGIT.search(line): last_lyric = i if last_lyric == -1: return sections # no transliteration markers — cannot locate a boundary trailer_lines = lines[last_lyric + 1 :] if len(trailer_lines) < _MIN_TRAILER_LINES: return sections # not a multi-line prose block — leave as lyric trailer = "\n".join(trailer_lines).strip() if len(trailer) < _MIN_TRAILER_CHARS or _PRONUNCIATION_DIGIT.search(trailer): return sections trimmed = "\n".join(lines[: last_lyric + 1]).rstrip() if not trimmed: return sections return sections[:-1] + [ DetectedSection( section_type=last.section_type, order=last.order, label=last.label, text=trimmed, start_pos=last.start_pos, end_pos=last.end_pos, ) ] class StructureParser: """Deterministic structure parser with Kotlin-parity heuristics.""" # When False, the translation-trailer strip is skipped. The repair tooling # parses twice (with and without) to isolate exactly what the strip removed and # confirm it is prose, never lyric. Production parsing leaves this True. _trailer_strip_enabled = True def parse(self, text: str) -> StructureParseResult: if not text.strip(): return StructureParseResult() source_text = text.replace("\\n", "\n") metadata_boundaries = self._find_metadata_boundaries(source_text) # TRACK-097: Guard against false metadata boundaries that truncate too early. # If the first boundary appears in the first 200 chars of a long document, # it's likely a navigation link ("Meaning of Kriti") not a real section header. # Skip it and use the next boundary, or fall back to full text. _MIN_LYRIC_WINDOW = 200 effective_boundaries = metadata_boundaries if ( effective_boundaries and effective_boundaries[0].start_pos < _MIN_LYRIC_WINDOW and len(source_text) > _MIN_LYRIC_WINDOW * 3 ): logger.info( "Skipping early metadata boundary '%s' at offset %d (likely navigation link)", effective_boundaries[0].label, effective_boundaries[0].start_pos, ) effective_boundaries = effective_boundaries[1:] lyric_window_end = effective_boundaries[0].start_pos if effective_boundaries else len(source_text) lyric_text = source_text[:lyric_window_end] blocks = self._build_blocks(lyric_text) sections, ragamalika_subsections = self._extract_sections(blocks, lyric_text) lyric_variants = self._extract_lyric_variants(blocks, lyric_text, sections) # TRACK-100: Extract Indic-script variants from post-boundary region. # On many blogs, Indic scripts appear AFTER the first metadata boundary. if effective_boundaries and sections: existing_scripts = {v.script for v in lyric_variants} post_variants = self._extract_post_boundary_variants( source_text, lyric_window_end, sections, existing_scripts ) lyric_variants.extend(post_variants) return StructureParseResult( sections=sections, lyric_variants=lyric_variants, metadata_boundaries=metadata_boundaries, ragamalika_subsections=ragamalika_subsections, ) def parse_sections(self, text: str) -> list[DetectedSection]: return self.parse(text).sections def parse_ragamalika_subsections(self, text: str) -> list[RagamalikaSubsection]: return self.parse(text).ragamalika_subsections def _build_blocks(self, raw_text: str) -> list[_TextBlock]: if not raw_text.strip(): return [] self._inline_pa_enabled = bool(_INLINE_CHARANAM_PROBE.search(raw_text)) self._inline_indic_pac_enabled = bool(_INLINE_INDIC_PAC_PROBE.search(raw_text)) self._inline_indic_swara_enabled = bool(_INLINE_INDIC_SWARA_PROBE.search(raw_text)) self._inline_indic_digit_enabled = bool(_INLINE_INDIC_DIGIT_PROBE.search(raw_text)) self._inline_indic_bare_enabled = bool(_INLINE_INDIC_BARE_CA_PROBE.search(raw_text)) # Pure-ragamalika segmentation: multiple " rAgaM" headers and no # ordinary P/A/C structure anywhere in the block. self._raga_segment_enabled = ( len(_RAGA_SEGMENT_HEADER_PROBE.findall(raw_text)) >= 2 and _STANDARD_SECTION_PROBE.search(raw_text) is None ) tokens: list[_LineToken] = [] offset = 0 preceded_by_blank = False for raw_line in raw_text.splitlines(keepends=True): line_without_newline = raw_line.rstrip("\n") stripped = line_without_newline.strip() if not stripped: preceded_by_blank = True offset += len(raw_line) continue left_padding = len(line_without_newline) - len(line_without_newline.lstrip()) start_pos = offset + left_padding end_pos = offset + len(line_without_newline) offset += len(raw_line) normalized = self._normalize_line(stripped) if not normalized or self._is_boilerplate(normalized): continue tokens.append( _LineToken( text=normalized, start_pos=start_pos, end_pos=end_pos, preceded_by_blank=preceded_by_blank, ) ) preceded_by_blank = False if not tokens: return [] blocks: list[_TextBlock] = [] current_label = "UNLABELED" current_start = tokens[0].start_pos current_lines: list[_LineToken] = [] current_raga_name: str | None = None current_is_viloma = False def flush() -> None: nonlocal current_lines, current_raga_name, current_is_viloma if current_lines or current_label in LANGUAGE_LABELS: block_start = current_start if not current_lines else current_lines[0].start_pos block_end = current_lines[-1].end_pos if current_lines else current_start blocks.append( _TextBlock( label=current_label, lines=current_lines, start_pos=block_start, end_pos=block_end, raga_name=current_raga_name, is_viloma=current_is_viloma, ) ) current_lines = [] current_raga_name = None current_is_viloma = False prev_hyphen = False for token in tokens: header = self._detect_header(token.text) if ( header is not None and prev_hyphen and not token.preceded_by_blank and header.label in _INLINE_PAC_LABELS and any(pattern.search(token.text) for pattern, _ in INLINE_PAC_PATTERNS) ): # Hyphenation wrap: previous *adjacent* line ended "-" and this # line's leading P/A/C is the rest of the broken word, not a header. header = None if header is not None: flush() current_label = header.label current_raga_name = header.raga_name current_is_viloma = header.is_viloma current_start = token.start_pos if header.remainder: current_lines.append( _LineToken( text=header.remainder, start_pos=token.start_pos, end_pos=token.end_pos, ) ) prev_hyphen = token.text.rstrip().endswith("-") continue current_lines.append(token) prev_hyphen = token.text.rstrip().endswith("-") flush() return blocks def _detect_header(self, line: str) -> _HeaderMatch | None: language = self._detect_language_header(line) if language is not None: return language return self._detect_section_header(line) def _detect_language_header(self, line: str) -> _HeaderMatch | None: lowered = line.lower() for key, label in LANGUAGE_HEADER_CANDIDATES: if ( lowered == key or lowered.startswith(f"{key}:") or lowered.startswith(f"{key} -") or lowered.startswith(f"{key} –") ): remainder = line[len(key) :].lstrip(":-– ") return _HeaderMatch(label=label, remainder=remainder) return None def _detect_section_header(self, line: str) -> _HeaderMatch | None: # TRACK-133: a line that opens with "caraNam" but continues with sahitya is # lyric (hyphenation continuation), not a charanam header — suppress it. caranam_lyric = _is_caranam_lyric_line(line) for pattern, label in SECTION_HEADER_PATTERNS: if pattern.search(line): if label == "CHARANAM" and caranam_lyric: continue remainder = pattern.sub("", line, count=1).strip() remainder = re.sub(r"^[:\-)\]\.\s]+", "", remainder).strip() # TRACK-101: Strip residual numbers from "caraNam 1" / "svara sAhitya 2" headers remainder = re.sub(r"^\d+\s*", "", remainder).strip() return _HeaderMatch(label=label, remainder=remainder) if getattr(self, "_inline_pa_enabled", False): for pattern, label in INLINE_PAC_PATTERNS: if pattern.search(line): remainder = pattern.sub("", line, count=1).strip() remainder = re.sub(r"^\d+\s*", "", remainder).strip() return _HeaderMatch(label=label, remainder=remainder) if getattr(self, "_inline_indic_pac_enabled", False): for pattern, label in INLINE_INDIC_PAC_PATTERNS: if pattern.search(line): remainder = pattern.sub("", line, count=1).strip() remainder = re.sub(r"^\d+\s*", "", remainder).strip() return _HeaderMatch(label=label, remainder=remainder) if getattr(self, "_inline_indic_swara_enabled", False): for pattern, label in INLINE_INDIC_SWARA_PATTERNS: if pattern.search(line): remainder = pattern.sub("", line, count=1).strip() remainder = re.sub(r"^\d+\s*", "", remainder).strip() return _HeaderMatch(label=label, remainder=remainder) # TRACK-133 form 3: digit-without-period inline Indic markers ("च4 सुर …"). if getattr(self, "_inline_indic_digit_enabled", False): for pattern, label in INLINE_INDIC_DIGIT_PATTERNS: if pattern.search(line): if label == "CHARANAM" and caranam_lyric: continue remainder = pattern.sub("", line, count=1).strip() remainder = re.sub(r"^\d+\s*", "", remainder).strip() return _HeaderMatch(label=label, remainder=remainder) # TRACK-133 form 1: bare single-akshara inline Indic markers ("च अल नाडु …"), # only when the document uses the bare-marker convention. if getattr(self, "_inline_indic_bare_enabled", False): for pattern, label in INLINE_INDIC_BARE_PATTERNS: if pattern.search(line): if label == "CHARANAM" and caranam_lyric: continue remainder = pattern.sub("", line, count=1).strip() remainder = re.sub(r"^\d+\s*", "", remainder).strip() return _HeaderMatch(label=label, remainder=remainder) # TRACK-133 WORK ITEM 2: pure-ragamalika " rAgaM" stanza boundary. if getattr(self, "_raga_segment_enabled", False): viloma = VILOMA_SUBSECTION_PATTERN.search(line) if viloma: return _HeaderMatch( label="RAGA_SEGMENT", remainder="", raga_name=viloma.group(1).strip(), is_viloma=True, ) if RAGA_SUBSECTION_PATTERN.search(line) or RAGA_SUBSECTION_INDIC_PATTERN.search(line): return _HeaderMatch( label="RAGA_SEGMENT", remainder="", raga_name=_raga_name_from_segment_header(line), ) return None def _extract_sections( self, blocks: list[_TextBlock], lyric_text: str ) -> tuple[list[DetectedSection], list[RagamalikaSubsection]]: ragamalika_subsections: list[RagamalikaSubsection] = [] if not blocks: trimmed = lyric_text.strip() if not trimmed: return [], [] return [ DetectedSection( section_type=SectionType.OTHER, order=1, label="Unknown", text=trimmed, start_pos=0, end_pos=len(lyric_text), ) ], [] sections: list[DetectedSection] = [] found_first_section = False charanam_counter = 0 for block in blocks: if block.label in LANGUAGE_LABELS: if found_first_section: break continue section_type = SECTION_LABEL_TO_TYPE.get(block.label) if section_type is None: continue found_first_section = True # Detect ragamalika subsections (metadata only, no splitting) detected_subs = self._detect_ragamalika_subsections(section_type, block) if detected_subs: ragamalika_subsections.extend(detected_subs) elif block.label == "RAGA_SEGMENT" and block.raga_name: ragamalika_subsections.append( RagamalikaSubsection( raga_name=block.raga_name, parent_section_type=SectionType.OTHER, is_viloma=block.is_viloma, order=0, ) ) # Always create ONE section per structural block block_text = "\n".join(line.text for line in block.lines).strip() if not block_text: continue order = len(sections) + 1 if section_type == SectionType.CHARANAM: charanam_counter += 1 canonical_label = "Charanam" if charanam_counter == 1 else f"Charanam {charanam_counter}" else: canonical_label = section_type.value.replace("_", " ").title() sections.append( DetectedSection( section_type=section_type, order=order, label=canonical_label, text=block_text, start_pos=block.start_pos, end_pos=block.end_pos, ) ) # Assign sequential order to ragamalika subsections for i, sub in enumerate(ragamalika_subsections): sub.order = i + 1 if sections: sections = self._demote_mks(sections) sections = self._merge_dual_format(sections) return sections, ragamalika_subsections trimmed = lyric_text.strip() if not trimmed: return [], [] return [ DetectedSection( section_type=SectionType.OTHER, order=1, label="Unknown", text=trimmed, start_pos=0, end_pos=len(lyric_text), ) ], [] def _demote_mks(self, sections: list[DetectedSection]) -> list[DetectedSection]: """Rule 1: MKS is never a top-level section — attach to preceding parent. If the MKS block contains an inline section header (e.g. చరణమ్ for Charanam), split it: text before the header is MKS of the parent, text after is a new section. """ result: list[DetectedSection] = [] for section in sections: if section.section_type == SectionType.MADHYAMA_KALA: # Check for inline section headers within MKS text split_sections = self._split_mks_inline_headers(section) if split_sections: mks_text, new_sections = split_sections # Attach MKS text to parent if result and mks_text.strip(): parent = result[-1] result[-1] = DetectedSection( section_type=parent.section_type, order=parent.order, label=parent.label, text=parent.text + "\n\n[Madhyama Kala Sahitya]\n" + mks_text, start_pos=parent.start_pos, end_pos=section.end_pos, ) elif mks_text.strip() and not result: # No parent — keep as OTHER result.append( DetectedSection( section_type=SectionType.OTHER, order=0, label="Madhyama Kala", text=mks_text, start_pos=section.start_pos, end_pos=section.end_pos, ) ) # Add the extracted sections result.extend(new_sections) elif result: parent = result[-1] result[-1] = DetectedSection( section_type=parent.section_type, order=parent.order, label=parent.label, text=parent.text + "\n\n[Madhyama Kala Sahitya]\n" + section.text, start_pos=parent.start_pos, end_pos=section.end_pos, ) # else: standalone MKS with no parent and no inline headers — drop it else: result.append(section) # Re-index order sequentially for i, s in enumerate(result): result[i] = DetectedSection( section_type=s.section_type, order=i + 1, label=s.label, text=s.text, start_pos=s.start_pos, end_pos=s.end_pos, ) return result def _split_mks_inline_headers(self, mks_section: DetectedSection) -> tuple[str, list[DetectedSection]] | None: """Check if MKS text contains inline section headers (e.g. చరణమ్). Returns (mks_text_before_header, [new_sections]) or None if no inline header found. """ lines = mks_section.text.splitlines() for i, line in enumerate(lines): header = self._detect_section_header(line.strip()) if header is not None and header.label not in METADATA_LABELS: section_type = SECTION_LABEL_TO_TYPE.get(header.label) if section_type and section_type != SectionType.MADHYAMA_KALA: mks_text = "\n".join(lines[:i]).strip() section_text_lines = [] if header.remainder: section_text_lines.append(header.remainder) section_text_lines.extend(lines[i + 1 :]) section_text = "\n".join(section_text_lines).strip() new_sections = [] if section_text: new_sections.append( DetectedSection( section_type=section_type, order=0, label=section_type.value.replace("_", " ").title(), text=section_text, start_pos=mks_section.start_pos, end_pos=mks_section.end_pos, ) ) return (mks_text, new_sections) return None def _merge_dual_format(self, sections: list[DetectedSection]) -> list[DetectedSection]: """Rule 3: Merge dual-format (continuous + word-division) duplicates.""" if len(sections) < 2: return sections result: list[DetectedSection] = [] skip_next = False for i, section in enumerate(sections): if skip_next: skip_next = False continue if i + 1 < len(sections): next_section = sections[i + 1] if section.section_type == next_section.section_type: norm_a = re.sub(r"\s+", "", section.text) norm_b = re.sub(r"\s+", "", next_section.text) if norm_a and norm_b: overlap = sum(1 for ca, cb in zip(norm_a, norm_b, strict=False) if ca == cb) max_len = max(len(norm_a), len(norm_b)) if max_len > 0 and overlap / max_len > 0.9: # Keep the longer one (word-division has more spaces) kept = next_section if len(next_section.text) >= len(section.text) else section result.append( DetectedSection( section_type=kept.section_type, order=len(result) + 1, label=kept.label, text=kept.text, start_pos=kept.start_pos, end_pos=kept.end_pos, ) ) skip_next = True continue result.append( DetectedSection( section_type=section.section_type, order=len(result) + 1, label=section.label, text=section.text, start_pos=section.start_pos, end_pos=section.end_pos, ) ) return result def _detect_ragamalika_subsections( self, section_type: SectionType, block: _TextBlock, ) -> list[RagamalikaSubsection]: """Scan block for ragamalika raga markers without splitting.""" if not block.lines: return [] subsections: list[RagamalikaSubsection] = [] for line in block.lines: viloma_match = VILOMA_SUBSECTION_PATTERN.search(line.text) raga_match = RAGA_SUBSECTION_PATTERN.search(line.text) if viloma_match is not None: subsections.append( RagamalikaSubsection( raga_name=viloma_match.group(1).strip(), parent_section_type=section_type, is_viloma=True, order=0, ) ) elif raga_match is not None: subsections.append( RagamalikaSubsection( raga_name=raga_match.group(1).strip(), parent_section_type=section_type, is_viloma=False, order=0, ) ) return subsections def _extract_lyric_variants( self, blocks: list[_TextBlock], lyric_text: str, canonical_sections: list[DetectedSection], ) -> list[DetectedLyricVariant]: language_blocks_seen = any(block.label in LANGUAGE_LABELS - METADATA_LABELS for block in blocks) if language_blocks_seen: return self._extract_language_header_variants(blocks, canonical_sections) return self._extract_script_split_variants(lyric_text, canonical_sections) def _extract_language_header_variants( self, blocks: list[_TextBlock], canonical_sections: list[DetectedSection], ) -> list[DetectedLyricVariant]: variants: list[DetectedLyricVariant] = [] current_label: str | None = None current_blocks: list[_TextBlock] = [] def flush() -> None: nonlocal current_label, current_blocks if current_label is None: return language, script = self._language_script_for_label(current_label) sections = self._sections_from_variant_blocks(current_blocks, canonical_sections) if sections: variants.append(DetectedLyricVariant(language=language, script=script, sections=sections)) current_label = None current_blocks = [] for block in blocks: if block.label in METADATA_LABELS: flush() break if block.label in LANGUAGE_LABELS - METADATA_LABELS: flush() current_label = block.label if block.lines: # Re-parse lines to detect inline section headers within variant text sub_blocks = self._reparse_lines_for_sections(block.lines, block.start_pos, block.end_pos) current_blocks.extend(sub_blocks) continue if current_label is not None: current_blocks.append(block) flush() return variants def _extract_post_boundary_variants( self, source_text: str, boundary_pos: int, canonical_sections: list[DetectedSection], existing_scripts: set[str], ) -> list[DetectedLyricVariant]: """TRACK-100: Extract Indic-script variants from text after the metadata boundary.""" all_blocks = self._build_blocks(source_text) post_blocks = [b for b in all_blocks if b.start_pos >= boundary_pos] if not post_blocks: return [] _INDIC_LABELS = {"DEVANAGARI", "TAMIL", "TELUGU", "KANNADA", "MALAYALAM", "SANSKRIT", "HINDI"} variants: list[DetectedLyricVariant] = [] current_label: str | None = None current_blocks: list[_TextBlock] = [] def flush() -> None: nonlocal current_label, current_blocks if current_label is None: return language, script = self._language_script_for_label(current_label) if script not in existing_scripts: sections = self._sections_from_variant_blocks(current_blocks, canonical_sections) if sections: variants.append(DetectedLyricVariant(language=language, script=script, sections=sections)) current_label = None current_blocks = [] for block in post_blocks: if block.label in METADATA_LABELS: flush() continue if block.label in _INDIC_LABELS: flush() current_label = block.label if block.lines: sub_blocks = self._reparse_lines_for_sections(block.lines, block.start_pos, block.end_pos) current_blocks.extend(sub_blocks) continue if block.label in {"ENGLISH", "LATIN"}: flush() continue if current_label is not None: current_blocks.append(block) flush() return variants def _merge_leading_prefix_into_first_raga_segment(self, blocks: list[_TextBlock]) -> list[_TextBlock]: """Fold any leading non-stanza blocks into the first RAGA_SEGMENT stanza. Only applies in pure-ragamalika mode. If the blocks contain no RAGA_SEGMENT (or the first block already is one) the list is returned unchanged. """ first_seg = next((i for i, b in enumerate(blocks) if b.label == "RAGA_SEGMENT"), None) if first_seg is None or first_seg == 0: return blocks prefix_lines: list[_LineToken] = [] for b in blocks[:first_seg]: prefix_lines.extend(b.lines) target = blocks[first_seg] merged = _TextBlock( label=target.label, lines=prefix_lines + target.lines, start_pos=blocks[0].start_pos, end_pos=target.end_pos, ) return [merged, *blocks[first_seg + 1 :]] def _split_charanam_pallavi_echoes( self, blocks: list[_TextBlock], canonical_sections: list[DetectedSection] ) -> list[_TextBlock]: """Recover missing Indic charanam headings only when echo evidence fills the deficit. TRACK-133: a line-final pallavi echo can close an unlabelled stanza inside a charanam. It remains part of that stanza. Require a closing echo on the final stanza too, and repair only a single glued block whose extra stanzas exactly equal the charanam deficit. Summing cuts across several charanams could keep a mixed true+false split. Split tokens before canonical mapping so later charanams retain their slots and each new section has source offsets. """ deficit = sum(s.section_type == SectionType.CHARANAM for s in canonical_sections) - sum( b.label == "CHARANAM" and bool(b.lines) for b in blocks ) if deficit <= 0: return blocks pallavi = next((b for b in blocks if b.label == "PALLAVI" and b.lines), None) if pallavi is None: return blocks script = self._detect_script(pallavi.lines[0].text) if script not in {"devanagari", "telugu", "kannada", "malayalam", "tamil"}: return blocks echo_match = re.search(r"\(([^()\s]+)\)\s*$", pallavi.lines[-1].text) if echo_match is None: return blocks echo = echo_match.group(1) opening = re.sub(r"^\d+\s*", "", pallavi.lines[0].text) if not opening.startswith(echo + " "): return blocks closing_echo = re.compile(r"\(" + re.escape(echo) + r"\)\s*$") glued: list[tuple[int, list[int]]] = [] for index, block in enumerate(blocks): if block.label != "CHARANAM" or not block.lines: continue if not closing_echo.search(block.lines[-1].text): continue boundaries = [ i + 1 for i, line in enumerate(block.lines[:-1]) if closing_echo.search(line.text) and self._detect_script(block.lines[i + 1].text) == script ] if boundaries: glued.append((index, boundaries)) if len(glued) != 1: return blocks index, boundaries = glued[0] if len(boundaries) != deficit: return blocks repaired: list[_TextBlock] = [] for i, block in enumerate(blocks): if i != index: repaired.append(block) continue start = 0 for end in (*boundaries, len(block.lines)): lines = block.lines[start:end] repaired.append( _TextBlock( label="CHARANAM", lines=lines, start_pos=lines[0].start_pos, end_pos=lines[-1].end_pos, ) ) start = end logger.info("TRACK-133 pallavi-echo split: restored %d %s charanam boundary(s)", deficit, script) return repaired def _sections_from_variant_blocks( self, blocks: list[_TextBlock], canonical_sections: list[DetectedSection], ) -> list[DetectedSection]: if not blocks: return [] # TRACK-133 follow-up: in a pure ragamalika the stanzas are the RAGA_SEGMENT # blocks. Any block before the first RAGA_SEGMENT (a title / "dasa raga # malika - tala" prefix line that carries no raga header of its own) is not a # stanza — the canonical builder drops such leading UNLABELED blocks. Here we # fold their text into the first stanza instead of dropping it, so the variant # keeps every stanza slot aligned 1:1 with canon (an extra leading section # would shift the mapping and drop the final stanza). if getattr(self, "_raga_segment_enabled", False): blocks = self._merge_leading_prefix_into_first_raga_segment(blocks) blocks = self._split_charanam_pallavi_echoes(blocks, canonical_sections) # Collect raw parsed sections from blocks, applying MKS demotion raw_sections: list[DetectedSection] = [] for block in blocks: text = "\n".join(line.text for line in block.lines).strip() if not text: continue section_type = SECTION_LABEL_TO_TYPE.get(block.label, SectionType.OTHER) raw_sections.append( DetectedSection( section_type=section_type, order=0, label=section_type.value.replace("_", " ").title() if section_type != SectionType.OTHER else "Other", text=text, start_pos=block.start_pos, end_pos=block.end_pos, ) ) # Apply MKS demotion and dual-format merging to variant too raw_sections = self._demote_mks(raw_sections) raw_sections = self._merge_dual_format(raw_sections) # Map to canonical structure by matching type and sequential occurrence if not canonical_sections: return raw_sections # Promote leading OTHER sections: when a variant starts with unlabeled # text (no section header before the first typed section), assign each # leading OTHER section the type of the next unmatched canonical section. # This handles pages where e.g. Devanagari Pallavi text appears directly # after the language header without a "पल्लवि" section header. # # TRACK-133 follow-up: in a *pure* ragamalika every canonical section is a # legitimately-distinct OTHER stanza (RAGA_SEGMENT). Promoting/merging leading # OTHER runs here would fold all N stanzas into one section — the exact # variant-side collapse this path is fixing. Gate the promotion off on the # same signal as the canonical RAGA_SEGMENT path so the N stanzas map 1:1. promoted_count = 0 if not getattr(self, "_raga_segment_enabled", False): canonical_iter = iter(canonical_sections) for s in raw_sections: if s.section_type != SectionType.OTHER: break canon = next(canonical_iter, None) if canon is not None: s.section_type = canon.section_type promoted_count += 1 # Merge promoted sections into the following typed section when both # share the same type (e.g. a title-only promoted PALLAVI followed by # an explicit PALLAVI block with the actual content). if promoted_count > 0: merged: list[DetectedSection] = [] for s in raw_sections: if merged and merged[-1].section_type == s.section_type and len(merged) <= promoted_count: prev = merged[-1] merged[-1] = DetectedSection( section_type=prev.section_type, order=prev.order, label=prev.label, text=prev.text + "\n" + s.text, start_pos=prev.start_pos, end_pos=s.end_pos, ) else: merged.append(s) raw_sections = merged type_queues: dict[SectionType, list[DetectedSection]] = {} for s in raw_sections: type_queues.setdefault(s.section_type, []).append(s) sections: list[DetectedSection] = [] for canonical in canonical_sections: queue = type_queues.get(canonical.section_type, []) if queue: matched = queue.pop(0) sections.append( DetectedSection( section_type=canonical.section_type, order=canonical.order, label=canonical.label, text=matched.text, start_pos=matched.start_pos, end_pos=matched.end_pos, ) ) return strip_refrain_trailer(sections) if self._trailer_strip_enabled else sections def _reparse_lines_for_sections( self, lines: list[_LineToken], block_start: int, block_end: int, ) -> list[_TextBlock]: """Re-parse lines within a language block to detect inline section headers.""" blocks: list[_TextBlock] = [] current_label = "UNLABELED" current_lines: list[_LineToken] = [] def flush() -> None: nonlocal current_lines if current_lines: blocks.append( _TextBlock( label=current_label, lines=current_lines, start_pos=current_lines[0].start_pos, end_pos=current_lines[-1].end_pos, ) ) current_lines = [] for line in lines: header = self._detect_section_header(line.text) if header is not None: flush() current_label = header.label if header.remainder: current_lines.append( _LineToken( text=header.remainder, start_pos=line.start_pos, end_pos=line.end_pos, ) ) else: current_lines.append(line) flush() return blocks def _extract_script_split_variants( self, lyric_text: str, canonical_sections: list[DetectedSection], ) -> list[DetectedLyricVariant]: if not canonical_sections: return [] script_sections: dict[str, dict[int, list[str]]] = {} for section in canonical_sections: for line in section.text.splitlines(): clean = line.strip() if not clean: continue script = self._detect_script(clean) if script is None: continue script_sections.setdefault(script, {}).setdefault(section.order, []).append(clean) if not script_sections: script = self._detect_script(lyric_text) or "latin" language = self._language_for_script(script) return [ DetectedLyricVariant( language=language, script=script, sections=canonical_sections, ) ] variants: list[DetectedLyricVariant] = [] for script, by_order in script_sections.items(): variant_sections: list[DetectedSection] = [] for section in canonical_sections: lines = by_order.get(section.order) if not lines: continue variant_sections.append( DetectedSection( section_type=section.section_type, order=section.order, label=section.label, text="\n".join(lines), start_pos=section.start_pos, end_pos=section.end_pos, ) ) if variant_sections: variants.append( DetectedLyricVariant( language=self._language_for_script(script), script=script, sections=variant_sections, ) ) variants.sort(key=lambda v: v.script) return variants def _find_metadata_boundaries(self, text: str) -> list[MetadataBoundary]: boundaries_by_pos: dict[int, MetadataBoundary] = {} for label, pattern in METADATA_BOUNDARY_PATTERNS: for match in pattern.finditer(text): start = match.start() existing = boundaries_by_pos.get(start) candidate = MetadataBoundary(label=label, start_pos=start, end_pos=match.end()) if existing is None or candidate.end_pos > existing.end_pos: boundaries_by_pos[start] = candidate return [boundaries_by_pos[pos] for pos in sorted(boundaries_by_pos)] def _normalize_line(self, line: str) -> str: if not line: return line return re.sub(r"[\u2080-\u2089]", "", line) def _is_boilerplate(self, line: str) -> bool: lowered = line.lower() # TRACK-103: Standalone "Back" navigation and "Meaning of Kriti" links if lowered == "back": return True if lowered.startswith("meaning of kriti"): return True if "a i i u u" in lowered or "ch j jh" in lowered or "ph b bh m" in lowered: return True if "pronunciation guide" in lowered: return True # Devanagari vowel/consonant mapping lines (e.g. "ऎ,कॆ,चॆ.. - e,ke,ce..(short)") if ("(short)" in lowered or "(long)" in lowered) and re.search(r"[\u0900-\u097F].*-\s*[a-zA-Z]", line): return True # Transliteration-key preamble lines (Modified HK chart at the top of each # script block on Govindan blogs \u2014 ADR-015 head-capture fix). Three forms, # all impossible in single-script lyric: # 1. a comma-separated Indic consonant chart ("\u0B95,\u0B9A,\u0B9F,\u0BA4,\u0BAA - 2-\u0916 \u2026") if re.match(r"^[\u0900-\u0D7F]\s*,\s*[\u0900-\u0D7F]", line): return True # 2. a cross-script mapping row \u2014 Devanagari mixed with another Indic # script ("\u0BB81 \u0936 - \u0936\u093F\u0935 - \u0B9A\u0BBF\u0BB5\u0BA9\u0BCD"); # lyric is always one script per line. if re.search(r"[\u0900-\u097F]", line) and re.search(r"[\u0B00-\u0D7F]", line): return True # 3. a parenthesised sandhi mapping, whole line ("(\u0B9A3 - \u0B9C)"); lyric # refrains like "(\u0BB5\u0BC7\u0B99\u0BCD\u0B95)" carry no ' - ' mapping arrow. if re.match(r"^\s*\([^)]*\s-\s[^)]*\)\s*$", line): return True # Filter nOTTu-svara header — it's metadata not lyric content if re.match(r"^\s*\(?n[oō]t+u[\s-]*svara\s+s[aā]hityam?\)?\.?\s*$", lowered): return True if lowered.startswith("updated on "): return True return any( marker in lowered for marker in ( "powered by blogger", "newer post", "older post", "subscribe to", "post a comment", "blog archive", "link to this post", "posted by", "all rights reserved", "copyright", "skip to main", "related posts", ) ) def _is_meta_line(self, line: str) -> bool: if RAGA_SUBSECTION_PATTERN.search(line) or VILOMA_SUBSECTION_PATTERN.search(line): return False lowered = line.lower() return any(keyword in lowered for keyword in METADATA_KEYWORDS) def _language_script_for_label(self, label: str) -> tuple[str, str]: mapping = { "SANSKRIT": ("sa", "devanagari"), "DEVANAGARI": ("sa", "devanagari"), "HINDI": ("hi", "devanagari"), "TAMIL": ("ta", "tamil"), "TELUGU": ("te", "telugu"), "KANNADA": ("kn", "kannada"), "MALAYALAM": ("ml", "malayalam"), "ENGLISH": ("en", "latin"), "LATIN": ("en", "latin"), } return mapping.get(label, ("en", "latin")) def _language_for_script(self, script: str) -> str: return { "devanagari": "sa", "tamil": "ta", "telugu": "te", "kannada": "kn", "malayalam": "ml", "latin": "en", }.get(script, "en") def _detect_script(self, text: str) -> str | None: counts = { "devanagari": 0, "tamil": 0, "telugu": 0, "kannada": 0, "malayalam": 0, "latin": 0, } for ch in text: cp = ord(ch) if 0x0900 <= cp <= 0x097F: counts["devanagari"] += 1 elif 0x0B80 <= cp <= 0x0BFF: counts["tamil"] += 1 elif 0x0C00 <= cp <= 0x0C7F: counts["telugu"] += 1 elif 0x0C80 <= cp <= 0x0CFF: counts["kannada"] += 1 elif 0x0D00 <= cp <= 0x0D7F: counts["malayalam"] += 1 elif 0x0041 <= cp <= 0x024F: counts["latin"] += 1 script, score = max(counts.items(), key=lambda kv: kv[1]) return script if score > 0 else None def to_canonical_sections(self, detected: list[DetectedSection]) -> list[CanonicalSection]: return [ CanonicalSection( type=d.section_type, order=d.order, label=d.label, ) for d in detected ] def to_canonical_lyric_sections(self, detected: list[DetectedSection]) -> list[CanonicalLyricSection]: return [CanonicalLyricSection(section_order=d.order, text=d.text) for d in detected if d.text] def to_canonical_lyric_variants( self, variants: list[DetectedLyricVariant], ) -> list[CanonicalLyricVariant]: return [ CanonicalLyricVariant( language=variant.language, script=variant.script, sections=self.to_canonical_lyric_sections(variant.sections), ) for variant in variants if variant.sections ] def to_canonical_metadata_boundaries( self, boundaries: list[MetadataBoundary], ) -> list[CanonicalMetadataBoundary]: return [ CanonicalMetadataBoundary( label=b.label, start_offset=b.start_pos, end_offset=b.end_pos, ) for b in boundaries ]