391 lines
14 KiB
Python
391 lines
14 KiB
Python
import httpx
|
|
import vobject
|
|
import logging
|
|
from typing import Optional
|
|
from xml.etree import ElementTree as ET
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CARDDAV_NAMESPACES = {
|
|
"d": "DAV:",
|
|
"cs": "http://calendarserver.org/ns/",
|
|
"card": "urn:ietf:params:xml:ns:carddav",
|
|
}
|
|
|
|
|
|
class CardDAVClient:
|
|
def __init__(self, server_url: str, username: str, password: str):
|
|
# Trailing slash entfernen
|
|
self.base_url = server_url.rstrip("/")
|
|
self.auth = (username, password)
|
|
|
|
def _client(self) -> httpx.Client:
|
|
return httpx.Client(auth=self.auth, verify=False, timeout=30)
|
|
|
|
def discover_principal(self) -> Optional[str]:
|
|
"""CardDAV Principal-URL ermitteln.
|
|
|
|
Strategie:
|
|
1. Wenn die Server-URL bereits einen Pfad enthält (z.B. /carddav/User),
|
|
wird sie direkt verwendet — der Admin weiß was er tut.
|
|
2. Wenn nur Host:Port angegeben ist, wird /.well-known/carddav probiert,
|
|
dann fällt es auf /carddav zurück.
|
|
"""
|
|
from urllib.parse import urlparse
|
|
parsed = urlparse(self.base_url)
|
|
path = (parsed.path or "").strip("/")
|
|
|
|
# Wenn URL schon einen Pfad hat -> direkt verwenden
|
|
if path:
|
|
return self.base_url
|
|
|
|
# Sonst: Discovery versuchen
|
|
try:
|
|
with self._client() as client:
|
|
resp = client.request(
|
|
"PROPFIND",
|
|
f"{self.base_url}/.well-known/carddav",
|
|
headers={"Depth": "0", "Content-Type": "application/xml"},
|
|
content="""<?xml version="1.0" encoding="utf-8"?>
|
|
<propfind xmlns="DAV:">
|
|
<prop>
|
|
<current-user-principal/>
|
|
<principal-URL/>
|
|
</prop>
|
|
</propfind>""",
|
|
follow_redirects=True,
|
|
)
|
|
if resp.status_code in (200, 207):
|
|
return str(resp.url).rstrip("/")
|
|
except Exception as e:
|
|
logger.warning(f"Well-known discovery failed: {e}")
|
|
|
|
return f"{self.base_url}/carddav"
|
|
|
|
def list_address_books(self) -> list[dict]:
|
|
"""Alle verfügbaren Adressbücher auflisten"""
|
|
base = self.discover_principal()
|
|
body = """<?xml version="1.0" encoding="utf-8"?>
|
|
<propfind xmlns="DAV:" xmlns:cs="http://calendarserver.org/ns/" xmlns:card="urn:ietf:params:xml:ns:carddav">
|
|
<prop>
|
|
<resourcetype/>
|
|
<displayname/>
|
|
<cs:getctag/>
|
|
<card:addressbook-description/>
|
|
</prop>
|
|
</propfind>"""
|
|
results = []
|
|
try:
|
|
with self._client() as client:
|
|
resp = client.request(
|
|
"PROPFIND",
|
|
base,
|
|
headers={"Depth": "1", "Content-Type": "application/xml"},
|
|
content=body,
|
|
)
|
|
if resp.status_code != 207:
|
|
logger.error(f"PROPFIND failed: {resp.status_code}")
|
|
return results
|
|
|
|
root = ET.fromstring(resp.text)
|
|
for response in root.findall(".//d:response", CARDDAV_NAMESPACES):
|
|
href_el = response.find("d:href", CARDDAV_NAMESPACES)
|
|
rt = response.find(".//d:resourcetype", CARDDAV_NAMESPACES)
|
|
is_ab = rt is not None and rt.find("card:addressbook", CARDDAV_NAMESPACES) is not None
|
|
|
|
if is_ab and href_el is not None:
|
|
href = href_el.text
|
|
name_el = response.find(".//d:displayname", CARDDAV_NAMESPACES)
|
|
name = name_el.text if name_el is not None else href.split("/")[-2]
|
|
|
|
# Absolute URL aufbauen
|
|
if href.startswith("/"):
|
|
from urllib.parse import urlparse
|
|
parsed = urlparse(self.base_url)
|
|
full_url = f"{parsed.scheme}://{parsed.netloc}{href}"
|
|
else:
|
|
full_url = href
|
|
|
|
results.append({"name": name, "url": full_url, "href": href})
|
|
except Exception as e:
|
|
logger.error(f"list_address_books error: {e}")
|
|
|
|
return results
|
|
|
|
def get_contacts(self, addressbook_url: str) -> list[dict]:
|
|
"""Alle vCards aus einem Adressbuch laden.
|
|
|
|
Verwendet REPORT addressbook-query (RFC 6352) — funktioniert mit Radicale,
|
|
SOGo, Nextcloud, Baikal und allen anderen RFC-konformen CardDAV-Servern.
|
|
"""
|
|
contacts = []
|
|
|
|
# REPORT addressbook-query holt URLs + vCard-Daten in einem Call
|
|
report_body = """<?xml version="1.0" encoding="utf-8"?>
|
|
<card:addressbook-query xmlns:d="DAV:" xmlns:card="urn:ietf:params:xml:ns:carddav">
|
|
<d:prop>
|
|
<d:getetag/>
|
|
<card:address-data/>
|
|
</d:prop>
|
|
<card:filter/>
|
|
</card:addressbook-query>"""
|
|
|
|
try:
|
|
with self._client() as client:
|
|
resp = client.request(
|
|
"REPORT",
|
|
addressbook_url,
|
|
headers={
|
|
"Depth": "1",
|
|
"Content-Type": "application/xml; charset=utf-8",
|
|
},
|
|
content=report_body,
|
|
)
|
|
|
|
if resp.status_code == 207:
|
|
contacts = self._parse_vcard_response(resp.text)
|
|
logger.info(f"REPORT returned {len(contacts)} contacts")
|
|
if contacts:
|
|
return contacts
|
|
else:
|
|
logger.warning(f"REPORT failed ({resp.status_code}), falling back to PROPFIND+GET")
|
|
|
|
# Fallback 1: PROPFIND mit address-data
|
|
propfind_body = """<?xml version="1.0" encoding="utf-8"?>
|
|
<d:propfind xmlns:d="DAV:" xmlns:card="urn:ietf:params:xml:ns:carddav">
|
|
<d:prop>
|
|
<d:getetag/>
|
|
<card:address-data/>
|
|
</d:prop>
|
|
</d:propfind>"""
|
|
resp = client.request(
|
|
"PROPFIND",
|
|
addressbook_url,
|
|
headers={"Depth": "1", "Content-Type": "application/xml; charset=utf-8"},
|
|
content=propfind_body,
|
|
)
|
|
if resp.status_code == 207:
|
|
contacts = self._parse_vcard_response(resp.text)
|
|
if contacts:
|
|
logger.info(f"PROPFIND returned {len(contacts)} contacts")
|
|
return contacts
|
|
|
|
# Fallback 2: PROPFIND nur für hrefs + einzelne GET-Requests
|
|
logger.info("Falling back to PROPFIND + individual GETs")
|
|
href_body = """<?xml version="1.0" encoding="utf-8"?>
|
|
<d:propfind xmlns:d="DAV:">
|
|
<d:prop>
|
|
<d:getetag/>
|
|
<d:getcontenttype/>
|
|
</d:prop>
|
|
</d:propfind>"""
|
|
resp = client.request(
|
|
"PROPFIND",
|
|
addressbook_url,
|
|
headers={"Depth": "1", "Content-Type": "application/xml; charset=utf-8"},
|
|
content=href_body,
|
|
)
|
|
if resp.status_code != 207:
|
|
logger.error(f"href PROPFIND failed: {resp.status_code}")
|
|
return contacts
|
|
|
|
# vCard URLs sammeln
|
|
from urllib.parse import urlparse
|
|
parsed_base = urlparse(self.base_url)
|
|
root = ET.fromstring(resp.text)
|
|
vcard_urls = []
|
|
for response in root.findall(".//d:response", CARDDAV_NAMESPACES):
|
|
href_el = response.find("d:href", CARDDAV_NAMESPACES)
|
|
ctype_el = response.find(".//d:getcontenttype", CARDDAV_NAMESPACES)
|
|
if href_el is None:
|
|
continue
|
|
href = href_el.text or ""
|
|
ctype = (ctype_el.text or "") if ctype_el is not None else ""
|
|
# Nur vCard-Ressourcen, keine Collection
|
|
if not href.endswith(".vcf") and "vcard" not in ctype.lower():
|
|
continue
|
|
if href.startswith("/"):
|
|
full_url = f"{parsed_base.scheme}://{parsed_base.netloc}{href}"
|
|
elif href.startswith("http"):
|
|
full_url = href
|
|
else:
|
|
full_url = f"{addressbook_url.rstrip('/')}/{href}"
|
|
vcard_urls.append((href, full_url))
|
|
|
|
logger.info(f"Fetching {len(vcard_urls)} vCards individually")
|
|
for href, url in vcard_urls:
|
|
try:
|
|
r = client.get(url)
|
|
if r.status_code == 200 and "BEGIN:VCARD" in r.text:
|
|
parsed = parse_vcard(r.text)
|
|
if parsed:
|
|
parsed["_href"] = href
|
|
parsed["_raw"] = r.text
|
|
contacts.append(parsed)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to GET vCard {url}: {e}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"get_contacts error: {e}", exc_info=True)
|
|
|
|
return contacts
|
|
|
|
def _parse_vcard_response(self, xml_text: str) -> list[dict]:
|
|
"""Parst eine 207 Multi-Status Antwort und extrahiert vCards"""
|
|
contacts = []
|
|
try:
|
|
root = ET.fromstring(xml_text)
|
|
except ET.ParseError as e:
|
|
logger.error(f"XML parse error: {e}")
|
|
return contacts
|
|
|
|
for response in root.findall(".//d:response", CARDDAV_NAMESPACES):
|
|
href_el = response.find("d:href", CARDDAV_NAMESPACES)
|
|
etag_el = response.find(".//d:getetag", CARDDAV_NAMESPACES)
|
|
vcard_el = response.find(".//card:address-data", CARDDAV_NAMESPACES)
|
|
|
|
if vcard_el is not None and vcard_el.text:
|
|
vcard_data = vcard_el.text.strip()
|
|
if "BEGIN:VCARD" in vcard_data:
|
|
parsed = parse_vcard(vcard_data)
|
|
if parsed:
|
|
parsed["_href"] = href_el.text if href_el is not None else ""
|
|
parsed["_etag"] = etag_el.text if etag_el is not None else ""
|
|
parsed["_raw"] = vcard_data
|
|
contacts.append(parsed)
|
|
return contacts
|
|
|
|
def test_connection(self) -> tuple[bool, str]:
|
|
"""Verbindungstest"""
|
|
try:
|
|
books = self.list_address_books()
|
|
if books is not None:
|
|
return True, f"{len(books)} Adressbuch/Adressbücher gefunden"
|
|
return False, "Keine Adressbücher gefunden"
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
|
|
def parse_vcard(vcard_text: str) -> Optional[dict]:
|
|
"""vCard → Dictionary für Microsoft Graph"""
|
|
try:
|
|
vcard = vobject.readOne(vcard_text)
|
|
except Exception as e:
|
|
logger.warning(f"vCard parse error: {e}")
|
|
return None
|
|
|
|
def get(attr):
|
|
try:
|
|
return getattr(vcard, attr).value
|
|
except AttributeError:
|
|
return None
|
|
|
|
def get_all(attr):
|
|
try:
|
|
return [x.value for x in getattr(vcard, attr + "_list", [])]
|
|
except (AttributeError, TypeError):
|
|
return []
|
|
|
|
result = {}
|
|
|
|
# Name
|
|
name = get("n")
|
|
if name:
|
|
result["givenName"] = name.given or ""
|
|
result["surname"] = name.family or ""
|
|
result["middleName"] = name.additional or ""
|
|
result["title"] = name.prefix or ""
|
|
result["generation"] = name.suffix or ""
|
|
|
|
fn = get("fn")
|
|
if fn:
|
|
result["displayName"] = fn
|
|
elif "givenName" in result or "surname" in result:
|
|
result["displayName"] = f"{result.get('givenName', '')} {result.get('surname', '')}".strip()
|
|
|
|
# E-Mails
|
|
emails = []
|
|
try:
|
|
for email_obj in getattr(vcard, "email_list", []):
|
|
addr = email_obj.value
|
|
params = email_obj.params
|
|
types = [t.lower() for t in params.get("TYPE", [])]
|
|
address_type = "work" if "work" in types else ("home" if "home" in types else "other")
|
|
emails.append({"address": addr, "name": result.get("displayName", ""), "type": address_type})
|
|
except Exception:
|
|
pass
|
|
if emails:
|
|
result["emailAddresses"] = emails
|
|
|
|
# Telefon
|
|
phones_business = []
|
|
phones_home = []
|
|
mobile = None
|
|
try:
|
|
for tel_obj in getattr(vcard, "tel_list", []):
|
|
num = tel_obj.value
|
|
types = [t.lower() for t in tel_obj.params.get("TYPE", [])]
|
|
if "cell" in types or "mobile" in types:
|
|
mobile = num
|
|
elif "home" in types:
|
|
phones_home.append(num)
|
|
else:
|
|
phones_business.append(num)
|
|
except Exception:
|
|
pass
|
|
|
|
if phones_business:
|
|
result["businessPhones"] = phones_business
|
|
if phones_home:
|
|
result["homePhones"] = phones_home
|
|
if mobile:
|
|
result["mobilePhone"] = mobile
|
|
|
|
# Organisation
|
|
org = get("org")
|
|
if org:
|
|
if isinstance(org, list):
|
|
result["companyName"] = org[0] if org else ""
|
|
result["department"] = org[1] if len(org) > 1 else ""
|
|
else:
|
|
result["companyName"] = str(org)
|
|
|
|
title = get("title")
|
|
if title:
|
|
result["jobTitle"] = title
|
|
|
|
# Adresse
|
|
try:
|
|
for adr_obj in getattr(vcard, "adr_list", []):
|
|
adr = adr_obj.value
|
|
types = [t.lower() for t in adr_obj.params.get("TYPE", [])]
|
|
addr_dict = {
|
|
"street": adr.street or "",
|
|
"city": adr.city or "",
|
|
"state": adr.region or "",
|
|
"postalCode": adr.code or "",
|
|
"countryOrRegion": adr.country or "",
|
|
}
|
|
if "home" in types:
|
|
result["homeAddress"] = addr_dict
|
|
else:
|
|
result["businessAddress"] = addr_dict
|
|
break # nur erste Adresse
|
|
except Exception:
|
|
pass
|
|
|
|
# Notizen
|
|
note = get("note")
|
|
if note:
|
|
result["personalNotes"] = note
|
|
|
|
# URL
|
|
url = get("url")
|
|
if url:
|
|
result["businessHomePage"] = url
|
|
|
|
# UID für Identifizierung
|
|
uid = get("uid")
|
|
result["_uid"] = uid or ""
|
|
|
|
return result
|