是否可以创建一个 python 脚本在每天给定时间查找目录中的文件？

Question

所以基本上，我正在创建一个允许用户将 csv 文件放入其中的目录。但是我想创建一个 python 脚本，它会在每天的给定时间（比如说中午）查看该文件夹，如果它不超过一天，则拿起放在其中的最新文件。但我不确定这是否可能。

如果应用程序在所需目录中找到新文件，我想运行它的这段代码：

def better_Match(results, best_percent = "Will need to get the match %"):
    result = {}
    result_list = [{item.name:item.text for item in result} for result in results]
    if result_list:
        score_list = [float(item['score']) for item in result_list]
        match_index = max(enumerate(score_list),key=lambda x: x[1])[0]
        logger.debug('MRCs:{}, Chosen MRC:{}'.format(score_list,score_list[match_index]))
        logger.debug(result_list[match_index])
        above_threshold = float(result_list[match_index]['score']) >= float(best_percent)
        if above_threshold:
            result = result_list[match_index]
    return result

def clean_plate_code(platecode):
    return str(platecode).lstrip('0').zfill(5)[:5]

def re_ch(file_path, orig_data, return_columns = ['ex_opbin']):
    list_of_chunk_files = list(file_path.glob('*.csv'))
    cb_ch = [pd.read_csv(f, sep=None, dtype=object, engine='python') for f in tqdm(list_of_chunk_files, desc='Combining ch', unit='chunk')]
    cb_ch = pd.concat(cb_ch)
    shared_columns = [column_name.replace('req_','') for column_name in cb_ch.columns if column_name.startswith('req_')]
    cb_ch.columns = cb_ch.columns.str.replace("req_", "")
    return_columns = return_columns + shared_columns
    cb_ch = cb_ch[return_columns]
    for column in shared_columns:
        cb_ch[column] = cb_ch[column].astype(str)
        orig_data[column] = orig_data[column].astype(str)
    final= orig_data.merge(cb_ch, how='left', on=shared_columns)
    return final

Answer 1

在特定时间运行脚本：

您可以在 linux 上使用 cron。 在windows中你可以使用windows scheduler

这是获取目录中最新文件的示例

files = os.listdir(output_folder)
files = [os.path.join(output_folder, file) for file in files]
files = [file for file in files if os.path.isfile(file)]
latest_file = max(files, key=os.path.getctime)

Answer 2

这样就可以了！

import os
import time
import threading
import pandas as pd

DIR_PATH = 'DIR_PATH_HERE'

def create_csv_file():
    # create files.csv file that will contains all the current files
    # This will run for one time only
    if not os.path.exists('files.csv'):
        list_of_files = os.listdir(DIR_PATH )
        list_of_files.append('files.csv')
        pd.DataFrame({'files':list_of_files}).to_csv('files.csv')
    else:
        None


def check_for_new_files():
    create_csv_file()
    files = pd.read_csv('files.csv')
    list_of_files = os.listdir(DIR_PATH )
    if len(files.files) != len(list_of_files):
        print('New file added')
        #do what you want
        #save your excel with the name sample.xslx
        #append your excel into list of files and get the set so you will not have the sample.xlsx twice if run again

        list_of_files.append('sample.xslx')
        list_of_files=list(set(list_of_files))

        #save again the curent list of files
        pd.DataFrame({'files':list_of_files}).to_csv('files.csv')
        print('Finished for the day!')



ticker = threading.Event()
# Run the program every 86400 seconds = 24h
while not ticker.wait(86400):
    check_for_new_files()

它基本上使用线程每 86400 秒（即 24 小时）检查新文件，并将所有当前文件保存在 py 文件所在的目录中，并检查 csv 文件中不存在的新文件并将它们附加到每天的 files.csv 文件。