|
pip install pysftp paramiko
import os
import pysftp
from pathlib import Path
import getpass
class SFTPUploader:
def __init__(self):
self.sftp = None
def connect(self, host, username, password=None, port=22):
"""
連接到 SFTP 伺服器
參數:
host (str): 伺服器地址
username (str): 使用者名稱
password (str): 密碼 (如果為 None 會提示輸入)
port (int): 端口號 (預設 22)
"""
if password is None:
password = getpass.getpass(f"請輸入 {username}@{host} 的密碼: ")
# 設定連線選項 (關閉主機密鑰檢查,生產環境應設定正確的密鑰)
cnopts = pysftp.CnOpts()
cnopts.hostkeys = None # 僅用於測試,生產環境應設定正確的 hostkeys
try:
self.sftp = pysftp.Connection(
host=host,
username=username,
password=password,
port=port,
cnopts=cnopts
)
print(f"成功連接到 {host}")
return True
except Exception as e:
print(f"連接失敗: {str(e)}")
return False
def upload_file(self, local_path, remote_path=None):
"""
上傳文件到 SFTP 伺服器
參數:
local_path (str): 本地文件路徑
remote_path (str): 遠程文件路徑 (如果為 None 會使用本地文件名)
返回:
bool: 上傳是否成功
"""
if not self.sftp:
print("尚未建立 SFTP 連接")
return False
if not Path(local_path).is_file():
print(f"本地文件不存在: {local_path}")
return False
try:
if remote_path is None:
remote_path = os.path.basename(local_path)
# 確保遠程目錄存在
remote_dir = os.path.dirname(remote_path)
if remote_dir:
self._ensure_remote_directory(remote_dir)
# 上傳文件
self.sftp.put(local_path, remote_path)
print(f"成功上傳: {local_path} -> {remote_path}")
return True
except Exception as e:
print(f"上傳失敗: {str(e)}")
return False
def _ensure_remote_directory(self, remote_dir):
"""確保遠程目錄存在"""
try:
self.sftp.chdir(remote_dir)
except IOError:
# 目錄不存在,需要創建
parts = remote_dir.split('/')
current = ''
for part in parts:
if not part:
continue
current += '/' + part
try:
self.sftp.chdir(current)
except IOError:
print(f"創建目錄: {current}")
self.sftp.mkdir(current)
self.sftp.chdir(current)
def disconnect(self):
"""關閉 SFTP 連接"""
if self.sftp:
self.sftp.close()
print("已斷開 SFTP 連接")
def main():
print("=== SFTP 圖片上傳程式 ===")
# 獲取連接資訊
host = input("SFTP 伺服器地址 (例如: sftp.example.com): ").strip()
username = input("使用者名稱: ").strip()
port = input("端口號 (預設 22): ").strip()
port = int(port) if port.isdigit() else 22
# 建立上傳器實例
uploader = SFTPUploader()
# 連接伺服器
if not uploader.connect(host, username, port=port):
return
try:
# 上傳循環
while True:
print("\n" + "="*40)
local_path = input("請輸入要上傳的圖片路徑 (或輸入 'quit' 退出): ").strip()
if local_path.lower() in ('quit', 'exit', 'q'):
break
remote_path = input("輸入遠程路徑 (留空使用本地文件名): ").strip() or None
# 上傳文件
uploader.upload_file(local_path, remote_path)
finally:
# 確保連接被關閉
uploader.disconnect()
if __name__ == "__main__":
main()
|
本帖子中包含更多資源
您需要 登錄 才可以下載或查看,沒有帳號?立即註冊
x
|