creates Taskwarrior tasks from unread emails fetched via IMAP, using a YAML configuration file for settings

twbymail

#!/bin/python

"""
This script allows you to automatically create tasks in Taskwarrior based on unread
emails fetched from an IMAP server. It fetches unread emails from a specified IMAP
server, parses the subject of each email, and creates a corresponding task in
Taskwarrior.

The script reads configuration parameters, such as IMAP server details, email address,
password command, IMAP folder, default project for tasks, and default tags for tasks,
from a YAML configuration file (`config.yaml`). It then uses these parameters to fetch
emails, parse their subjects, and create tasks accordingly.

The subject of each email is parsed to extract the task title and any project or tags
specified in the subject. If the subject starts with 'project:', the word that follows
the colon is treated as the value of the project parameter, replacing the default
project parameter. Any words that follow and start with a plus sign are treated as tags,
replacing the default tags parameter.

To use the script, you need to set up the `config.yaml` file with your IMAP and
Taskwarrior parameters. Then, you can run the script to automatically create tasks based
on unread emails.

Note: Ensure that you handle sensitive information, such as passwords, securely. It's
recommended to use tools like `pass` (passwordstore) to retrieve passwords securely.

Confguration file as an example : 
# config.yaml

imap:
  server: "imap.example.com"
  email_address: "your_email@example.com"
  password_command: "pass show myimappassword"
  folder: "inbox"

task:
  default_project: "inbox"
  default_tags: "email,todo"

"""

# GistID:9e8210b6af27e8f021bf295b0d4ab6e6

import os
import subprocess
import imaplib
import email
from email.header import decode_header
import yaml

def fetch_emails(server, email_address, password, folder):
    """
    Fetches unread emails from the specified IMAP server.

    Args:
        server (str): IMAP server address.
        email_address (str): Email address.
        password (str): Email password.
        folder (str): IMAP folder to search for unread emails.

    Returns:
        list: List of unread emails.
    """
    # Connect to the IMAP server
    mail = imaplib.IMAP4_SSL(server)
    mail.login(email_address, password)
    mail.select(folder)

    # Search for unread emails
    _, data = mail.search(None, "UNSEEN")
    email_ids = data[0].split()

    emails = []
    for email_id in email_ids:
        _, data = mail.fetch(email_id, "(RFC822)")
        raw_email = data[0][1]
        msg = email.message_from_bytes(raw_email)
        emails.append(msg)

    mail.close()
    mail.logout()
    return emails

def parse_email_subject(msg):
    """
    Parses the subject of the email.

    Args:
        msg (email.message.Message): Email message.

    Returns:
        str: Subject of the email.
    """
    subject = decode_header(msg["Subject"])[0][0]
    if isinstance(subject, bytes):
        try:
            subject = subject.decode()
        except UnicodeDecodeError:
            subject = subject.decode('latin1', errors='replace')  # Ignore invalid bytes
    return subject

def create_task(subject, default_project=None, default_tags=None):
    """
    Creates a task in Taskwarrior.

    Args:
        subject (str): Subject of the task.
        default_project (str, optional): Default project for the task. Defaults to None.
        default_tags (str, optional): Default tags for the task. Defaults to None.
    """
    project = default_project
    tags = default_tags.split(",") if default_tags else []

    # Check if subject starts with 'project:' or 'Project:'
    if subject.lower().startswith("project:"):
        parts = subject.split()
        project = parts[0].split(":")[1]
        subject = " ".join(parts[1:])

    # Parse tags
    parts = subject.split()
    new_subject = ""
    for part in parts:
        if part.startswith("+"):
            tags.append(part[1:])
        else:
            new_subject += part + " "
    new_subject = new_subject.strip()

    # Remove tags from the subject
    subject_parts = new_subject.split()
    new_subject_without_tags = []
    for part in subject_parts:
        if not part.startswith("+"):
            new_subject_without_tags.append(part)

    command = ["task", "add", " ".join(new_subject_without_tags)]
    if project:
        command.extend(["project:" + project])
    for tag in tags:
        command.append("+" + tag)
    subprocess.run(command)

def load_config():
    """
    Loads configuration from the YAML file.

    Returns:
        dict: Configuration parameters.
    """
    config_file = os.path.join(os.path.expanduser("~"), ".config", "twbymail", "config.yaml")
    with open(config_file, "r") as file:
        config = yaml.safe_load(file)
    return config

def main():
    """
    Main function to orchestrate the process of fetching unread emails
    from an IMAP server and creating corresponding tasks in Taskwarrior
    based on the email subjects.

    This function loads configuration parameters, retrieves passwords,
    fetches unread emails, parses their subjects, and creates tasks
    in Taskwarrior based on the parsed subjects.

    """
    # Load configuration
    config = load_config()

    # Retrieve password from command line
    password_process = subprocess.Popen(config["imap"]["password_command"], shell=True, stdout=subprocess.PIPE)
    password = password_process.stdout.readline().decode().strip()

    # Fetch emails
    emails = fetch_emails(config["imap"]["server"], config["imap"]["email_address"], password, config["imap"]["folder"])

    for email in emails:
        subject = parse_email_subject(email)
        create_task(subject, default_project=config["task"]["default_project"], default_tags=config["task"]["default_tags"])

if __name__ == "__main__":
    main()