159 lines
4.6 KiB
Python
159 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
ics_mail_importer_env.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.
|
|
|
|
Konfiguration über .env-Datei mit python-dotenv.
|
|
"""
|
|
|
|
import imaplib
|
|
import email
|
|
import logging
|
|
import hashlib
|
|
import os
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
|
|
from dotenv import load_dotenv
|
|
import caldav
|
|
from icalendar import Calendar
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
load_dotenv(BASE_DIR / ".env")
|
|
|
|
LOG_FILE = BASE_DIR / "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__)
|
|
|
|
SEEN_FILE = BASE_DIR / "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")
|
|
|
|
|
|
def fetch_ics_attachments():
|
|
host = os.getenv("IMAP_HOST", "imap.mailbox.org")
|
|
port = int(os.getenv("IMAP_PORT", "993"))
|
|
username = os.getenv("IMAP_USERNAME")
|
|
password = os.getenv("IMAP_PASSWORD")
|
|
folder = os.getenv("IMAP_FOLDER", "INBOX")
|
|
|
|
if not username or not password:
|
|
raise RuntimeError("IMAP_USERNAME oder IMAP_PASSWORD fehlt in .env")
|
|
|
|
log.info(f"Verbinde mit IMAP {host}:{port} als {username} ...")
|
|
conn = imaplib.IMAP4_SSL(host, port)
|
|
conn.login(username, password)
|
|
conn.select(folder)
|
|
|
|
unseen_only = os.getenv("IMAP_UNSEEN_ONLY", "true").lower() == "true"
|
|
criteria = "(UNSEEN)" if 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))
|
|
|
|
mark_read = os.getenv("IMAP_MARK_AS_READ", "false").lower() == "true"
|
|
if mark_read:
|
|
conn.store(mid, "+FLAGS", "\\Seen")
|
|
|
|
conn.close()
|
|
conn.logout()
|
|
log.info(f" {len(found)} .ics-Anhang/-Anhänge gefunden.")
|
|
return found
|
|
|
|
|
|
def import_to_caldav(ics_bytes: bytes, uid_hash: str):
|
|
caldav_url = os.getenv("CALDAV_URL")
|
|
username = os.getenv("CALDAV_USERNAME")
|
|
password = os.getenv("CALDAV_PASSWORD")
|
|
|
|
if not caldav_url or not username or not password:
|
|
raise RuntimeError(
|
|
"CALDAV_URL, CALDAV_USERNAME oder CALDAV_PASSWORD fehlt in .env"
|
|
)
|
|
|
|
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
|
|
|
|
|
|
def main():
|
|
log.info("=" * 60)
|
|
log.info(f"ICS-Importer gestartet ({datetime.now(timezone.utc).isoformat()})")
|
|
|
|
seen_uids = load_seen_uids()
|
|
attachments = fetch_ics_attachments()
|
|
|
|
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(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() |