# weiyongfu # sftp sync import paramiko import os import stat from datetime import datetime import sys import threading from concurrent.futures import ThreadPoolExecutor, as_completed import time import hashlib import re def check_remote_complete_file(sftp, current_date): """ 检查远程complete文件中是否包含当前日期的时间戳 """ remote_complete_path = "/list/10.99.9.75_sftp/L2/complete" try: # 尝试读取远程complete文件 with sftp.open(remote_complete_path, 'r') as remote_file: content = remote_file.read().decode('utf-8') # 检查是否包含当前日期 if current_date in content: print(f"远程complete文件已包含当前日期 {current_date},开始同步") return True else: print(f"远程complete文件未包含当前日期 {current_date},等待中...") return False except FileNotFoundError: print(f"远程complete文件不存在: {remote_complete_path},等待中...") return False except Exception as e: print(f"检查远程complete文件时出错: {e},等待中...") return False def create_local_complete_file(current_date, verification_passed=True, error_messages=None): """ 在本地创建complete文件 只有在验证成功时才创建文件 """ local_complete_dir = r"E:\成功" local_complete_path = os.path.join(local_complete_dir, current_date) try: # 确保目录存在 os.makedirs(local_complete_dir, exist_ok=True) # 只有在验证通过时才创建文件 if verification_passed: # 创建文件并写入验证状态 with open(local_complete_path, 'w') as f: timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') f.write(f"COMPLETE - {timestamp} - MD5 verification PASSED\n") print(f"本地complete文件创建成功: {local_complete_path}") print(f"MD5验证状态: 通过") return True else: # 验证失败时,如果文件已存在,删除它 if os.path.exists(local_complete_path): os.remove(local_complete_path) print(f"删除失败的complete文件: {local_complete_path}") return False except Exception as e: print(f"操作本地complete文件失败: {e}") return False def wait_for_remote_complete(hostname, port, username, private_key_path, current_date): """ 等待远程complete文件包含当前日期的时间戳 """ print(f"等待远程complete文件包含日期: {current_date}") while True: ssh_client = paramiko.SSHClient() ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: # 加载私钥 private_key = load_private_key(private_key_path) if not private_key: print("私钥加载失败,1分钟后重试...") time.sleep(60) continue # 建立连接 ssh_client.connect(hostname, port, username, pkey=private_key, timeout=30) sftp = ssh_client.open_sftp() # 检查complete文件 if check_remote_complete_file(sftp, current_date): sftp.close() ssh_client.close() return True sftp.close() ssh_client.close() except Exception as e: print(f"连接检查失败: {e}") # 等待1分钟 print("等待60秒后重试...") time.sleep(60) def load_private_key(private_key_path): """ 加载私钥文件 """ try: # 首先尝试作为RSA密钥加载 return paramiko.RSAKey.from_private_key_file(private_key_path) except paramiko.ssh_exception.PasswordRequiredException: # 如果私钥有密码保护,需要输入密码 key_password = input("请输入私钥密码: ") return paramiko.RSAKey.from_private_key_file(private_key_path, password=key_password) except Exception as e1: print(f"RSA密钥加载失败: {e1}") try: # 尝试作为ECDSA密钥加载 return paramiko.ECDSAKey.from_private_key_file(private_key_path) except Exception as e2: print(f"ECDSA密钥加载失败: {e2}") try: # 尝试作为Ed25519密钥加载 return paramiko.Ed25519Key.from_private_key_file(private_key_path) except Exception as e3: print(f"所有密钥格式尝试失败: {e3}") return None def calculate_file_md5(file_path, chunk_size=8192): """ 计算文件的MD5值 """ md5_hash = hashlib.md5() try: with open(file_path, 'rb') as f: # 逐块读取文件以节省内存 for chunk in iter(lambda: f.read(chunk_size), b""): md5_hash.update(chunk) return md5_hash.hexdigest().lower() # 返回小写十六进制字符串 except Exception as e: print(f"计算文件MD5时出错 {file_path}: {e}") return None def parse_md5_file(md5_file_path): """ 解析MD5.txt文件,提取文件名和MD5值 支持多种编码格式 """ md5_dict = {} # 尝试多种编码格式 encodings = ['utf-8', 'gbk', 'gb2312', 'gb18030', 'latin1', 'iso-8859-1'] for encoding in encodings: try: with open(md5_file_path, 'r', encoding=encoding) as f: content = f.read() # 使用正则表达式提取文件名和MD5值 # 格式: MD5 哈希(文件 StockTick.7z.065): # 17 9a a5 ff f3 a4 8b 76 ee 2e 45 84 68 77 1b 8e pattern = r'MD5\s*哈希\s*\(\s*文件\s*([^)]+)\s*\)\s*:\s*([0-9a-fA-F ]+)' matches = re.findall(pattern, content, re.MULTILINE) if not matches: # 尝试另一种可能的格式 pattern2 = r'MD5\s*哈希\s*\(\s*文件\s*([^)]+)\s*\)\s*:\s*([0-9a-fA-F ]+)' matches = re.findall(pattern2, content, re.MULTILINE) if matches: for filename, md5_hex in matches: # 清理MD5值:移除空格并转换为小写 cleaned_md5 = md5_hex.replace(' ', '').lower() md5_dict[filename.strip()] = cleaned_md5 print(f"使用编码 {encoding} 从 {md5_file_path} 中解析出 {len(md5_dict)} 个MD5记录") return md5_dict except UnicodeDecodeError: continue # 尝试下一种编码 except Exception as e: print(f"使用编码 {encoding} 解析MD5文件失败 {md5_file_path}: {e}") continue # 如果所有编码都失败,尝试二进制读取 try: with open(md5_file_path, 'rb') as f: content = f.read() # 将二进制内容转换为字符串,使用errors='ignore'忽略无法解码的字符 content_str = content.decode('utf-8', errors='ignore') pattern = r'MD5\s*哈希\s*\(\s*文件\s*([^)]+)\s*\)\s*:\s*([0-9a-fA-F ]+)' matches = re.findall(pattern, content_str, re.MULTILINE) if not matches: pattern2 = r'MD5\s*哈希\s*\(\s*文件\s*([^)]+)\s*\)\s*:\s*([0-9a-fA-F ]+)' matches = re.findall(pattern2, content_str, re.MULTILINE) if matches: for filename, md5_hex in matches: cleaned_md5 = md5_hex.replace(' ', '').lower() md5_dict[filename.strip()] = cleaned_md5 print(f"使用二进制方式从 {md5_file_path} 中解析出 {len(md5_dict)} 个MD5记录") return md5_dict except Exception as e: print(f"二进制方式解析MD5文件失败 {md5_file_path}: {e}") print(f"所有编码尝试都失败,无法解析MD5文件: {md5_file_path}") return None def verify_files(target_directory, md5_dict): """ 验证目录中的文件MD5值 """ results = [] verified_count = 0 total_files = len(md5_dict) for filename, expected_md5 in md5_dict.items(): file_path = os.path.join(target_directory, filename) if not os.path.exists(file_path): error_msg = f"文件不存在: {filename}" print(f" ✗ {error_msg}") results.append(error_msg) continue # 计算实际MD5值 actual_md5 = calculate_file_md5(file_path) if actual_md5 is None: error_msg = f"MD5计算失败: {filename}" print(f" ✗ {error_msg}") results.append(error_msg) elif actual_md5 == expected_md5: print(f" ✓ {filename}: MD5验证通过") verified_count += 1 else: error_msg = f"{filename}: MD5不匹配 (期望: {expected_md5}, 实际: {actual_md5})" print(f" ✗ {error_msg}") results.append(error_msg) return results, verified_count, total_files def test_all_file_md5(filepath): """ 验证指定目录下的所有文件MD5值 返回: (是否全部验证通过, 错误消息列表) """ md5_file_path = os.path.join(filepath, "MD5.txt") target_directory = filepath print(f"\n开始验证目录: {filepath}") # 检查MD5文件是否存在 if not os.path.exists(md5_file_path): error_msg = f"MD5.txt文件不存在: {md5_file_path}" print(f" ✗ {error_msg}") return False, [error_msg] print("解析MD5文件...") try: md5_dict = parse_md5_file(md5_file_path) if not md5_dict: error_msg = f"解析MD5文件失败或文件为空: {md5_file_path}" print(f" ✗ {error_msg}") return False, [error_msg] print(f"成功解析 {len(md5_dict)} 个文件的MD5记录") except Exception as e: error_msg = f"解析MD5文件失败: {str(e)}" print(f" ✗ {error_msg}") return False, [error_msg] print("\n开始验证文件...") results, verified_count, total_files = verify_files(target_directory, md5_dict) print(f"\n验证完成: {filepath}") print(f"总文件数: {total_files}") print(f"通过验证: {verified_count}") print(f"问题文件: {len(results)}") if results: print("\n发现问题:") for msg in results: print(f" - {msg}") return False, results else: print("\n所有文件验证通过!") return True, [] def get_failed_files_from_results(filepath, results): """ 从验证结果中提取失败的文件名 """ failed_files = [] for result in results: # 从错误消息中提取文件名 # 格式可能是: "文件不存在: StockTick.7z.001" 或 "StockTick.7z.001: MD5不匹配..." if "文件不存在:" in result: filename = result.split("文件不存在: ")[1].strip() elif "MD5不匹配" in result or "MD5计算失败" in result: filename = result.split(":")[0].strip() else: continue failed_files.append(filename) return failed_files def retry_failed_files(hostname, port, username, private_key, sync_paths, date_ymd, date_md, failed_files_dict): """ 重新下载验证失败的文件 """ print(f"\n{'=' * 60}") print("开始重新下载验证失败的文件") print(f"{'=' * 60}") # 统计信息 stats = { 'total_files': 0, 'downloaded': 0, 'skipped': 0, 'failed': 0 } stats_lock = threading.Lock() # 为每个失败的文件找到对应的远程路径 all_files_to_retry = [] for path_info in sync_paths: remote_path = path_info["remote"] local_base_path = path_info["local"] name = path_info["name"] if name not in failed_files_dict or not failed_files_dict[name]: continue # 获取该路径的失败文件列表 failed_files_in_path = failed_files_dict[name] print(f"\n重新下载 {name} 的 {len(failed_files_in_path)} 个失败文件") # 为每个失败文件构建文件信息 for filename in failed_files_in_path: remote_file_path = os.path.join(remote_path, filename).replace('\\', '/') local_file_path = os.path.join(local_base_path, filename) # 获取远程文件属性 try: ssh_client = paramiko.SSHClient() ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh_client.connect(hostname, port, username, pkey=private_key, timeout=30) sftp = ssh_client.open_sftp() remote_attr = sftp.stat(remote_file_path) all_files_to_retry.append((remote_file_path, local_file_path, remote_attr)) sftp.close() ssh_client.close() except Exception as e: print(f" 获取远程文件信息失败 {filename}: {e}") if not all_files_to_retry: print("没有需要重新下载的文件") return stats stats['total_files'] = len(all_files_to_retry) print(f"\n总共需要重新下载 {stats['total_files']} 个文件") # 使用多线程重新下载文件 max_workers = 10 print(f"开始多线程重新下载 (线程数: {max_workers})...") with ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交所有下载任务 future_to_file = {} for file_info in all_files_to_retry: future = executor.submit( download_file_with_new_connection, file_info, stats, stats_lock, hostname, port, username, private_key ) future_to_file[future] = file_info # 处理完成的任务 completed = 0 for future in as_completed(future_to_file): result = future.result() completed += 1 if completed % 5 == 0: # 每完成5个文件打印一次进度 print(f" 重试进度: {completed}/{stats['total_files']}") # 打印重试统计信息 print(f"\n重试完成!") print("=" * 50) print(f"重试文件数: {stats['total_files']}") print(f"成功下载: {stats['downloaded']}") print(f"跳过文件: {stats['skipped']}") print(f"失败文件: {stats['failed']}") print("=" * 50) return stats def verify_and_retry(hostname, port, username, private_key, sync_paths, date_ymd, date_md, max_retries=10): """ 验证所有目录的MD5,并在失败时重试下载 """ print(f"\n{'=' * 60}") print(f"开始MD5验证与重试 - 日期: {date_ymd}") print(f"{'=' * 60}") all_passed = False all_error_messages = [] failed_files_dict = {} # 按目录存储失败文件 for retry_count in range(max_retries + 1): if retry_count > 0: print(f"\n{'=' * 60}") print(f"第 {retry_count} 次重试下载验证失败的文件") print(f"{'=' * 60}") # 重新下载失败的文件 retry_stats = retry_failed_files( hostname, port, username, private_key, sync_paths, date_ymd, date_md, failed_files_dict ) # 如果有文件重试失败,直接退出 if retry_stats['failed'] > 0: print(f"重试下载失败,无法继续验证") all_error_messages.append(f"第{retry_count}次重试下载失败") break # 清空之前的失败记录 all_error_messages.clear() failed_files_dict.clear() current_failed = False # 验证每个目录 for path_info in sync_paths: local_path = path_info["local"] name = path_info["name"] if not os.path.exists(local_path): error_msg = f"目录不存在: {local_path}" print(f" ✗ {error_msg}") all_error_messages.append(error_msg) current_failed = True failed_files_dict[name] = [] # 目录不存在,无法获取失败文件列表 continue # 验证该目录的MD5 passed, error_msgs = test_all_file_md5(local_path) if not passed: current_failed = True all_error_messages.extend([f"{name}: {msg}" for msg in error_msgs]) # 提取失败的文件名 failed_files = get_failed_files_from_results(local_path, error_msgs) failed_files_dict[name] = failed_files else: failed_files_dict[name] = [] # 检查是否全部通过 if not current_failed: all_passed = True print(f"\n所有MD5验证通过!") break elif retry_count < max_retries: print(f"\n第 {retry_count + 1} 次验证失败,准备重试...") time.sleep(2) # 等待2秒再进行重试 else: print(f"\n已达到最大重试次数 ({max_retries}),停止重试") return all_passed, all_error_messages def sync_remote_to_local(): # 服务器连接信息 hostname = "59.36.23.197" port = 16888 username = "boyigo" # 私钥文件路径 private_key_path = r"C:\Users\admin\Desktop\boyigo" # 获取当天日期 today = datetime.now() year = today.strftime("%Y") month = today.strftime("%m") day = today.strftime("%d") date_ymd = today.strftime("%Y%m%d") # 格式: 20251114 date_md = today.strftime("%m%d") # 格式: 1114 # 等待远程complete文件包含当前日期 if not wait_for_remote_complete(hostname, port, username, private_key_path, date_ymd): print("等待远程complete文件超时或失败") return False # 根据日期动态生成同步路径 sync_paths = [ { "remote": f"/list/10.99.9.75_sftp/L2/{date_ymd}SH", "local": fr"E:\data\SHL2\{year}\{date_ymd}", "name": "SHL2" }, { "remote": f"/list/10.99.9.75_sftp/L2/{date_ymd}SZ", "local": fr"E:\data\SZL2\{year}\{date_md}", "name": "SZL2" } ] # 统计信息 stats = { 'total_files': 0, 'downloaded': 0, 'skipped': 0, 'failed': 0 } stats_lock = threading.Lock() # 创建SSH客户端 ssh_client = paramiko.SSHClient() ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - 开始连接服务器进行同步...") print(f"主机: {hostname}:{port}, 用户: {username}") # 加载私钥文件 private_key = load_private_key(private_key_path) if not private_key: print("私钥加载失败") return False # 使用密钥建立连接 ssh_client.connect(hostname, port, username, pkey=private_key, timeout=30) sftp_main = ssh_client.open_sftp() print("连接成功,开始同步文件...") print(f"同步日期: {date_ymd}") # 收集所有需要同步的文件 all_files = [] for path_info in sync_paths: remote_path = path_info["remote"] local_base_path = path_info["local"] name = path_info["name"] print(f"\n扫描目录: {remote_path} ({name})") files_in_path = scan_remote_directory(sftp_main, remote_path, local_base_path) print(f" 找到 {len(files_in_path)} 个文件") all_files.extend(files_in_path) stats['total_files'] = len(all_files) print(f"\n总共发现 {stats['total_files']} 个文件需要同步") if stats['total_files'] == 0: print("没有文件需要同步") # 不创建complete文件 return False # 使用多线程下载文件 - 10个线程 max_workers = 10 print(f"开始多线程下载 (线程数: {max_workers})...") with ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交所有下载任务 future_to_file = {} for file_info in all_files: future = executor.submit( download_file_with_new_connection, file_info, stats, stats_lock, hostname, port, username, private_key ) future_to_file[future] = file_info # 处理完成的任务 completed = 0 for future in as_completed(future_to_file): result = future.result() completed += 1 if completed % 10 == 0: # 每完成10个文件打印一次进度 print(f" 进度: {completed}/{stats['total_files']}") # 打印统计信息 print(f"\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - 文件同步完成!") print("=" * 50) print(f"总文件数: {stats['total_files']}") print(f"成功下载: {stats['downloaded']}") print(f"跳过文件: {stats['skipped']}") print(f"失败文件: {stats['failed']}") print("=" * 50) # 检查是否有失败的文件 if stats['failed'] > 0: print(f"警告: 有 {stats['failed']} 个文件下载失败,MD5验证可能不完整") # 同步完成后进行MD5验证,并在失败时重试 md5_verification_passed, error_messages = verify_and_retry( hostname, port, username, private_key, sync_paths, date_ymd, date_md, max_retries=10 ) # 根据验证结果创建complete文件 if md5_verification_passed: print(f"\n所有MD5验证通过,创建成功标记文件") # 只在成功时创建文件 create_local_complete_file(date_ymd, verification_passed=True) return True else: print(f"\nMD5验证失败,不创建成功标记文件") # 验证失败时不创建文件 create_local_complete_file(date_ymd, verification_passed=False, error_messages=error_messages) return False except Exception as e: error_msg = f"同步过程中出现错误: {e}" print(error_msg) import traceback traceback.print_exc() # 发生异常时不创建文件 return False finally: # 关闭连接 try: sftp_main.close() except: pass try: ssh_client.close() except: pass return True def scan_remote_directory(sftp, remote_path, local_base_path): """扫描远程目录,返回所有需要同步的文件列表""" files_list = [] try: remote_items = sftp.listdir_attr(remote_path) except Exception as e: print(f" 无法访问远程目录 {remote_path}: {e}") return files_list for item in remote_items: remote_item_path = os.path.join(remote_path, item.filename).replace('\\', '/') if stat.S_ISDIR(item.st_mode): # 如果是目录,递归扫描 sub_local_path = os.path.join(local_base_path, item.filename) sub_files = scan_remote_directory(sftp, remote_item_path, sub_local_path) files_list.extend(sub_files) else: # 如果是文件,添加到列表 local_file_path = os.path.join(local_base_path, item.filename) files_list.append((remote_item_path, local_file_path, item)) return files_list def download_file(sftp, file_info, stats, stats_lock): """下载单个文件""" remote_file_path, local_file_path, remote_attr = file_info try: # 确保本地目录存在 local_dir = os.path.dirname(local_file_path) os.makedirs(local_dir, exist_ok=True) # 检查本地文件是否存在 if os.path.exists(local_file_path): local_stat = os.stat(local_file_path) local_size = local_stat.st_size local_mtime = local_stat.st_mtime # 获取远程文件修改时间 remote_mtime = remote_attr.st_mtime # 如果文件大小和修改时间都相同,则跳过 if local_size == remote_attr.st_size and abs(local_mtime - remote_mtime) < 2: with stats_lock: stats['skipped'] += 1 return "skipped" else: print(f" 更新: {os.path.basename(remote_file_path)} (大小: {local_size} -> {remote_attr.st_size})") else: print(f" 下载: {os.path.basename(remote_file_path)} (大小: {remote_attr.st_size} bytes)") # 下载文件 sftp.get(remote_file_path, local_file_path) # 验证下载的文件大小 downloaded_size = os.path.getsize(local_file_path) if downloaded_size != remote_attr.st_size: # 如果大小不匹配,删除文件并重试一次 os.remove(local_file_path) sftp.get(remote_file_path, local_file_path) downloaded_size = os.path.getsize(local_file_path) if downloaded_size != remote_attr.st_size: raise Exception(f"文件大小不匹配: 期望 {remote_attr.st_size}, 实际 {downloaded_size}") # 保持文件修改时间 os.utime(local_file_path, (remote_attr.st_atime, remote_attr.st_mtime)) with stats_lock: stats['downloaded'] += 1 return "downloaded" except Exception as e: print(f" 下载失败 {os.path.basename(remote_file_path)}: {e}") with stats_lock: stats['failed'] += 1 return "failed" def download_file_with_new_connection(file_info, stats, stats_lock, hostname, port, username, private_key): """为多线程创建的下载函数,每个线程有自己的SFTP连接""" remote_file_path, local_file_path, remote_attr = file_info # 为这个线程创建独立的SFTP连接 ssh_client = paramiko.SSHClient() ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) sftp = None try: ssh_client.connect(hostname, port, username, pkey=private_key, timeout=30) sftp = ssh_client.open_sftp() # 调用原有的下载逻辑 return download_file(sftp, file_info, stats, stats_lock) except Exception as e: print(f" 连接失败 {os.path.basename(remote_file_path)}: {e}") with stats_lock: stats['failed'] += 1 return "failed" finally: # 关闭连接 try: if sftp: sftp.close() except: pass try: ssh_client.close() except: pass def main(): """主函数""" try: print("=" * 60) print("远程文件同步工具 (10线程SFTP + 密钥认证 + MD5验证 + 失败重试)") print("=" * 60) success = sync_remote_to_local() if success: print("\n同步成功且MD5验证通过!") return 0 else: print("\n同步失败或MD5验证未通过!") return 1 except Exception as e: print(f"程序发生未捕获异常: {e}") import traceback traceback.print_exc() return 1 if __name__ == "__main__": try: exit_code = main() except KeyboardInterrupt: print("\n程序被用户中断") exit_code = 1 except Exception as e: print(f"程序发生严重错误: {e}") import traceback traceback.print_exc() exit_code = 1 # 添加延时,确保所有输出都能显示 time.sleep(1) print("\n程序执行完成,即将退出...") input("按Enter键退出...") sys.exit(exit_code)