Function _read_programs_at_offsets

apps/epg/tasks.py:4660–4745

Seek to each offset, extract elements for *tvg_id*, return the first one currently airing. Chunk-based so it works on minified XML.

(file_path, tvg_id, offsets, now)

def _read_programs_at_offsets(file_path, tvg_id, offsets, now):
    """
    Seek to each offset, extract <programme> elements for *tvg_id*, return the
    first one currently airing. Chunk-based so it works on minified XML.
    """
    PROG_CLOSE = b'</programme>'
    CLOSE_LEN = len(PROG_CLOSE)
    READ_SIZE = 2 * 1024 * 1024  # 2MB per read

    with open(file_path, 'rb') as f:
        for offset in offsets:
            f.seek(offset)
            buf = bytearray()
            done = False

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

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

                    # Check channel before searching for close tag
                    m = _CHANNEL_ATTR_RE.search(
                        buf,
                        tag_start,
                        tag_end + 1 if tag_end != -1 else tag_start + _MAX_START_TAG,
                    )
                    if not m:
                        search_from = (
                            (tag_end + 1)
                            if tag_end != -1
                            else (tag_start + _PROGRAMME_TAG_LEN)
                        )
                        continue

                    ch = _decode_channel_id(m.group(1) or m.group(2))
                    if ch != tvg_id:
                        done = True  # different channel, end of block
                        break

                    # Find the closing </programme> tag
                    close_pos = buf.find(
                        PROG_CLOSE, tag_end + 1 if tag_end != -1 else m.end()
                    )
                    if close_pos == -1:
                        if not chunk:
                            done = True  # EOF with no close tag
                        break  # need more data
                    close_end = close_pos + CLOSE_LEN