mit config.ini
This commit is contained in:
@@ -0,0 +1,169 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
ics_mail_importer.py
|
||||
--------------------
|
||||
Sucht in einem mailbox.org IMAP-Postfach nach E-Mails mit .ics-Anhängen
|
||||
und importiert die enthaltenen Termine automatisch in einen CalDAV-Kalender.
|
||||
|
||||
Benötigte Pakete:
|
||||
pipenv install caldav icalendar
|
||||
|
||||
Einrichtung:
|
||||
1. config.ini ausfüllen (liegt im gleichen Verzeichnis)
|
||||
2. Optional: Als Cronjob einrichten (z.B. alle 30 Minuten)
|
||||
*/30 * * * * /pfad/zu/python3 /pfad/zum/ics_mail_importer.py
|
||||
"""
|
||||
|
||||
import imaplib
|
||||
import email
|
||||
import os
|
||||
import logging
|
||||
import hashlib
|
||||
import configparser
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import caldav
|
||||
from icalendar import Calendar
|
||||
|
||||
# ─────────────────────────── Logging ───────────────────────────────────────────
|
||||
LOG_FILE = Path(__file__).parent / "ics_importer.log"
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||||
handlers=[
|
||||
logging.FileHandler(LOG_FILE),
|
||||
logging.StreamHandler(),
|
||||
],
|
||||
)
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# ─────────────────────────── Konfiguration laden ───────────────────────────────
|
||||
CONFIG_FILE = Path(__file__).parent / "config.ini"
|
||||
|
||||
def load_config():
|
||||
if not CONFIG_FILE.exists():
|
||||
log.error("config.ini nicht gefunden. Bitte config.ini.example umbenennen und ausfüllen.")
|
||||
raise FileNotFoundError(str(CONFIG_FILE))
|
||||
cfg = configparser.ConfigParser()
|
||||
cfg.read(CONFIG_FILE)
|
||||
return cfg
|
||||
|
||||
# ─────────────────────────── Duplikat-Tracking ────────────────────────────────
|
||||
SEEN_FILE = Path(__file__).parent / "imported_uids.txt"
|
||||
|
||||
def load_seen_uids():
|
||||
if not SEEN_FILE.exists():
|
||||
return set()
|
||||
return set(SEEN_FILE.read_text().splitlines())
|
||||
|
||||
def save_uid(uid: str):
|
||||
with SEEN_FILE.open("a") as f:
|
||||
f.write(uid + "\n")
|
||||
|
||||
# ─────────────────────────── IMAP-Anhänge holen ───────────────────────────────
|
||||
def fetch_ics_attachments(cfg):
|
||||
host = cfg["imap"]["host"]
|
||||
port = int(cfg["imap"].get("port", "993"))
|
||||
username = cfg["imap"]["username"]
|
||||
password = cfg["imap"]["password"]
|
||||
folder = cfg["imap"].get("folder", "INBOX")
|
||||
|
||||
log.info(f"Verbinde mit IMAP {host}:{port} als {username} ...")
|
||||
conn = imaplib.IMAP4_SSL(host, port)
|
||||
conn.login(username, password)
|
||||
conn.select(folder)
|
||||
|
||||
search_unseen_only = cfg["imap"].getboolean("unseen_only", fallback=True)
|
||||
criteria = "(UNSEEN)" if search_unseen_only else "ALL"
|
||||
_, msg_ids = conn.search(None, criteria)
|
||||
|
||||
found = []
|
||||
for mid in msg_ids[0].split():
|
||||
_, data = conn.fetch(mid, "(RFC822)")
|
||||
raw = data[0][1]
|
||||
msg = email.message_from_bytes(raw)
|
||||
|
||||
for part in msg.walk():
|
||||
ct = part.get_content_type()
|
||||
fn = part.get_filename() or ""
|
||||
|
||||
is_ics = (
|
||||
ct in ("text/calendar", "application/ics")
|
||||
or fn.lower().endswith(".ics")
|
||||
)
|
||||
if not is_ics:
|
||||
continue
|
||||
|
||||
ics_bytes = part.get_payload(decode=True)
|
||||
if not ics_bytes:
|
||||
continue
|
||||
|
||||
uid_hash = hashlib.sha256(ics_bytes).hexdigest()
|
||||
log.info(f" ics-Anhang gefunden: {fn!r} (Hash {uid_hash[:12]}…)")
|
||||
found.append((uid_hash, ics_bytes))
|
||||
|
||||
if cfg["imap"].getboolean("mark_as_read", fallback=False):
|
||||
conn.store(mid, "+FLAGS", "\\Seen")
|
||||
|
||||
conn.close()
|
||||
conn.logout()
|
||||
log.info(f" {len(found)} .ics-Anhang/-Anhänge gefunden.")
|
||||
return found
|
||||
|
||||
# ─────────────────────────── CalDAV-Import ────────────────────────────────────
|
||||
def import_to_caldav(cfg, ics_bytes: bytes, uid_hash: str):
|
||||
caldav_url = cfg["caldav"]["url"]
|
||||
username = cfg["caldav"]["username"]
|
||||
password = cfg["caldav"]["password"]
|
||||
|
||||
client = caldav.DAVClient(url=caldav_url, username=username, password=password)
|
||||
calendar = client.calendar(url=caldav_url)
|
||||
|
||||
cal = Calendar.from_ical(ics_bytes)
|
||||
|
||||
imported = 0
|
||||
for component in cal.walk():
|
||||
if component.name != "VEVENT":
|
||||
continue
|
||||
|
||||
single_cal = Calendar()
|
||||
single_cal.add("prodid", "-//ics_mail_importer//DE")
|
||||
single_cal.add("version", "2.0")
|
||||
single_cal.add_component(component)
|
||||
|
||||
uid = str(component.get("uid", uid_hash + f"-{imported}"))
|
||||
try:
|
||||
calendar.add_event(single_cal.to_ical().decode("utf-8"))
|
||||
log.info(f" ✓ Termin importiert: {component.get('summary', '(kein Titel)')} [{uid[:20]}]")
|
||||
imported += 1
|
||||
except Exception as e:
|
||||
log.warning(f" ✗ Fehler beim Import von {uid[:20]}: {e}")
|
||||
|
||||
return imported
|
||||
|
||||
# ─────────────────────────── Hauptprogramm ────────────────────────────────────
|
||||
def main():
|
||||
log.info("=" * 60)
|
||||
log.info(f"ICS-Importer gestartet ({datetime.now(timezone.utc).isoformat()})")
|
||||
|
||||
cfg = load_config()
|
||||
seen_uids = load_seen_uids()
|
||||
|
||||
attachments = fetch_ics_attachments(cfg)
|
||||
|
||||
total_imported = 0
|
||||
for uid_hash, ics_bytes in attachments:
|
||||
if uid_hash in seen_uids:
|
||||
log.info(f" Überspringe bereits importierten Anhang {uid_hash[:12]}…")
|
||||
continue
|
||||
|
||||
n = import_to_caldav(cfg, ics_bytes, uid_hash)
|
||||
total_imported += n
|
||||
save_uid(uid_hash)
|
||||
|
||||
log.info(f"Fertig. {total_imported} Termine insgesamt importiert.")
|
||||
log.info("=" * 60)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user