eric: comparison src/eric7/WebBrowser/Sync/FtpSyncHandler.py

-:e9e7eca7efee
+:bf71ee032bb4
 """
 Module implementing a synchronization handler using FTP.
 """
-import ftplib           # secok
+import ftplib  # secok
 import io
 import contextlib
 import pathlib
 from PyQt6.QtCore import pyqtSignal, QTimer, QCoreApplication, QByteArray
 class FtpSyncHandler(SyncHandler):
 """
 Class implementing a synchronization handler using FTP.
 @signal syncStatus(type_, message) emitted to indicate the synchronization
 status (string one of "bookmarks", "history", "passwords",
 "useragents" or "speeddial", string)
 @signal syncError(message) emitted for a general error with the error
 message (string)
 synchronization (string)
 @signal syncFinished(type_, done, download) emitted after a
 synchronization has finished (string one of "bookmarks", "history",
 "passwords", "useragents" or "speeddial", boolean, boolean)
 """
 syncStatus = pyqtSignal(str, str)
 syncError = pyqtSignal(str)
 syncMessage = pyqtSignal(str)
 syncFinished = pyqtSignal(str, bool, bool)
 def __init__(self, parent=None):
 """
 Constructor
 @param parent reference to the parent object (QObject)
 """
 super().__init__(parent)
 self.__state = "idle"
 self.__forceUpload = False
 self.__connected = False
 self.__remoteFilesFound = {}
 def initialLoadAndCheck(self, forceUpload):
 """
 Public method to do the initial check.
 @param forceUpload flag indicating a forced upload of the files
 (boolean)
 """
 if not Preferences.getWebBrowser("SyncEnabled"):
 return
 self.__state = "initializing"
 self.__forceUpload = forceUpload
 self.__dirLineParser = FtpDirLineParser()
 self.__remoteFilesFound = {}
 self.__idleTimer = QTimer(self)
 self.__idleTimer.setInterval(
-Preferences.getWebBrowser("SyncFtpIdleTimeout") * 1000)
+Preferences.getWebBrowser("SyncFtpIdleTimeout") * 1000
+)
 self.__idleTimer.timeout.connect(self.__idleTimeout)
 self.__ftp = EricFtp()
 # do proxy setup
 proxyType = (
 EricFtpProxyType.NO_PROXY
-if not Preferences.getUI("UseProxy") else
+if not Preferences.getUI("UseProxy")
-Preferences.getUI("ProxyType/Ftp")
+else Preferences.getUI("ProxyType/Ftp")
 )
 if proxyType != EricFtpProxyType.NO_PROXY:
 self.__ftp.setProxy(
 proxyType,
 Preferences.getUI("ProxyHost/Ftp"),
-Preferences.getUI("ProxyPort/Ftp"))
+Preferences.getUI("ProxyPort/Ftp"),
+)
 if proxyType != EricFtpProxyType.NON_AUTHORIZING:
 self.__ftp.setProxyAuthentication(
 Preferences.getUI("ProxyUser/Ftp"),
 Preferences.getUI("ProxyPassword/Ftp"),
-Preferences.getUI("ProxyAccount/Ftp"))
+Preferences.getUI("ProxyAccount/Ftp"),
+)
 QTimer.singleShot(0, self.__doFtpCommands)
 def __doFtpCommands(self):
 """
 Private slot executing the sequence of FTP commands.
 """
 try:
 self.__initialSync()
 self.__state = "idle"
 self.__idleTimer.start()
 except (ftplib.all_errors + (EricFtpProxyError,)) as err:
 self.syncError.emit(str(err))
 def __connectAndLogin(self):
 """
 Private method to connect to the FTP server and log in.
 @return flag indicating a successful log in (boolean)
 """
 self.__ftp.connect(
 Preferences.getWebBrowser("SyncFtpServer"),
 Preferences.getWebBrowser("SyncFtpPort"),
-timeout=5)
+timeout=5,
+)
 self.__ftp.login(
 Preferences.getWebBrowser("SyncFtpUser"),
-Preferences.getWebBrowser("SyncFtpPassword"))
+Preferences.getWebBrowser("SyncFtpPassword"),
+)
 self.__connected = True
 return True
 def __changeToStore(self):
 """
 Private slot to change to the storage directory.
 This action will create the storage path on the server, if it
 does not exist. Upon return, the current directory of the server
 is the sync directory.
 """
-storePathList = Preferences.getWebBrowser("SyncFtpPath").replace(
+storePathList = (
-"\\", "/").split("/")
+Preferences.getWebBrowser("SyncFtpPath").replace("\\", "/").split("/")
+)
 if storePathList[0] == "":
 storePathList.pop(0)
 while storePathList:
 path = storePathList[0]
 try:
 self.__ftp.mkd(path)
 self.__ftp.cwd(path)
 else:
 raise
 storePathList.pop(0)
 def __dirListCallback(self, line):
 """
 Private slot handling the receipt of directory listing lines.
 @param line the received line of the directory listing (string)
 """
 try:
 urlInfo = self.__dirLineParser.parseLine(line)
 except FtpDirLineParserError:
 # silently ignore parser errors
 urlInfo = None
 if (
-urlInfo and
+urlInfo
-urlInfo.isValid() and
+and urlInfo.isValid()
-urlInfo.isFile() and
+and urlInfo.isFile()
-urlInfo.name() in self._remoteFiles.values()
+and urlInfo.name() in self._remoteFiles.values()
 ):
 self.__remoteFilesFound[urlInfo.name()] = urlInfo.lastModified()
 QCoreApplication.processEvents()
 def __downloadFile(self, type_, fileName, timestamp):
 """
 Private method to downlaod the given file.
 @param type_ type of the synchronization event (string one
 of "bookmarks", "history", "passwords", "useragents" or
 "speeddial")
 @param fileName name of the file to be downloaded (string)
 @param timestamp time stamp in seconds of the file to be downloaded
 self.syncStatus.emit(type_, self._messages[type_]["RemoteExists"])
 buffer = io.BytesIO()
 try:
 self.__ftp.retrbinary(
 "RETR {0}".format(self._remoteFiles[type_]),
-lambda x: self.__downloadFileCallback(buffer, x))
+lambda x: self.__downloadFileCallback(buffer, x),
+)
 ok, error = self.writeFile(
-QByteArray(buffer.getvalue()), fileName, type_, timestamp)
+QByteArray(buffer.getvalue()), fileName, type_, timestamp
+)
 if not ok:
 self.syncStatus.emit(type_, error)
 self.syncFinished.emit(type_, ok, True)
 except ftplib.all_errors as err:
 self.syncStatus.emit(type_, str(err))
 self.syncFinished.emit(type_, False, True)
 def __downloadFileCallback(self, buffer, data):
 """
 Private method receiving the downloaded data.
 @param buffer reference to the buffer (io.BytesIO)
 @param data byte string to store in the buffer (bytes)
 @return number of bytes written to the buffer (integer)
 """
 res = buffer.write(data)
 QCoreApplication.processEvents()
 return res
 def __uploadFile(self, type_, fileName):
 """
 Private method to upload the given file.
 @param type_ type of the synchronization event (string one
 of "bookmarks", "history", "passwords", "useragents" or
 "speeddial")
 @param fileName name of the file to be uploaded (string)
 @return flag indicating success (boolean)
 buffer = io.BytesIO(data.data())
 try:
 self.__ftp.storbinary(
 "STOR {0}".format(self._remoteFiles[type_]),
 buffer,
-callback=lambda x: QCoreApplication.processEvents())
+callback=lambda x: QCoreApplication.processEvents(),
+)
 self.syncFinished.emit(type_, True, False)
 res = True
 except ftplib.all_errors as err:
 self.syncStatus.emit(type_, str(err))
 self.syncFinished.emit(type_, False, False)
 return res
 def __initialSyncFile(self, type_, fileName):
 """
 Private method to do the initial synchronization of the given file.
 @param type_ type of the synchronization event (string one
 of "bookmarks", "history", "passwords", "useragents" or
 "speeddial")
 @param fileName name of the file to be synchronized (string)
 """
 if (
-not self.__forceUpload and
+not self.__forceUpload
-self._remoteFiles[type_] in self.__remoteFilesFound
+and self._remoteFiles[type_] in self.__remoteFilesFound
 ):
 if (
-not pathlib.Path(fileName).exists() or
+not pathlib.Path(fileName).exists()
-pathlib.Path(fileName).stat().st_mtime <
+or pathlib.Path(fileName).stat().st_mtime
-self.__remoteFilesFound[
+< self.__remoteFilesFound[self._remoteFiles[type_].toSecsSinceEpoch()]
-self._remoteFiles[type_].toSecsSinceEpoch()]
 ):
 self.__downloadFile(
-type_, fileName,
+type_,
-self.__remoteFilesFound[self._remoteFiles[type_]]
+fileName,
-.toTime_t())
+self.__remoteFilesFound[self._remoteFiles[type_]].toTime_t(),
+)
 else:
-self.syncStatus.emit(
+self.syncStatus.emit(type_, self.tr("No synchronization required."))
-type_, self.tr("No synchronization required."))
 self.syncFinished.emit(type_, True, True)
 else:
 if self._remoteFiles[type_] not in self.__remoteFilesFound:
-self.syncStatus.emit(
+self.syncStatus.emit(type_, self._messages[type_]["RemoteMissing"])
-type_, self._messages[type_]["RemoteMissing"])
 else:
-self.syncStatus.emit(
+self.syncStatus.emit(type_, self._messages[type_]["LocalNewer"])
-type_, self._messages[type_]["LocalNewer"])
 self.__uploadFile(type_, fileName)
 def __initialSync(self):
 """
 Private slot to do the initial synchronization.
 """
 # Bookmarks
 if Preferences.getWebBrowser("SyncBookmarks"):
 self.__initialSyncFile(
-"bookmarks",
+"bookmarks", WebBrowserWindow.bookmarksManager().getFileName()
-WebBrowserWindow.bookmarksManager().getFileName())
+)
 # History
 if Preferences.getWebBrowser("SyncHistory"):
 self.__initialSyncFile(
-"history",
+"history", WebBrowserWindow.historyManager().getFileName()
-WebBrowserWindow.historyManager().getFileName())
+)
 # Passwords
 if Preferences.getWebBrowser("SyncPasswords"):
 self.__initialSyncFile(
-"passwords",
+"passwords", WebBrowserWindow.passwordManager().getFileName()
-WebBrowserWindow.passwordManager().getFileName())
+)
 # User Agent Settings
 if Preferences.getWebBrowser("SyncUserAgents"):
 self.__initialSyncFile(
-"useragents",
+"useragents", WebBrowserWindow.userAgentsManager().getFileName()
-WebBrowserWindow.userAgentsManager().getFileName())
+)
 # Speed Dial Settings
 if Preferences.getWebBrowser("SyncSpeedDial"):
 self.__initialSyncFile(
-"speeddial",
+"speeddial", WebBrowserWindow.speedDial().getFileName()
-WebBrowserWindow.speedDial().getFileName())
+)
 self.__forceUpload = False
 def __syncFile(self, type_, fileName):
 """
 Private method to synchronize the given file.
 @param type_ type of the synchronization event (string one
 of "bookmarks", "history", "passwords", "useragents" or
 "speeddial")
 @param fileName name of the file to be synchronized (string)
 """
 if self.__state == "initializing":
 return
 # use idle timeout to check, if we are still connected
 if self.__connected:
 self.__idleTimeout()
 if not self.__connected or self.__ftp.sock is None:
 ok = self.__connectAndLogin()
 if not ok:
-self.syncStatus.emit(
+self.syncStatus.emit(type_, self.tr("Cannot log in to FTP host."))
-type_, self.tr("Cannot log in to FTP host."))
 return
 # upload the changed file
 self.__state = "uploading"
 self.syncStatus.emit(type_, self._messages[type_]["Uploading"])
 if self.__uploadFile(type_, fileName):
-self.syncStatus.emit(
+self.syncStatus.emit(type_, self.tr("Synchronization finished."))
-type_, self.tr("Synchronization finished."))
 self.__state = "idle"
 def syncBookmarks(self):
 """
 Public method to synchronize the bookmarks.
 """
+self.__syncFile("bookmarks", WebBrowserWindow.bookmarksManager().getFileName())
+def syncHistory(self):
+"""
+Public method to synchronize the history.
+"""
+self.__syncFile("history", WebBrowserWindow.historyManager().getFileName())
+def syncPasswords(self):
+"""
+Public method to synchronize the passwords.
+"""
+self.__syncFile("passwords", WebBrowserWindow.passwordManager().getFileName())
+def syncUserAgents(self):
+"""
+Public method to synchronize the user agents.
+"""
 self.__syncFile(
-"bookmarks",
+"useragents", WebBrowserWindow.userAgentsManager().getFileName()
-WebBrowserWindow.bookmarksManager().getFileName())
+)
-def syncHistory(self):
-"""
-Public method to synchronize the history.
-"""
-self.__syncFile(
-"history",
-WebBrowserWindow.historyManager().getFileName())
-def syncPasswords(self):
-"""
-Public method to synchronize the passwords.
-"""
-self.__syncFile(
-"passwords",
-WebBrowserWindow.passwordManager().getFileName())
-def syncUserAgents(self):
-"""
-Public method to synchronize the user agents.
-"""
-self.__syncFile(
-"useragents",
-WebBrowserWindow.userAgentsManager().getFileName())
 def syncSpeedDial(self):
 """
 Public method to synchronize the speed dial data.
 """
-self.__syncFile(
+self.__syncFile("speeddial", WebBrowserWindow.speedDial().getFileName())
-"speeddial",
-WebBrowserWindow.speedDial().getFileName())
 def shutdown(self):
 """
 Public method to shut down the handler.
 """
 if self.__idleTimer.isActive():
 self.__idleTimer.stop()
 with contextlib.suppress(ftplib.all_errors):
 if self.__connected:
 self.__ftp.quit()
 self.__connected = False
 def __idleTimeout(self):
 """
 Private slot to prevent a disconnect from the server.
 """
 if self.__state == "idle" and self.__connected:

Mercurial Repositories > eric / file comparison

comparison: src/eric7/WebBrowser/Sync/FtpSyncHandler.py

src/eric7/WebBrowser/Sync/FtpSyncHandler.py