자동으로 sftp다운로드 함수를 10분 간격으로 실행하고, 버튼으로 자동 실행을 On/Off 관리하는 데는 Python GUI 라이브러리 tkinter를 활용하여 구현하였습니다.
아래는 tkinter를 활용하여 10분 간격으로 해당 함수를 실행하고, 간단한 UI에서 On/Off 버튼으로 자동 실행을 제어할 수 있는 예제 코드입니다.
import tkinter as tk
from tkinter import messagebox
import threading
import time
from datetime import datetime
# --- 필요한 모듈 및 함수 가져오기 ---
import paramiko
import os
def create_sftp_client(host, port, username, password):
"""SFTP 클라이언트를 생성하고 반환"""
try:
transport = paramiko.Transport((host, port))
transport.connect(username=username, password=password)
return paramiko.SFTPClient.from_transport(transport), transport
except Exception as e:
print(f"SFTP 연결에 실패했습니다: {e}")
return None, None
def sftp_download_directory(host, port, username, password, remote_directory, local_directory):
"""원격 디렉토리의 모든 파일을 다운로드"""
sftp, transport = create_sftp_client(host, port, username, password)
if not sftp or not transport:
return
try:
today_date = datetime.now().strftime("%Y%m%d")
target_date = today_date[2:]
remote_files = sftp.listdir(remote_directory)
if not os.path.exists(local_directory):
os.makedirs(local_directory)
for file_name in remote_files:
if target_date not in file_name:
print(f"스킵: 파일 '{file_name}'은 당일 날짜에 해당되지 않습니다.")
continue
remote_file_path = os.path.join(remote_directory, file_name)
local_file_path = os.path.join(local_directory, file_name)
if os.path.exists(local_file_path):
print(f"스킵: 중복 파일 '{file_name}'이(가) 이미 로컬에 존재합니다.")
continue
print(f"다운로드 중: {remote_file_path} -> {local_file_path}")
sftp.get(remote_file_path, local_file_path)
print("모든 파일을 성공적으로 다운로드했습니다!")
except Exception as e:
print(f"파일 다운로드 중 오류 발생: {e}")
finally:
if sftp:
sftp.close()
if transport:
transport.close()
# --- 전역 변수 ---
running = False # 자동 실행 상태 (On/Off 제어)
thread = None # 백그라운드 스레드
# --- 자동 실행 함수 ---
def auto_download():
global running
while running:
print("파일 다운로드 작업 실행 중...")
# SFTP 다운로드 호출
host = "xxx.xxx.xx.xxx"
port = 22
username = "user"
password = "password"
remote_directory = "/TEST/db/"
local_directory = "D:/SFTP/db/"
sftp_download_directory(host, port, username, password, remote_directory, local_directory)
print("10분 후 다시 실행됩니다...")
time.sleep(600) # 10분 대기 (600초)
# --- 버튼 제어 함수 ---
def start_auto_download():
global running, thread
if not running:
running = True
thread = threading.Thread(target=auto_download, daemon=True)
thread.start()
print("자동 실행이 시작되었습니다!")
def stop_auto_download():
global running
if running:
running = False
print("자동 실행이 중지되었습니다!")
# --- 종료 처리 ---
def on_closing():
if running:
stop_auto_download()
root.destroy()
# --- UI 구성 ---
root = tk.Tk()
root.title("SFTP 자동 다운로드")
root.geometry("300x150")
# 설명 텍스트
label = tk.Label(root, text="10분 간격으로 SFTP 파일 다운로드 실행", wraplength=250)
label.pack(pady=10)
# 버튼 구성
start_button = tk.Button(root, text="자동 실행 시작", command=start_auto_download)
start_button.pack(pady=5)
stop_button = tk.Button(root, text="자동 실행 중지", command=stop_auto_download)
stop_button.pack(pady=5)
# 종료 이벤트 처리
root.protocol("WM_DELETE_WINDOW", on_closing)
# UI 실행
root.mainloop()
#1. 코드 설명
1-1. 자동 실행 로직
- auto_download() 함수는 while running: 루프 안에서 주어진 작업을 10분 간격 (time.sleep(600))으로 반복 실행합니다. 상태 변수 running으로 자동 실행의 활성화/비활성화 상태를 관리합니다.
1-2. 스레드를 사용한 비동기 처리
- threading.Thread를 사용하여 다운로드 작업이 백그라운드에서 비동기로 수행되도록 설정합니다.
UI 응답성을 유지하면서 다운로드 작업을 백그라운드에서 처리합니다.
1-3. UI 구성
"자동 실행 시작" 버튼: 다운로드 반복 작업 시작
"자동 실행 중지" 버튼: 반복 작업 종료
각각의 버튼은 start_auto_download()와 stop_auto_download()에 연결되어 자동 실행 기능을 제어합니다.
1-4. 프로그램 종료 처리
사용자가 프로그램 창을 닫으면 on_closing() 함수가 호출되어, 실행 중인 스레드를 안전하게 종료합니다.
#2. 기능 확장 방안
2-1. SFTP 연결 정보 입력 기능
- host, port, username, password 등을 입력할 필드를 추가하여 동적으로 설정할 수 있도록 변경.
2-2. 프로세스 로그 표시
- 현재 다운로드 중인 파일과 작업 로그를 UI의 텍스트 박스에 출력
'Python' 카테고리의 다른 글
[Python] Python에서 SFTP 구현하기 (0) | 2025.03.12 |
---|