#!/usr/bin/env python3 # -*- coding: utf-8 -*- import os import sys import paramiko import argparse from datetime import datetime, timedelta import time import warnings # 忽略paramiko的警告 warnings.filterwarnings("ignore", category=DeprecationWarning, module="paramiko") def parse_arguments(): """解析命令行参数""" parser = argparse.ArgumentParser(description='SFTP数据同步工具') parser.add_argument('-d', '--date', help='指定日期 (格式: YYYYMMDD), 例如: 20250205', type=str, default=None) parser.add_argument('-y', '--yesterday', help='下载昨天的数据', action='store_true') parser.add_argument('-t', '--test', help='测试模式，不实际传输文件', action='store_true') parser.add_argument('-l', '--list', help='仅列出远程文件，不传输', action='store_true') return parser.parse_args() def validate_date(date_str): """验证日期格式是否有效""" try: # 尝试将字符串解析为日期 datetime.strptime(date_str, '%Y%m%d') return True except ValueError: return False def get_target_date(args): """获取目标日期""" if args.date: # 使用命令行指定的日期 if validate_date(args.date): return args.date else: print(f"错误: 日期格式不正确: {args.date}") print("请使用 YYYYMMDD 格式，例如: 20250205") sys.exit(1) elif args.yesterday: # 使用昨天的日期 yesterday = datetime.now() - timedelta(days=1) return yesterday.strftime('%Y%m%d') else: # 默认使用今天 return datetime.now().strftime('%Y%m%d') def sftp_transfer(target_date, test_mode=False, list_only=False): """ SFTP文件传输函数 参数: target_date: 目标日期 (格式: YYYYMMDD) test_mode: 测试模式，不实际传输文件 list_only: 仅列出文件，不传输 """ print(f"目标日期: {target_date}") # sftp服务器 host = '220.203.233.97' port = 16888 username = 'sysop' password = 'swjg_Boyizc1212' # 远程路径 - 更新为v4 remote_base_path = f'/home/sysop/kr_md_save_centos7_9_v4/data/stockdata/{target_date}' # 本地路径 local_base_path = f'/mnt/data1/data_new/data/stockdata/{target_date}' # 如果本地目录不存在，则创建（如果不是仅列出模式） if not list_only and not test_mode: if not os.path.exists(local_base_path): os.makedirs(local_base_path, exist_ok=True) print(f"创建本地目录: {local_base_path}") # 连接SFTP服务器 try: transport = paramiko.Transport((host, port)) transport.connect(username=username, password=password) sftp = paramiko.SFTPClient.from_transport(transport) print(f"成功连接到SFTP服务器: {host}:{port}") print(f"远程目录: {remote_base_path}") print(f"本地目录: {local_base_path}") # 列出远程目录中的所有文件 try: remote_files = sftp.listdir(remote_base_path) except FileNotFoundError: print(f"错误: 远程目录不存在: {remote_base_path}") print("可能原因:") print(" 1) 日期格式错误") print(" 2) 远程服务器上该日期的数据还未生成") print(" 3) 远程服务器上没有该日期的数据") sftp.close() transport.close() return False, 0, 0, [] if not remote_files: print("警告: 远程目录为空，没有文件需要传输") sftp.close() transport.close() return True, 0, 0, [] print(f"找到 {len(remote_files)} 个文件需要传输") # 按文件名排序，便于查看 remote_files.sort() print("远程文件列表:") for i, filename in enumerate(remote_files, 1): print(f" {i:3d}. {filename}") # 如果是仅列出模式，直接返回 if list_only: print(f"\n仅列出模式，不进行文件传输") sftp.close() transport.close() return True, 0, len(remote_files), [] # 如果是测试模式，模拟传输 if test_mode: print(f"\n测试模式，模拟传输 {len(remote_files)} 个文件") print("实际传输将被跳过") transferred_files = remote_files sftp.close() transport.close() return True, len(remote_files), len(remote_files), transferred_files # 实际传输文件 print(f"\n开始传输文件...") transferred_count = 0 transferred_files = [] failed_files = [] for filename in remote_files: # 修复路径拼接问题，确保使用正确的路径分隔符 remote_file_path = f"{remote_base_path}/{filename}" local_file_path = os.path.join(local_base_path, filename) try: print(f"[{transferred_count + 1}/{len(remote_files)}] 传输: {filename}") # 检查文件是否存在且可读 try: file_stat = sftp.stat(remote_file_path) print(f" 远程文件大小: {file_stat.st_size:,} 字节") except Exception as e: print(f" 警告: 无法获取文件状态: {str(e)}") # 传输文件 sftp.get(remote_file_path, local_file_path) # 验证本地文件是否成功创建 if os.path.exists(local_file_path): file_size = os.path.getsize(local_file_path) transferred_count += 1 transferred_files.append(filename) print(f" 完成 - 本地文件大小: {file_size:,} 字节") else: failed_files.append(filename) print(f" 失败: 本地文件未创建") except Exception as e: failed_files.append(filename) print(f" 传输失败: {str(e)}") print(f"\n传输完成: 成功传输 {transferred_count}/{len(remote_files)} 个文件") # 输出详细结果 if transferred_files: print(f"成功传输的文件 ({len(transferred_files)}):") for i, f in enumerate(transferred_files, 1): print(f" {i:3d}. ✓ {f}") if failed_files: print(f"传输失败的文件 ({len(failed_files)}):") for i, f in enumerate(failed_files, 1): print(f" {i:3d}. ✗ {f}") # 关闭连接 sftp.close() transport.close() return transferred_count > 0, transferred_count, len(remote_files), transferred_files except Exception as e: print(f"连接或传输过程中发生错误: {str(e)}") import traceback traceback.print_exc() return False, 0, 0, [] def create_success_marker(date_str, success=True, transferred=0, total=0, elapsed_time=0, transferred_files=None): """ 在本地创建成功标记文件 只有在传输成功时才创建文件 """ # 成功标记目录 success_dir = r"/mnt/data1/data_new/data/success" marker_file = os.path.join(success_dir, date_str) try: # 确保目录存在 os.makedirs(success_dir, exist_ok=True) # 只有在传输成功且传输文件数大于0时才创建文件 if success and transferred > 0: # 创建文件并写入传输信息 with open(marker_file, 'w', encoding='utf-8') as f: timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') content = f"传输成功 - {timestamp}\n" content += f"传输日期: {date_str}\n" content += f"传输文件数: {transferred}/{total}\n" content += f"传输耗时: {elapsed_time:.2f}秒\n" content += f"传输速率: {total / elapsed_time:.2f} 文件/秒\n" if elapsed_time > 0 else "" content += "=" * 50 + "\n" # 添加传输的文件列表（如果有） if transferred_files and len(transferred_files) > 0: content += f"成功传输的文件列表 ({len(transferred_files)}):\n" for i, file_name in enumerate(transferred_files, 1): content += f" {i:3d}. {file_name}\n" else: content += "文件列表: 无\n" content += "=" * 50 + "\n" content += "任务完成时间: " + timestamp + "\n" f.write(content) print(f"\n成功标记文件创建成功: {marker_file}") print(f"传输状态: 成功，传输 {transferred}/{total} 个文件") return True else: # 传输失败时，如果文件已存在，删除它 if os.path.exists(marker_file): os.remove(marker_file) print(f"删除失败的标记文件: {marker_file}") # 根据失败原因给出不同的提示 if not success: print("传输失败: 连接或传输过程中出现错误") elif transferred == 0: print("传输失败: 没有成功传输任何文件") else: print("传输失败: 未知原因") return False except Exception as e: print(f"操作成功标记文件失败: {e}") return False def main(): """主函数""" # 解析命令行参数 args = parse_arguments() print("=" * 70) print("SFTP数据同步工具") print("=" * 70) # 获取目标日期 target_date = get_target_date(args) # 显示操作模式 mode_info = [] if args.test: mode_info.append("测试模式") if args.list: mode_info.append("仅列出模式") if args.yesterday: mode_info.append("下载昨天数据") elif args.date: mode_info.append(f"指定日期: {args.date}") else: mode_info.append("下载今天数据") if mode_info: print(f"模式: {' | '.join(mode_info)}") print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"目标日期: {target_date}") print("-" * 70) # 记录开始时间 start_time = time.time() # 执行传输任务 success, transferred_count, total_count, transferred_files = sftp_transfer( target_date, test_mode=args.test, list_only=args.list ) # 记录结束时间 end_time = time.time() elapsed_time = end_time - start_time # 如果不是仅列出模式，创建成功标记文件 if not args.list: marker_created = create_success_marker( target_date, success, transferred_count, total_count, elapsed_time, transferred_files ) print("-" * 70) # 输出总结 if args.list: print(f"列表完成! 找到 {total_count} 个文件，耗时: {elapsed_time:.2f}秒") elif args.test: print(f"测试完成! 模拟传输 {total_count} 个文件，耗时: {elapsed_time:.2f}秒") elif success and transferred_count > 0: print(f"任务执行成功! 传输 {transferred_count}/{total_count} 个文件，耗时: {elapsed_time:.2f}秒") print(f"平均传输速率: {transferred_count / elapsed_time:.2f} 文件/秒") else: print( f"任务执行失败或没有文件传输! 成功: {success}, 传输文件数: {transferred_count}, 耗时: {elapsed_time:.2f}秒") print("=" * 70) # 返回退出代码 if args.list or args.test: return 0 # 列表和测试模式总是返回成功 else: return 0 if success and transferred_count > 0 else 1 if __name__ == '__main__': sys.exit(main())