Source code for langchain_google_community.gmail.utils
"""Gmail tool utils."""
from __future__ import annotations
import logging
import os
from typing import TYPE_CHECKING, List, Optional, Tuple
if TYPE_CHECKING:
from google.auth.transport.requests import Request # type: ignore[import]
from google.oauth2.credentials import Credentials # type: ignore[import]
from google.oauth2.service_account import Credentials as ServiceCredentials
from google_auth_oauthlib.flow import InstalledAppFlow # type: ignore[import]
from googleapiclient.discovery import Resource # type: ignore[import]
from googleapiclient.discovery import build as build_resource
logger = logging.getLogger(__name__)
[docs]def import_google() -> Tuple[Request, Credentials, ServiceCredentials]:
"""Import google libraries.
Returns:
Tuple[Request, Credentials]: Request and Credentials classes.
"""
try:
from google.auth.transport.requests import Request # noqa: F401
from google.oauth2.credentials import Credentials # noqa: F401
from google.oauth2.service_account import Credentials as ServiceCredentials
except ImportError:
raise ImportError(
"You need to install gmail dependencies to use this toolkit. "
"Try running poetry install --with gmail"
)
return Request, Credentials, ServiceCredentials # type: ignore[return-value]
[docs]def import_installed_app_flow() -> InstalledAppFlow:
"""Import InstalledAppFlow class.
Returns:
InstalledAppFlow: InstalledAppFlow class.
"""
try:
from google_auth_oauthlib.flow import InstalledAppFlow
except ImportError:
raise ImportError(
"You need to install gmail dependencies to use this toolkit. "
"Please, install bigquery dependency group: "
"`pip install langchain-google-community[gmail]`"
)
return InstalledAppFlow
[docs]def import_googleapiclient_resource_builder() -> build_resource:
"""Import googleapiclient.discovery.build function.
Returns:
build_resource: googleapiclient.discovery.build function.
"""
try:
from googleapiclient.discovery import build
except ImportError:
raise ImportError(
"You need to install all dependencies to use this toolkit. "
"Try running pip install langchain-google-community"
)
return build
DEFAULT_SCOPES = ["https://mail.google.com/"]
DEFAULT_SERVICE_SCOPES = ["https://www.googleapis.com/auth/gmail.readonly"]
DEFAULT_CREDS_TOKEN_FILE = "token.json"
DEFAULT_CLIENT_SECRETS_FILE = "credentials.json"
DEFAULT_SERVICE_ACCOUNT_FILE = "service_account.json"
[docs]def get_gmail_credentials(
token_file: Optional[str] = None,
client_secrets_file: Optional[str] = None,
service_account_file: Optional[str] = None,
scopes: Optional[List[str]] = None,
use_domain_wide: bool = False,
delegated_user: Optional[str] = None,
) -> Credentials:
"""Get credentials."""
if use_domain_wide:
_, _, ServiceCredentials = import_google()
service_account_file = service_account_file or DEFAULT_SERVICE_ACCOUNT_FILE
scopes = scopes or DEFAULT_SERVICE_SCOPES
credentials = ServiceCredentials.from_service_account_file(
service_account_file, scopes=scopes
)
if delegated_user:
credentials = credentials.with_subject(delegated_user)
return credentials
else:
# From https://developers.google.com/gmail/api/quickstart/python
Request, Credentials, _ = import_google()
InstalledAppFlow = import_installed_app_flow()
creds = None
scopes = scopes or DEFAULT_SCOPES
token_file = token_file or DEFAULT_CREDS_TOKEN_FILE
client_secrets_file = client_secrets_file or DEFAULT_CLIENT_SECRETS_FILE
# The file token.json stores the user's access and refresh tokens, and is
# created automatically when the authorization flow completes for the first
# time.
if os.path.exists(token_file):
creds = Credentials.from_authorized_user_file(token_file, scopes)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request()) # type: ignore[call-arg]
else:
# https://developers.google.com/gmail/api/quickstart/python#authorize_credentials_for_a_desktop_application # noqa
flow = InstalledAppFlow.from_client_secrets_file(
client_secrets_file, scopes
)
creds = flow.run_local_server(port=0)
with open(token_file, "w") as token:
token.write(creds.to_json())
return creds
[docs]def build_resource_service(
credentials: Optional[Credentials] = None,
service_name: str = "gmail",
service_version: str = "v1",
use_domain_wide: bool = False,
delegated_user: Optional[str] = None,
service_account_file: Optional[str] = None,
scopes: Optional[List[str]] = None,
) -> Resource:
"""Build a Gmail service."""
credentials = credentials or get_gmail_credentials(
use_domain_wide=use_domain_wide,
delegated_user=delegated_user,
service_account_file=service_account_file,
scopes=scopes,
)
builder = import_googleapiclient_resource_builder()
return builder(service_name, service_version, credentials=credentials)
[docs]def clean_email_body(body: str) -> str:
"""Clean email body."""
try:
from bs4 import BeautifulSoup
try:
soup = BeautifulSoup(str(body), "html.parser")
body = soup.get_text()
return str(body)
except Exception as e:
logger.error(e)
return str(body)
except ImportError:
logger.warning("BeautifulSoup not installed. Skipping cleaning.")
return str(body)