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

作者: 高鸿祥 | 来源:发表于2022-07-07 17:36 被阅读0次

需求：使用程序来自动化处理邮箱，比如获取指定邮件、获取验证码等业务

两个方案

模拟点击，比如使用selenium打开网页来处理。但是这种方式不可控因素多，效率低，访问外网网速慢需要时间比较久
走协议，程序访问 gmail api 来获取

建议程序使用走协议的方案，不可控因素少，效率高。

前提：gmail客户端开启邮件访问协议POP和IMAP

协议访问需要授权，权限认证三种方式

用户名+密码+开启安全性较低的应用的访问权限，从2020.05.30开始，google已不再支持此权限
用户名+应用密码+开启二次验证，比较麻烦
用户名+OAuth2验证

建议使用用户名+OAuth2验证的方式

编程层面也存在多种解决方案

使用python内置库，smtplib库发送邮件。imaplib库接收邮件。底层库，处理比较琐碎
使用第三方库，imbox、yagmail等，别人写好的处理mail的库，方便好用，但是目前不支持OAuth2验证，gmail邮箱不好用，qq邮箱等其他邮箱可以用这个库。
谷歌开发的库，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/oauthplayground/

从 Google API Console 获取 OAuth 2.0 客户端凭据

Google API Console网址：https://console.cloud.google.com/

1、创建并选择项目

创建并选择项目.jpeg

2、创建证书

创建凭据.jpeg

证书下载完放在项目目录下，可以重命名为client_secret.json

3、编辑OAuth consent screen，添加测试用户

OAuth consent screen添加测试用户.jpeg

4、启用gmail api

启用gmail api.jpeg

客户端应用从 Google 授权服务器请求访问令牌，从响应中提取令牌

安装 Google 客户端库

  pip install --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib

官方库github地址

生成刷新令牌并存储备用

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 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-%E8%AF%BB%E5%8F%96gmail-python-%E6%90%9C%E7%B4%A2gmail-python%E6%93%8D%E4%BD%9Cgmail-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.html
Google 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)

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

从 Google API Console 获取 OAuth 2.0 客户端凭据

客户端应用从 Google 授权服务器请求访问令牌，从响应中提取令牌

安装 Google 客户端库

生成刷新令牌并存储备用

将该令牌发送到Google Gmail API，完成业务

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读