Function build_programme_index

apps/epg/tasks.py:4486–4580

Scan the XML file with raw binary I/O to build a {tvg_id: [byte_offset, ...]} map. Persists the result to the EPGSourceIndex table. Most XMLTV files group programmes by channel, but some split a channel across multiple non-contiguous blocks, so we record block starts up to _OFFSET_C

(source_id)

def build_programme_index(source_id):
    """
    Scan the XML file with raw binary I/O to build a {tvg_id: [byte_offset, ...]} map.
    Persists the result to the EPGSourceIndex table. Most XMLTV files group programmes
    by channel, but some split a channel across multiple non-contiguous blocks, so we
    record block starts up to _OFFSET_CAP and mark only channels that exceed the cap
    as interleaved.
    """
    try:
        source = EPGSource.objects.get(id=source_id)
    except EPGSource.DoesNotExist:
        logger.error(f'[build_programme_index] EPGSource {source_id} not found')
        return

    file_path = _resolve_source_file(source)
    if not file_path or not os.path.exists(file_path):
        logger.warning(
            f'[build_programme_index] File not found for source {source_id}: {file_path}'
        )
        return

    logger.debug(
        f'[build_programme_index] Building byte-offset index for source {source_id} from {file_path}'
    )
    start = time.monotonic()
    index = {}
    prev_channel = None
    interleaved_channels = set()

    CHUNK = 8 * 1024 * 1024  # 8MB

    with open(file_path, 'rb') as f:
        buf = bytearray()
        buf_offset = 0  # absolute file offset of buf[0]

        while True:
            chunk = f.read(CHUNK)
            if not chunk and not buf:
                break
            buf.extend(chunk)
            search_from = 0

            while True:
                idx, tag_end = _find_programme_tag(buf, search_from)
                if idx == -1:
                    break
                if tag_end == -1 and chunk:
                    break  # incomplete tag at buffer edge, need more data

                abs_pos = buf_offset + idx
                m = _CHANNEL_ATTR_RE.search(
                    buf, idx, tag_end + 1 if tag_end != -1 else idx + _MAX_START_TAG
                )
                if m:
                    channel_id = _decode_channel_id(m.group(1) or m.group(2))
                    if channel_id not in index:
                        index[channel_id] = [abs_pos]
                    elif channel_id != prev_channel: