Luigi Python脚本依赖报错：运行时未满足依赖问题排查

让我帮你一步步解决这个Luigi任务依赖报错的问题，你的代码里有几个关键问题导致了Unfulfilled dependency错误：

1. 错误使用luigi.ExternalTask #

luigi.ExternalTask是用来标记已经存在的外部资源（比如手动生成的文件），它不会执行run()方法。但你的Generate_TV_WebScraping_File任务需要主动爬取数据生成CSV，所以应该继承普通的luigi.Task，而不是ExternalTask。

2. 方法名拼写错误（致命问题） #

两个任务里的输出方法都写错了：ouptut() → 正确的应该是output()（少了一个t）。这个错误会让Luigi无法识别任务的输出目标，进而认为依赖的文件从未生成，导致依赖检查失败。

3. 硬编码文件路径，未使用Luigi的目标管理 #

在Generate_TV_WebScraping_File的run()方法中，你直接硬编码路径打开文件，而没有通过self.output().open()来操作。这样Luigi无法跟踪文件的生成状态，无法确认任务是否完成。

4. requires()方法中参数传递错误 #

Insert_TV_WebScraping_To_Db的requires()里，你尝试传递self.input_path，但这个参数并不属于当前任务。因为Generate_TV_WebScraping_File已经定义了带默认值的input_path参数，直接实例化即可。

修正后的完整代码 #

import requests
from bs4 import BeautifulSoup
import re
import pyodbc
import luigi
import arrow

class Generate_TV_WebScraping_File(luigi.Task):  # 改为继承luigi.Task
    input_path = luigi.Parameter('X:/somefilepath/Televised_Football_Staging.csv')
    
    def output(self):  # 修正方法名拼写
        return luigi.LocalTarget(self.input_path)
    
    def run(self):
        ################################ GET DATA FROM WEBSITE ###############################################
        page_link = 'https://www.somewebsite.html'
        page_response = requests.get(page_link, timeout=5)
        page_content = BeautifulSoup(page_response.content, "html.parser")
        clean_date = ''
        
        # 使用self.output().open()来操作文件，让Luigi跟踪状态
        with self.output().open('w') as f:
            for rows in page_content.findAll('div', attrs={"class":"row-fluid"}):
                if rows.findAll('div', attrs={"class": "span12 matchdate"}):
                    date_row = rows.findAll('div', attrs={"class": "span12 matchdate"})
                    concat_rows = ",".join(str(x) for x in date_row)
                    clean_date = re.sub("<.*?>", " ", concat_rows)
                elif rows.findAll('div', attrs={"class": "span4 matchfixture"}):
                    concat_rows = ",".join(str(x) for x in rows)
                    clean_rows = re.sub("<.*?>", " ", concat_rows)
                    f.write('%s\n' % (clean_rows + "," + clean_date))
        #######################################################################################################

class Insert_TV_WebScraping_To_Db(luigi.Task):
    def requires(self):
        # 直接实例化依赖任务，不需要错误传递参数
        return Generate_TV_WebScraping_File()
    
    def output(self):  # 修正方法名拼写
        sys_date = arrow.now().format('YYYYMMDD')
        return luigi.LocalTarget('X:/somefilepath/tv_webscrape_log_' + sys_date + '.txt')
    
    def run(self):
        ############################### INSERT DATA INTO DATABASE ###################################################
        cnxn = pyodbc.connect(driver="{SQL Server}", server="someserver", database="somedatabase", autocommit=True)
        cursor = cnxn.cursor()
        cursor.execute('EXEC somedatabase.someschema.somedbproc')
        cnxn.close()
        
        # 生成输出的日志文件，让Luigi确认任务完成
        with self.output().open('w') as log_file:
            log_file.write(f"Task completed successfully at {arrow.now().format('YYYY-MM-DD HH:mm:ss')}")
        #############################################################################################################

# Run Luigi Tasks
luigi.run(main_task_cls=Insert_TV_WebScraping_To_Db)

额外补充 #

• 第二个任务的run()方法里必须生成output()定义的日志文件，否则Luigi会认为任务未完成，下次运行会重复执行。
• 使用with语句操作文件是更安全的方式，无需手动调用close()，还能避免资源泄漏。

内容的提问来源于stack exchange，提问作者DB-93