如何在Luigi中循环处理输出，批量保存爬虫抓取的图片

核心问题原因 #

Luigi 框架的 Task.output() 是内置调度接口，不支持传入自定义参数，框架调用该方法时也不会传递任何额外入参，因此你自定义的 image_id 形参无法生效。

方案1：单Task批量处理（适合小型爬虫场景） #

如果不需要细粒度管控单张图片的下载状态，直接在 run 方法内构造存储路径即可，可额外增加哨兵文件标记整个任务的完成状态：

import luigi
import requests
from bs4 import BeautifulSoup
from PIL import Image
import os

class DownloadImages(luigi.Task):
    def requires(self):
        pass 

    def output(self):
        # 哨兵文件，标记整个批量下载任务完成
        return luigi.LocalTarget("img/download_complete.txt")

    def run(self):
        os.makedirs("img", exist_ok=True)
        resp = requests.get("https://my-site.com")
        soup = BeautifulSoup(resp.content, "html.parser")
        images_list = soup.select("img")
        # 从1开始计数，匹配img_1、img_2的命名规则
        for image_id, img_tag in enumerate(images_list, start=1):
            image_url = img_tag["src"]
            img = Image.open(requests.get(image_url, stream=True).raw)
            save_path = f"img/img_{image_id}.jpeg"
            img.save(save_path)
        # 全部下载完成后写入哨兵文件
        with self.output().open("w") as f:
            f.write(f"成功下载{len(images_list)}张图片")

方案2：拆分单图下载子Task（符合Luigi设计规范，支持断点续存） #

如果需要单张图片下载失败后自动重试、不重复下载已成功的图片，可以把单张图片下载拆分为独立子Task，通过动态依赖实现需求：

import luigi
import requests
from bs4 import BeautifulSoup
from PIL import Image
import os

# 单张图片下载Task
class DownloadSingleImage(luigi.Task):
    image_id = luigi.IntParameter()
    image_url = luigi.Parameter()

    def output(self):
        return luigi.LocalTarget(f"img/img_{self.image_id}.jpeg")

    def run(self):
        os.makedirs("img", exist_ok=True)
        img = Image.open(requests.get(self.image_url, stream=True).raw)
        img.save(self.output().path)

# 主调度Task
class DownloadAllImages(luigi.Task):
    def requires(self):
        resp = requests.get("https://my-site.com")
        soup = BeautifulSoup(resp.content, "html.parser")
        images_list = soup.select("img")
        # 动态生成所有单图下载的子Task
        for idx, img_tag in enumerate(images_list, start=1):
            yield DownloadSingleImage(image_id=idx, image_url=img_tag["src"])

    def output(self):
        return luigi.LocalTarget("img/download_complete.txt")

    def run(self):
        total_count = len(list(self.requires()))
        with self.output().open("w") as f:
            f.write(f"全部下载完成，共{total_count}张图片")

方案选择建议 #

• 方案1代码更简洁，适合图片量小、不需要细粒度状态管控的场景
• 方案2符合Luigi的任务拆分设计理念，支持断点续传，单张下载失败不会影响已完成的任务，适合图片量较大的生产场景

内容的提问来源于stack exchange，提问作者Andrea