<think>我们使用watchdog来监控文件夹的变化。我们将监控一级子文件夹（纯数字文件夹）的创建和修改事件。当事件发生时，我们检查该数字文件夹下的新二级子文件夹（即新出现的文件夹），然后对每个新出现的二级子文件夹执行以下操作： 1. 修改TEST.py中的target_path为这个二级子文件夹的路径。 2. 运行TEST.py。 注意：我们只关心一级子文件夹（纯数字）的创建和修改事件。在修改事件中，我们通过记录每个一级子文件夹的最后修改时间来判断是否真的有更新（避免重复触发）。 同时，我们需要记录每个一级子文件夹下已经处理过的二级子文件夹，以避免重复处理。 步骤： 1. 初始化时，记录所有已有的一级数字子文件夹及其最后修改时间，并记录这些一级子文件夹下已有的二级子文件夹（标记为已处理）。 2. 当有新的数字文件夹创建时，将其加入监控，并处理该文件夹下所有的二级子文件夹（因为都是新的）。 3. 当已有的数字文件夹被修改时，检查其最后修改时间是否更新，如果更新则检查该文件夹下新增的二级子文件夹（对比之前记录的状态），然后处理这些新增的二级子文件夹。 我们将使用一个字典来存储一级数字子文件夹的状态：路径 -> (最后修改时间, 已处理的二级子文件夹集合) 代码结构： - 使用FileSystemEventHandler来处理事件（on_created和on_modified）。 - 在初始化时，扫描WATCH_DIR下所有一级子文件夹，如果是纯数字的，则记录其状态。 - 在事件处理中，对于创建事件，初始化该文件夹的状态，并处理其下所有二级子文件夹。 - 对于修改事件，检查时间戳变化，然后找出新增的二级子文件夹进行处理。 修改TEST.py的方式：我们通过读取TEST.py文件内容，用正则表达式替换target_path的值，然后写入一个临时文件（避免在修改时被其他进程使用），再执行临时文件。 注意：由于TEST.py可能正在被使用（比如之前运行未结束），我们不要直接修改原文件，而是生成一个临时副本执行。执行完成后可以删除临时文件（可选）。 我们假设TEST.py中有一行类似：target_path = "原路径" 的代码，我们将替换这个路径。 具体实现如下： </think> 我将基于您提供的代码逻辑进行修改，实现监控数字子文件夹变化并处理新二级子文件夹的需求。以下是完整的解决方案： ```python import os import time import re import subprocess import warnings warnings.simplefilter("ignore") # 忽略所有警告 # 配置参数 WATCH_DIR = r"D:/Users/x00708/Nano/2026" # 监控的主目录 TEST_SCRIPT = r"D:\Users\x00708\Nano\TEST.py" # 要修改和执行的脚本 PYTHON_EXE = r"D:\Users\x00708\Nano\.venv\Scripts\python.exe" # Python解释器路径 # 存储状态的数据结构 folder_states = {} # {一级数字文件夹路径: (最后修改时间, 已处理的二级子文件夹集合)} last_check_time = time.time() # 上次检查时间 def is_digit_folder(path): """检查是否为纯数字名称的文件夹""" return os.path.isdir(path) and os.path.basename(path).isdigit() def scan_initial_folders(): """初始化扫描所有数字子文件夹的状态""" global folder_states folder_states = {} for entry in os.listdir(WATCH_DIR): entry_path = os.path.join(WATCH_DIR, entry) if is_digit_folder(entry_path): mtime = os.path.getmtime(entry_path) processed_subfolders = set() # 扫描并记录所有现有的二级子文件夹 for sub_entry in os.listdir(entry_path): sub_path = os.path.join(entry_path, sub_entry) if os.path.isdir(sub_path): processed_subfolders.add(sub_path) folder_states[entry_path] = (mtime, processed_subfolders) print(f"初始化完成，监控 {len(folder_states)} 个数字文件夹") def find_new_subfolders(primary_folder): """查找primary_folder下所有新的二级子文件夹""" current_subfolders = set() new_subfolders = [] # 获取当前所有的二级子文件夹 for name in os.listdir(primary_folder): full_path = os.path.join(primary_folder, name) if os.path.isdir(full_path): current_subfolders.add(full_path) # 获取已处理记录 last_mtime, processed_set = folder_states.get(primary_folder, (0, set())) # 找出之前未处理过的子文件夹 new_subfolders = list(current_subfolders - processed_set) # 更新状态 current_mtime = os.path.getmtime(primary_folder) folder_states[primary_folder] = (current_mtime, current_subfolders) return new_subfolders def modify_and_run_script(target_path): """修改TEST.py中的target_path并执行""" try: # 读取原始脚本内容 try: with open(TEST_SCRIPT, 'r', encoding='utf-8') as f: content = f.read() except UnicodeDecodeError: with open(TEST_SCRIPT, 'r', encoding='gbk') as f: content = f.read() # 替换target_path - 使用正则表达式匹配不同格式 new_content = re.sub( r'(target_path\s*=\s*[ur]?[\'"]).*?([\'"])', rf'\g<1>{target_path}\g<2>', content ) # 创建临时脚本路径 temp_script = os.path.join(os.path.dirname(TEST_SCRIPT), "TEMP_" + os.path.basename(TEST_SCRIPT)) # 写入临时脚本 with open(temp_script, 'w', encoding='utf-8') as f: f.write(new_content) print(f"正在处理路径: {target_path}") # 执行临时脚本 result = subprocess.run( [PYTHON_EXE, temp_script], capture_output=True, text=True, check=True ) print(f"脚本执行成功:\n{result.stdout}") if result.stderr: print(f"警告信息:\n{result.stderr}") # 删除临时脚本 os.remove(temp_script) return True except subprocess.CalledProcessError as e: print(f"脚本执行失败:\n{e.stderr}") return False except Exception as e: print(f"处理脚本时出错: {e}") return False def check_folder_changes(): """检查文件夹变化并处理""" global folder_states, last_check_time current_time = time.time() print(f"\n检查文件夹变化 ({time.strftime('%Y-%m-%d %H:%M:%S')})...") changes_detected = False # 1. 检查是否有新的一级数字文件夹 current_entries = set(os.listdir(WATCH_DIR)) for entry in current_entries: entry_path = os.path.join(WATCH_DIR, entry) # 如果是数字文件夹且不在当前状态中 if is_digit_folder(entry_path) and entry_path not in folder_states: print(f"检测到新数字文件夹: {entry_path}") mtime = os.path.getmtime(entry_path) processed_subfolders = set() folder_states[entry_path] = (mtime, processed_subfolders) changes_detected = True # 处理该文件夹下的所有二级子文件夹 new_subfolders = find_new_subfolders(entry_path) for subfolder in new_subfolders: modify_and_run_script(subfolder) # 2. 检查已有数字文件夹的更新 for folder_path, (last_mtime, processed_set) in list(folder_states.items()): # 确保文件夹仍然存在 if not os.path.exists(folder_path): print(f"文件夹已删除: {folder_path}") del folder_states[folder_path] continue current_mtime = os.path.getmtime(folder_path) # 检查是否发生变化 if current_mtime > last_mtime: print(f"检测到文件夹更新: {folder_path}") changes_detected = True # 处理新增的二级子文件夹 new_subfolders = find_new_subfolders(folder_path) for subfolder in new_subfolders: modify_and_run_script(subfolder) if not changes_detected: print("未检测到变化") last_check_time = current_time def main(): # 确保监控目录存在 if not os.path.exists(WATCH_DIR): os.makedirs(WATCH_DIR) print(f"创建了监控目录: {WATCH_DIR}") # 初始化状态 scan_initial_folders() print(f"开始监控文件夹: {WATCH_DIR}") print(f"监控规则: 仅处理纯数字名称的一级子文件夹") print(f"检测到变化时将修改并执行: {TEST_SCRIPT}") print("按 Ctrl+C 停止监控...") try: while True: check_folder_changes() time.sleep(60) # 每分钟检查一次 except KeyboardInterrupt: print("\n监控已停止") if __name__ == "__main__": main() ``` ### 代码详细解释： 1. **核心功能**： - 定期扫描监控目录（每分钟一次） - 检测新创建的数字文件夹（一级子文件夹） - 检测已有数字文件夹的更新 - 处理数字文件夹下的新二级子文件夹 2. **状态管理**： - `folder_states` 字典： - 键：一级数字文件夹路径（如 "D:/Users/x00708/Nano/2026/202603"） - 值：元组 `(最后修改时间, 已处理的二级子文件夹集合)` - `last_check_time`：记录上次检查时间 3. **关键函数**： - `is_digit_folder()`：验证是否为纯数字文件夹 - `scan_initial_folders()`：初始化时扫描所有现有数字文件夹 - `find_new_subfolders()`：找出未处理过的二级子文件夹 - `modify_and_run_script()`： - 修改目标脚本中的路径 - 创建并执行临时脚本 - 清理临时文件 4. **变化检测流程**： 1. 检查是否有新的一级数字文件夹 2. 检查已有数字文件夹是否更新（通过修改时间） 3. 对于每个变化的数字文件夹，找出新二级子文件夹 4. 为每个新二级子文件夹修改并执行脚本 5. **健壮性设计**： - 处理文件编码问题（UTF-8 和 GBK） - 自动创建监控目录（如果不存在） - 捕获并显示脚本执行结果 - 优雅处理键盘中断（Ctrl+C） - 自动清理临时文件 ### 使用说明： 1. **配置参数**： - `WATCH_DIR`：监控的主目录 - `TEST_SCRIPT`：要修改和执行的Python脚本 - `PYTHON_EXE`：Python解释器路径 2. **TEST.py 要求**： - 必须包含 `target_path` 变量 - 示例格式： ```python # TEST.py target_path = "默认路径" # 将被自动替换 def main(): print(f"处理路径: {target_path}") # 您的处理逻辑... if __name__ == "__main__": main() ``` 3. **运行程序**： ```bash python folder_monitor.py ``` 4. **测试方法**： - 在监控目录下创建数字文件夹（如 "202603"） - 在该文件夹下创建二级子文件夹（如 "PDE202603050272"） - 程序将自动检测变化并执行处理 ### 注意事项： 1. 程序每分钟检查一次变化，可根据需要调整 `time.sleep(60)` 2. 确保Python解释器路径正确 3. 程序会创建临时脚本文件，执行后自动删除 4. 只处理纯数字名称的一级子文件夹 5. 二级子文件夹不需要特定命名格式