353 lines
12 KiB
Python
353 lines
12 KiB
Python
#!/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()) |