问题场景

应用日志通常按时间追加，单个文件可能从几百 MB 到数十 GB。若要按时间范围导出或分析，逐行扫描并解析全部记录会很浪费。这里的前提是：记录按时间非递减排列，每条记录第一行以固定格式时间戳开头，后续多行堆栈或详情行不以时间戳开头。

本文把文件当作字节序列，用二分查找定位两个边界：

• left：第一条时间戳大于等于 start 的记录起始偏移；
• right：第一条时间戳大于 end 的记录起始偏移。

读取 [left, right) 即可得到时间在 [start, end] 内的完整记录。因为右边界是第一条大于结束时间的记录，所以时间等于 end 的多条记录不会被截断。

核心思路

二分时，mid 很可能落在某一行中间。不能直接把 mid 当成行首，否则可能读到半行时间戳。实现中先检查 mid 前一个字节：如果不是换行符，就调用 readline() 丢弃当前残行，再从下一行开始读取。

如果读到的行没有时间戳，说明它可能是多行记录的续行，继续向后跳到下一条时间戳行。这样二分采样始终落在记录起始行上。多行记录本身不需要在查找阶段解析，最终按字节区间整体读取即可。

最小实现

from datetime import datetime
from pathlib import Path
from typing import BinaryIO, Iterator, Optional, Tuple
import os
import re
import tempfile

TS_RE = re.compile(rb'^([0-9]{4}-[0-9]{2}-[0-9]{2}[ T][0-9]{2}:[0-9]{2}:[0-9]{2})')
DT_FMT = '%Y-%m-%dT%H:%M:%S'
NL = bytes([10])


def parse_ts(line: bytes) -> Optional[datetime]:
    m = TS_RE.match(line)
    if not m:
        return None
    text = m.group(1).decode('ascii').replace(' ', 'T')
    return datetime.strptime(text, DT_FMT)


def line_at_or_after(f: BinaryIO, pos: int) -> Tuple[int, bytes]:
    if pos > 0:
        f.seek(pos - 1)
        prev = f.read(1)
        f.seek(pos)
        if prev != NL:
            f.readline()
    else:
        f.seek(0)
    start = f.tell()
    return start, f.readline()


def next_record(f: BinaryIO, pos: int):
    while True:
        start, line = line_at_or_after(f, pos)
        if not line:
            return None
        ts = parse_ts(line)
        if ts is not None:
            return start, line, ts
        pos = start + len(line)


def first_record(f: BinaryIO, size: int, target: datetime, strict: bool) -> int:
    lo, hi = 0, size
    ans = size
    while lo < hi:
        mid = (lo + hi) // 2
        got = next_record(f, mid)
        if got is None:
            hi = mid
            continue
        off, line, ts = got
        ok = (ts > target) if strict else (ts >= target)
        if ok:
            ans = off
            hi = mid
        else:
            lo = off + len(line)
    return ans


def find_range(path: str, start: datetime, end: datetime):
    size = Path(path).stat().st_size
    with open(path, 'rb') as f:
        left = first_record(f, size, start, False)
        right = first_record(f, size, end, True)
    if left >= right:
        return None
    return left, right


def iter_range(path: str, start: int, end: int, chunk: int = 1 << 20) -> Iterator[bytes]:
    with open(path, 'rb') as f:
        f.seek(start)
        left = end - start
        while left > 0:
            data = f.read(min(chunk, left))
            if not data:
                break
            left -= len(data)
            yield data

运行与验证

把代码保存为 demo.py，下面的自测会生成一个包含多行记录的临时日志，查询 10:00:01 到 10:00:02 的范围：

if __name__ == '__main__':
    sample = b'''2024-05-01T10:00:00 INFO start
2024-05-01T10:00:01 ERROR boom
    at demo.py:1
    at main.py:2
2024-05-01T10:00:02 INFO ok
2024-05-01T10:00:03 WARN tail
'''
    fd, name = tempfile.mkstemp(suffix='.log')
    os.close(fd)
    try:
        Path(name).write_bytes(sample)
        t0 = datetime.strptime('2024-05-01T10:00:01', DT_FMT)
        t1 = datetime.strptime('2024-05-01T10:00:02', DT_FMT)
        r = find_range(name, t0, t1)
        print('range:', r)
        if r:
            data = b''.join(iter_range(name, r[0], r[1]))
            print(data.decode('utf-8'))
    finally:
        os.unlink(name)

执行 python demo.py 后应观察到：输出包含 10:00:01 的 ERROR 记录及其两行续行，也包含 10:00:02 的 INFO 记录，但不包含 10:00:03 的 WARN 记录。还可以用几个边界用例验证：

1. start 等于 end 且该时间存在记录时，应输出该时间点的全部记录；
2. 时间范围完全晚于文件最后一条记录时，find_range 返回 None；
3. 时间范围完全早于文件第一条记录时，返回空范围；
4. 在小文件上逐行扫描计算期望起止偏移，与二分结果对照。

边界与限制

• 日志必须按时间非递减排序。乱序文件不能用该方法。
• 每条记录第一行必须以可解析时间戳开头，续行不能以同样格式开头；若格式不同，需要调整 TS_RE 和 parse_ts。
• 示例只解析到秒，忽略毫秒和时区。若日志带毫秒或时区，应统一解析规则，并保证查询时间与日志时间处于同一时区。
• 如果某条多行记录极大，readline() 可能一次读入很长的一行；对超长单行日志应改用有界分块扫描。
• 文件在查找期间不应被追加或重写；需要处理滚动日志时，应先复制快照或在稳定文件上操作。
• 读取阶段按字节块输出，不保证每个块恰好以换行结尾；若下游按行处理，需要在块之间保留残余缓冲。

小结

这个实现把时间范围查询转化为字节偏移上的二分查找，通过处理中间残行、跳过无时间戳续行、用严格大于 end 的右边界，保证多行记录和等值时间戳边界不被误截。它适合时间有序、记录起始行格式稳定的日志文件。