-
-
Save timrichardson/e6ee6640a8b7fe664f3a5a80406ca980 to your computer and use it in GitHub Desktop.
from string import Template | |
import time | |
import pytest | |
from google.oauth2 import service_account | |
from googleapiclient.discovery import build | |
from googleapiclient.errors import * | |
from google.auth.exceptions import * | |
import easygui | |
import pandas as pd | |
def get_api_credentials(key_path): | |
API_scopes =['https://www.googleapis.com/auth/gmail.settings.basic', | |
'https://www.googleapis.com/auth/gmail.settings.sharing'] | |
credentials = service_account.Credentials.from_service_account_file(key_path,scopes=API_scopes) | |
return credentials | |
def update_sig(full_name, job_title, telephone, username, sig_template, credentials,live=False): | |
sig = sig_template.substitute(full_name=full_name, job_title=job_title, | |
telephone=telephone) | |
print("Username: %s Fullname: %s Job title: %s" % (username, full_name, job_title)) | |
if live: | |
credentials_delegated = credentials.with_subject(username) | |
gmail_service = build("gmail", "v1", credentials=credentials_delegated) | |
addresses = gmail_service.users().settings().sendAs().list(userId='me', | |
fields='sendAs(isPrimary,sendAsEmail)').execute().get('sendAs') | |
# this way of getting the primary address is copy & paste from google example | |
address = None | |
for address in addresses: | |
if address.get('isPrimary'): | |
break | |
if address: | |
rsp = gmail_service.users().settings().sendAs().patch(userId='me', | |
sendAsEmail=address['sendAsEmail'], | |
body={'signature':sig}).execute() | |
print(f"Signature changed for: {username}") | |
else: | |
print(f"Could not find primary address for: {username}") | |
def main(): | |
excelfile = easygui.fileopenbox(msg="Please open the .xlsx file with signature",title="Choose signatures .xlsx file") | |
if not excelfile: | |
print("No signature .xlsx file selected, so stopping") | |
return | |
user_data = pd.ExcelFile(excelfile) | |
# df = user_data.parse("testsheet") | |
df = user_data.parse("livedata") | |
key_path = easygui.fileopenbox(msg="Please open the confidential Google secret .json file", | |
title="Choose Google json secrets") | |
credentials = get_api_credentials(key_path=key_path) | |
if not credentials: | |
print("No credential file selected, so stopping") | |
return | |
try: | |
sig_file = open("template.html", "r") | |
sig_template = Template(sig_file.read()) | |
except (FileNotFoundError,IOError): | |
print("Could not open the template file") | |
raise | |
for r in df.values: | |
username = r[0] | |
first_name = "%s" % r[1] | |
if first_name == "nan": | |
first_name = '' | |
second_name = "%s" % r[2] | |
if second_name == "nan": | |
second_name = '' | |
full_name = "%s %s" % (first_name, second_name) | |
job_title = "%s" % r[3] | |
if job_title == 'nan': | |
job_title = '' | |
telephone = "%s" % r[4] | |
if telephone == 'nan': | |
telephone = "1300 863 824" | |
retry_count = 0 | |
while retry_count < 3: | |
try: | |
update_sig(full_name=full_name,job_title=job_title,username=username,telephone=telephone, | |
sig_template=sig_template,credentials=credentials,live=True) | |
break | |
except (RefreshError,TransportError) as e: | |
retry_count += 1 | |
print(f"Error encountered for: {username}, retrying (attempt {retry_count}). Error was: {e}") | |
time.sleep(2) | |
continue | |
except Exception as e: | |
raise | |
else: | |
print(f"Failed to update {username}") | |
if __name__ == '__main__': | |
main() | |
@pytest.fixture | |
def get_test_api_credentials(): | |
return get_api_credentials(key_path='private/gmailsignatureproject-86b504154ef1.json') | |
def test_fetch_user_info(credentials): | |
credentials_delegated = credentials.with_subject("[email protected]") | |
gmail_service = build("gmail","v1",credentials=credentials_delegated) | |
addresses = gmail_service.users().settings().sendAs().list(userId='me').execute() | |
assert gmail_service | |
def test_fetch_another_user(credentials): | |
credentials_delegated = credentials.with_subject("[email protected]") | |
gmail_service = build("gmail", "v1", credentials=credentials_delegated) | |
addresses = gmail_service.users().settings().sendAs().list(userId='me').execute() | |
assert gmail_service | |
hi guys...can anyone help me with email delegation using gmail api?
@timrichardson
This is my code so far:
https://dpaste.org/82GB
Could you provide an example HTML file? I am trying to make my code work.
Could you provide an example HTML file? I am trying to make my code work.
I think you can put the html as a string.
results = gmail.users().settings().sendAs().patch(userId='me', sendAsEmail=address['sendAsEmail'], body=signature).execute()
signature
from above is signature = {'signature': '<div>something</div>' }
hi guys...can anyone help me with email delegation using gmail api?
@timrichardson
This is my code so far:
https://dpaste.org/82GB
404 from the link
Could you provide an example HTML file? I am trying to make my code work.
I think you can put the html as a string.
results = gmail.users().settings().sendAs().patch(userId='me', sendAsEmail=address['sendAsEmail'], body=signature).execute()
signature
from above issignature = {'signature': '<div>something</div>' }
Thanks, I will give another try!
this is great. has anyone gotten it to work with information within workspace and not a separate Excel file?
@BigEHead you could use google sheets to fill in the information. a variant i created:
general_phone_number = ""
company = ""
signature_template = "signature_template.html"
signature_sheet_id = ""
import logging
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.errors import *
from google.auth.exceptions import *
from jinja2 import Environment, FileSystemLoader
import pandas as pd
logging.basicConfig(filename="signature_updates.log", level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
dagen_afkortingen = {
"ma": "maandag",
"di": "dinsdag",
"wo": "woensdag",
"do": "donderdag",
"vr": "vrijdag",
"vr.morgen": "vrijdag morgen",
"za": "zaterdag",
"zo": "zondag",
}
def get_api_credentials(key_path):
try:
API_scopes = [
"https://www.googleapis.com/auth/gmail.settings.basic",
"https://www.googleapis.com/auth/gmail.settings.sharing",
"https://www.googleapis.com/auth/spreadsheets",
]
credentials = service_account.Credentials.from_service_account_file(key_path, scopes=API_scopes)
return credentials
except FileNotFoundError:
logging.error(f"Could not find or open the credentials file: {key_path}")
raise # Re-raise the exception to stop execution
except Exception as e: # Catch other potential exceptions during credential loading
logging.error(f"An error occurred while loading credentials: {e}")
raise
def get_users_file(sheet_id, credentials):
"""Retrieves staff data from a Google Sheet using the Sheets API."""
try:
service = build("sheets", "v4", credentials=credentials)
# Get the spreadsheet values
result = (
service.spreadsheets()
.values()
.get(
spreadsheetId=sheet_id,
range="signature_data!A1:I200", # Replace with the *actual* range of your data!
valueRenderOption="UNFORMATTED_VALUE",
dateTimeRenderOption="SERIAL_NUMBER",
)
.execute()
)
# print(result)
values = result.get("values", [])
if values:
headers = values[0]
data = []
for row in values[1:]:
# Pad rows with empty strings to match header length
while len(row) < len(headers):
row.append("")
data.append(row)
df = pd.DataFrame(data, columns=headers)
if "handtekening_ingesteld" not in df.columns:
df["handtekening_ingesteld"] = ""
df = df.query("handtekening_ingesteld.isna() | (handtekening_ingesteld == '')")
return df
else:
print("Geen data gevonden.")
return pd.DataFrame()
except HttpError as error:
logging.error(f"An error occurred: {error}")
return pd.DataFrame()
except Exception as e:
logging.error(f"An error occurred: {e}")
return pd.DataFrame()
def create_and_update_signatures(df, signature_template, credentials, sheet_id):
"""
Creates email signatures from an Excel file using a Jinja2 template
and updates them in Gmail using the Google API.
Args:
signature_template (str): Path to the Jinja2 template file.
"""
env = Environment(loader=FileSystemLoader("."))
template = env.get_template(signature_template)
for index, row in df.iterrows():
login_name = row["login"] # login/or mail adres are the same
volledige_naam = f"{row["Roepnaam"].strip()} {row["Voorvoegsel"].strip() + ' ' if row["Voorvoegsel"] else ''}{row["Achternaam"].strip()}"
job_title = row["klas(en)_of_functie"].strip()
gebouw = row["locatie"].strip().lower()
dagen = [dag.strip() for dag in (row["aanwezige_dagen"].lower().split(",") if "," in row["aanwezige_dagen"].lower() else [row["aanwezige_dagen"].lower().strip()])]
# ondersteuningteam = row["ot"]
match gebouw:
case "vgl" | "vgl/vl":
gebouw_naam = "Van Gaverenlaan"
exact_location = ""
case "vl" | "vl/vgl":
gebouw_naam = "Viandenlaan"
exact_location = ""
# if ondersteuningteam:
# email = "[email protected]"
# else:
email = login_name
match dagen:
case ["" | None]:
aanwezige_dagen = "maandag - dinsdag - woensdag - donderdag - vrijdag ?"
case x if "fulltime" in " ".join(dagen):
aanwezige_dagen = "maandag - dinsdag - woensdag - donderdag - vrijdag"
case [*_]: #meerdere items in de lijst.
aanwezige_dagen = f"aanwezig {gebouw_naam}: {' - '.join(map(lambda dag: dagen_afkortingen.get(dag, dag), dagen))}"
case _:
aanwezige_dagen = "maandag - dinsdag - woensdag - donderdag - vrijdag"
signature = template.render(
volledige_naam=volledige_naam,
job_title=job_title,
email=email,
company=company,
algemene_nummer=general_phone_number,
exact_location_school=exact_location,
dagen_aanwezig=aanwezige_dagen,
)
# Update the signature using the Gmail API
try:
credentials_delegated = credentials.with_subject(login_name)
gmail_service = build("gmail", "v1", credentials=credentials_delegated)
sheet_service = build("sheets", "v4", credentials=credentials)
primary_address = next(
(
addr
for addr in gmail_service.users()
.settings()
.sendAs()
.list(userId="me")
.execute()
.get(
"sendAs",
)
if addr.get("isPrimary")
),
None,
)
if primary_address:
gmail_service.users().settings().sendAs().patch(
userId="me",
sendAsEmail=primary_address["sendAsEmail"],
body={"signature": signature},
).execute()
logging.info(f"Signature updated for: {volledige_naam}: {signature}")
print(f"Signature updated for: {volledige_naam}")
try:
headers = df.columns.tolist()
col_index = headers.index("handtekening_ingesteld") # for selecting the correct colum
col_letter = chr(col_index + 65)
row_index_in_sheet = index + 2
sheet_service.spreadsheets().values().update(
spreadsheetId=sheet_id,
range=f"signature_data!{col_letter}{row_index_in_sheet}",
valueInputOption="USER_ENTERED",
body={"values": [["ja"]]},
).execute()
except Exception as e:
logging.error(f"Error updating Google Sheet for {volledige_naam}: {e}")
else:
logging.error(f"Could not find primary address for {volledige_naam}")
except (HttpError, RefreshError, DefaultCredentialsError) as e:
logging.error(f"An error occurred for {volledige_naam}: {e}")
def main():
credentials = get_api_credentials(key_path="service_account_signatures.json")
new_users = get_users_file(signature_sheet_id, credentials)
if not new_users.empty:
create_and_update_signatures(new_users, signature_template, credentials, signature_sheet_id)
if __name__ == "__main__":
main()
Thanks a lot. Used your example and my code is working now.