用 Python 直接读取 tar.gz 中的 CSV：免解包统计与错误检查

python3 --version
mkdir tar-csv-demo
cd tar-csv-demo
python3 - <<'PY_DATA'
import io
import tarfile
payload = b'\xef\xbb\xbfid,value\n001,10\n"sample,002",12\n003,20\n'
with tarfile.open("daily.tar.gz", "x:gz") as archive:
    info = tarfile.TarInfo("reports/daily.csv")
    info.size = len(payload)
    archive.addfile(info, io.BytesIO(payload))
PY_DATA
python3 -m tarfile -l daily.tar.gz

"""Summarize one CSV member in a completed, trusted tar.gz. Python 3.9+."""
import argparse
import csv
import io
import json
import sys
import tarfile
import zlib


def summarize(path, member, max_bytes):
    with tarfile.open(path, "r:gz") as archive:
        matches = [item for item in archive.getmembers() if item.name == member]
        if not matches:
            raise ValueError(f"member not found: {member!r}")
        if len(matches) != 1:
            raise ValueError(f"duplicate member: {member!r}")
        info = matches[0]
        if not info.isfile():
            raise ValueError(f"member is not a regular file: {member!r}")
        if info.size > max_bytes:
            raise ValueError(f"member size {info.size} exceeds limit {max_bytes}")
        records = total = 0
        with archive.extractfile(info) as raw:
            with io.TextIOWrapper(raw, encoding="utf-8-sig", newline="") as text:
                reader = csv.reader(text, strict=True)
                if next(reader, None) != ["id", "value"]:
                    raise ValueError("expected header: id,value")
                for row in reader:
                    where = f"physical line {reader.line_num}"
                    if len(row) != 2 or not row[0].strip():
                        raise ValueError(f"{where}: expected nonempty id and value")
                    try:
                        value = int(row[1])
                    except ValueError as exc:
                        raise ValueError(f"{where}: value must be an integer") from exc
                    records += 1
                    total += value
    return {"member": member, "records": records, "total": total}


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("archive", help="completed local tar.gz")
    parser.add_argument("member", help="exact member name, as listed by tarfile")
    parser.add_argument("--max-bytes", type=int, default=64 * 1024 * 1024,
                        help="maximum declared CSV size; default: 67108864")
    args = parser.parse_args()
    if args.max_bytes <= 0:
        parser.error("--max-bytes must be positive")
    try:
        result = summarize(args.archive, args.member, args.max_bytes)
    except (OSError, EOFError, tarfile.TarError, zlib.error,
            UnicodeError, csv.Error, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(json.dumps(result, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python3 read_tar_csv.py --help
python3 read_tar_csv.py daily.tar.gz reports/daily.csv

{"member": "reports/daily.csv", "records": 3, "total": 42}

python3 read_tar_csv.py daily.tar.gz missing.csv
printf 'exit=%s\n' "$?"
python3 read_tar_csv.py daily.tar.gz reports/daily.csv --max-bytes 10
printf 'exit=%s\n' "$?"

python3 - <<'PY_DUP'
import io
import tarfile
payload = b"id,value\n001,1\n"
with tarfile.open("duplicate.tar.gz", "x:gz") as archive:
    for _ in range(2):
        info = tarfile.TarInfo("reports/daily.csv")
        info.size = len(payload)
        archive.addfile(info, io.BytesIO(payload))
PY_DUP
python3 read_tar_csv.py duplicate.tar.gz reports/daily.csv
printf 'exit=%s\n' "$?"

python3 - <<'PY_CHECK'
from pathlib import Path
assert not Path("reports").exists()
assert not Path("daily.csv").exists()
print("OK: no extracted CSV")
PY_CHECK