121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
#weiyongfu
|
|
#sftp sync test
|
|
|
|
import paramiko
|
|
import os
|
|
from datetime import datetime
|
|
import time
|
|
|
|
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 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')
|
|
print(f"远程complete文件内容: {content}")
|
|
# 检查是否包含当前日期
|
|
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 test_complete_and_date():
|
|
"""
|
|
测试远程complete文件内容和获取当前日期
|
|
"""
|
|
# 服务器连接信息
|
|
hostname = "59.36.23.197"
|
|
port = 16888
|
|
username = "boyigo"
|
|
private_key_path = r"C:\Users\admin\Desktop\运维\SFTP\boyigo"
|
|
|
|
# 获取当前日期
|
|
today = datetime.now()
|
|
date_ymd = today.strftime("%Y%m%d")
|
|
|
|
print("=" * 50)
|
|
print("测试开始")
|
|
print("=" * 50)
|
|
print(f"系统当前时间: {today}")
|
|
print(f"获取的日期字符串: {date_ymd}")
|
|
print(f"日期格式验证: YYYY={date_ymd[0:4]}, MM={date_ymd[4:6]}, DD={date_ymd[6:8]}")
|
|
print("=" * 50)
|
|
|
|
# 建立连接
|
|
ssh_client = paramiko.SSHClient()
|
|
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
|
|
try:
|
|
print("正在连接服务器...")
|
|
private_key = load_private_key(private_key_path)
|
|
if not private_key:
|
|
print("私钥加载失败")
|
|
return
|
|
|
|
ssh_client.connect(hostname, port, username, pkey=private_key, timeout=30)
|
|
sftp = ssh_client.open_sftp()
|
|
print("连接成功!")
|
|
|
|
# 检查complete文件
|
|
print("\n检查远程complete文件...")
|
|
result = check_remote_complete_file(sftp, date_ymd)
|
|
|
|
if result:
|
|
print(f"✓ 远程complete文件包含当前日期 {date_ymd}")
|
|
else:
|
|
print(f"✗ 远程complete文件不包含当前日期 {date_ymd}")
|
|
|
|
sftp.close()
|
|
ssh_client.close()
|
|
|
|
except Exception as e:
|
|
print(f"连接或检查失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
try:
|
|
ssh_client.close()
|
|
except:
|
|
pass
|
|
|
|
if __name__ == "__main__":
|
|
test_complete_and_date()
|
|
print("\n测试完成")
|
|
input("按Enter键退出...") |