Wenn ich im Internet unterwegs bin, sei es mit dem Handy, meinem MacBook Pro oder meinem Dienstrechner im Büro, stoße ich immer wieder auf Dinge, die ich zwar interessant finde, mit denen ich mich aber später beschäftigen möchte. Für solche Anlässe brauche ich eine Möglichkeit, meine Trouvaillen zuverlässig zu speichern. Vor einiger Zeit griff ich dafür auf DevonThink zurück. Das Programm läuft aber nicht plattformübergreifend. Eine andere Lösung musste also her.
Ein hervorragender Anlass, selbst etwas zu basteln! Da das Ganze schnell recht groß und übersichtlich werden kann, beschloss ich, in kleinen Schritten vorzugehen. Den ersten möchte ich heute vorstellen.1
Mithilfe eines Python-Skriptes werde ich mich in einem E-Mail-Konto einloggen, prüfen, ob es neue Nachrichten gibt, diese herunterladen und schließlich in einer SQL-Datenbank auf meinem Raspberry Pi abspeichern. Von dort aus nämlich kann ich immer wieder auf sie zugreifen.
Als Datenbank nehme ich SQLite. Sie wird standardmäßig von Python unterstützt und reicht für meine Zwecke vollständig aus. Um auf die E-Mails zugreifen zu können, habe ich mich für imapclient entschieden. Da E-Mails in einem vertrackten Format auf dem Server liegen, nutze ich außerdem pyzmail, um sie zu parsen und in ein dictionary umzuwandeln.
In Pythons Standardbibliothek ist bereits ein Adapter für SQLite enthalten, deshalb brauchte ich dafür nichts weiter zu installieren. Anders sieht es bei imapclient und pyzmail aus. Ich schaffe zuerst eine virtuelle Umgebung für mein kleines Projekt und installiere dann in dieser die nötigen Pakete:
python3 -m venv ~/.virtualenvs/email
source ~/.virtualenvs/email/bin/activate
pip install imapclient pyzmail36
Da das Paket pyzmail vom Entwickler nicht mehr weiterentwickelt und an die neueren Python-Versionen angepasst wird, muss ein Fork namens “pyzmail36” über pip installiert werden.
Danach lege ich ein neues Verzeichnis an, erstelle eine Skript-Datei mit touch email.py und öffne den Editor meiner Wahl. In das Skript kommt als erstes die sogenannte shebang und die Definition des character encodings, dann binde ich die notwendigen Pakete ein:
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sqlite3
import imapclient
import pyzmail # pip3 install pyzmail36
Anschließend kümmere ich mich um die Datenbankanbindung. Damit sich die einzelne Bausteine leichter erweitern und in andere Projekte einbinden lassen, erstelle ich dafür Objekte. Über den ‘Bauplan’ für die jeweiligen Objekte geben die Klassen mit ihren Methoden Auskunft.
Die erste Klasse ist für ein ‘Storage’-Objekt zuständig und kümmert sich um die Speicherung der Daten.
class Storage(object):
def __init__(self):
self.connection = sqlite3.connect('storage.db')
self.cursor = self.connection.cursor()
Wenn das Storage-Objekt erstellt wird, legt es eine SQLite-Datenbank namens ‘storage.db’ an und baut eine Verbindung zu ihr auf. Mithilfe des sogenannten cursors, den ich daraufhin deklariere, können SQL-Befehle programmatisch an die Datenbank übermittelt werden.
In der Methode ‘create_table’ lege ich die Tabelle ‘email’ in der Datenbank an:
def create_table(self):
try:
self.cursor.execute("""CREATE TABLE IF NOT EXISTS email
(id INTEGER PRIMARY KEY,
sender VARCHAR(100),
receiver VARCHAR(100),
subject VARCHAR(200),
text TEXT,
html TEXT,
created TIMESTAMP DEFAULT CURRENT_TIMESTAMP
NOT NULL)
""")
print('Creating table...')
self.cursor.execute("""CREATE UNIQUE INDEX sender_subject ON email
(sender, subject)""")
except sqlite3.OperationalError:
print('... but table already exists.')
Um etwaige Fehler abzufangen, packe ich die eigentlichen Anweisungen in ein try/except-Konstrukt. Die Tabelle soll erstellt werden, falls sie noch nicht existiert, und die Spalten ‘id’, ‘sender’, ‘receiver’, ‘subject’, ‘text’, ‘html’ und ‘created’ umfassen. Die erste Spalte ist bei SQLite zwar nicht zwingend notwendig, weil die Datenbank von sich aus eine rowid vergibt. Ich finde es aber transparenter, die Spalte explizit zu deklarieren. Der Wert der Spalte erhöht sich automatisch mit jeder Zeile und dient als primärer Identifikator des Datensatzes. Die nächsten drei Spalten erwarten Strings. Mit ‘VARCHAR(X)’ gebe ich ihnen aber eine maximale Länge vor.
Anders sieht es mit den Spalten ‘text’ und ‘html’ aus. Hier soll der Inhalt der E-Mail abgespeichert werden, und zwar einmal als reiner Text und einmal mit HTML-Tags. Deswegen sind mit dem Datentyp ‘TEXT’ keine Begrenzungen vorgegeben. Die letzte Spalte ‘created’ hat den Datentyp ‘TIMESTAMP’ und darf nicht leer bleiben. In ihr wird automatisch die aktuelle Systemzeit der Datenbank eingetragen, wenn eine Zeile ergänzt oder verändert wird.
Mit der SQL-Anweisung CREATE UNIQUE INDEX sender_subject ON email (sender, subject) stelle ich sicher, dass kein Eintrag vorgenommen wird, wenn zwei E-Mails nicht nur denselben Sender, sondern auch denselben Betreff aufweisen. Dann handelt es sich nämlich aller Wahrscheinlichkeit nach um eine Dublette.
Nun implementiere ich eine Methode, um die E-Mail-Informationen in der Datenbank abzuspeichern. Dafür rufe ich zuerst die gerade erstellte Methode create_table() auf, um vorher die Tabelle anzulegen, so sie noch nicht vorhanden ist. Danach stelle ich mit with self.connection: sicher, dass die eingangs geöffnete Verbindung zur Datenbank genutzt wird, um die E-Mails einzutragen. Die eigentliche SQL-Anweisung lege ich in der Variable ‘sql’ ab. Falls die zuvor eingerichtete constraint-Bedingung verletzt wird, soll der Eintrag ignoriert und nicht übernommen werden.
def save_to_db(self, mails):
self.create_table()
with self.connection:
print('Inserting only new mails.')
sql = """INSERT OR IGNORE INTO email(sender, receiver, subject,
text, html) VALUES (:sender, :receiver, :subject, :text_content,
:html_content)"""
self.cursor.executemany(sql, mails)
Auf die jeweiligen Werte verweise ich, indem ich mit (:sender, :receiver, :subject, :text_content, :html_content) die Schlüssel eines dictionary angebe. Wo kommt das dictionary her? Beim Aufruf der Methode übergebe ich als Argument die Liste mails, in der für jede einzulesende E-Mail ein dictionary mit den jeweiligen Informationen enthalten ist. Es handelt sich also um eine Liste von dictionaries. Um die IDs und den Zeitstempel muss ich mich nicht mehr kümmern, das übernimmt die Datenbank für mich. Schließlich sorgt die Anweisung self.cursor.executemany(sql, mails) für die Ausführung des dafür definierten SQL-Befehls, und zwar für jedes dictionary in der Liste mails.
Als nächstes geht es darum, die E-Mails herunterzuladen und im Arbeitsspeicher vorzuhalten. Dafür erstelle ich eine neue Klasse namens ‘Downloader’:
class Downloader(object):
def __init__(self):
print('Trying to connect to IMAP client...')
self.imap = imapclient.IMAPClient('imap.server.com', ssl=True)
self.imap.login('irgendwas@domain.de', 'ultraGeheimesPasswort')
Wenn das Objekt geschaffen wird, baut es mit der Hilfe von IMAPClient eine Verbindung zum IMAP-Server auf und legt sie in der Variable imap ab. Danach übersendet es die login credentials, also Benutzername und Passwort.
Als nächstes erstelle ich eine Methode, der eine Liste mit den heruntergeladenen E-Mail-Objekten als Argument übergeben wird und deren Aufgabe darin besteht, aus den einzelnen Objekten die Informationen herauszufiltern, die in der Datenbank abgelegt werden sollen:
def save_as_dict(self, messages):
email_list = []
for m in messages:
mail = {}
mail['sender'] = m.get_address('from')[1]
mail['receiver'] = m.get_address('to')[1]
mail['subject'] = m.get_subject()
if m.text_part != None:
mail['text_content'] = m.text_part.get_payload().decode(
m.text_part.charset)
else:
mail['text_content'] = "None"
if m.html_part != None:
mail['html_content'] = m.html_part.get_payload().decode(
m.html_part.charset)
else:
mail['html_content'] = "None"
email_list.append(mail)
print('Number of downloaded emails: ' + str(len(email_list)))
return email_list
Zuerst lege ich eine leere Liste namens ‘email_list’ an. Danach gehe ich jedes E-Mail-Objekt m durch, das sich in der Liste messages befindet. Für jedes Objekt erstelle ich ein leeres dictionary mit dem Namen ‘mail’. Dann greife auf die Absender-Adresse des E-Mail-Objektes und speichere sie mit dem Schlüssel ‘sender’ im dictionary ab. Analog gehe ich mit der Empfänger-Adresse und dem Betreff vor. Darauf wird es etwas komplizierter: Ich prüfe zunächst mit if m.text_part != None:, ob in dem E-Mail-Objekt der eigentliche Inhalt der Nachricht im reinen Textformat vorliegt. Falls das der Fall sein sollte, lege ich den Text im dictionary ab, nachdem er mit dem Encoding dekodiert worden ist, in dem er übermittelt worden ist. Falls es keinen Inhalt im reinen Text gibt, wird None eingetragen. Die gleiche Prüfung erfolgt anschließend dafür, ob es einen HTML-Inhalt im Objekt gibt. Das heißt, ich stelle sicher, dass im dictionary der Text der Nachricht und die HTML-Version abgespeichert wird, falls sie vorhanden sind. Als letztes wird das erstellte und angereicherte dictionary in die Liste email_list eingefügt, die mit return email_list zum Schluss der Methode zurückgegeben wird.
Jetzt muss noch geklärt werden, wie ich die eigentlichen E-Mails herunterladen und in die Liste messages abspeichern kann. Dafür erstelle ich eine Methode get_emails():
def get_emails(self):
try:
print('Opening INBOX and looking for unseen mail.')
self.imap.select_folder('INBOX', readonly=False)
mails = self.imap.search(['UNSEEN'])
raw_messages = self.imap.fetch(mails, ['BODY[]'])
messages = [pyzmail.PyzMessage.factory(raw_messages[n][b'BODY[]'])
for n in raw_messages]
email_list = self.save_as_dict(messages)
finally:
print('Logging out.')
self.imap.logout()
return email_list
In einer try/finally-Konstruktion wähle ich zuerst den Ordner ‘INBOX’ in dem E-Mail-Konto aus, das eingangs in der Variable imap angegeben wurde. Die Option readonly=False legt fest, dass die abgerufenen E-Mails als gelesen markiert werden. Danach definiere ich, wonach der Ordner durchsucht werden soll. In diesem Fall geht es darum, alle ungelesenen Nachrichten zu finden und in der Variable namens ‘mails’ abzulegen. Mit self.imap.search(['FROM', 'kontakt@domain.de']) ließe sich beispielsweise nach allen Absendern mit der Adresse ‘kontakt@domain.de’ suchen. Nachdem ich eine Liste mit den Suchergebnissen erhalten habe, rufe ich von den entsprechenden Nachrichten den eigentlichen ‘E-Mail-Körper’ ab und speichere diese in der Variable ‘raw_messages’. Für jede darin enthaltene Roh-Nachricht lasse ich PyzMessage ein E-Mail-Objekt erstellen, das ich mithilfe einer list comprehension in eine Liste namens ‘messages’ packe. Daraufhin übergebe ich diese Liste als Argument an die oben definierte Methode save_as_dict() und erhalte eine Liste mit dictionaries anstelle der E-Mail-Objekte zurück. Das finally: stellt sicher, dass auf jeden Fall, auch wenn es einen Fehler geben sollte, die Verbindung zum Imap-Server getrennt wird. Schließlich gibt die Methode die erstellte Liste mit dictionaries zurück.
Der letzte Teil meines Skriptes bindet die einzelnen Bauteile zusammen. In der Funktion main() erstelle ich zunächst ein Objekt Downloader() und weise es der Variable downloader zu. Anschließend rufe ich seine Methode get_emails() auf und lege die zurückgegebene Liste mit E-Mail-Informationen (als dictionary) in der Variable emails ab. Danach erstelle ich ein Objekt Storage() und rufe seine Methode save_to_db() auf, der ich emails als Argument übergebe.
def main():
downloader = Downloader()
emails = downloader.get_emails()
storage = Storage()
storage.save_to_db(emails)
Um das Skript sowohl vom Terminal als auch von einem anderen Programm aufrufen zu können, prüfe ich zu guter Letzt, ob die Python-Variable __name__ der Variable __main__ entspricht, ob also der aufgerufene Programmname mit dem Hauptprozess übereinstimmt:
if __name__ == "__main__":
main()
Ist dem so, dann habe ich es von der Kommandozeile aus aufgerufen, und es wird die Funktion main() gestartet, die das eigentliche Programm zum Laufen bringt.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
This little script checks for new emails in a specified account, downloads
the emails and safes them in a dedicated sqlite database.
It aims at serving as a convenient way to store information you came across
while surfing the internet.
Author: Ramon Voges
Version 0.1
"""
import sqlite3
import imapclient
import pyzmail
class Storage(object):
"""Manages the storage for the downloaded mails."""
def __init__(self):
"""Initializes the storage object. """
self.connection = sqlite3.connect('storage.db')
self.cursor = self.connection.cursor()
def create_table(self):
"""Creates, if necessary, the table.
"""
try:
self.cursor.execute("""CREATE TABLE IF NOT EXISTS email
(id INTEGER PRIMARY KEY,
sender VARCHAR(100),
receiver VARCHAR(100),
subject VARCHAR(200),
text TEXT,
html TEXT,
created TIMESTAMP DEFAULT CURRENT_TIMESTAMP
NOT NULL)
""")
print('Creating table...')
# Sorgt dafür, dass keine doppelten Werte eingegeben werden können!
self.cursor.execute("""CREATE UNIQUE INDEX sender_subject ON email
(sender, subject)""")
print('Creating constraint...')
except sqlite3.OperationalError:
print('... but table already exists.')
def save_to_db(self, mails):
"""Saves the emails to the database.
:emails: a list of dictionaries with information on the emails
"""
self.create_table()
with self.connection:
print('Inserting only new mails.')
sql = """INSERT OR IGNORE INTO email(sender, receiver, subject,
text, html) VALUES (:sender, :receiver, :subject, :text_content,
:html_content)"""
self.cursor.executemany(sql, mails)
class Downloader(object):
"""Responsible for downloading the emails."""
def __init__(self):
"""Creating the downloader. """
print('Trying to connect to IMAP client...')
self.imap = imapclient.IMAPClient('imap.server.com', ssl=True)
self.imap.login('speicher@domain.de', 'UltraGeheimesPasswort')
def save_as_dict(self, messages):
"""Saves information on the emails in a list of dictionaries.
:messages: a list of parsed emails
:returns: a list of dictionaries with information on each email
"""
email_list = []
for m in messages:
mail = {}
mail['sender'] = m.get_address('from')[1]
mail['receiver'] = m.get_address('to')[1]
mail['subject'] = m.get_subject()
if m.text_part != None:
mail['text_content'] = m.text_part.get_payload().decode(
m.text_part.charset)
else:
mail['text_content'] = "None"
if m.html_part != None:
mail['html_content'] = m.html_part.get_payload().decode(
m.html_part.charset)
else:
mail['html_content'] = "None"
email_list.append(mail)
print('Number of downloaded emails: ' + str(len(email_list)))
return email_list
def get_emails(self):
"""Looks for new mails and saves them in memory as a list of
dictionaries.
:returns: a list of parsed emails as dictionaries
"""
try:
print('Opening INBOX and looking for unseen mail.')
self.imap.select_folder('INBOX', readonly=True)
mails = self.imap.search(['UNSEEN'])
raw_messages = self.imap.fetch(mails, ['BODY[]'])
messages = [pyzmail.PyzMessage.factory(raw_messages[n][b'BODY[]'])
for n in raw_messages]
email_list = self.save_as_dict(messages)
finally:
print('Logging out.')
self.imap.logout()
return email_list
def main():
"""Starts the script if run from the command line.
"""
downloader = Downloader()
emails = downloader.get_emails()
storage = Storage()
storage.save_to_db(emails)
if __name__ == "__main__":
main()
Besonders hilfreich mit vielen weiterführenden Hinweisen war für mich Automate the Boring Stuff with Python von Al Sweigart. ↩