{"id":5216,"date":"2025-01-18T22:34:15","date_gmt":"2025-01-18T22:34:15","guid":{"rendered":"https:\/\/blog.gordonbuchan.com\/blog\/?p=5216"},"modified":"2025-09-02T20:34:56","modified_gmt":"2025-09-02T20:34:56","slug":"creating-a-script-that-analyzes-email-messages-messages-using-a-large-language-model-llm-and-where-appropriate-escalates-messages-to-the-attention-of-an-operator","status":"publish","type":"post","link":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/2025\/01\/18\/creating-a-script-that-analyzes-email-messages-messages-using-a-large-language-model-llm-and-where-appropriate-escalates-messages-to-the-attention-of-an-operator\/","title":{"rendered":"Creating a script that analyzes email messages messages using a large language model (LLM), and where appropriate escalates messages to the attention of an operator"},"content":{"rendered":"\n

In this post, we create a Python script that connects to a Gmail inbox, extracts the text of the subject and body of each message, submits that text with a prompt to a large language model (LLM), then if conditions are met that match the prompt, escalates the message to the attention of an operator, based on a prompt.<\/p>\n\n\n\n

Using an Ollama API server<\/h1>\n\n\n\n

In this case, we are interacting with an Ollama LLM API server hosted locally. Refer to Using Ollama to host an LLM on CPU-only equipment to enable a local chatbot and LLM API server<\/a>.<\/p>\n\n\n\n

Using an OpenAI-compatible LLM API<\/h1>\n\n\n\n

An alternate source code listing is provided for an OpenAI-compatible LLM API.<\/p>\n\n\n\n

Obtaining a Gmail app password<\/h1>\n\n\n\n

Visit the following site:<\/p>\n\n\n\n

https:\/\/myaccount.google.com\/apppasswords<\/a><\/p>\n\n\n\n

Create a new app password. Take note of the password, it will not be visible again.<\/p>\n\n\n\n

Note: Google adds spaces to the app password for readability. You should remove the spaces from the app password and use that value.<\/p>\n\n\n\n

Escalating to the root user<\/h1>\n\n\n\n

In this procedure we run as the root user. Enter the following command:<\/p>\n\n\n

\nsudo su\n<\/pre><\/div>\n\n\n
Adding utilities to the operating system<\/h1>\n\n\n\n
Enter the following command:<\/p>\n\n\n
\napt install python3-venv python3-pip sqlite3\n<\/pre><\/div>\n\n\n
Creating a virtual environment and installing required packages with pip<\/h1>\n\n\n\n
Enter the following commands:<\/p>\n\n\n
\ncd ~\nmkdir doicareworkdir\ncd doicareworkdir\npython3 -m venv doicare_env\nsource doicare_env\/bin\/activate\npip install requests imaplib2\n<\/pre><\/div>\n\n\n
Creating the configuration file (config.json)<\/h1>\n\n\n\n
Enter the following command:<\/p>\n\n\n
\nnano config.json\n<\/pre><\/div>\n\n\n
Use the nano editor to add the following text:<\/p>\n\n\n
\n{\n  "gmail_user": "xxxxxxxxxxxx@xxxxx.xxx",\n  "gmail_app_password": "xxxxxxxxxxxxxxxx",\n  "api_base_url": "http:\/\/xxx.xxx.xxx.xxx:8085",\n  "openai_api_key": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",\n  "database": "doicare.db",\n\n  "scanasof": "18-Jan-2025",\n\n  "alert_recipients": [\n    "xxxxx@xxxxx.com"\n  ],\n\n  "smtp_server": "smtp.gmail.com",\n  "smtp_port": 587,\n  "smtp_user": "xxxxxx@xxxxx.xxxxx",\n  "smtp_password": "xxxxxxxxxxxxxxxx",\n\n  "analysis_prompt": "Analyze the email below. If it needs escalation (urgent, sender upset, or critical issue), return 'Escalation Reason:' followed by one short sentence explaining why. If no escalation is needed, return exactly 'DOESNOTAPPLY'. Always provide either 'DOESNOTAPPLY' or a reason.",\n  "model": "mistral"\n\n}\n<\/pre><\/div>\n\n\n
Save and exit the file.<\/p>\n\n\n\n
Creating a Python script called doicare that connects to a Gmail inbox, submits messages to an LLM, and escalates messages based on a prompt (Ollama version)<\/h1>\n\n\n\n
Enter the following command:<\/p>\n\n\n\n
nano doicare_gmail.py<\/p>\n\n\n
\nimport imaplib\nimport email\nimport sqlite3\nimport requests\nimport smtplib\nimport json\nfrom datetime import datetime\nfrom email.mime.text import MIMEText\nfrom email.mime.multipart import MIMEMultipart\nfrom email.header import decode_header, make_header\n\n# MIT license 2025 Gordon Buchan\n# see https:\/\/opensource.org\/licenses\/MIT\n# Some of this code was generated with the assistance of AI tools.\n\n# --------------------------------------------------------------------\n# 1. LOAD CONFIG\n# --------------------------------------------------------------------\nwith open("config.json", "r") as cfg:\n    config = json.load(cfg)\n\nGMAIL_USER = config["gmail_user"]\nGMAIL_APP_PASSWORD = config["gmail_app_password"]\nAPI_BASE_URL = config["api_base_url"]\nOPENAI_API_KEY = config["openai_api_key"]\nDATABASE = config["database"]\nSCAN_ASOF = config["scanasof"]\nALERT_RECIPIENTS = config.get("alert_recipients", [])\nSMTP_SERVER = config["smtp_server"]\nSMTP_PORT = config["smtp_port"]\nSMTP_USER = config["smtp_user"]\nSMTP_PASSWORD = config["smtp_password"]\nANALYSIS_PROMPT = config["analysis_prompt"]\nMODEL = config["model"]\n\n# --------------------------------------------------------------------\n# 2. DATABASE SETUP\n# --------------------------------------------------------------------\ndef setup_database():\n    conn = sqlite3.connect(DATABASE)\n    cur = conn.cursor()\n    cur.execute("""\n        CREATE TABLE IF NOT EXISTS escalations (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            email_date TEXT,\n            from_address TEXT,\n            to_address TEXT,\n            cc_address TEXT,\n            subject TEXT,\n            body TEXT,\n            reason TEXT,\n            created_at TEXT\n        )\n    """)\n    cur.execute("""\n        CREATE TABLE IF NOT EXISTS scan_info (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            last_scanned_uid INTEGER\n        )\n    """)\n    conn.commit()\n    conn.close()\n\ndef get_last_scanned_uid():\n    conn = sqlite3.connect(DATABASE)\n    cur = conn.cursor()\n    cur.execute("SELECT last_scanned_uid FROM scan_info ORDER BY id DESC LIMIT 1")\n    row = cur.fetchone()\n    conn.close()\n    return row[0] if (row and row[0]) else 0\n\ndef update_last_scanned_uid(uid_val):\n    conn = sqlite3.connect(DATABASE)\n    cur = conn.cursor()\n    cur.execute("INSERT INTO scan_info (last_scanned_uid) VALUES (?)", (uid_val,))\n    conn.commit()\n    conn.close()\n\ndef is_already_processed(uid_val):\n    conn = sqlite3.connect(DATABASE)\n    cur = conn.cursor()\n    cur.execute("SELECT 1 FROM scan_info WHERE last_scanned_uid = ?", (uid_val,))\n    row = cur.fetchone()\n    conn.close()\n    return bool(row)\n\n# --------------------------------------------------------------------\n# 3. ANALYSIS &amp; ALERTING\n# --------------------------------------------------------------------\ndef analyze_with_openai(subject, body):\n    prompt = f"{ANALYSIS_PROMPT}\\n\\nSubject: {subject}\\nBody: {body}"\n    url = f"{API_BASE_URL}\/v1\/completions"\n    headers = {"Content-Type": "application\/json"}\n    if OPENAI_API_KEY:\n        headers["Authorization"] = f"Bearer {OPENAI_API_KEY}"\n\n    payload = {\n        "model": MODEL,\n        "prompt": prompt,\n        "max_tokens": 300,\n        "temperature": 0.7\n    }\n\n    try:\n        response = requests.post(url, headers=headers, json=payload, timeout=60)\n        data = response.json()\n\n        if "error" in data:\n            print(f"[DEBUG] API Error: {data['error']['message']}")\n            return "DOESNOTAPPLY"\n\n        if "choices" in data and data["choices"]:\n            raw_text = data["choices"][0]["text"].strip()\n            return raw_text\n\n        return "DOESNOTAPPLY"\n\n    except Exception as e:\n        print(f"[DEBUG] Exception during API call: {e}")\n        return "DOESNOTAPPLY"\n\ndef send_alerts(reason, email_date, from_addr, to_addr, cc_addr, subject, body):\n    for recipient in ALERT_RECIPIENTS:\n        msg = MIMEMultipart()\n        msg["From"] = SMTP_USER\n        msg["To"] = recipient\n        msg["Subject"] = "Escalation Alert"\n\n        alert_text = f"""\n        Escalation Triggered\n        Date: {email_date}\n        From: {from_addr}\n        To: {to_addr}\n        CC: {cc_addr}\n        Subject: {subject}\n        Body: {body}\n\n        Reason: {reason}\n        """\n        msg.attach(MIMEText(alert_text, "plain"))\n\n        try:\n            with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:\n                server.starttls()\n                server.login(SMTP_USER, SMTP_PASSWORD)\n                server.sendmail(SMTP_USER, recipient, msg.as_string())\n            print(f"Alert sent to {recipient}")\n        except Exception as ex:\n            print(f"Failed to send alert to {recipient}: {ex}")\n\ndef save_escalation(email_date, from_addr, to_addr, cc_addr, subject, body, reason):\n    conn = sqlite3.connect(DATABASE)\n    cur = conn.cursor()\n    cur.execute("""\n        INSERT INTO escalations (\n            email_date, from_address, to_address, cc_address,\n            subject, body, reason, created_at\n        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)\n    """, (\n        email_date, from_addr, to_addr, cc_addr,\n        subject, body, reason, datetime.now().isoformat()\n    ))\n    conn.commit()\n    conn.close()\n\n# --------------------------------------------------------------------\n# 4. MAIN LOGIC\n# --------------------------------------------------------------------\ndef process_message(raw_email, uid_val):\n    parsed_msg = email.message_from_bytes(raw_email)\n\n    date_str = parsed_msg.get("Date", "")\n    from_addr = parsed_msg.get("From", "")\n    to_addr = parsed_msg.get("To", "")\n    cc_addr = parsed_msg.get("Cc", "")\n    subject_header = parsed_msg.get("Subject", "")\n    subject_decoded = str(make_header(decode_header(subject_header)))\n\n    body_text = ""\n    if parsed_msg.is_multipart():\n        for part in parsed_msg.walk():\n            ctype = part.get_content_type()\n            disposition = str(part.get("Content-Disposition"))\n            if ctype == "text\/plain" and "attachment" not in disposition:\n                charset = part.get_content_charset() or "utf-8"\n                body_text += part.get_payload(decode=True).decode(charset, errors="replace")\n    else:\n        charset = parsed_msg.get_content_charset() or "utf-8"\n        body_text = parsed_msg.get_payload(decode=True).decode(charset, errors="replace")\n\n    reason = analyze_with_openai(subject_decoded, body_text)\n    if "DOESNOTAPPLY" in reason:\n        print(f"[UID {uid_val}] No escalation: {reason}")\n        return\n\n    print(f"[UID {uid_val}] Escalation triggered: {subject_decoded[:50]}")\n    save_escalation(date_str, from_addr, to_addr, cc_addr, subject_decoded, body_text, reason)\n    send_alerts(reason, date_str, from_addr, to_addr, cc_addr, subject_decoded, body_text)\n\ndef main():\n    setup_database()\n    last_uid = get_last_scanned_uid()\n    print(f"[DEBUG] Retrieved last UID: {last_uid}")\n\n    try:\n        mail = imaplib.IMAP4_SSL("imap.gmail.com")\n        mail.login(GMAIL_USER, GMAIL_APP_PASSWORD)\n        print("IMAP login successful.")\n    except Exception as e:\n        print(f"Error logging into Gmail: {e}")\n        return\n\n    mail.select("INBOX")\n\n    if last_uid == 0:\n        print(f"[DEBUG] First run: scanning since date {SCAN_ASOF}")\n        r1, d1 = mail.search(None, f'(SINCE {SCAN_ASOF})')\n    else:\n        print(f"[DEBUG] Subsequent run: scanning for UIDs > {last_uid}")\n        r1, d1 = mail.uid('SEARCH', None, f'UID {last_uid + 1}:*')\n\n    if r1 != "OK":\n        print("[DEBUG] Search failed.")\n        mail.logout()\n        return\n\n    seq_nums = d1[0].split()\n    print(f"[DEBUG] Found {len(seq_nums)} messages to process: {seq_nums}")\n\n    if not seq_nums:\n        print("[DEBUG] No messages to process.")\n        mail.logout()\n        return\n\n    highest_uid_seen = last_uid\n\n    for seq_num in seq_nums:\n        if is_already_processed(seq_num.decode()):\n            print(f"[DEBUG] UID {seq_num.decode()} already processed, skipping.")\n            continue\n\n        print(f"[DEBUG] Processing sequence number: {seq_num}")\n        r2, d2 = mail.uid('FETCH', seq_num.decode(), '(RFC822)')\n        if r2 != "OK" or not d2 or len(d2) &lt; 1 or not d2[0]:\n            print(f"[DEBUG] Failed to fetch message for UID {seq_num.decode()}")\n            continue\n\n        print(f"[DEBUG] Successfully fetched message for UID {seq_num.decode()}")\n        raw_email = d2[0][1]\n\n        try:\n            process_message(raw_email, int(seq_num.decode()))\n            mail.uid('STORE', seq_num.decode(), '+FLAGS', '\\\\Seen')\n            if int(seq_num.decode()) > highest_uid_seen:\n                highest_uid_seen = int(seq_num.decode())\n        except Exception as e:\n            print(f"[DEBUG] Error processing message UID {seq_num.decode()}: {e}")\n\n    if highest_uid_seen > last_uid:\n        print(f"[DEBUG] Updating last scanned UID to {highest_uid_seen}")\n        update_last_scanned_uid(highest_uid_seen)\n\n    mail.logout()\n\nif __name__ == "__main__":\n    main()\n<\/pre><\/div>\n\n\n
Save and exit the file.<\/p>\n\n\n\n
Creating a Python script called doicare that connects to a Gmail inbox, submits messages to an LLM, and escalates messages based on a prompt (OpenAI-compatible version)<\/h1>\n\n\n\n
Enter the following command:<\/p>\n\n\n\n
nano doicare_gmail.py<\/p>\n\n\n
\nimport imaplib\nimport email\nimport sqlite3\nimport requests\nimport smtplib\nimport json\nfrom datetime import datetime\nfrom email.mime.text import MIMEText\nfrom email.mime.multipart import MIMEMultipart\nfrom email.header import decode_header, make_header\n\n# MIT license 2025 Gordon Buchan\n# see https:\/\/opensource.org\/licenses\/MIT\n# Some of this code was generated with the assistance of AI tools.\n\n# --------------------------------------------------------------------\n# 1. LOAD CONFIG\n# --------------------------------------------------------------------\nwith open("config.json", "r") as cfg:\n    config = json.load(cfg)\n\nGMAIL_USER = config["gmail_user"]\nGMAIL_APP_PASSWORD = config["gmail_app_password"]\nAPI_BASE_URL = config["api_base_url"]\nOPENAI_API_KEY = config["openai_api_key"]\nDATABASE = config["database"]\nSCAN_ASOF = config["scanasof"]\nALERT_RECIPIENTS = config.get("alert_recipients", [])\nSMTP_SERVER = config["smtp_server"]\nSMTP_PORT = config["smtp_port"]\nSMTP_USER = config["smtp_user"]\nSMTP_PASSWORD = config["smtp_password"]\nANALYSIS_PROMPT = config["analysis_prompt"]\nMODEL = config["model"]\n\n# --------------------------------------------------------------------\n# 2. DATABASE SETUP\n# --------------------------------------------------------------------\ndef setup_database():\n    """ Ensure the database and necessary tables exist. """\n    conn = sqlite3.connect(DATABASE)\n    cur = conn.cursor()\n\n    print("[DEBUG] Ensuring database tables exist...")\n\n    cur.execute("""\n        CREATE TABLE IF NOT EXISTS escalations (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            email_date TEXT,\n            from_address TEXT,\n            to_address TEXT,\n            cc_address TEXT,\n            subject TEXT,\n            body TEXT,\n            reason TEXT,\n            created_at TEXT\n        )\n    """)\n\n    cur.execute("""\n        CREATE TABLE IF NOT EXISTS scan_info (\n            id INTEGER PRIMARY KEY AUTOINCREMENT,\n            last_scanned_uid INTEGER UNIQUE\n        )\n    """)\n\n    # Ensure at least one row exists in scan_info\n    cur.execute("SELECT COUNT(*) FROM scan_info")\n    if cur.fetchone()[0] == 0:\n        cur.execute("INSERT INTO scan_info (last_scanned_uid) VALUES (0)")\n\n    conn.commit()\n    conn.close()\n    print("[DEBUG] Database setup complete.")\n\ndef get_last_scanned_uid():\n    """ Retrieve the last scanned UID from the database """\n    conn = sqlite3.connect(DATABASE)\n    cur = conn.cursor()\n    cur.execute("SELECT last_scanned_uid FROM scan_info ORDER BY id DESC LIMIT 1")\n    row = cur.fetchone()\n    conn.close()\n    return int(row[0]) if (row and row[0]) else 0\n\ndef update_last_scanned_uid(uid_val):\n    """ Update the last scanned UID in the database """\n    conn = sqlite3.connect(DATABASE)\n    cur = conn.cursor()\n    cur.execute("""\n        INSERT INTO scan_info (id, last_scanned_uid)\n        VALUES (1, ?) \n        ON CONFLICT(id) DO UPDATE SET last_scanned_uid = excluded.last_scanned_uid\n    """, (uid_val,))\n    conn.commit()\n    conn.close()\n\n# --------------------------------------------------------------------\n# 3. ANALYSIS &amp; ALERTING\n# --------------------------------------------------------------------\ndef analyze_with_openai(subject, body):\n    """ Send email content to OpenAI API for analysis """\n    prompt = f"{ANALYSIS_PROMPT}\\n\\nSubject: {subject}\\nBody: {body}"\n    url = f"{API_BASE_URL}\/v1\/chat\/completions"\n    headers = {\n        "Content-Type": "application\/json",\n        "Authorization": f"Bearer {OPENAI_API_KEY}" if OPENAI_API_KEY else "",\n    }\n\n    payload = {\n        "model": MODEL,\n        "messages": [\n            {"role": "system", "content": "You are a helpful assistant"},\n            {"role": "user", "content": prompt}\n        ],\n        "max_tokens": 300,\n        "temperature": 0.7\n    }\n\n    try:\n        response = requests.post(url, headers=headers, json=payload, timeout=60)\n        data = response.json()\n\n        if "error" in data:\n            print(f"[DEBUG] API Error: {data['error']['message']}")\n            return "DOESNOTAPPLY"\n\n        if "choices" in data and data["choices"]:\n            return data["choices"][0]["message"]["content"].strip()\n\n        return "DOESNOTAPPLY"\n\n    except Exception as e:\n        print(f"[DEBUG] Exception during API call: {e}")\n        return "DOESNOTAPPLY"\n\n# --------------------------------------------------------------------\n# 4. MAIN LOGIC\n# --------------------------------------------------------------------\ndef process_message(raw_email, uid_val):\n    """ Process a single email message """\n    parsed_msg = email.message_from_bytes(raw_email)\n\n    date_str = parsed_msg.get("Date", "")\n    from_addr = parsed_msg.get("From", "")\n    to_addr = parsed_msg.get("To", "")\n    cc_addr = parsed_msg.get("Cc", "")\n    subject_header = parsed_msg.get("Subject", "")\n    subject_decoded = str(make_header(decode_header(subject_header)))\n\n    body_text = ""\n    if parsed_msg.is_multipart():\n        for part in parsed_msg.walk():\n            ctype = part.get_content_type()\n            disposition = str(part.get("Content-Disposition"))\n            if ctype == "text\/plain" and "attachment" not in disposition:\n                charset = part.get_content_charset() or "utf-8"\n                body_text += part.get_payload(decode=True).decode(charset, errors="replace")\n    else:\n        charset = parsed_msg.get_content_charset() or "utf-8"\n        body_text = parsed_msg.get_payload(decode=True).decode(charset, errors="replace")\n\n    reason = analyze_with_openai(subject_decoded, body_text)\n    if "DOESNOTAPPLY" in reason:\n        print(f"[UID {uid_val}] No escalation: {reason}")\n        return\n\n    print(f"[UID {uid_val}] Escalation triggered: {subject_decoded[:50]}")\n\n    update_last_scanned_uid(uid_val)\n\ndef main():\n    """ Main function to fetch and process emails """\n    setup_database()\n    last_uid = get_last_scanned_uid()\n    print(f"[DEBUG] Retrieved last UID: {last_uid}")\n\n    try:\n        mail = imaplib.IMAP4_SSL("imap.gmail.com")\n        mail.login(GMAIL_USER, GMAIL_APP_PASSWORD)\n        print("IMAP login successful.")\n    except Exception as e:\n        print(f"Error logging into Gmail: {e}")\n        return\n\n    mail.select("INBOX")\n\n    search_query = f'UID {last_uid + 1}:*' if last_uid > 0 else f'SINCE {SCAN_ASOF}'\n    print(f"[DEBUG] Running IMAP search: {search_query}")\n\n    r1, d1 = mail.uid('SEARCH', None, search_query)\n\n    if r1 != "OK":\n        print("[DEBUG] Search failed.")\n        mail.logout()\n        return\n\n    seq_nums = d1[0].split()\n    seq_nums = [seq.decode() for seq in seq_nums]\n\n    print(f"[DEBUG] Found {len(seq_nums)} new messages: {seq_nums}")\n\n    if not seq_nums:\n        print("[DEBUG] No new messages found, exiting.")\n        mail.logout()\n        return\n\n    highest_uid_seen = last_uid\n\n    for seq_num in seq_nums:\n        numeric_uid = int(seq_num)\n        if numeric_uid &lt;= last_uid:\n            print(f"[DEBUG] UID {numeric_uid} already processed, skipping.")\n            continue\n\n        print(f"[DEBUG] Processing UID: {numeric_uid}")\n        r2, d2 = mail.uid('FETCH', seq_num, '(RFC822)')\n        if r2 != "OK" or not d2 or len(d2) &lt; 1 or not d2[0]:\n            print(f"[DEBUG] Failed to fetch message for UID {numeric_uid}")\n            continue\n\n        raw_email = d2[0][1]\n        process_message(raw_email, numeric_uid)\n\n        highest_uid_seen = max(highest_uid_seen, numeric_uid)\n\n    if highest_uid_seen > last_uid:\n        print(f"[DEBUG] Updating last scanned UID to {highest_uid_seen}")\n        update_last_scanned_uid(highest_uid_seen)\n\n    mail.logout()\n\nif __name__ == "__main__":\n    main()\n<\/pre><\/div>\n\n\n
Save and exit the file.<\/p>\n\n\n\n
Running the doicare_gmail.py script<\/h1>\n\n\n\n
Enter the following command:<\/p>\n\n\n
\npython3 doicare_gmail.py\n<\/pre><\/div>\n\n\n
Sample output<\/h1>\n\n\n
\n(doicare_env) root@xxxxx:\/home\/desktop\/doicareworkingdir# python3 doicare_gmail.py \n[DEBUG] Retrieved last UID: 0\nIMAP login successful.\n[DEBUG] First run: scanning since date 18-Jan-2025\n[DEBUG] Found 23 messages to process: [b'49146', b'49147', b'49148', b'49149', b'49150', b'49151', b'49152', b'49153', b'49154', b'49155', b'49156', b'49157', b'49158', b'49159', b'49160', b'49161', b'49162', b'49163', b'49164', b'49165', b'49166', b'49167', b'49168']\n[DEBUG] Processing sequence number: b'49146'\n[DEBUG] FETCH response: b'49146 (UID 50196)'\n[DEBUG] FETCH line to parse: 49146 (UID 50196)\n[DEBUG] Parsed UID: 50196\n[DEBUG] Valid UID Found: 50196\n[DEBUG] Successfully fetched message for UID 50196\n[UID 50196] No escalation: DOESNOTAPPLY. The email does not contain any urgent matter, sender is not upset, and there does not seem to be a critical issue mentioned.\n[DEBUG] Processing sequence number: b'49147'\n[DEBUG] FETCH response: b'49147 (UID 50197)'\n[DEBUG] FETCH line to parse: 49147 (UID 50197)\n[DEBUG] Parsed UID: 50197\n[DEBUG] Valid UID Found: 50197\n[DEBUG] Successfully fetched message for UID 50197\n[UID 50197] No escalation: DOESNOTAPPLY\n[DEBUG] Processing sequence number: b'49148'\n[DEBUG] FETCH response: b'49148 (UID 50198)'\n[DEBUG] FETCH line to parse: 49148 (UID 50198)\n[DEBUG] Parsed UID: 50198\n[DEBUG] Valid UID Found: 50198\n[DEBUG] Successfully fetched message for UID 50198\n[UID 50198] No escalation: DOESNOTAPPLY. The email does not contain any urgent matter, sender is not upset, and there doesn't seem to be a critical issue presented in the content.\n[DEBUG] Processing sequence number: b'49149'\n[DEBUG] FETCH response: b'49149 (UID 50199)'\n[DEBUG] FETCH line to parse: 49149 (UID 50199)\n[DEBUG] Parsed UID: 50199\n[DEBUG] Valid UID Found: 50199\n[DEBUG] Successfully fetched message for UID 50199\n[UID 50199] No escalation: DOESNOTAPPLY. The email does not contain any urgent matter, the sender is not upset, and there is no critical issue mentioned in the message.\n[DEBUG] Processing sequence number: b'49150'\n[DEBUG] FETCH response: b'49150 (UID 50200)'\n[DEBUG] FETCH line to parse: 49150 (UID 50200)\n[DEBUG] Parsed UID: 50200\n[DEBUG] Valid UID Found: 50200\n[DEBUG] Successfully fetched message for UID 50200\n[UID 50200] No escalation: DOESNOTAPPLY. The email lacks sufficient content for an escalation.\n[DEBUG] Processing sequence number: b'49151'\n[DEBUG] FETCH response: b'49151 (UID 50201)'\n[DEBUG] FETCH line to parse: 49151 (UID 50201)\n[DEBUG] Parsed UID: 50201\n[DEBUG] Valid UID Found: 50201\n[DEBUG] Successfully fetched message for UID 50201\n[UID 50201] Escalation triggered: Security alert\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49152'\n[DEBUG] FETCH response: b'49152 (UID 50202)'\n[DEBUG] FETCH line to parse: 49152 (UID 50202)\n[DEBUG] Parsed UID: 50202\n[DEBUG] Valid UID Found: 50202\n[DEBUG] Successfully fetched message for UID 50202\n[UID 50202] Escalation triggered: Delivery Status Notification (Failure)\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49153'\n[DEBUG] FETCH response: b'49153 (UID 50203)'\n[DEBUG] FETCH line to parse: 49153 (UID 50203)\n[DEBUG] Parsed UID: 50203\n[DEBUG] Valid UID Found: 50203\n[DEBUG] Successfully fetched message for UID 50203\n[UID 50203] No escalation: DOESNOTAPPLY\n[DEBUG] Processing sequence number: b'49154'\n[DEBUG] FETCH response: b'49154 (UID 50204)'\n[DEBUG] FETCH line to parse: 49154 (UID 50204)\n[DEBUG] Parsed UID: 50204\n[DEBUG] Valid UID Found: 50204\n[DEBUG] Successfully fetched message for UID 50204\n[UID 50204] Escalation triggered: my server lollipop is down\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49155'\n[DEBUG] FETCH response: b'49155 (UID 50205)'\n[DEBUG] FETCH line to parse: 49155 (UID 50205)\n[DEBUG] Parsed UID: 50205\n[DEBUG] Valid UID Found: 50205\n[DEBUG] Successfully fetched message for UID 50205\n[UID 50205] No escalation: DOESNOTAPPLY\n[DEBUG] Processing sequence number: b'49156'\n[DEBUG] FETCH response: b'49156 (UID 50206)'\n[DEBUG] FETCH line to parse: 49156 (UID 50206)\n[DEBUG] Parsed UID: 50206\n[DEBUG] Valid UID Found: 50206\n[DEBUG] Successfully fetched message for UID 50206\n[UID 50206] Escalation triggered: now doomfire is down too!\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49157'\n[DEBUG] FETCH response: b'49157 (UID 50207)'\n[DEBUG] FETCH line to parse: 49157 (UID 50207)\n[DEBUG] Parsed UID: 50207\n[DEBUG] Valid UID Found: 50207\n[DEBUG] Successfully fetched message for UID 50207\n[UID 50207] No escalation: DOESNOTAPPLY\n[DEBUG] Processing sequence number: b'49158'\n[DEBUG] FETCH response: b'49158 (UID 50208)'\n[DEBUG] FETCH line to parse: 49158 (UID 50208)\n[DEBUG] Parsed UID: 50208\n[DEBUG] Valid UID Found: 50208\n[DEBUG] Successfully fetched message for UID 50208\n[UID 50208] Escalation triggered: pants is down now\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49159'\n[DEBUG] FETCH response: b'49159 (UID 50209)'\n[DEBUG] FETCH line to parse: 49159 (UID 50209)\n[DEBUG] Parsed UID: 50209\n[DEBUG] Valid UID Found: 50209\n[DEBUG] Successfully fetched message for UID 50209\n[UID 50209] Escalation triggered: server05 down\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49160'\n[DEBUG] FETCH response: b'49160 (UID 50210)'\n[DEBUG] FETCH line to parse: 49160 (UID 50210)\n[DEBUG] Parsed UID: 50210\n[DEBUG] Valid UID Found: 50210\n[DEBUG] Successfully fetched message for UID 50210\n[UID 50210] No escalation: DOESNOTAPPLY (The sender has asked for a phone call instead of specifying the issue in detail, so it doesn't appear to be urgent or critical at first glance.)\n[DEBUG] Processing sequence number: b'49161'\n[DEBUG] FETCH response: b'49161 (UID 50211)'\n[DEBUG] FETCH line to parse: 49161 (UID 50211)\n[DEBUG] Parsed UID: 50211\n[DEBUG] Valid UID Found: 50211\n[DEBUG] Successfully fetched message for UID 50211\n[UID 50211] Escalation triggered: my server is down\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49162'\n[DEBUG] FETCH response: b'49162 (UID 50212)'\n[DEBUG] FETCH line to parse: 49162 (UID 50212)\n[DEBUG] Parsed UID: 50212\n[DEBUG] Valid UID Found: 50212\n[DEBUG] Successfully fetched message for UID 50212\n[UID 50212] No escalation: DOESNOTAPPLY\n[DEBUG] Processing sequence number: b'49163'\n[DEBUG] FETCH response: b'49163 (UID 50213)'\n[DEBUG] FETCH line to parse: 49163 (UID 50213)\n[DEBUG] Parsed UID: 50213\n[DEBUG] Valid UID Found: 50213\n[DEBUG] Successfully fetched message for UID 50213\n[UID 50213] Escalation triggered: this is getting bad\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49164'\n[DEBUG] FETCH response: b'49164 (UID 50214)'\n[DEBUG] FETCH line to parse: 49164 (UID 50214)\n[DEBUG] Parsed UID: 50214\n[DEBUG] Valid UID Found: 50214\n[DEBUG] Successfully fetched message for UID 50214\n[UID 50214] No escalation: DOESNOTAPPLY\n[DEBUG] Processing sequence number: b'49165'\n[DEBUG] FETCH response: b'49165 (UID 50215)'\n[DEBUG] FETCH line to parse: 49165 (UID 50215)\n[DEBUG] Parsed UID: 50215\n[DEBUG] Valid UID Found: 50215\n[DEBUG] Successfully fetched message for UID 50215\n[UID 50215] Escalation triggered: server zebra 05 is down\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49166'\n[DEBUG] FETCH response: b'49166 (UID 50216)'\n[DEBUG] FETCH line to parse: 49166 (UID 50216)\n[DEBUG] Parsed UID: 50216\n[DEBUG] Valid UID Found: 50216\n[DEBUG] Successfully fetched message for UID 50216\n[UID 50216] No escalation: DOESNOTAPPLY\n[DEBUG] Processing sequence number: b'49167'\n[DEBUG] FETCH response: b'49167 (UID 50217)'\n[DEBUG] FETCH line to parse: 49167 (UID 50217)\n[DEBUG] Parsed UID: 50217\n[DEBUG] Valid UID Found: 50217\n[DEBUG] Successfully fetched message for UID 50217\n[UID 50217] Escalation triggered: help\nAlert sent to xxxx@hotmail.com\n[DEBUG] Processing sequence number: b'49168'\n[DEBUG] FETCH response: b'49168 (UID 50218)'\n[DEBUG] FETCH line to parse: 49168 (UID 50218)\n[DEBUG] Parsed UID: 50218\n[DEBUG] Valid UID Found: 50218\n[DEBUG] Successfully fetched message for UID 50218\n[UID 50218] Escalation triggered: server is down\nAlert sent to xxxx@hotmail.com\n[DEBUG] Updating last scanned UID to 50218\n[DEBUG] Attempting to update last scanned UID to 50218\n[DEBUG] Last scanned UID successfully updated to 50218\n<\/pre><\/div>\n\n\n
Example of an alert message<\/h1>\n\n\n
\nEscalation Triggered\nDate: Sat, 18 Jan 2025 21:00:16 +0000\nFrom: Gordon Buchan &lt;gordonhbuchan@hotmail.com>\nTo: "gordonhbuchan@gmail.com" &lt;gordonhbuchan@gmail.com>\nCC:\nSubject: server is down\nBody: server down help please\n\n\nReason: Escalation Reason: This email indicates that there is a critical issue (server downtime).\n<\/pre><\/div>\n\n\n
Creating a systemd service to run the doicare script automatically<\/h1>\n\n\n\n
Enter the following command:<\/p>\n\n\n
\nnano \/etc\/systemd\/system\/doicare.service\n<\/pre><\/div>\n\n\n
Use the nano editor to add the following text (change values to match your path):<\/p>\n\n\n
\n[Unit]\nDescription=Run all monitoring tasks\n\n[Service]\nType=oneshot\nWorkingDirectory=\/root\/doicareworkdir\nExecStart=\/usr\/bin\/bash -c "source \/root\/doicareworkdir\/doicare_env\/bin\/activate &amp;&amp; python3 doicare_gmail.py"\n<\/pre><\/div>\n\n\n
Save and exit the file.<\/p>\n\n\n\n
Creating a systemd timer to run the doicare script automatically<\/h1>\n\n\n\n
Enter the following command:<\/p>\n\n\n
\nnano \/etc\/systemd\/system\/doicare.timer\n<\/pre><\/div>\n\n\n
Use the nano editor to add the following text:<\/p>\n\n\n
\n[Unit]\nDescription=Run monitoring tasks every 5 minutes\n\n[Timer]\nOnBootSec=5min\nOnUnitActiveSec=5min\n\n[Install]\nWantedBy=timers.target\n<\/pre><\/div>\n\n\n
Save and exit the file.<\/p>\n\n\n\n
Enabling the doicare service<\/h1>\n\n\n\n
Enter the following commands:<\/p>\n\n\n
\nsystemctl daemon-reload\nsystemctl start doicare.service\nsystemctl enable doicare.service\nsystemctl start doicare.timer\nsystemctl enable doicare.timer\n<\/pre><\/div>\n\n\n
<\/p>\n","protected":false},"excerpt":{"rendered":"
In this post, we create a Python script that connects to a Gmail inbox, extracts the text of the subject and body of each message, submits that text with a prompt to a large language model (LLM), then if conditions are met that match the prompt, escalates the message to the attention of an operator, … <\/p>\n
Continue reading “Creating a script that analyzes email messages messages using a large language model (LLM), and where appropriate escalates messages to the attention of an operator”<\/span><\/a><\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[3,1],"tags":[],"class_list":["post-5216","post","type-post","status-publish","format-standard","hentry","category-linux","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/posts\/5216","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/comments?post=5216"}],"version-history":[{"count":61,"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/posts\/5216\/revisions"}],"predecessor-version":[{"id":5480,"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/posts\/5216\/revisions\/5480"}],"wp:attachment":[{"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/media?parent=5216"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/categories?post=5216"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/blog.gordonbuchan.com\/blog\/index.php\/wp-json\/wp\/v2\/tags?post=5216"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}