使用 SFTPSensor 的气流 Xcom 比较时间戳

如何解决使用 SFTPSensor 的气流 Xcom 比较时间戳

我需要在 sftp 中找到来自 target_dir 的新登陆文件，并且只有 id 有最新的文件。

例如，上次上传的文件是 5 分钟前，在这几分钟内，poke 将返回 False 直到有新文件进来（因为它具有最新的时间戳）。

满足条件：

1. 文件存在于 target_dir 和
2. 仅当有最近时间戳的文件时才执行

需要使用 XCom 来完成。

这是我不完整的代码（抱歉！）：

class MySensor(BaseSensorOperator):

    template_fields = ('path',)

    @apply_defaults
    def __init__(self,path,sftp_conn_id='sftp_default',*args,**kwargs): # pylint: disable=keyword-arg-before-vararg
        super(MySensor,self).__init__(*args,**kwargs) # pylint: disable=super-with-arguments
        self.path = path
        self.hook = None
        self.sftp_conn_id = sftp_conn_id

    def poke(self,context): # pylint: disable=unused-argument
        self.hook = SFTPHook(self.sftp_conn_id)
        self.log.info('Poking for %s',self.path)
        try:
            current_time = datetime.Now(tz=timezone.utc)
            current_time_str = current_time.strftime("%Y%m%d%H%M")

            prev_time_str = current_time_str - timedelta(minutes=1)

            files_list = self.hook.list_directory(self.path)
            files_desc = self.hook.describe_directory(self.path)

            self.log.info("Description: ",files_desc)
            self.log.info("List of files: ",files_list)
            self.log.info("LENGTH OF FILES {}".format(len(files_list)))

            files_time = [files_desc[file]['modify'] for file in files_list]

            self.xcom_push(key=current_time_str,value=files_time)
            prevIoUs_mod_time = self.xcom_pull(key=prev_time_str)

            check_latest_file = any(current_time_str in string for string in files_time)
            
            self.log.info(check_latest_file)

            return False
            # return len(files_list) > 0 and check_latest_file
        except IOError as e: # pylint: disable=invalid-name
            if e.errno != SFTP_NO_SUCH_FILE:
                raise e
            return False
        finally:
            self.hook.close_conn()

sftp_sensor = MySensor(task_id='my_task',path='/target_dir/',sftp_conn_id='my_conn',poke_interval=60,dag=dag)