Я написал следующий код для синхронизации c моего рабочего пространства с моим FTP-сервером:
import ftplib
import time
import tkinter as tk
import sys
import os, os.path
import argparse
import threading
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class FTPFSHandler(FileSystemEventHandler):
def __init__(self, ftp, args):
self.ftp = ftp
self.args = args
def on_created(self, event):
print(f'event type: {event.event_type} path : {event.src_path}')
print(f'\033[32;1mNew File will be written to: \033[34;1m{event.src_path}\033[0m')
cur = self.ftp.pwd()
print(cur)
print(event.src_path)
if not cur == '/' + os.path.dirname(event.src_path):
self.ftp.cwd(os.path.dirname(event.src_path))
try:
self.ftp.storbinary('STOR %s' % (os.path.basename(event.src_path)), open(event.src_path, 'rb'))
except IsADirectoryError:
self.ftp.mkd('/' + event.src_path)
self.ftp.cwd(cur)
def on_deleted(self, event):
print(f'event type: {event.event_type} path : {event.src_path}')
print(f'\033[31;1mFile will be deleted at: \033[34;1m{event.src_path}\033[0m')
cur = self.ftp.pwd()
print(cur)
print(event.src_path)
try:
if not cur == '/' + os.path.dirname(event.src_path):
self.ftp.cwd(os.path.dirname(event.src_path))
self.ftp.delete(os.path.basename(event.src_path))
except ftplib.error_perm:
if not cur == '/' + '/'.join(event.src_path.split('/')[:(len(event.src_path.split('/')) - 1)]):
self.ftp.cwd('/' + '/'.join(event.src_path.split('/')
[:(len(event.src_path.split('/')) - 1)]))
self.ftp.rmd(event.src_path.split(
'/')[len(event.src_path.split('/'))-1])
self.ftp.cwd(cur)
def on_modified(self, event):
time.sleep(1)
print(f'event type: {event.event_type} path : {event.src_path}')
print(f'\033[32;1mFile will be updated at: \033[34;1m{event.src_path}\033[0m')
cur = self.ftp.pwd()
print(cur)
print(event.src_path)
try:
if not cur == '/' + os.path.dirname(event.src_path):
self.ftp.cwd(os.path.dirname(event.src_path))
self.ftp.storbinary('STOR %s' % (os.path.basename(event.src_path)), open(event.src_path, 'rb'))
except:
pass
def is_file(filename):
current = ftp.pwd()
try:
ftp.cwd(filename)
except:
ftp.cwd(current)
return True
ftp.cwd(current)
return False
class Login():
def __init__(self):
self.root = tk.Tk()
self.root.geometry("480x200+0+0")
self.create_widgets()
self.root.mainloop()
def create_widgets(self):
self.l = tk.Label(self.root, text="Log in", font=('Arial', 44))
self.l.pack()
self.l1 = tk.Label(self.root, text="FTP Username:")
self.l1.pack()
self.e = tk.Entry(self.root)
self.e.pack()
self.l2 = tk.Label(self.root, text="FTP Password:")
self.l2.pack()
self.p = tk.Entry(self.root, show="\u25CF")
self.p.pack()
self.b = tk.Button(self.root, text="Log In", command=self.logon)
self.b.pack()
def logon(self):
self.password = self.p.get()
self.uname = self.e.get()
self.root.destroy()
x = Login()
try: #Ensure That a username and Password were entered
print("Username: " + x.uname)
except:
sys.exit(1)
def createfiles(ftp, b='.', ignore=[], bdir='..'):
files = ftp.nlst()
print(files)
for f in files:
rf = b + '/' + f
if f.startswith('.') or rf in ignore:
continue
if is_file(f):
if not os.path.isfile(rf):
o = open(rf, 'wb')
print('\033[32;1mWriting file in directory: \033[34;1m' + ftp.pwd() + '\033[0m')
ftp.retrbinary('RETR %s' % f, o.write)
o.close()
else:
if not os.path.isdir(rf):
os.mkdir(rf)
ftp.cwd(f)
try:
createfiles(ftp, rf, '..')
except:
pass
try:
ftp.cwd(bdir)
except:
pass
parser = argparse.ArgumentParser()
parser.add_argument('--base', '-b', help="the working directory on the server")
parser.add_argument('--ignore', '-i', nargs='+', help="do not download the listed directories")
parser.add_argument('--host', '-s', help="the host to connect to")
args = parser.parse_args()
host = args.host if args.host else 'ftpupload.net'
def testFtp(ftp):
t = threading.currentThread()
while getattr(t, "do_run", True):
try:
ftp.nlst()
except ftplib.error_temp:
ftp.login(user=x.uname, passwd=x.password)
time.sleep(15)
with ftplib.FTP(host) as ftp:
try:
ftp.login(user=x.uname, passwd=x.password)
except Exception as e:
print('\033[31;1mLOGIN FAILED: \033[0m' + str(e))
sys.exit(1)
if args.base:
try:
os.mkdir(args.base)
except:
pass
ftp.cwd(args.base)
if args.ignore:
createfiles(ftp, args.base, args.ignore)
else:
createfiles(ftp, args.base)
else:
if args.ignore:
createfiles(ftp, args.base, args.ignore)
else:
createfiles(ftp)
event_handler = FTPFSHandler(ftp, args)
observer = Observer()
path_ = args.base if args.base else '.'
observer.schedule(event_handler, path_, recursive=True)
observer.start()
try:
tester = threading.Thread(target=testFtp, args=(ftp,))
tester.start()
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
tester.do_run = False
print('\033[35;1mShutting down\033[0m: Please Wait up to 15 Seconds.')
observer.join()
tester.join()
После того, как я нажал кнопку «Войти» в окне, скрипт работает отлично, регистрирует вывод и делает все, что я хочу, даже закрывает окно, но значок запуска python просто сидит там, " Не отвечает ". Как я могу закрыть это окно, а затем продолжить выполнение моего кода без иконки Python Launcher, сидящей там и говорящей мне, что она "не отвечает" ????
Thx заранее!