Python 批量导出任务怎么做过期清理：状态文件、临时结果和查询边界

报表导出刚上线时，接口往往只做一件事：查数据、写CSV、返回下载地址。跑个几天，问题就陆续冒出来：用户重复点击生成多份没用的结果、后台任务中断留下半截不完整的文件、旧文件没人清理，磁盘慢慢被一堆没业务价值的导出占满。

比较稳妥的实现思路，是把单次导出当成一个有完整生命周期的任务，而不是写完就丢的一次性文件操作。下面用Python标准库搭一套最小可用方案：每个任务分配独立文件夹存放状态和结果，临时文件先落盘写完再转正，查询接口只认就绪状态的结果，清理器只会回收超出保留时长的已结束任务。

先把导出结果拆成独立的小型任务档案

所有导出文件全塞在同一个exports/存储路径下，很快就会乱掉。这里按任务ID单独建文件夹：元数据里记录任务状态、创建时间和用户可见的文件名，最终结果文件只有生成完成后才会出现。文件夹本身就是单条任务的边界，后续的查询、下载、清理操作都围绕这个边界展开。

work/export-jobs/
└── 8f639b7a-6f59-4a9a-8d13-1ed6e4d5ee24/
    ├── meta.json
    ├── writing.lock
    └── result.csv

创建任务：只接收可映射到业务查询的合法参数

导出请求不应该直接接收任意拼接的SQL字符串，也不应该让调用方自定义存储路径。请求侧传入合规的报表类型和日期范围，服务层把这些参数校验规范化后写入任务元数据。下面的示例代码省略了Web框架相关逻辑，核心规则可以直接放到现有的Flask、Django项目或者本地定时脚本里使用。

from __future__ import annotations

import json
import uuid
from datetime import UTC, datetime
from pathlib import Path

ROOT = Path("work/export-jobs")
ALLOWED_REPORTS = {"orders", "refunds"}

def now_text() -> str:
    return datetime.now(UTC).isoformat()

def write_json(path: Path, data: dict) -> None:
    temp = path.with_suffix(path.suffix + ".part")
    temp.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
    temp.replace(path)

def create_job(report: str, start: str, end: str) -> str:
    if report not in ALLOWED_REPORTS:
        raise ValueError("不支持的报表类型")
    if start > end:
        raise ValueError("开始日期不能晚于结束日期")

    job_id = str(uuid.uuid4())
    job_dir = ROOT / job_id
    job_dir.mkdir(parents=True, exist_ok=False)
    (job_dir / "writing.lock").write_text(now_text(), encoding="utf-8")
    write_json(job_dir / "meta.json", {
        "id": job_id,
        "report": report,
        "start": start,
        "end": end,
        "status": "queued",
        "created_at": now_text(),
        "updated_at": now_text()
    })
    return job_id

这部分看似只是几行简单的文件操作，其实定好了两个关键规则：任务ID不会由前端传入生成，存储根路径也不会随请求动态变化。后续做任务查询时，只需要验证ID格式是否合法，直接在固定根路径下查找即可，完全避免用户传入恶意参数拼接出路径遍历漏洞。

写结果：半成品始终保留在临时文件名下

写CSV最容易踩的坑是先把状态改成完成，结果文件还在写入中途就断了。更稳妥的执行顺序是：先把状态标为writing，往result.csv.part写全部内容，文件完全关闭后再把临时文件重命名为正式名，最后才把元数据状态更新为ready。同一文件夹内的重命名操作是原子的，完全避免查询端看到正式文件名却读到残缺内容的问题。

import csv

def generate_csv(job_id: str, rows: list[dict[str, str]]) -> None:
    job_dir = ROOT / job_id
    meta_path = job_dir / "meta.json"
    meta = json.loads(meta_path.read_text(encoding="utf-8"))
    meta["status"] = "writing"
    meta["updated_at"] = now_text()
    write_json(meta_path, meta)

    part = job_dir / "result.csv.part"
    final = job_dir / "result.csv"
    try:
        with part.open("w", newline="", encoding="utf-8") as fp:
            writer = csv.DictWriter(fp, fieldnames=["order_id", "amount"])
            writer.writeheader()
            writer.writerows(rows)
        part.replace(final)
        meta["status"] = "ready"
        meta["rows"] = len(rows)
    except OSError as err:
        meta["status"] = "failed"
        meta["error"] = f"文件写入失败：{err.__class__.__name__}"
        part.unlink(missing_ok=True)
    finally:
        meta["updated_at"] = now_text()
        write_json(meta_path, meta)
        (job_dir / "writing.lock").unlink(missing_ok=True)

不要觉得示例里数据来自rows就忽略了数据查询的边界校验。实际项目里这部分要调用提前封装好的受控查询函数：报表类型对应固定的字段列表和权限过滤规则，传入的时间范围用参数化查询拼接。导出接口绝对不能做成绕过权限、直接扫全表数据的后门。

查询与下载：状态和文件必须同时校验通过

结果查询不能只看meta.json的状态字段，也不能只判断文件存不存在，两个条件同时满足才允许返回下载：任务状态必须是ready，同时result.csv是可读的普通文件。缺任意一项都按未完成或异常处理，避免边缘逻辑分支把旧的无效文件误发给用户。

def get_ready_file(job_id: str) -> Path | None:
    try:
        uuid.UUID(job_id)
    except ValueError:
        return None

    job_dir = ROOT / job_id
    meta_path = job_dir / "meta.json"
    result = job_dir / "result.csv"
    if not meta_path.is_file() or not result.is_file():
        return None

    meta = json.loads(meta_path.read_text(encoding="utf-8"))
    if meta.get("status") != "ready":
        return None
    return result

Web层拿到这个合法的文件路径后，再做一遍当前用户和任务归属的权限校验，再调用框架自带的下载响应接口返回文件。不要把除了job_id之外的内部路径参数暴露到浏览器地址栏；返回给用户的真实文件名可以用报表类型加日期范围动态生成，不要暴露内部存储的路径结构规则。

过期清理：先跳过还在运行写入的任务

写清理逻辑最忌讳的操作是扫一遍存储路径，直接按创建时间批量删文件夹。运行时长很长的慢报表可能还在后台跑，机器重启后也可能留下过期的旧锁文件。下面这套规则比较保守安全：只识别格式合法的UUID任务文件夹；只清理状态为ready或者failed的文件夹；如果writing.lock的修改时间还在宽限期内，直接跳过该任务；所有校验通过后才删除整个任务文件夹。

import shutil
from datetime import timedelta

def cleanup_jobs(retention_hours: int = 24, writing_grace_minutes: int = 30) -> list[str]:
    removed: list[str] = []
    now = datetime.now(UTC)
    cutoff = now - timedelta(hours=retention_hours)
    grace = now - timedelta(minutes=writing_grace_minutes)

    for job_dir in ROOT.iterdir():
        if not job_dir.is_dir():
            continue
        try:
            uuid.UUID(job_dir.name)
            meta = json.loads((job_dir / "meta.json").read_text(encoding="utf-8"))
            updated = datetime.fromisoformat(meta["updated_at"])
        except (ValueError, KeyError, OSError, json.JSONDecodeError):
            continue

        lock = job_dir / "writing.lock"
        if lock.exists() and datetime.fromtimestamp(lock.stat().st_mtime, UTC) > grace:
            continue
        if meta.get("status") not in {"ready", "failed"} or updated > cutoff:
            continue

        shutil.rmtree(job_dir)
        removed.append(job_dir.name)
    return removed

文件保留时长不要在代码里写死。财务导出、客服导出、临时数据分析文件的保留要求完全不同，建议把配置项放到服务配置中心，每次执行清理都记录本次处理的任务数、删除失败的路径和耗时。刚上线的前几天可以先开启模拟执行，只输出待清理的候选任务，人工核对没问题再打开真实删除逻辑，这个阶段很容易发现之前没注意到的时间字段、时区处理不一致的问题。

上线前完成这几项校验

1. 手动让导出生成逻辑抛出OSError，确认任务状态正常变为failed，文件夹里不会生成正式的结果文件。
2. 在文件写入的中途去查询该任务，确认下载入口不会返回result.csv文件。
3. 手动把一条状态为ready的任务文件夹更新时间改成超出保留时长，确认清理器只会删掉这个测试文件夹，不会误删其他任务。
4. 给正在写入的任务文件夹留一个更新时间很近的writing.lock文件，确认清理逻辑会直接跳过该文件夹。

这套最小实现没有覆盖大文件流式写入、对象存储对接、分布式锁和异步队列调度的场景，但它把最容易混在一起的三个核心判断拆解开了：结果文件是否完整、任务是否允许下载、文件是否符合回收条件。先把这三个判断逻辑做扎实，后续要换成Celery异步任务、对接对象存储或者把状态迁移到数据库里，都很容易平滑过渡。

相关问题

为什么不能直接按文件修改时间清理？

文件修改时间只能说明这个文件最近被操作过，完全没法判断对应任务是不是已经执行完成。状态文件加写入标记的组合，才能准确区分可以回收的历史结果，和还在生成中的未完成任务。

任务状态一定要存在JSON文件里吗？

不一定。单机部署、任务量级小的场景，用本地JSON存状态非常直观好维护；多进程或者多机器分布式部署的场景，更适合把状态存在数据库或者Redis里。不管存在哪里，返回结果前都要同时核对状态和对应文件是否真实存在。

单条任务清理失败会不会影响整个清理流程？

某个文件夹删除失败时，直接记录下任务ID和对应异常信息，继续处理其他待清理任务就可以。不要因为单个被系统占用的文件卡住，导致整轮回收逻辑完全停止运行。

导出结果文件可以长期保留吗？

可以，但要把长期归档文件和临时下载文件分开处理。长期留存的导出文件需要配套完整的业务归属、权限校验和审计规则；临时生成的导出文件只适合在短保留期内供用户下载。