First Commit with complete Setup coded by Claude 4.7 Opus

backend/carddav_client.py

@@ -0,0 +1,391 @@
+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