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)