#!/usr/bin/env python # -*- coding: utf-8 -*- """文件与目录操作模块""" import os import json import tkinter as tk from tkinter import messagebox, filedialog, scrolledtext from .logging_ui import add_to_log from .ui_widgets import center_window from app.config.settings import ConfigManager def select_file(log_widget, file_types=None, title="选择文件"): """通用文件选择对话框""" if file_types is None: file_types = [("所有文件", "*.*")] file_path = filedialog.askopenfilename(title=title, filetypes=file_types) if file_path: add_to_log(log_widget, f"已选择文件: {file_path}\n", "info") return file_path def select_excel_file(log_widget): """选择Excel文件""" return select_file( log_widget, [("Excel文件", "*.xlsx *.xls"), ("所有文件", "*.*")], "选择Excel文件" ) def ensure_directories(): """确保必要的目录结构存在""" config = ConfigManager() directories = [ config.get_path('Paths', 'input_folder', fallback='data/input', create=True), config.get_path('Paths', 'output_folder', fallback='data/output', create=True), config.get_path('Paths', 'result_folder', fallback='data/result', create=True), config.get_path('Paths', 'temp_folder', fallback='data/temp', create=True), os.path.join(config.app_root, 'logs') ] for directory in directories: if not os.path.exists(directory): os.makedirs(directory, exist_ok=True) print(f"创建目录: {directory}") def clean_cache(log_widget): """清除处理缓存""" from .command_runner import set_running_task try: config = ConfigManager() processed_record = config.get_path('Paths', 'processed_record', fallback='data/processed_files.json') output_folder = config.get_path('Paths', 'output_folder', fallback='data/output') cache_files = [ processed_record, os.path.join(output_folder, "processed_files.json"), os.path.join(output_folder, "merged_files.json") ] for cache_file in cache_files: if os.path.exists(cache_file): os.remove(cache_file) add_to_log(log_widget, f"已清除缓存文件: {cache_file}\n", "success") temp_dir = config.get_path('Paths', 'temp_folder', fallback='data/temp') if os.path.exists(temp_dir): for file in os.listdir(temp_dir): file_path = os.path.join(temp_dir, file) try: if os.path.isfile(file_path): os.remove(file_path) add_to_log(log_widget, f"已清除临时文件: {file_path}\n", "info") except Exception as e: add_to_log(log_widget, f"清除文件时出错: {file_path}, 错误: {str(e)}\n", "error") log_dir = os.path.join(config.app_root, 'logs') if os.path.exists(log_dir): for file in os.listdir(log_dir): if file.endswith(".active"): file_path = os.path.join(log_dir, file) try: os.remove(file_path) add_to_log(log_widget, f"已清除活动日志标记: {file_path}\n", "info") except Exception as e: add_to_log(log_widget, f"清除文件时出错: {file_path}, 错误: {str(e)}\n", "error") set_running_task(None) add_to_log(log_widget, "缓存清除完成，系统将重新处理所有文件\n", "success") messagebox.showinfo("缓存清除", "缓存已清除，系统将重新处理所有文件。") except Exception as e: add_to_log(log_widget, f"清除缓存时出错: {str(e)}\n", "error") messagebox.showerror("错误", f"清除缓存时出错: {str(e)}") def open_result_directory(): try: config = ConfigManager() result_dir = config.get_path('Paths', 'result_folder', fallback='data/result', create=True) os.startfile(result_dir) except Exception as e: messagebox.showerror("错误", f"无法打开结果目录: {str(e)}") def _open_directory_from_settings(config_key, default_path, label): """通用的从配置读取路径并打开目录""" try: config = ConfigManager() path = config.get_path('Paths', config_key, fallback=default_path, create=True) os.startfile(path) except Exception as e: messagebox.showerror("错误", f"无法打开{label}: {str(e)}") def open_input_directory_from_settings(): _open_directory_from_settings('input_folder', 'data/input', '输入目录') def open_output_directory_from_settings(): _open_directory_from_settings('output_folder', 'data/output', '输出目录') def open_result_directory_from_settings(): _open_directory_from_settings('result_folder', 'data/result', '结果目录') def clean_data_files(log_widget): """清理数据文件（仅清理input和output目录）""" try: if not messagebox.askyesno("确认清理", "确定要清理input和output目录的文件吗？这将删除所有输入和输出数据。"): add_to_log(log_widget, "操作已取消\n", "info") return config = ConfigManager() files_cleaned = 0 input_dir = config.get_path('Paths', 'input_folder', fallback='data/input') if os.path.exists(input_dir): for file in os.listdir(input_dir): file_path = os.path.join(input_dir, file) if os.path.isfile(file_path): os.remove(file_path) files_cleaned += 1 add_to_log(log_widget, "已清理input目录\n", "info") output_dir = config.get_path('Paths', 'output_folder', fallback='data/output') if os.path.exists(output_dir): for file in os.listdir(output_dir): file_path = os.path.join(output_dir, file) if os.path.isfile(file_path): os.remove(file_path) files_cleaned += 1 add_to_log(log_widget, "已清理output目录\n", "info") add_to_log(log_widget, f"清理完成，共清理 {files_cleaned} 个文件\n", "success") messagebox.showinfo("清理完成", f"已成功清理 {files_cleaned} 个文件") except Exception as e: add_to_log(log_widget, f"清理数据文件时出错: {str(e)}\n", "error") messagebox.showerror("错误", f"清理数据文件时出错: {str(e)}") def clean_result_files(log_widget): try: if not messagebox.askyesno("确认清理", "确定要清理result目录的文件吗？这将删除所有已生成的采购单文件。"): add_to_log(log_widget, "操作已取消\n", "info") return config = ConfigManager() count = 0 result_dir = config.get_path('Paths', 'result_folder', fallback='data/result') if os.path.exists(result_dir): for file in os.listdir(result_dir): file_path = os.path.join(result_dir, file) if os.path.isfile(file_path): os.remove(file_path) count += 1 add_to_log(log_widget, f"已清理result目录，共 {count} 个文件\n", "success") messagebox.showinfo("清理完成", f"已清理result目录 {count} 个文件") except Exception as e: add_to_log(log_widget, f"清理result目录时出错: {str(e)}\n", "error") messagebox.showerror("错误", f"清理result目录时出错: {str(e)}") def validate_unit_price_against_item_data(result_path: str, log_widget=None): try: from app.services.order_service import OrderService service = OrderService() bad_results = service.validate_unit_price(result_path) if bad_results: display_count = min(len(bad_results), 10) msg = f"存在{len(bad_results)}条单价与商品资料进货价差异超过1元:\n" + "\n".join(bad_results[:display_count]) if len(bad_results) > 10: msg += f"\n...(其余 {len(bad_results) - 10} 条已省略)" messagebox.showwarning("单价校验提示", msg) if log_widget is not None: add_to_log(log_widget, f"单价校验发现{len(bad_results)}条差异>1元\n", "warning") else: if log_widget is not None: add_to_log(log_widget, "单价校验通过（差异<=1元）\n", "success") except Exception as e: if log_widget is not None: add_to_log(log_widget, f"单价校验出错: {str(e)}\n", "error")