eric: comparison src/eric7/WebBrowser/AdBlock/AdBlockSubscription.py

-:e9e7eca7efee
+:bf71ee032bb4
 import re
 import hashlib
 import base64
 from PyQt6.QtCore import (
-pyqtSignal, Qt, QObject, QByteArray, QDateTime, QUrl, QUrlQuery,
+pyqtSignal,
-QCryptographicHash, QDate, QTime
+Qt,
+QObject,
+QByteArray,
+QDateTime,
+QUrl,
+QUrlQuery,
+QCryptographicHash,
+QDate,
+QTime,
 )
 from PyQt6.QtNetwork import QNetworkReply, QNetworkRequest
 from EricWidgets import EricMessageBox
 class AdBlockSubscription(QObject):
 """
 Class implementing the AdBlock subscription.
 @signal changed() emitted after the subscription has changed
 @signal rulesChanged() emitted after the subscription's rules have changed
 @signal enabledChanged(bool) emitted after the enabled state was changed
 @signal rulesEnabledChanged() emitted after a rule enabled state was
 changed
 """
 changed = pyqtSignal()
 rulesChanged = pyqtSignal()
 enabledChanged = pyqtSignal(bool)
 rulesEnabledChanged = pyqtSignal()
 def __init__(self, url, custom, parent=None, default=False):
 """
 Constructor
 @param url AdBlock URL for the subscription (QUrl)
 @param custom flag indicating a custom subscription (boolean)
 @param parent reference to the parent object (QObject)
 @param default flag indicating a default subscription (boolean)
 """
 super().__init__(parent)
 self.__custom = custom
 self.__url = url.toEncoded()
 self.__enabled = False
 self.__downloading = None
 self.__defaultSubscription = default
 self.__title = ""
 self.__location = QByteArray()
 self.__lastUpdate = QDateTime()
 self.__requiresLocation = ""
 self.__requiresTitle = ""
-self.__updatePeriod = 0     # update period in hours, 0 = use default
+self.__updatePeriod = 0  # update period in hours, 0 = use default
 self.__remoteModified = QDateTime()
-self.__rules = []   # list containing all AdBlock rules
+self.__rules = []  # list containing all AdBlock rules
 self.__checksumRe = re.compile(
 r"""^\s*!\s*checksum[\s\-:]+([\w\+\/=]+).*\n""",
-re.IGNORECASE | re.MULTILINE)
+re.IGNORECASE | re.MULTILINE,
+)
 self.__expiresRe = re.compile(
-r"""(?:expires:|expires after)\s*(\d+)\s*(hour|h)?""",
+r"""(?:expires:|expires after)\s*(\d+)\s*(hour|h)?""", re.IGNORECASE
-re.IGNORECASE)
+)
 self.__remoteModifiedRe = re.compile(
 r"""!\s*(?:Last modified|Updated):\s*(\d{1,2})\s*"""
 r"""(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s*"""
 r"""(\d{2,4})\s*((\d{1,2}):(\d{2}))?""",
-re.IGNORECASE)
+re.IGNORECASE,
+)
 self.__monthNameToNumber = {
 "Jan": 1,
 "Feb": 2,
 "Mar": 3,
 "Apr": 4,
 "Jul": 7,
 "Aug": 8,
 "Sep": 9,
 "Oct": 10,
 "Nov": 11,
-"Dec": 12
+"Dec": 12,
 }
 self.__parseUrl(url)
 def __parseUrl(self, url):
 """
 Private method to parse the AdBlock URL for the subscription.
 @param url AdBlock URL for the subscription
 @type QUrl
 """
 if url.scheme() != "abp":
 return
 if url.path() != "subscribe":
 return
 urlQuery = QUrlQuery(url)
 self.__title = QUrl.fromPercentEncoding(
-QByteArray(urlQuery.queryItemValue("title").encode()))
+QByteArray(urlQuery.queryItemValue("title").encode())
+)
 self.__enabled = urlQuery.queryItemValue("enabled") != "false"
-self.__location = QByteArray(QUrl.fromPercentEncoding(
+self.__location = QByteArray(
-QByteArray(urlQuery.queryItemValue("location").encode()))
+QUrl.fromPercentEncoding(
-.encode("utf-8"))
+QByteArray(urlQuery.queryItemValue("location").encode())
+).encode("utf-8")
+)
 # Check for required subscription
 self.__requiresLocation = QUrl.fromPercentEncoding(
-QByteArray(urlQuery.queryItemValue(
+QByteArray(urlQuery.queryItemValue("requiresLocation").encode())
-"requiresLocation").encode()))
+)
 self.__requiresTitle = QUrl.fromPercentEncoding(
-QByteArray(urlQuery.queryItemValue("requiresTitle").encode()))
+QByteArray(urlQuery.queryItemValue("requiresTitle").encode())
+)
 if self.__requiresLocation and self.__requiresTitle:
 from WebBrowser.WebBrowserWindow import WebBrowserWindow
 WebBrowserWindow.adBlockManager().loadRequiredSubscription(
-self.__requiresLocation, self.__requiresTitle)
+self.__requiresLocation, self.__requiresTitle
+)
 lastUpdateString = urlQuery.queryItemValue("lastUpdate")
-self.__lastUpdate = QDateTime.fromString(lastUpdateString,
+self.__lastUpdate = QDateTime.fromString(
-Qt.DateFormat.ISODate)
+lastUpdateString, Qt.DateFormat.ISODate
+)
 self.__loadRules()
 def url(self):
 """
 Public method to generate the URL for this subscription.
 @return AdBlock URL for the subscription
 @rtype QUrl
 """
 url = QUrl()
 url.setScheme("abp")
 url.setPath("subscribe")
 queryItems = []
 queryItems.append(("location", bytes(self.__location).decode()))
 queryItems.append(("title", self.__title))
 if self.__requiresLocation and self.__requiresTitle:
 queryItems.append(("requiresLocation", self.__requiresLocation))
 queryItems.append(("requiresTitle", self.__requiresTitle))
 if not self.__enabled:
 queryItems.append(("enabled", "false"))
 if self.__lastUpdate.isValid():
 queryItems.append(
-("lastUpdate",
+("lastUpdate", self.__lastUpdate.toString(Qt.DateFormat.ISODate))
-self.__lastUpdate.toString(Qt.DateFormat.ISODate))
 )
 query = QUrlQuery()
 query.setQueryItems(queryItems)
 url.setQuery(query)
 return url
 def isEnabled(self):
 """
 Public method to check, if the subscription is enabled.
 @return flag indicating the enabled status
 @rtype bool
 """
 return self.__enabled
 def setEnabled(self, enabled):
 """
 Public method to set the enabled status.
 @param enabled flag indicating the enabled status
 @type bool
 """
 if self.__enabled == enabled:
 return
 self.__enabled = enabled
 self.enabledChanged.emit(enabled)
 def title(self):
 """
 Public method to get the subscription title.
 @return subscription title
 @rtype string
 """
 return self.__title
 def setTitle(self, title):
 """
 Public method to set the subscription title.
 @param title subscription title
 @type str
 """
 if self.__title == title:
 return
 self.__title = title
 self.changed.emit()
 def location(self):
 """
 Public method to get the subscription location.
 @return URL of the subscription location
 @rtype QUrl
 """
 return QUrl.fromEncoded(self.__location)
 def setLocation(self, url):
 """
 Public method to set the subscription location.
 @param url URL of the subscription location
 @type QUrl
 """
 if url == self.location():
 return
 self.__location = url.toEncoded()
 self.__lastUpdate = QDateTime()
 self.changed.emit()
 def requiresLocation(self):
 """
 Public method to get the location of a required subscription.
 @return location of a required subscription
 @rtype str
 """
 return self.__requiresLocation
 def lastUpdate(self):
 """
 Public method to get the date and time of the last update.
 @return date and time of the last update
 @rtype QDateTime
 """
 return self.__lastUpdate
 def rulesFileName(self):
 """
 Public method to get the name of the rules file.
 @return name of the rules file
 @rtype str
 """
 if self.location().scheme() == "file":
 return self.location().toLocalFile()
 if self.__location.isEmpty():
 return ""
-sha1 = bytes(QCryptographicHash.hash(
+sha1 = bytes(
-self.__location, QCryptographicHash.Algorithm.Sha1).toHex()
+QCryptographicHash.hash(
+self.__location, QCryptographicHash.Algorithm.Sha1
+).toHex()
 ).decode()
-dataDir = os.path.join(
+dataDir = os.path.join(Utilities.getConfigDir(), "web_browser", "subscriptions")
-Utilities.getConfigDir(), "web_browser", "subscriptions")
 if not os.path.exists(dataDir):
 os.makedirs(dataDir)
-fileName = os.path.join(
+fileName = os.path.join(dataDir, "adblock_subscription_{0}".format(sha1))
-dataDir, "adblock_subscription_{0}".format(sha1))
 return fileName
 def __loadRules(self):
 """
 Private method to load the rules of the subscription.
 """
 fileName = self.rulesFileName()
 header = f.readline().strip()
 if not header.startswith("[Adblock"):
 EricMessageBox.warning(
 None,
 self.tr("Load subscription rules"),
-self.tr("""AdBlock file '{0}' does not start"""
+self.tr(
-""" with [Adblock.""")
+"""AdBlock file '{0}' does not start"""
-.format(fileName))
+""" with [Adblock."""
+).format(fileName),
+)
 f.close()
 os.unlink(fileName)
 self.__lastUpdate = QDateTime()
 else:
 from .AdBlockRule import AdBlockRule
 self.__updatePeriod = 0
 self.__remoteModified = QDateTime()
 self.__rules = []
 self.__rules.append(AdBlockRule(header, self))
 for line in f.readlines():
 # hours
 self.__updatePeriod = int(period)
 else:
 # days
 self.__updatePeriod = int(period) * 24
-remoteModified = self.__remoteModifiedRe.search(
+remoteModified = self.__remoteModifiedRe.search(line)
-line)
 if remoteModified:
-day, month, year, time, hour, minute = (
+(
-remoteModified.groups()
+day,
-)
+month,
+year,
+time,
+hour,
+minute,
+) = remoteModified.groups()
 self.__remoteModified.setDate(
-QDate(int(year),
+QDate(
-self.__monthNameToNumber[month],
+int(year),
-int(day))
+self.__monthNameToNumber[month],
+int(day),
+)
 )
 if time:
 self.__remoteModified.setTime(
-QTime(int(hour), int(minute)))
+QTime(int(hour), int(minute))
+)
 else:
 # no time given, set it to 23:59
-self.__remoteModified.setTime(
+self.__remoteModified.setTime(QTime(23, 59))
-QTime(23, 59))
 self.changed.emit()
 except OSError as err:
 EricMessageBox.warning(
 None,
 self.tr("Load subscription rules"),
 self.tr(
-"""Unable to read AdBlock file '{0}'.\nReason: {1}""")
+"""Unable to read AdBlock file '{0}'.\nReason: {1}"""
-.format(fileName, str(err))
+).format(fileName, str(err)),
 )
 elif not fileName.endswith("_custom"):
 self.__lastUpdate = QDateTime()
 self.checkForUpdate()
 def checkForUpdate(self):
 """
 Public method to check for an update.
 """
 updatePeriod = (
 self.__updatePeriod
-if self.__updatePeriod else
+if self.__updatePeriod
-Preferences.getWebBrowser("AdBlockUpdatePeriod") * 24
+else Preferences.getWebBrowser("AdBlockUpdatePeriod") * 24
 )
 if (
-not self.__lastUpdate.isValid() or
+not self.__lastUpdate.isValid()
-(self.__remoteModified.isValid() and
+or (
-self.__remoteModified.addSecs(updatePeriod * 3600) <
+self.__remoteModified.isValid()
-QDateTime.currentDateTime()) or
+and self.__remoteModified.addSecs(updatePeriod * 3600)
-self.__lastUpdate.addSecs(updatePeriod * 3600) <
+< QDateTime.currentDateTime()
-QDateTime.currentDateTime()
+)
+or self.__lastUpdate.addSecs(updatePeriod * 3600)
+< QDateTime.currentDateTime()
 ):
 self.updateNow()
 def updateNow(self):
 """
 Public method to update the subscription immediately.
 """
 if self.__downloading is not None:
 return
 if not self.location().isValid():
 return
 if self.location().scheme() == "file":
 self.__lastUpdate = QDateTime.currentDateTime()
 self.__loadRules()
 return
 from WebBrowser.WebBrowserWindow import WebBrowserWindow
-reply = WebBrowserWindow.networkManager().get(
-QNetworkRequest(self.location()))
+reply = WebBrowserWindow.networkManager().get(QNetworkRequest(self.location()))
-reply.finished.connect(
+reply.finished.connect(lambda: self.__rulesDownloaded(reply))
-lambda: self.__rulesDownloaded(reply))
 self.__downloading = reply
 def __rulesDownloaded(self, reply):
 """
 Private slot to deal with the downloaded rules.
 @param reply reference to the network reply
 @type QNetworkReply
 """
 response = bytes(reply.readAll())
 reply.close()
 self.__downloading = None
 if reply.error() != QNetworkReply.NetworkError.NoError:
 if not self.__defaultSubscription:
 # don't show error if we try to load the default
 EricMessageBox.warning(
 None,
 self.tr("Downloading subscription rules"),
 self.tr(
 """<p>Subscription rules could not be"""
-""" downloaded.</p><p>Error: {0}</p>""")
+""" downloaded.</p><p>Error: {0}</p>"""
-.format(reply.errorString()))
+).format(reply.errorString()),
+)
 else:
 # reset after first download attempt
 self.__defaultSubscription = False
 return
 if not response:
 EricMessageBox.warning(
 None,
 self.tr("Downloading subscription rules"),
-self.tr("""Got empty subscription rules."""))
+self.tr("""Got empty subscription rules."""),
-return
+)
+return
 fileName = self.rulesFileName()
 try:
 with open(fileName, "wb") as f:
 from WebBrowser.WebBrowserWindow import WebBrowserWindow
 if (
-WebBrowserWindow.adBlockManager().useLimitedEasyList() and
+WebBrowserWindow.adBlockManager().useLimitedEasyList()
-self.url().toString().startswith(
+and self.url()
-WebBrowserWindow.adBlockManager()
+.toString()
-.getDefaultSubscriptionUrl())
+.startswith(
+WebBrowserWindow.adBlockManager().getDefaultSubscriptionUrl()
+)
 ):
 limited = True
 # ignore Third-party advertisers rules for performance
 # whitelist rules at the end will be used
 index = response.find(
 b"!---------------------------"
 b"Third-party advertisers"
-b"---------------------------!")
+b"---------------------------!"
+)
 part1 = response[:index]
 index = response.find(
 b"!-----------------------"
 b"Whitelists to fix broken sites"
-b"------------------------!")
+b"------------------------!"
+)
 part2 = response[index:]
 f.write(part1)
 f.write(part2)
 else:
 limited = False
 f.write(response)
 f.close()
 self.__lastUpdate = QDateTime.currentDateTime()
 if limited or self.__validateCheckSum(fileName):
 self.__loadRules()
 else:
 os.unlink(fileName)
 except OSError:
 EricMessageBox.warning(
 None,
 self.tr("Downloading subscription rules"),
-self.tr("""Unable to write to AdBlock file '{0}'.""")
+self.tr("""Unable to write to AdBlock file '{0}'.""").file(fileName),
-.file(fileName))
+)
 self.__downloading = None
 reply.deleteLater()
 def __validateCheckSum(self, fileName):
 """
 Private method to check the subscription file's checksum.
 @param fileName name of the file containing the subscription
 @type str
 @return flag indicating a valid file. A file is considered
 valid, if the checksum is OK, the file does not contain a
 checksum (i.e. cannot be checked) or we are using the limited
 try:
 with open(fileName, "r", encoding="utf-8") as f:
 data = f.read()
 except (OSError, OSError):
 return False
 match = re.search(self.__checksumRe, data)
 if match:
 expectedChecksum = match.group(1)
 else:
 # consider it as valid
 return True
 # normalize the data
-data = re.sub(r"\r", "", data)              # normalize eol
+data = re.sub(r"\r", "", data)  # normalize eol
-data = re.sub(r"\n+", "\n", data)           # remove empty lines
+data = re.sub(r"\n+", "\n", data)  # remove empty lines
 data = re.sub(self.__checksumRe, "", data)  # remove checksum line
 # calculate checksum
-md5 = hashlib.md5()             # secok
+md5 = hashlib.md5()  # secok
 md5.update(data.encode("utf-8"))
-calculatedChecksum = (
+calculatedChecksum = base64.b64encode(md5.digest()).decode().rstrip("=")
-base64.b64encode(md5.digest()).decode().rstrip("=")
-)
 if calculatedChecksum == expectedChecksum:
 return True
 else:
 res = EricMessageBox.yesNo(
 None,
 self.tr(
 """<p>AdBlock subscription <b>{0}</b> has a wrong"""
 """ checksum.<br/>"""
 """Found: {1}<br/>"""
 """Calculated: {2}<br/>"""
-"""Use it anyway?</p>""")
+"""Use it anyway?</p>"""
-.format(self.__title, expectedChecksum,
+).format(self.__title, expectedChecksum, calculatedChecksum),
-calculatedChecksum))
+)
 return res
 def saveRules(self):
 """
 Public method to save the subscription rules.
 """
 fileName = self.rulesFileName()
 if not fileName:
 return
 try:
 with open(fileName, "w", encoding="utf-8") as f:
 if not self.__rules or not self.__rules[0].isHeader():
 f.write("[Adblock Plus 2.0]\n")
 for rule in self.__rules:
 f.write(rule.filter() + "\n")
 except OSError:
 EricMessageBox.warning(
 None,
 self.tr("Saving subscription rules"),
-self.tr("""Unable to write to AdBlock file '{0}'.""")
+self.tr("""Unable to write to AdBlock file '{0}'.""").format(fileName),
-.format(fileName))
+)
 def rule(self, offset):
 """
 Public method to get a specific rule.
 @param offset offset of the rule
 @type int
 @return requested rule
 @rtype AdBlockRule
 """
 if offset >= len(self.__rules):
 return None
 return self.__rules[offset]
 def allRules(self):
 """
 Public method to get the list of rules.
 @return list of rules
 @rtype list of AdBlockRule
 """
 return self.__rules[:]
 def addRule(self, rule):
 """
 Public method to add a rule.
 @param rule reference to the rule to add
 @type AdBlockRule
 @return offset of the rule
 @rtype int
 """
 self.__rules.append(rule)
 self.rulesChanged.emit()
 return len(self.__rules) - 1
 def removeRule(self, offset):
 """
 Public method to remove a rule given the offset.
 @param offset offset of the rule to remove
 @type int
 """
 if offset < 0 or offset > len(self.__rules):
 return
 del self.__rules[offset]
 self.rulesChanged.emit()
 def replaceRule(self, rule, offset):
 """
 Public method to replace a rule given the offset.
 @param rule reference to the rule to set
 @type AdBlockRule
 @param offset offset of the rule to remove
 @type int
 @return requested rule
 @rtype AdBlockRule
 """
 if offset >= len(self.__rules):
 return None
 self.__rules[offset] = rule
 self.rulesChanged.emit()
 return self.__rules[offset]
 def canEditRules(self):
 """
 Public method to check, if rules can be edited.
 @return flag indicating rules may be edited
 @rtype bool
 """
 return self.__custom
 def canBeRemoved(self):
 """
 Public method to check, if the subscription can be removed.
 @return flag indicating removal is allowed
 @rtype bool
 """
 return not self.__custom and not self.__defaultSubscription
 def setRuleEnabled(self, offset, enabled):
 """
 Public method to enable a specific rule.
 @param offset offset of the rule
 @type int
 @param enabled new enabled state
 @type bool
 @return reference to the changed rule
 @rtype AdBlockRule
 """
 if offset >= len(self.__rules):
 return None
 rule = self.__rules[offset]
 rule.setEnabled(enabled)
 self.rulesEnabledChanged.emit()
 if rule.isCSSRule():
 from WebBrowser.WebBrowserWindow import WebBrowserWindow
 WebBrowserWindow.mainWindow().reloadUserStyleSheet()
 return rule

Mercurial Repositories > eric / file comparison

comparison: src/eric7/WebBrowser/AdBlock/AdBlockSubscription.py

src/eric7/WebBrowser/AdBlock/AdBlockSubscription.py