383 lines
14 KiB
Python
383 lines
14 KiB
Python
#weiyongfu
|
||
#sftp sync
|
||
#多线程，可计划任务完成webhook通知
|
||
|
||
import paramiko
|
||
import os
|
||
import stat
|
||
from datetime import datetime
|
||
import sys
|
||
import threading
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
import requests
|
||
import json
|
||
import time
|
||
|
||
def send_webhook_notification(stats, status, error_message=None):
|
||
"""
|
||
发送Webhook通知 - 企业微信机器人版本
|
||
"""
|
||
# 您的企业微信机器人Webhook URL
|
||
webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=eb38eaac-c2a2-40a0-b64e-9a78edc67ebf"
|
||
|
||
# 构建企业微信机器人要求的消息格式
|
||
if status == "success":
|
||
content = f"✅ 文件同步成功\n\n" \
|
||
f" 同步统计:\n" \
|
||
f"• 总文件数: {stats['total_files']}\n" \
|
||
f"• 成功下载: {stats['downloaded']}\n" \
|
||
f"• 跳过文件: {stats['skipped']}\n" \
|
||
f"• 失败文件: {stats['failed']}\n" \
|
||
f"• 完成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
||
elif status == "failed":
|
||
content = f"❌ 文件同步失败\n\n" \
|
||
f" 同步统计:\n" \
|
||
f"• 总文件数: {stats['total_files']}\n" \
|
||
f"• 成功下载: {stats['downloaded']}\n" \
|
||
f"• 跳过文件: {stats['skipped']}\n" \
|
||
f"• 失败文件: {stats['failed']}\n" \
|
||
f"• 错误信息: {error_message}\n" \
|
||
f"• 失败时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
||
elif status == "no_files":
|
||
content = f" 没有文件需要同步\n\n" \
|
||
f"• 检查时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
||
else:
|
||
content = f"文件同步状态: {status}\n" \
|
||
f"总文件数: {stats['total_files']}\n" \
|
||
f"完成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
||
|
||
# 企业微信机器人要求的消息格式
|
||
notification_data = {
|
||
"msgtype": "text",
|
||
"text": {
|
||
"content": content
|
||
}
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
webhook_url,
|
||
json=notification_data,
|
||
headers={'Content-Type': 'application/json'},
|
||
timeout=10
|
||
)
|
||
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get('errcode') == 0:
|
||
print("企业微信通知发送成功")
|
||
else:
|
||
print(f"企业微信通知发送失败，错误码: {result.get('errcode')}, 错误信息: {result.get('errmsg')}")
|
||
else:
|
||
print(f"企业微信通知发送失败，HTTP状态码: {response.status_code}")
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"企业微信通知发送错误: {e}")
|
||
|
||
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
|
||
|
||
# 根据日期动态生成同步路径
|
||
sync_paths = [
|
||
{
|
||
"remote": f"/list/10.99.9.75_sftp/L2/{date_ymd}SH",
|
||
"local": fr"E:\data\SHL2\{year}\{date_ymd}"
|
||
},
|
||
{
|
||
"remote": f"/list/10.99.9.75_sftp/L2/{date_ymd}SZ",
|
||
"local": fr"E:\data\SZL2\{year}\{date_md}"
|
||
}
|
||
]
|
||
|
||
# 统计信息
|
||
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}")
|
||
|
||
# 尝试加载私钥文件
|
||
try:
|
||
# 首先尝试作为RSA密钥加载
|
||
private_key = paramiko.RSAKey.from_private_key_file(private_key_path)
|
||
print(f"使用RSA私钥文件: {private_key_path}")
|
||
except paramiko.ssh_exception.PasswordRequiredException:
|
||
# 如果私钥有密码保护，需要输入密码
|
||
key_password = input("请输入私钥密码: ")
|
||
private_key = paramiko.RSAKey.from_private_key_file(private_key_path, password=key_password)
|
||
except Exception as e1:
|
||
print(f"RSA密钥加载失败: {e1}")
|
||
try:
|
||
# 尝试作为ECDSA密钥加载
|
||
private_key = paramiko.ECDSAKey.from_private_key_file(private_key_path)
|
||
print(f"使用ECDSA私钥文件: {private_key_path}")
|
||
except Exception as e2:
|
||
print(f"ECDSA密钥加载失败: {e2}")
|
||
try:
|
||
# 尝试作为Ed25519密钥加载
|
||
private_key = paramiko.Ed25519Key.from_private_key_file(private_key_path)
|
||
print(f"使用Ed25519私钥文件: {private_key_path}")
|
||
except Exception as e3:
|
||
print(f"所有密钥格式尝试失败: {e3}")
|
||
print("请检查私钥文件路径和格式")
|
||
# 发送失败通知
|
||
send_webhook_notification(stats, "failed", f"密钥加载失败: {e3}")
|
||
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"]
|
||
|
||
print(f"\n扫描目录: {remote_path}")
|
||
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("没有文件需要同步")
|
||
# 发送无文件通知
|
||
send_webhook_notification(stats, "no_files")
|
||
return True
|
||
|
||
# 使用多线程下载文件 - 5个线程
|
||
max_workers = 5
|
||
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)
|
||
|
||
# 发送成功通知
|
||
send_webhook_notification(stats, "success")
|
||
|
||
except Exception as e:
|
||
error_msg = f"同步过程中出现错误: {e}"
|
||
print(error_msg)
|
||
import traceback
|
||
traceback.print_exc()
|
||
# 发送失败通知
|
||
send_webhook_notification(stats, "failed", error_msg)
|
||
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("远程文件同步工具 (5线程SFTP + 密钥认证 + Webhook通知)")
|
||
print("=" * 60)
|
||
|
||
success = sync_remote_to_local()
|
||
|
||
if success:
|
||
print("\n同步成功!")
|
||
return 0
|
||
else:
|
||
print("\n同步失败!")
|
||
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) |