Envoyez et recevez Gmail via l'API Gmail en utilisant Python
introduction
Comme vous le savez, Gmail dispose d'une API Gmail qui vous permet de profiter des fonctionnalités uniques de Gmail telles que les capacités de recherche sans utiliser SMTP ou POP3. Cet article fournit des instructions détaillées sur l'utilisation de Python pour activer l'API Gmail, créer des scripts et les exécuter.
Avant d'utiliser l'API Gmail, il est assez difficile de se préparer, comme la création d'un projet, l'activation de l'API, la définition de la portée et la création d'informations d'authentification, donc je l'ai écrit sous forme de mémorandum.
- Contenu de l'API Gmail à partir d'avril 2020.
- En supposant que Python 3.7.x.
Préparation avant utilisation
Créer un projet qui utilise l'API Google
Ouvrez la console Google Cloud Platform (https://console.cloud.google.com/) dans le navigateur qui ouvre Gmail pour votre compte Gmail.
L'écran suivant apparaîtra: Acceptez les termes et conditions et cliquez sur «Accepter et exécuter».
Sélectionnez "Nouveau projet" dans "Sélectionner un projet"
Définissez le nom du projet et cliquez sur "Créer"
Si vous pouvez le créer, l'écran suivant s'affichera
Activer l'API Gmail
Cliquez sur "Bibliothèque"
Recherchez «API Gmail» dans «Rechercher des API et des services»
Cliquez sur l '"API Gmail" affichée
Cliquez sur "Activer"
Créer un écran de consentement OAuth
Cliquez sur "Écran de consentement OAuth" dans "API et services"
Sélectionnez "Externe" et cliquez sur "Créer"
Définissez un nom approprié pour "Nom de l'application" et cliquez sur "Ajouter une portée"
Sélectionnez une portée comme indiqué ci-dessous et cliquez sur "Ajouter"
Cliquez sur "Enregistrer"
Créer des identifiants
Sélectionnez "Credentials", cliquez sur "Create Credentials" et cliquez sur "Select with Wizard" dans le menu qui apparaît.
Sur l'écran «Ajouter des informations d'identification au projet», définissez comme indiqué ci-dessous et cliquez sur «Informations d'identification requises».
Cliquez sur "Créer un ID client OAuth"
Téléchargez les informations d'identification (client_id.json) et cliquez sur "Terminer"
Lorsque vous appuyez sur le bouton Terminer, l'écran passe à celui illustré ci-dessous.
Le fichier d'informations d'identification téléchargé client_id.json sera utilisé ultérieurement dans le script Python.
Supplément: à propos des informations d'authentification
Vous pouvez créer une clé API, un ID client OAuth2.0 ou une clé de compte de service dans vos informations d'identification. Lorsqu'il s'agit de données utilisateur, il semble courant d'utiliser un ID client OAuth2.0. La page Présentation de l'authentification des documents Google Cloud (https://cloud.google.com/docs/authentication?hl=ja) décrit les utilisations de chaque méthode d'authentification.
La date d'expiration du jeton d'actualisation des informations d'identification est décrite ci-dessous. Une fois authentifié et le jeton d'actualisation émis, il ne semble pas expirer à moins qu'il ne soit laissé pendant 6 mois ou que 50 jetons ou plus soient émis. Il semble que les mécanismes d'automatisation tels que la RPA soient également acceptables. https://developers.google.com/identity/protocols/OAuth2#expiration
You must write your code to anticipate the possibility that a granted refresh token might no longer work. A refresh token might stop working for one of these reasons:
* The user has revoked your app's access.
* The refresh token has not been used for six months.
* The user changed passwords and the refresh token contains Gmail scopes.
* The user account has exceeded a maximum number of granted (live) refresh tokens.
* There is currently a limit of 50 refresh tokens per user account per client. If the limit is reached, creating a new refresh token automatically invalidates the oldest refresh token without warning. This limit does not apply to service accounts.
Préparation du script Python
Maintenant que nous sommes prêts, nous aurons un script Python à exécuter.
Préparer l'environnement Python
(Je pense que beaucoup de gens ont déjà vécu dans cette partie)
Installation de Python
Installez selon ici. Veuillez installer 3.7.x.
Installation de pipenv
Suivez les étapes ici (http://pipenv-ja.readthedocs.io/ja/translate-ja/install.html#installing-pipenv) pour installer.
Créer un script Python
Préparez un répertoire approprié et préparez un script python. Enregistrez le client_id.json enregistré à l'étape précédente dans le même répertoire. A rejoint les exemples de l'API Gmail: sweat_drops:
Vous pouvez trouver un exemple de code dans la référence de l'API Gmail. C'est facile à utiliser lorsque vous venez ici!
gmail_credential.py
import os
import pickle
from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow
#Définir la portée de l'API Gmail
SCOPES = [
"https://www.googleapis.com/auth/gmail.compose",
"https://www.googleapis.com/auth/gmail.readonly",
"https://www.googleapis.com/auth/gmail.labels",
"https://www.googleapis.com/auth/gmail.modify",
]
def get_credential():
"""
Obtenez un jeton d'accès
Enregistrez le jeton au format pickle dans le répertoire actuel afin qu'il puisse être réutilisé. (Désolé pour les divers ...)
"""
creds = None
if os.path.exists("token.pickle"):
with open("token.pickle", "rb") as token:
creds = pickle.load(token)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file("client_id.json", SCOPES)
# creds = flow.run_local_server()
creds = flow.run_console()
with open("token.pickle", "wb") as token:
pickle.dump(creds, token)
return creds
listmail.py
"""
list GMail Inbox.
Usage:
listmail.py <query> <tag> <count>
listmail.py -h | --help
listmail.py --version
Options:
-h --help Show this screen.
--version Show version.
"""
import pickle
import base64
import json
import io
import csv
import os.path
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import base64
from email.mime.text import MIMEText
from apiclient import errors
import logging
from docopt import docopt
from gmail_credential import get_credential
logger = logging.getLogger(__name__)
def list_labels(service, user_id):
"""
Obtenez une liste d'étiquettes
"""
labels = []
response = service.users().labels().list(userId=user_id).execute()
return response["labels"]
def decode_base64url_data(data):
"""
Décodage base64url
"""
decoded_bytes = base64.urlsafe_b64decode(data)
decoded_message = decoded_bytes.decode("UTF-8")
return decoded_message
def list_message(service, user_id, query, label_ids=[], count=3):
"""
Obtenez une liste de courriels
Parameters
----------
service : googleapiclient.discovery.Resource
Ressources pour communiquer avec Gmail
user_id : str
Identifiant d'utilisateur
query : str
Chaîne de requête d'e-mail. est:non lu etc.
label_ids : list
Liste des identifiants indiquant l'étiquette à rechercher
count : str
Nombre maximum d'informations par e-mail à renvoyer
Returns
----------
messages : list
id, body, subject,Liste des données du dictionnaire avec des clés telles que de
"""
messages = []
try:
message_ids = (
service.users()
.messages()
.list(userId=user_id, maxResults=count, q=query, labelIds=label_ids)
.execute()
)
if message_ids["resultSizeEstimate"] == 0:
logger.warning("no result data!")
return []
#Vérifiez le contenu du message en fonction de l'identifiant du message
for message_id in message_ids["messages"]:
message_detail = (
service.users()
.messages()
.get(userId="me", id=message_id["id"])
.execute()
)
message = {}
message["id"] = message_id["id"]
message["body"] = decode_base64url_data(
message_detail["payload"]["body"]["data"]
)
# payload.headers[name: "Subject"]
message["subject"] = [
header["value"]
for header in message_detail["payload"]["headers"]
if header["name"] == "Subject"
][0]
# payload.headers[name: "From"]
message["from"] = [
header["value"]
for header in message_detail["payload"]["headers"]
if header["name"] == "From"
][0]
logger.info(message_detail["snippet"])
messages.append(message)
return messages
except errors.HttpError as error:
print("An error occurred: %s" % error)
def remove_labels(service, user_id, messages, remove_labels):
"""
Supprimez l'étiquette. Utilisé pour marquer comme lu(is:Si vous supprimez l'étiquette non lue, elle devient lue)
"""
message_ids = [message["id"] for message in messages]
labels_mod = {
"ids": message_ids,
"removeLabelIds": remove_labels,
"addLabelIds": [],
}
# import pdb;pdb.set_trace()
try:
message_ids = (
service.users()
.messages()
.batchModify(userId=user_id, body=labels_mod)
.execute()
)
except errors.HttpError as error:
print("An error occurred: %s" % error)
#Traitement principal
def main(query="is:unread", tag="daily_report", count=3):
creds = get_credential()
service = build("gmail", "v1", credentials=creds, cache_discovery=False)
#Liste d'étiquettes
labels = list_labels(service, "me")
target_label_ids = [label["id"] for label in labels if label["name"] == tag]
#Liste de courrier[{'body': 'xxx', 'subject': 'xxx', 'from': 'xxx'},]
messages = list_message(service, "me", query, target_label_ids, count=count)
# unread label
unread_label_ids = [label["id"] for label in labels if label["name"] == "UNREAD"]
# remove labels form messages
remove_labels(service, "me", messages, remove_labels=unread_label_ids)
logger.info(json.dumps(messages, ensure_ascii=False))
if messages:
return json.dumps(messages, ensure_ascii=False)
else:
return None
#Partie exécution du programme
if __name__ == "__main__":
arguments = docopt(__doc__, version="0.1")
query = arguments["<query>"]
tag = arguments["<tag>"]
count = arguments["<count>"]
logging.basicConfig(level=logging.DEBUG)
messages_ = main(query=query, tag=tag, count=count)
print(messages_)
Le script d'envoi ressemble à ceci.
- Dans l'exemple de code Python de Users.messages.send,
sendmail.py
"""
Send E-Mail with GMail.
Usage:
sendmail.py <sender> <to> <subject> <message_text_file_path> [--attach_file_path=<file_path>] [--cc=<cc>]
sendmail.py -h | --help
sendmail.py --version
Options:
-h --help Show this screen.
--version Show version.
--attach_file_path=<file_path> Path of file attached to message.
--cc=<cc> cc email address list(separated by ','). Default None.
"""
import pickle
import os.path
from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
import base64
from email.mime.base import MIMEBase
from email.mime.text import MIMEText
from email.mime.image import MIMEImage
from email.mime.audio import MIMEAudio
from pathlib import Path
from email.mime.multipart import MIMEMultipart
import mimetypes
from apiclient import errors
from gmail_credential import get_credential
from docopt import docopt
import logging
logger = logging.getLogger(__name__)
def create_message(sender, to, subject, message_text, cc=None):
"""
Encoder en base64 MIMEText
"""
enc = "utf-8"
message = MIMEText(message_text.encode(enc), _charset=enc)
message["to"] = to
message["from"] = sender
message["subject"] = subject
if cc:
message["Cc"] = cc
encode_message = base64.urlsafe_b64encode(message.as_bytes())
return {"raw": encode_message.decode()}
def create_message_with_attachment(
sender, to, subject, message_text, file_path, cc=None
):
"""
Base64 encoder MIMEText avec des pièces jointes
"""
message = MIMEMultipart()
message["to"] = to
message["from"] = sender
message["subject"] = subject
if cc:
message["Cc"] = cc
# attach message text
enc = "utf-8"
msg = MIMEText(message_text.encode(enc), _charset=enc)
message.attach(msg)
content_type, encoding = mimetypes.guess_type(file_path)
if content_type is None or encoding is not None:
content_type = "application/octet-stream"
main_type, sub_type = content_type.split("/", 1)
if main_type == "text":
with open(file_path, "rb") as fp:
msg = MIMEText(fp.read(), _subtype=sub_type)
elif main_type == "image":
with open(file_path, "rb") as fp:
msg = MIMEImage(fp.read(), _subtype=sub_type)
elif main_type == "audio":
with open(file_path, "rb") as fp:
msg = MIMEAudio(fp.read(), _subtype=sub_type)
else:
with open(file_path, "rb") as fp:
msg = MIMEBase(main_type, sub_type)
msg.set_payload(fp.read())
p = Path(file_path)
msg.add_header("Content-Disposition", "attachment", filename=p.name)
message.attach(msg)
encode_message = base64.urlsafe_b64encode(message.as_bytes())
return {"raw": encode_message.decode()}
def send_message(service, user_id, message):
"""
envoyer un e-mail
Parameters
----------
service : googleapiclient.discovery.Resource
Ressources pour communiquer avec Gmail
user_id : str
Identifiant d'utilisateur
message : dict
"raw"Clé,Un dict avec un objet MIME encodé en base64 comme valeur
Returns
----------
Aucun
"""
try:
sent_message = (
service.users().messages().send(userId=user_id, body=message).execute()
)
logger.info("Message Id: %s" % sent_message["id"])
return None
except errors.HttpError as error:
logger.info("An error occurred: %s" % error)
raise error
#Traitement principal
def main(sender, to, subject, message_text, attach_file_path, cc=None):
#Obtention de jetons d'accès et de services de construction
creds = get_credential()
service = build("gmail", "v1", credentials=creds, cache_discovery=False)
if attach_file_path:
#Création du corps de l'e-mail
message = create_message_with_attachment(
sender, to, subject, message_text, attach_file_path, cc=cc
)
else:
message = create_message(
sender, to, subject, message_text, cc=cc
)
#envoyer un e-mail
send_message(service, "me", message)
#Partie exécution du programme
if __name__ == "__main__":
arguments = docopt(__doc__, version="0.1")
sender = arguments["<sender>"]
to = arguments["<to>"]
cc = arguments["--cc"]
subject = arguments["<subject>"]
message_text_file_path = arguments["<message_text_file_path>"]
attach_file_path = arguments["--attach_file_path"]
logging.basicConfig(level=logging.DEBUG)
with open(message_text_file_path, "r", encoding="utf-8") as fp:
message_text = fp.read()
main(
sender=sender,
to=to,
subject=subject,
message_text=message_text,
attach_file_path=attach_file_path,
cc=cc,
)
Préparation à l'exécution d'un script python
Accédez au répertoire dans lequel vous avez créé le script python et installez les modules requis.
% pipenv install google-api-python-client oauth2client google-auth-httplib2 google-auth-oauthlib docopt
Exécuter le script Python
% pipenv run python listmail.py is:non lu Demander un devis 10
Si vous exécutez tel que, l'URL de l'écran d'authentification OAuth sera affichée sur l'écran de cmd.exe lors de la première exécution, alors ouvrez-la à partir du navigateur et approuvez-la. L'URL de l'écran d'authentification OAuth ne sera pas affichée à partir de la deuxième fois.
% pipenv run python listmail.py is:non lu Demander un devis 10
Please visit this URL to authorize this application: https://accounts.google.com/o/oauth2/auth?response_type=code&client_id=AAAAAAAAAAAAAA
Enter the authorization code:
Lorsque vous ouvrez l'URL affichée sur la console ci-dessus, l'écran suivant s'affiche.
Cliquez sur "Détails" et cliquez sur le lien "Aller à XX (page non sécurisée)" qui apparaît.
La boîte de dialogue d'autorisation s'affichera plusieurs fois, alors autorisez chacune d'elles.
À la fin, un écran de confirmation s'affichera, alors permettez-le.
Copiez le code de l'écran ci-dessous et collez-le après ```Entrez le code d'autorisation: `` `pour exécuter le script.
à la fin
L'API Gmail est intéressante car vous pouvez utiliser les puissants filtres, les fonctionnalités d'étiquetage et les requêtes de Gmail tels quels.
Recommended Posts