Python 临时目录实战：自动清理中间文件，保留成功报告

python3 --version
mkdir tempfile-report-demo
cd tempfile-report-demo
mkdir output
python3 - <<'PY_DATA'
from pathlib import Path
Path("notes.txt").write_text("  first task  \n\n第二个任务\n", encoding="utf-8")
PY_DATA

"""Build a numbered UTF-8 report in a private temporary workspace."""
import argparse
import os
from pathlib import Path
import sys
import tempfile


def build_report(source, output, fail=False):
    source = source.resolve(strict=True)
    if not source.is_file():
        raise ValueError("input must be a regular file")
    output = output.parent.resolve(strict=True) / output.name
    if output.is_symlink():
        raise ValueError("output must not be a symlink")
    if output.exists() and output.samefile(source):
        raise ValueError("input and output must be different files")

    with tempfile.TemporaryDirectory(prefix=".report-", dir=output.parent) as name:
        work = Path(name)
        print(f"WORK={work}", file=sys.stderr, flush=True)
        staged = work / "report.txt"
        count = 0
        with source.open(encoding="utf-8") as incoming:
            with staged.open("x", encoding="utf-8", newline="\n") as outgoing:
                for line in incoming:
                    text = line.strip()
                    if text:
                        count += 1
                        outgoing.write(f"{count}\t{text}\n")
        if fail:
            raise RuntimeError("simulated failure before replacement")
        os.replace(staged, output)
    return count


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("source", type=Path, help="existing UTF-8 text file")
    parser.add_argument("output", type=Path, help="report path; replaces an existing file")
    parser.add_argument("--fail", action="store_true", help="fail before replacement")
    args = parser.parse_args()
    try:
        count = build_report(args.source, args.output, args.fail)
    except (OSError, UnicodeError, ValueError, RuntimeError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(f"OK: {count} nonempty lines")
    return 0


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

python3 task_report.py --help
python3 task_report.py notes.txt output/report.txt
cat output/report.txt

OK: 2 nonempty lines
1	first task
2	第二个任务

"""Verify cleanup and output preservation without leaving test data."""
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import re
import subprocess
import sys
import tempfile

script = Path("task_report.py").resolve(strict=True)
with tempfile.TemporaryDirectory(prefix="check-", dir=Path.cwd()) as name:
    root = Path(name)
    source = root / "notes.txt"
    output = root / "report.txt"
    source.write_text("  first task  \n\n第二个任务\n", encoding="utf-8")

    def run(target, *options):
        result = subprocess.run(
            [sys.executable, str(script), str(source), str(target), *options],
            capture_output=True, text=True, encoding="utf-8", timeout=10,
        )
        match = re.search(r"^WORK=(.+)$", result.stderr, re.MULTILINE)
        if match:
            assert not Path(match[1]).exists(), result.stderr
        return result

    result = run(output)
    assert result.returncode == 0, result.stderr
    assert output.read_text(encoding="utf-8") == "1\tfirst task\n2\t第二个任务\n"
    previous = output.read_bytes()
    result = run(output, "--fail")
    assert result.returncode == 1 and not result.stdout
    assert output.read_bytes() == previous
    result = run(root / "never-created.txt", "--fail")
    assert result.returncode == 1
    assert not (root / "never-created.txt").exists()

    source.write_bytes(b"valid\n\xffinvalid\n")
    result = run(output)
    assert result.returncode == 1 and output.read_bytes() == previous
    source.write_text("ready\n", encoding="utf-8")
    assert run(source).returncode == 1
    assert run(output, "--fai").returncode == 2

    with ThreadPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(run, [root / f"report-{i}.txt" for i in range(4)]))
    assert all(result.returncode == 0 for result in results)
    paths = [re.search(r"^WORK=(.+)$", r.stderr, re.MULTILINE)[1] for r in results]
    assert len(set(paths)) == 4
    assert not list(root.glob(".report-*"))
print("PASS: cleanup, failure preservation, UTF-8 errors, isolated workspaces")

python3 verify_report.py

python3 - <<'PY_CHILD'
from pathlib import Path
import subprocess
import sys
import tempfile

with tempfile.TemporaryDirectory(prefix="child-", dir=Path.cwd()) as work:
    subprocess.run(
        [sys.executable, "-c",
         "from pathlib import Path; Path('intermediate.txt').write_text('done', encoding='utf-8')"],
        cwd=work, check=True, timeout=10,
    )
    assert (Path(work) / "intermediate.txt").read_text(encoding="utf-8") == "done"
assert not Path(work).exists()
print("PASS: child finished before cleanup")
PY_CHILD