secstate/dotfiles
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[bin] Some minor fixes to checking work email

Browse Source
This commit is contained in:
Colin Powell
2025-07-24 02:53:22 -04:00
parent a4f29b3048
commit 939eec9cd7
1 changed files with 15 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
bin/.bin/checkworkmail.py
View File

@ -26,6 +26,7 @@ NTFY_TOPIC = os.getenv("NTFY_TOPIC", "dev-notifications")
NTFY_SERVER = os.getenv("NTFY_SERVER", "https://ntfy.sh") NTFY_SERVER = os.getenv("NTFY_SERVER", "https://ntfy.sh")
CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "60")) CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "60"))
JIRA_SENDERS = ["jira@yourcompany.com", "jira@atlassian.net"] JIRA_SENDERS = ["jira@yourcompany.com", "jira@atlassian.net"]
SLACK_SENDER = "notification@slack.com"
GITHUB_SENDER = "notifications@github.com" GITHUB_SENDER = "notifications@github.com"
JIRA_BASE_URL = os.getenv("JIRA_BASE_URL", "https://yourcompany.atlassian.net/browse") JIRA_BASE_URL = os.getenv("JIRA_BASE_URL", "https://yourcompany.atlassian.net/browse")
@ -73,7 +74,6 @@ def extract_jira_link(subject):
def format_message(source, subject, body, link=None): def format_message(source, subject, body, link=None):
return body.strip() or "" return body.strip() or ""
return "\n".join(lines)
def send_ntfy_notification(title, message): def send_ntfy_notification(title, message):
url = f"{NTFY_SERVER.rstrip('/')}/{NTFY_TOPIC}" url = f"{NTFY_SERVER.rstrip('/')}/{NTFY_TOPIC}"
@ -85,15 +85,14 @@ def is_github_email(from_email):
def is_jira_email(from_email, subject): def is_jira_email(from_email, subject):
return any(s in from_email.lower() for s in JIRA_SENDERS) or re.search(r'[A-Z]+-\d+', subject) return any(s in from_email.lower() for s in JIRA_SENDERS) or re.search(r'[A-Z]+-\d+', subject)
def is_slack_email(from_email, subject):
return SLACK_SENDER in from_email.lower()
def archive_message(mail, email_id): def archive_message(mail, email_id):
try: try:
if isinstance(email_id, bytes): if isinstance(email_id, bytes):
email_id = email_id.decode() email_id = email_id.decode()
mail.store(email_id, '+X-GM-LABELS', 'processed')
# Properly remove the \Inbox label (archives the email)
# X-GM-LABELS is a Gmail IMAP extension
mail.store(email_id, '-X-GM-LABELS', r'(\Inbox)')
print(f"Archived message {email_id}")
except Exception as e: except Exception as e:
print(f"Error archiving message {email_id}: {e}") print(f"Error archiving message {email_id}: {e}")
@ -106,7 +105,12 @@ def check_notifications():
mail.login(GMAIL_USER, GMAIL_APP_PASSWORD) mail.login(GMAIL_USER, GMAIL_APP_PASSWORD)
mail.select("inbox") mail.select("inbox")
result, data = mail.search(None, '(UNSEEN)') #result, data = mail.search(None, '(UNSEEN)')
ok, data = mail.search(None, 'UNSEEN X-GM-LABELS', "inbox")
#search_criteria = r'(UNSEEN X-GM-LABELS "\\Inbox")'
#result, data = mail.uid('search', None, search_criteria)
mail_ids = data[0].split() mail_ids = data[0].split()
if not mail_ids: if not mail_ids:
@ -129,6 +133,9 @@ def check_notifications():
elif is_jira_email(from_email, subject): elif is_jira_email(from_email, subject):
source = "Jira" source = "Jira"
link = extract_jira_link(subject) link = extract_jira_link(subject)
elif is_slack_email(from_email, subject):
source = "Slack"
link = ""
else: else:
print(f"Skipping non-matching email: {subject}") print(f"Skipping non-matching email: {subject}")
continue continue
@ -138,6 +145,7 @@ def check_notifications():
send_ntfy_notification(subject, message) send_ntfy_notification(subject, message)
archive_message(mail, num) archive_message(mail, num)
mail.expunge()
mail.logout() mail.logout()
# ---------------------- # ----------------------