path: root/main.py

from email.message import EmailMessage
from email.headerregistry import Address
from urllib.parse import urlencode
import http.cookies
import password
from password import Passwords
from uuid import uuid4
import os
from typing import (
    Optional,
    cast,
)
from mu import mu_search, get_mail
from html_render import HTML, render_document


from flask import (
    Flask,
    request,
    redirect,
    url_for,
    flash,
    get_flashed_messages
)


def mailto(addr: str) -> HTML:
    return ('a', {'href': f'mailto:{addr}'}, addr)


def format_email(addr: Address) -> list[HTML]:
    mail_addr = f'{addr.username}@{addr.domain}'
    return [addr.display_name, ' <', mailto(mail_addr), '>']


def header_format(key: str, value) -> HTML:
    if key in ['to', 'cc', 'bcc']:
        return ('ul', *[('li', *format_email(addr))
                        for addr in value.addresses])
    elif key == 'from':
        return format_email(value.addresses[0])
    elif key == 'in-reply-to':
        # type(value) == email.headerregistry._UnstructuredHeader
        id = str(value).strip("<>")
        return ['<', ('a', {'href': '?' + urlencode({'id': id})}, id), '>']
    else:
        return value


style: HTML = lambda: """
    nav {
        display: block;
        width: 100%;
        height: 4em;
        color: white;
        background-color: darkgrey;
    }

    dl {
        display: grid;
        grid-template-columns: 10ch auto;
    }
    dt {
        font-weight: bold;
    }
    dd {
        font-family: mono;
        font-size: 80%;
    }
    dd > * {
        margin: 0;
    }

    """


def attachement_tree(mail: EmailMessage) -> HTML:
    ct = mail.get_content_type()
    fn = mail.get_filename()

    children = []
    for child in mail.iter_parts():
        children.append(attachement_tree(cast(EmailMessage, child)))

    content: HTML
    if children:
        content = ('ul', *children)
    else:
        content = []

    if fn:
        body = f'{ct} {fn}'
    else:
        body = str(ct)
    return ('li', body, content)

# --------------------------------------------------


def login_page(returnto: Optional[str] = None) -> HTML:
    return ('form', {'action': '/login', 'method': 'POST'},
            ('input', {'name': 'username', 'placeholder': 'Username'}),
            ('input', {'type': 'password',
                       'placeholder': 'Password',
                       'name': 'password'}),
            ('input', {'type': 'hidden',
                       'name': 'returnto',
                       'value': returnto})
            if returnto else [],
            ('input', {'type': 'submit'}),
            )


def user_info(username: str) -> HTML:
    return [('span', username),
            ('form', {'action': '/logout', 'method': 'POST'},
                ('input', {'type': 'submit', 'value': 'Logga ut'}))]


def login_prompt() -> HTML:
    return ('a', {'href': '/login'}, 'Logga in')


def flashed_messages() -> HTML:
    return ('ul', {'class': 'flashes'},
            *[('li', msg) for msg in get_flashed_messages()])


def page_base(title: Optional[str] = None,
              body: HTML = [],
              username: Optional[str] = None) -> HTML:
    return ('html',
            ('head',
             ('meta', {'charset': 'utf-8'}),
             ('title', title),
             ('style', style),
             ),
            ('body',
             ('nav',
              user_info(username) if username else login_prompt()
              ),
             flashed_messages(),
             body))


def response_for(id: str, username: Optional[str] = None) -> str:

    mail = cast(EmailMessage, get_mail(id))

    headers = {}
    for (key, value) in mail.items():
        headers[key.lower()] = value

    head = []
    for h in ['date', 'from', 'to', 'cc', 'bcc', 'subject', 'x-original-to',
              'in-reply-to']:
        if x := headers.get(h.lower()):
            head += [('dt', h.title()),
                     ('dd', header_format(h.lower(), x))]

    body_part = mail.get_body(preferencelist=('html', 'plain'))
    if not body_part:
        raise ValueError("No suitable body in email")
    ct = body_part.get_content_type()
    body: HTML
    if ct == 'text/html':
        body = lambda: cast(EmailMessage, body_part).get_content()
    else:
        body = ('pre', cast(EmailMessage, body_part).get_content())

    if t := headers.get('subject'):
        title = f'Mail — {t}'
    else:
        title = 'Mail'

    main_body = [('dl', *head),
                 ('hr',),
                 ('main', body),
                 ('hr',),
                 ('ul', attachement_tree(mail)),
                 ]
    html_str = render_document(page_base(title=title,
                                         body=main_body,
                                         username=username))

    return html_str


def search_field(q: str) -> HTML:
    return ('form', {'action': '/search', 'method': 'GET'},
            ('label', {'for': 'search'},
             'Mu Search Query'),
            ('textarea', {'id': 'search', 'name': 'q'},
             q),
            ('input', {'type': 'Submit'}))


def search_result(q, by, reverse):

    keys = ['From', 'To', 'Subject', 'Date', 'Size', 'Maildir', 'Msgid']

    rows = mu_search(q, by, reverse)
    body = []
    for row in rows:
        rowdata = ['tr']
        for key in keys:
            rowdata.append(row[key])
        body.append(rowdata)

    return ('table',
            ('thead',
             ('tr',
              [('th', m) for m in keys])),
            ('tbody',
             body
             ))


def search_page(q, by, username=None):
    main_body = [search_field(q)]

    if q:
        main_body.append(search_result(q, by, False))

    return render_document(page_base(title='Serach',
                                     body=main_body,
                                     username=username))


def index_page(username):
    ids = [
        'CAEzixGsw-4zJ8_ejK_vDgmcQ9s-MbBc-ho+HL4arV4a+ghOOPg@mail.gmail.com',
        'CA+pcBt-gLb0GtbFOjJ5_7Q_WXtqApVPQ9w-3O7GH=VqCEQat6g@mail.gmail.com',
    ]

    body = [('h1', "Sample ID's"),
            ('ul',
             [('li', ('a', {'href': '?' + urlencode({'id': id})}, id))
              for id in ids]
             ),
            ]

    return render_document(page_base(title='Mail index',
                                     body=body,
                                     username=username
                                     ))


valid_session_cookies: dict[str, str] = {}

passwords: Passwords = password.Passwords(cast(os.PathLike, 'passwords.json'))


def is_logged_in():
    c = request.cookies.get('session')
    if c and valid_session_cookies.get(c):
        return valid_session_cookies[c]
    return False


app = Flask(__name__)


@app.route('/')
def index():
    login = is_logged_in()
    if not login:
        return redirect(url_for('login_page_', returnto=request.path))
    if id := request.args.get('id'):
        print("id =", id)
        response = response_for(''.join(id).replace(' ', '+'),
                                login)
    else:
        response = index_page(login)
    return response


@app.route('/search')
def search_page_():
    login = is_logged_in()
    if not login:
        return redirect(url_for('login_page_', returnto=request.path))
    return search_page(request.args.get('q'),
                       request.args.get('by'),
                       login)


@app.route('/login', methods=['GET'])
def login_page_():
    if not is_logged_in():
        body = login_page(request.args.get('returnto'))
        return render_document(page_base(title='Login', body=body))
    else:
        # TODO do something sensible here
        pass


@app.route('/login', methods=['POST'])
def login_form():
    global valid_session_cookies
    logged_in = is_logged_in()

    resp = redirect(request.args.get('returnto', url_for('index')))
    if logged_in:
        flash('Already loged in')
        return resp

    username = request.form['username']
    password = request.form['password']
    if passwords.validate(username, password):
        unique = str(uuid4())
        valid_session_cookies[unique] = username
        resp.set_cookie('session', unique)
    else:
        flash('Invalid username or password')
    return resp


@app.route('/logout', methods=['POST'])
def logout_form():
    global valid_session_cookies
    logged_in = is_logged_in()
    if not logged_in:
        flash('Not logged in')
        return redirect(url_for('index'))
    c = request.cookies.get('session')
    if valid_session_cookies.get(c):
        del valid_session_cookies[c]
    resp = redirect(url_for('index'))
    resp.set_cookie('session', '')
    return resp


if __name__ == '__main__':
    app.run(debug=True, port=8090)