【Python】巨大なテキストを丸ごと読まない：行単位で抽出・集計する

from pathlib import Path
from tempfile import TemporaryDirectory
import math


def filter_rows(source, destination, rejected, threshold=2.0):
    paths = [Path(path).resolve() for path in (source, destination, rejected)]
    if len(set(paths)) != 3:
        raise ValueError("input and outputs must be distinct")
    total = kept = errors = 0
    with paths[0].open(encoding="utf-8") as incoming, \
         paths[1].open("x", encoding="utf-8", newline="\n") as outgoing, \
         paths[2].open("x", encoding="utf-8", newline="\n") as bad:
        for number, raw in enumerate(incoming, start=1):
            total += 1
            fields = raw.rstrip("\r\n").split("\t")
            try:
                if len(fields) != 2 or not fields[0]:
                    raise ValueError("columns")
                value = float(fields[1])
                if not math.isfinite(value):
                    raise ValueError("non-finite")
            except ValueError:
                errors += 1
                bad.write(f"{number}\tinvalid row\n")
                continue
            if value >= threshold:
                outgoing.write("\t".join(fields) + "\n")
                kept += 1
    return {"total": total, "kept": kept, "errors": errors}


with TemporaryDirectory() as temporary:
    root = Path(temporary)
    source, output, bad = (root / name for name in ("input.tsv", "selected.tsv", "rejected.tsv"))
    source.write_text("A\t1.5\nB\t2\nC\t3.5\nbroken\nD\tnan\nE\t4", encoding="utf-8")
    counts = filter_rows(source, output, bad)
    assert counts == {"total": 6, "kept": 3, "errors": 2}
    assert output.read_text() == "B\t2\nC\t3.5\nE\t4\n"
    assert bad.read_text() == "4\tinvalid row\n5\tinvalid row\n"
    assert source.read_text().endswith("E\t4")
    print(counts)
    print(output.read_text(), end="")