# OAuth2认证+谷歌客户端库调用gmail api处理业务

**Published by:** [gaohongxiang](https://paragraph.com/@gaohongxiang-2/)
**Published on:** 2023-02-13
**URL:** https://paragraph.com/@gaohongxiang-2/oauth2-gmail-api

## Content

OAuth2认证+谷歌客户端库调用gmail api处理业务需求：使用程序来自动化处理邮箱，比如获取指定邮件、获取验证码等业务 两个方案模拟点击，但使用方式不打开网页来处理。这种可控因素多，效率低，访问外网速度慢，需要时间比较久走协议，访问gmail api 来获取建议程序使用走协议的方案，不可控因素少，效率高。POP协议和gmail客户端开启邮件访问授权访问授权，授权方式用户名+密码+开启安全性较低的应用的访问权限，从2020.05.30开始，google已不再支持此权限用户名+应用密码+开启二次验证，比较麻烦用户名+OAuth2验证建议使用用户名+OAuth2验证的方式 编程层次也存在分析解决方案使用python编写，smtplib库发送邮件。imaplib库接收邮件。垃圾邮件库，处理比较垃圾库使用，qq等库，但是现在imbox，yagmail其他人写邮件的库，方便好用的邮箱2验证，gmail邮箱不好用，其他邮箱可以用这个库。谷歌开发的库，google_auth_oauthlib（OAuth2验证）、google-api-python-client（谷歌客户端库）综上，解决方案为：python程序走协议通过用户名+OAuth2的验证方式，使用google_auth_oauthlib、google-api-python-client等谷歌客户端库来访问gmail api，实现业务需求。以下为官方示例教程使用 OAuth 2.0 访问 Google API：https://developers.google.com/identity/protocols/oauth2快速入门：https://developers.google.com/gmail/api/quickstart/python具体流程如下Google API Console 获取 OAuth 2.0 从客户端连接。您的客户端应用程序从 Google 授权服务器请求访问令牌，从响应中获取令牌，的业务通知发送到您要访问的 Google API，验证后处理。为了更好地理解自己的逻辑，官方提供了将 OAuth 2.0 与 Google 使用结合的互动式视频（包括使用您的客户端连接的选项）：https://developers.google.com/authplayground/Google API Console 获取 OAuth 2.0 客户端请求 从谷歌API解析地址：https://console.google.com/ 1、创建并选择项目2、创建证书证书下载完整搜索项目目录下，可以重命名为client_secret.json 3、编辑OAuth同意画面，添加测试用户4、启用gmail api客户端应用从 Google 授权服务器请求访问令牌，从响应中添加令牌安装谷歌客户端库 pip install --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib 官方库github地址https://github.com/googleapis/google-auth-library-python-oauthlibhttps://github.com/googleapis/google-api-python-client生成并存储备用标记from google_auth_oauthlib.flow import Flow import google.oauth2.credentials from googleapiclient.discovery import build import html2text import base64 import json, sys, os sys.path.append(os.getcwd()) # 根目录 # 导入模块都从根目录开始写 from accounts.adspower import * class OAuth2Unit(AdsPowerUnit): def __init__(self, ads_id): super(OAuth2Unit, self).__init__(ads_id) def create_refresh_token(self, gmail): """所有邮箱授权，生成刷新令牌，保存起来，以后直接用 """ # # 使用谷歌API的客户秘密文件创建流程 flow = Flow.from_client_secrets_file( './client_secret.json', # 阅读、撰写、发送和永久删除你在Gmail的所有邮件 scopes=['openid','https://mail.google.com/', 'https://www.googleapis.com/auth/userinfo.email', 'https://www.googleapis.com/auth/userinfo.profile'], redirect_uri='urn:ietf:wg:oauth:2.0:oob') # 告诉用户要去授权的URL auth_url, _ = flow.authorization_url( # 离线访问所请求范围的权限，可以刷新访问令牌，而无需重新提示用户授予权限 access_type='offline', # 增量授权，建议打开 include_granted_scopes='true') # 利用selenium自动获取code，这样可以全自动 url = format(auth_url) self.driver.get(url) try: WebDriverWait(self.driver,10).until(EC.visibility_of_element_located((By.XPATH, "//div[text()='"+gmail+"']"))).click() WebDriverWait(self.driver,10).until(EC.visibility_of_element_located((By.XPATH, "//div/div/div/div/div/div/div/div/div/div/div/div[2]"))).click() time.sleep(2) # 勾选gmail权限 WebDriverWait(self.driver,10).until(lambda x:self.driver.find_elements(by=By.CSS_SELECTOR, value="input"))[3].click() WebDriverWait(self.driver,10).until(lambda x:self.driver.find_elements(by=By.CSS_SELECTOR, value="button"))[2].click() except Exception as e: pass code = WebDriverWait(self.driver,10).until(EC.visibility_of_element_located((By.TAG_NAME, "textarea"))).get_attribute("value") print(code) # # 手动复制url得到code再粘贴到终端的方式。 # # 终端获得的url复制到新邮箱所在adspower浏览器，新邮箱同意授权 # print('Please go to this URL: {}'.format(auth_url)) # # 用户得到的授权码填入此处，来获取访问令牌。 # code = input('Enter the authorization code: ') flow.fetch_token(code=code) # credentials包含token（访问令牌）、refresh_token（刷新令牌）、client_id、client_secret等信息 credentials = flow.credentials # 将刷新令牌保存到json文件 # json.load()：将已编码的 JSON 字符串解码为 Python 对象 # json.dump()：将 Python 对象编码成 JSON 字符串 new_refresh_token = {gmail: credentials.refresh_token} with open('./accounts/mail/credential_tokens.json', 'r') as f: content = json.load(f) old_refresh_token = content['refresh_token'] refresh_token = {**old_refresh_token, **new_refresh_token} content.update({"refresh_token": refresh_token}) with open('./accounts/mail/credential_tokens.json', 'w') as f: json.dump(content, f, indent=4, ensure_ascii=False) if __name__ == '__main__': oauth = OAuth2Unit('adspower_id') # 这个谷歌邮箱是添加进OAuth同意屏幕的测试用户 oauth.create_refresh_token("gmail_address") 此段代码结合了adspower指纹浏览器使用，实现完全自动化。如果不需要，把adspower相关部分删掉，使用手动复制url得到code再粘贴到终端的方式来生成刷新令牌（refresh_token） 授权范围（scopes）决定了应用的权限，具体请参考：https://developers.google.com/gmail/api/auth/scopes?hl=en 将刷新令牌保存到credential_tokens.json文件中备用，下次应用访问此用户就不用征求意见了。文件格式如下{ "client_id": "xxx", "client_secret": "xxx", "token_uri": "https://oauth2.googleapis.com/token", "refresh_token": { "xxx@gmail.com": "xxx", "yyy@gmail.com": "yyy" } } 相关方法示例google_auth_oauthlib.flow module：https://google-auth-oauthlib.readthedocs.io/en/latest/reference/google_auth_oauthlib.flow.html#google_auth_oauthlib.flow.Flow.from_client_secrets_filecredentials示例：https://google-auth.readthedocs.io/en/latest/reference/google.oauth2.credentials.html#google.oauth2.credentials.Credentials将该令牌发送到Google Gmail API，完成业务class GmailUnit(): def __init__(self, gmail, file_='./credential_tokens.json'): """获取授权,调用google gmail api """ data = json.load(open(file_,'r',encoding="utf-8")) credentials = google.oauth2.credentials.Credentials( token=None, refresh_token=data['refresh_token'][gmail], client_id=data['client_id'], client_secret=data['client_secret'], token_uri=data['token_uri'] ) self.service = build('gmail', 'v1', credentials=credentials) def get_messages_by_query(self, user_id='me', label_ids=['INBOX'], query=''): try: response = self.service.users().messages().list(userId=user_id, labelIds=label_ids, q=query).execute() messages = [] if 'messages' in response: messages.extend(response['messages']) while 'nextPageToken' in response: page_token = response['nextPageToken'] response = self.service.users().messages().list(userId=user_id, labelIds=label_ids, q=query, pageToken=page_token).execute() messages.extend(response['messages']) # print(messages) # print(len(messages)) for message in messages: msg = self.service.users().messages().get(userId=user_id, id=message['id']).execute() # print(msg) # 正文 msg = msg['payload']['body']['data'] # print(msg) # base64url解码 msg = base64.urlsafe_b64decode(msg).decode('utf-8') print(msg) except Exception as e: print('An error occurred: %s' % e) if __name__ == '__main__': gmail = GmailUnit("xxx@gmail.com") #user_id='me', label_ids=['INBOX'], query='' gmail.get_messages_by_query(query='from:xxx subject:xxx') __init__函数从credential_tokens.json文件中读取数据，获取授权 get_messages_by_query函数调用gmail api处理业务 具体业务查看官方gmail api文档google gmail api：https://developers.google.com/gmail/api/reference/rest?apix=true、https://developers.google.com/resources/api-libraries/documentation/gmail/v1/python/latest/index.html可用于 Gmail 的搜索运算符（查询条件）：https://support.google.com/mail/answer/7190?hl=en文件结构如下client_secret.json # 下载的OAuth 2.0 客户端凭据 credential_tokens.json # 存放refresh_token等信息，获取授权用 mail.py # OAuth2Unit获取refresh_token，GmailUnit处理业务 adspower.py # adspower指纹浏览器相关，用不到请忽略，文末列出相关代码 其他参考示例Python 读取gmail代码示例: https://justcode.ikeepstudying.com/2019/09/python-读取gmail-python-搜索gmail-python操作gmail-how-to-access-gmail-using-python/操作方法：在 Python 中使用 OAuth2 和 Gmail发送HTML邮件: https://blog.macuyiko.com/post/2016/how-to-send-html-mails-with-oauth2-and-gmail-in-python.htmlGoogle OAuth 2.0 中文认证指南：https://wiki.jikexueyuan.com/project/google-oauth-2/adspower相关代码from selenium import webdriver from selenium.webdriver.chrome.service import Service from selenium.webdriver.chrome.options import Options from selenium.webdriver.support.wait import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.common.by import By from selenium.webdriver.common.keys import Keys from selenium.webdriver.common.action_chains import ActionChains from selenium.webdriver.common.desired_capabilities import DesiredCapabilities import requests,time class AdsPowerUnit(object): """selenium操作adspower指纹浏览器 """ def __init__(self, ads_id): """启动浏览器（webdriver为adspower自带的，不必单独下载） 通过adspower提供的api和自己的浏览器id来启动浏览器 Attributes： ads_id:adspower浏览器id """ self.ads_id = ads_id status = 'http://local.adspower.com:50325/status' open_url = "http://local.adspower.net:50325/api/v1/browser/start?user_id=" + self.ads_id self.close_url = "http://local.adspower.net:50325/api/v1/browser/stop?user_id=" + self.ads_id resp = requests.get(open_url).json() if resp["code"] != 0: print(resp["msg"]) print("please check ads_id") # 启动浏览器后在返回值中拿到对应的Webdriver的路径resp["data"]["webdriver"] chrome_driver = Service(str(resp["data"]["webdriver"])) # selenium启动的chrome浏览器是一个空白的浏览器。chromeOptions是一个配置chrome启动属性的类，用来配置参数。 chrome_options = webdriver.ChromeOptions() # adspower提供的debug接口,用于执行selenium自动化 chrome_options.add_experimental_option("debuggerAddress", resp["data"]["ws"]["selenium"]) # add_extension设置应用扩展。但是注意，adspower的扩展不能通过代码，是通过在客户端的应用程序里上传的，会应用到所有浏览器 # chrome_options.add_extension("./metamask.crx") chrome_options.add_argument('--disable-gpu') #get直接返回，再也不等待界面加载完成 desired_capabilities = DesiredCapabilities.CHROME desired_capabilities["pageLoadStrategy"] = "none" self.driver = webdriver.Chrome(service=chrome_driver, options=chrome_options) # 全屏 self.driver.maximize_window() self.close_other_windows() def close_other_windows(self): """关闭无关窗口，只留当前窗口 理论上下面代码会保留当前窗口，句柄没错。但是实际窗口却没有达到预期。不清楚具体原因。后续再研究。目前能做到的就是只保留一个窗口 """ current_handle = self.driver.current_window_handle all_handles = self.driver.window_handles for handle in all_handles: self.driver.switch_to.window(handle) if handle != current_handle: self.driver.close() def quit(self): """关闭浏览器 """ time.sleep(5) self.driver.quit() requests.get(self.close_url)

## Publication Information

- [gaohongxiang](https://paragraph.com/@gaohongxiang-2/): Publication homepage
- [All Posts](https://paragraph.com/@gaohongxiang-2/): More posts from this publication
- [RSS Feed](https://api.paragraph.com/blogs/rss/@gaohongxiang-2): Subscribe to updates
- [Twitter](https://twitter.com/gaohongxiang): Follow on Twitter