Python 多线程检查 HTTP 状态码并删除 403 文件

批量检查 data/*.json 里的图片 URL，删除所有返回 403 的 JSON 文件。

单线程版本

import json
from pathlib import Path

import requests

for json_file in Path("data").glob("*.json"):
    try:
        data = json.loads(json_file.read_text(encoding="utf-8"))
        if not data:
            continue

        url = data[0].get("imgHttpUrl")
        if not url:
            continue

        r = requests.get(url, stream=True, timeout=10)
        if r.status_code == 403:
            print(f"删除 {json_file}")
            json_file.unlink()

    except Exception as e:
        print(json_file, e)

只检查数组第一个元素的 URL，够快。若一个文件里所有 URL 都要检查，把 data[0] 改成循环即可。

多线程版本（100 并发）

import json
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

import requests

DATA_DIR = Path("data")
MAX_WORKERS = 100

session = requests.Session()


def check_file(json_file: Path):
    try:
        data = json.loads(json_file.read_text(encoding="utf-8"))
        if not data:
            return

        url = data[0].get("imgHttpUrl")
        if not url:
            return

        # 优先用 HEAD，减少带宽
        r = session.head(url, timeout=10, allow_redirects=True)

        # 部分服务器不支持 HEAD，降级到 GET
        if r.status_code == 405:
            r = session.get(url, stream=True, timeout=10)

        if r.status_code == 403:
            print(f"[删除] {json_file}")
            json_file.unlink()

    except Exception as e:
        print(f"[错误] {json_file}: {e}")


files = list(DATA_DIR.glob("*.json"))

with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
    futures = [executor.submit(check_file, f) for f in files]
    for _ in as_completed(futures):
        pass

print("完成")

requests.Session 在多线程中是线程安全的，复用 TCP 连接比每次创建新 session 快。

并发数选择

文件数	建议 workers
< 1000	50
1000~10000	100
> 10000	考虑 aiohttp + asyncio

HTTP 请求是 IO 密集型，100 线程一般没问题。若服务端有限流，调小 MAX_WORKERS。

aiohttp 异步版（超大规模）

文件数超过 10 万时，aiohttp 比线程池快几倍：

import asyncio
import json
from pathlib import Path

import aiohttp
import aiofiles

DATA_DIR = Path("data")
CONCURRENCY = 500


async def check_file(session, json_file: Path):
    try:
        async with aiofiles.open(json_file, encoding="utf-8") as f:
            content = await f.read()
        data = json.loads(content)
        if not data:
            return

        url = data[0].get("imgHttpUrl")
        if not url:
            return

        async with session.head(url, allow_redirects=True, timeout=aiohttp.ClientTimeout(total=10)) as r:
            if r.status == 403:
                print(f"[删除] {json_file}")
                json_file.unlink()

    except Exception as e:
        print(f"[错误] {json_file}: {e}")


async def main():
    files = list(DATA_DIR.glob("*.json"))
    sem = asyncio.Semaphore(CONCURRENCY)

    async def bounded(f):
        async with sem:
            await check_file(session, f)

    async with aiohttp.ClientSession() as session:
        await asyncio.gather(*(bounded(f) for f in files))

asyncio.run(main())

按 answer 目录过滤删除（补充场景）

如果是 downloads/ 里没有对应 answer/ 文件就删除，可以改为本地比对：

from pathlib import Path

downloads_dir = Path("downloads")
answer_dir = Path("answer")

# answer 目录中所有文件名（不带后缀）
answer_stems = {f.stem for f in answer_dir.iterdir() if f.is_file()}

for file in downloads_dir.iterdir():
    if not file.is_file():
        continue
    if file.stem not in answer_stems:
        print("删除", file)
        file.unlink()

不需要网络请求，纯本地文件对比，速度极快。