diff -r e9e7eca7efee -r bf71ee032bb4 src/eric7/EricNetwork/EricGoogleMail.py
--- a/src/eric7/EricNetwork/EricGoogleMail.py Wed Jul 13 11:16:20 2022 +0200
+++ b/src/eric7/EricNetwork/EricGoogleMail.py Wed Jul 13 14:55:47 2022 +0200
@@ -25,7 +25,10 @@
import Globals
from .EricGoogleMailHelpers import (
- CLIENT_SECRET_FILE, SCOPES, TOKEN_FILE, APPLICATION_NAME
+ CLIENT_SECRET_FILE,
+ SCOPES,
+ TOKEN_FILE,
+ APPLICATION_NAME,
)
@@ -33,46 +36,47 @@
"""
Class implementing a simple web browser to perform the OAuth2
authentication process.
-
+
@signal approvalCodeReceived(str) emitted to indicate the receipt of the
approval code
"""
+
approvalCodeReceived = pyqtSignal(str)
-
+
def __init__(self, parent=None):
"""
Constructor
-
+
@param parent reference to the parent widget
@type QWidget
"""
super().__init__(parent)
-
+
self.__layout = QVBoxLayout(self)
-
+
from PyQt6.QtWebEngineWidgets import QWebEngineView
+
self.__browser = QWebEngineView(self)
self.__browser.titleChanged.connect(self.__titleChanged)
self.__browser.loadFinished.connect(self.__pageLoadFinished)
self.__layout.addWidget(self.__browser)
-
- self.__buttonBox = QDialogButtonBox(
- QDialogButtonBox.StandardButton.Close)
+
+ self.__buttonBox = QDialogButtonBox(QDialogButtonBox.StandardButton.Close)
self.__buttonBox.rejected.connect(self.reject)
self.__layout.addWidget(self.__buttonBox)
-
+
self.resize(600, 700)
-
+
@pyqtSlot(str)
def __titleChanged(self, title):
"""
Private slot handling changes of the web page title.
-
+
@param title web page title
@type str
"""
self.setWindowTitle(title)
-
+
@pyqtSlot()
def __pageLoadFinished(self):
"""
@@ -80,18 +84,20 @@
"""
url = self.__browser.url()
if url.toString().startswith(
- "https://accounts.google.com/o/oauth2/approval/v2"):
+ "https://accounts.google.com/o/oauth2/approval/v2"
+ ):
urlQuery = QUrlQuery(url)
approvalCode = urlQuery.queryItemValue(
- "approvalCode", QUrl.ComponentFormattingOption.FullyDecoded)
+ "approvalCode", QUrl.ComponentFormattingOption.FullyDecoded
+ )
if approvalCode:
self.approvalCodeReceived.emit(approvalCode)
self.close()
-
+
def load(self, url):
"""
Public method to start the authorization flow by loading the given URL.
-
+
@param url URL to be laoded
@type str or QUrl
"""
@@ -101,46 +107,47 @@
class EricGoogleMail(QObject):
"""
Class implementing the logic to send emails via Google Mail.
-
+
@signal sendResult(bool, str) emitted to indicate the transmission result
and a result message
"""
+
sendResult = pyqtSignal(bool, str)
-
+
def __init__(self, parent=None):
"""
Constructor
-
+
@param parent reference to the parent object
@type QObject
"""
super().__init__(parent=parent)
-
+
self.__messages = []
-
+
self.__session = None
self.__clientConfig = {}
-
+
self.__browser = None
-
+
def sendMessage(self, message):
"""
Public method to send a message via Google Mail.
-
+
@param message email message to be sent
@type email.mime.text.MIMEBase
"""
self.__messages.append(message)
-
+
if not self.__session:
self.__startSession()
else:
self.__doSendMessages()
-
+
def __prepareMessage(self, message):
"""
Private method to prepare the message for sending.
-
+
@param message message to be prepared
@type email.mime.text.MIMEBase
@return prepared message dictionary
@@ -148,178 +155,183 @@
"""
messageAsBase64 = base64.urlsafe_b64encode(message.as_bytes())
raw = messageAsBase64.decode()
- return {'raw': raw}
-
+ return {"raw": raw}
+
def __startSession(self):
"""
Private method to start an authorized session and optionally start the
authorization flow.
"""
# check for availability of secrets file
- if not os.path.exists(os.path.join(Globals.getConfigDir(),
- CLIENT_SECRET_FILE)):
+ if not os.path.exists(os.path.join(Globals.getConfigDir(), CLIENT_SECRET_FILE)):
self.sendResult.emit(
False,
- self.tr("The client secrets file is not present. Has the Gmail"
- " API been enabled?")
+ self.tr(
+ "The client secrets file is not present. Has the Gmail"
+ " API been enabled?"
+ ),
)
return
-
- with open(os.path.join(Globals.getConfigDir(), CLIENT_SECRET_FILE),
- "r") as clientSecret:
+
+ with open(
+ os.path.join(Globals.getConfigDir(), CLIENT_SECRET_FILE), "r"
+ ) as clientSecret:
clientData = json.load(clientSecret)
- self.__clientConfig = clientData['installed']
+ self.__clientConfig = clientData["installed"]
token = self.__loadToken()
if token is None:
# no valid OAuth2 token available
self.__session = OAuth2Session(
- self.__clientConfig['client_id'],
+ self.__clientConfig["client_id"],
scope=SCOPES,
- redirect_uri=self.__clientConfig['redirect_uris'][0]
+ redirect_uri=self.__clientConfig["redirect_uris"][0],
)
authorizationUrl, _ = self.__session.authorization_url(
- self.__clientConfig['auth_uri'],
+ self.__clientConfig["auth_uri"],
access_type="offline",
- prompt="select_account"
+ prompt="select_account",
)
if self.__browser is None:
with contextlib.suppress(ImportError):
self.__browser = EricGoogleMailAuthBrowser()
self.__browser.approvalCodeReceived.connect(
- self.__processAuthorization)
+ self.__processAuthorization
+ )
if self.__browser:
self.__browser.show()
self.__browser.load(QUrl(authorizationUrl))
else:
from PyQt6.QtGui import QDesktopServices
+
QDesktopServices.openUrl(QUrl(authorizationUrl))
ok, authCode = EricTextInputDialog.getText(
None,
self.tr("OAuth2 Authorization Code"),
- self.tr("Enter the OAuth2 authorization code:"))
+ self.tr("Enter the OAuth2 authorization code:"),
+ )
if ok and authCode:
self.__processAuthorization(authCode)
else:
self.__session = None
else:
self.__session = OAuth2Session(
- self.__clientConfig['client_id'],
+ self.__clientConfig["client_id"],
scope=SCOPES,
- redirect_uri=self.__clientConfig['redirect_uris'][0],
+ redirect_uri=self.__clientConfig["redirect_uris"][0],
token=token,
auto_refresh_kwargs={
- 'client_id': self.__clientConfig['client_id'],
- 'client_secret': self.__clientConfig['client_secret'],
+ "client_id": self.__clientConfig["client_id"],
+ "client_secret": self.__clientConfig["client_secret"],
},
- auto_refresh_url=self.__clientConfig['token_uri'],
- token_updater=self.__saveToken)
+ auto_refresh_url=self.__clientConfig["token_uri"],
+ token_updater=self.__saveToken,
+ )
self.__doSendMessages()
-
+
@pyqtSlot(str)
def __processAuthorization(self, authCode):
"""
Private slot to process the received authorization code.
-
+
@param authCode received authorization code
@type str
"""
self.__session.fetch_token(
- self.__clientConfig['token_uri'],
- client_secret=self.__clientConfig['client_secret'],
- code=authCode)
+ self.__clientConfig["token_uri"],
+ client_secret=self.__clientConfig["client_secret"],
+ code=authCode,
+ )
self.__saveToken(self.__session.token)
-
+
# authorization completed; now send all queued messages
self.__doSendMessages()
-
+
def __doSendMessages(self):
"""
Private method to send all queued messages.
"""
if not self.__session:
- self.sendResult.emit(
- False,
- self.tr("No authorized session available.")
- )
+ self.sendResult.emit(False, self.tr("No authorized session available."))
return
-
+
try:
results = []
credentials = self.__credentialsFromSession()
- service = discovery.build('gmail', 'v1', credentials=credentials,
- cache_discovery=False)
+ service = discovery.build(
+ "gmail", "v1", credentials=credentials, cache_discovery=False
+ )
count = 0
while self.__messages:
count += 1
message = self.__messages.pop(0)
message1 = self.__prepareMessage(message)
- service.users().messages().send(
- userId="me", body=message1).execute()
+ service.users().messages().send(userId="me", body=message1).execute()
results.append(self.tr("Message #{0} sent.").format(count))
self.sendResult.emit(True, "\n\n".join(results))
except Exception as error:
self.sendResult.emit(False, str(error))
-
+
def __loadToken(self):
"""
Private method to load a token from the token file.
-
+
@return loaded token
@rtype dict or None
"""
- homeDir = os.path.expanduser('~')
- credentialsDir = os.path.join(homeDir, '.credentials')
+ homeDir = os.path.expanduser("~")
+ credentialsDir = os.path.join(homeDir, ".credentials")
if not os.path.exists(credentialsDir):
os.makedirs(credentialsDir)
tokenPath = os.path.join(credentialsDir, TOKEN_FILE)
-
+
if os.path.exists(tokenPath):
with open(tokenPath, "r") as tokenFile:
return json.load(tokenFile)
else:
return None
-
+
def __saveToken(self, token):
"""
Private method to save a token to the token file.
-
+
@param token token to be saved
@type dict
"""
- homeDir = os.path.expanduser('~')
- credentialsDir = os.path.join(homeDir, '.credentials')
+ homeDir = os.path.expanduser("~")
+ credentialsDir = os.path.join(homeDir, ".credentials")
if not os.path.exists(credentialsDir):
os.makedirs(credentialsDir)
tokenPath = os.path.join(credentialsDir, TOKEN_FILE)
-
+
with open(tokenPath, "w") as tokenFile:
json.dump(token, tokenFile)
-
+
def __credentialsFromSession(self):
"""
Private method to create a credentials object.
-
+
@return created credentials object
@rtype google.oauth2.credentials.Credentials
"""
credentials = None
-
+
if self.__clientConfig and self.__session:
token = self.__session.token
if token:
credentials = Credentials(
- token['access_token'],
- refresh_token=token.get('refresh_token'),
- id_token=token.get('id_token'),
- token_uri=self.__clientConfig['token_uri'],
- client_id=self.__clientConfig['client_id'],
- client_secret=self.__clientConfig['client_secret'],
- scopes=SCOPES
+ token["access_token"],
+ refresh_token=token.get("refresh_token"),
+ id_token=token.get("id_token"),
+ token_uri=self.__clientConfig["token_uri"],
+ client_id=self.__clientConfig["client_id"],
+ client_secret=self.__clientConfig["client_secret"],
+ scopes=SCOPES,
)
credentials.expiry = datetime.datetime.fromtimestamp(
- token['expires_at'])
-
+ token["expires_at"]
+ )
+
return credentials
@@ -327,7 +339,7 @@
"""
Module function to get some help about how to enable the Google Mail
OAuth2 service.
-
+
@return help text
@rtype str
"""
@@ -354,6 +366,6 @@
"https://console.developers.google.com/start/api?id=gmail",
APPLICATION_NAME,
Globals.getConfigDir(),
- CLIENT_SECRET_FILE
+ CLIENT_SECRET_FILE,
)
)
