Hacker-Software/minmal-file-cloud-email-pim-passmanager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: Kontakte mit Outlook-Feldern + CardDAV-Server + Sharing

Browse Source 
Komplette Kontakte-Ueberarbeitung analog zum Kalender-Ausbau.

Backend-Model:
* AddressBook: color (pro Buch), ausserdem Per-User-Color via
  AddressBookShare.color wie bei CalendarShare.
* Contact: volle Outlook-artige Struktur - prefix/first/middle/
  last/suffix, display_name, nickname, organization, department,
  job_title, birthday, anniversary, notes, photo sowie JSON-
  Spalten fuer mehrfach vorhandene Felder (emails, phones,
  addresses mit allen Adressteilen, websites, impp, categories).

Backend-API:
* REST CRUD uebernimmt die neuen Felder und generiert vCard 3.0
  als Source of Truth fuer CardDAV. Voller vCard-Parser +
  -Builder mit Escape/Unescape, TYPE-Parametern, Line-Folding.
* Neuer Endpoint PUT /addressbooks/<id>/my-color - persoenliche
  Farbe pro Buch ohne den Besitzer zu beeinflussen.
* SSE-Events vom Typ 'addressbook' an Besitzer + alle Share-
  Empfaenger bei jeder Aenderung.

CardDAV-Server (backend/app/dav/carddav.py):
* Volle Discovery via principal - addressbook-home-set wird
  neben calendar-home-set annonciert.
* PROPFIND/REPORT/GET/PUT/DELETE/MKCOL fuer
  /dav/<user>/ab-<id>/ und /<...>/{uid}.vcf
* addressbook-query + addressbook-multiget REPORTs
* ETag-basierte Konfliktpruefung via If-Match/If-None-Match

Frontend (ContactsView.vue):
* Komplett neuer Editor mit vier Tabs: Allgemein (Name, Org),
  Kommunikation (Emails/Phones/Websites/IMPP dynamisch),
  Adressen (mehrere mit allen Teilen), Details (Geburtstag,
  Jahrestag, Kategorien, Notizen).
* Avatar mit Fotoauswahl oder Initialen-Farbkreis.
* Kalender-Sharing-Flow 1:1 uebernommen: Autocomplete fuer
  Benutzersuche, Share-Liste mit Stift zum Bearbeiten, Muelleimer
  zum Entfernen, Per-User-Farbe, CardDAV-URL-Info-Block pro
  Adressbuch, Live-Refresh via SSE.
* Suche durchsucht Displayname, E-Mail und Firma.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Stefan Hacker
2026-04-12 15:16:01 +02:00
parent fbf10197d7
commit 9c102823e4
6 changed files with 1401 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/contacts.py
+328 -106
View File
@@ -1,13 +1,32 @@
import json
import re
import uuid
from datetime import datetime, timezone
from flask import request, jsonify
from flask import request, jsonify, Response
from app.api import api_bp
from app.api.auth import token_required
from app.extensions import db
from app.models.contact import AddressBook, Contact, AddressBookShare
from app.models.user import User
from app.services.events import broadcaster
def _notify_addressbook(owner_id: int, book_id: int, change: str, shared_with=()):
"""SSE event for a vcard or share change. Re-uses the calendar event
infrastructure with a separate 'addressbook' type."""
recipients = [owner_id, *shared_with]
broadcaster.publish(recipients, {
'type': 'addressbook',
'change': change,
'address_book_id': book_id,
})
def _book_recipients(book: AddressBook):
return [s.shared_with_id for s in
AddressBookShare.query.filter_by(address_book_id=book.id).all()]
def _get_addressbook_or_err(book_id, user, need_write=False):
@@ -26,7 +45,224 @@ def _get_addressbook_or_err(book_id, user, need_write=False):
return book, None
# --- Address Books ---
# ---------------------------------------------------------------------------
# vCard helpers
# ---------------------------------------------------------------------------
def _escape(s):
if s is None:
return ''
return str(s).replace('\\', '\\\\').replace(',', '\\,').replace(';', '\\;').replace('\n', '\\n')
def _unescape(s):
if not s:
return ''
return s.replace('\\n', '\n').replace('\\;', ';').replace('\\,', ',').replace('\\\\', '\\')
def _apply_fields_to_contact(contact: Contact, data: dict):
"""Copy fields from a JSON request into a Contact model instance."""
for field in ('prefix', 'first_name', 'middle_name', 'last_name', 'suffix',
'nickname', 'organization', 'department', 'job_title',
'notes', 'photo', 'birthday', 'anniversary'):
if field in data:
value = data[field]
setattr(contact, field, (value.strip() if isinstance(value, str) else value) or None)
if 'display_name' in data:
contact.display_name = (data['display_name'] or '').strip() or None
for jsonfield in ('emails', 'phones', 'addresses', 'websites', 'impp', 'categories'):
if jsonfield in data:
value = data[jsonfield] or []
setattr(contact, jsonfield, json.dumps(value) if value else None)
# Denormalised primary fields for list display
emails = data.get('emails') if 'emails' in data else json.loads(contact.emails) if contact.emails else []
phones = data.get('phones') if 'phones' in data else json.loads(contact.phones) if contact.phones else []
contact.primary_email = (emails[0]['value'] if emails else None)
contact.primary_phone = (phones[0]['value'] if phones else None)
# Legacy columns
contact.email = contact.primary_email
contact.phone = contact.primary_phone
# Compose display name if not provided
if not contact.display_name:
parts = [contact.prefix, contact.first_name, contact.middle_name,
contact.last_name, contact.suffix]
contact.display_name = ' '.join(p for p in parts if p) or contact.organization or None
def _build_vcard(contact: Contact) -> str:
"""Render a Contact into vCard 3.0 text."""
lines = ['BEGIN:VCARD', 'VERSION:3.0', f'UID:{contact.uid}']
if contact.display_name:
lines.append(f'FN:{_escape(contact.display_name)}')
# N: lastname;firstname;middle;prefix;suffix
n_parts = [_escape(contact.last_name), _escape(contact.first_name),
_escape(contact.middle_name), _escape(contact.prefix),
_escape(contact.suffix)]
if any(n_parts):
lines.append('N:' + ';'.join(n_parts))
if contact.nickname:
lines.append(f'NICKNAME:{_escape(contact.nickname)}')
if contact.organization or contact.department:
lines.append(f'ORG:{_escape(contact.organization or "")};{_escape(contact.department or "")}')
if contact.job_title:
lines.append(f'TITLE:{_escape(contact.job_title)}')
for e in (json.loads(contact.emails) if contact.emails else []):
typ = (e.get('type') or 'home').upper()
lines.append(f'EMAIL;TYPE={typ}:{_escape(e.get("value", ""))}')
for p in (json.loads(contact.phones) if contact.phones else []):
typ = (p.get('type') or 'cell').upper()
lines.append(f'TEL;TYPE={typ}:{_escape(p.get("value", ""))}')
for a in (json.loads(contact.addresses) if contact.addresses else []):
typ = (a.get('type') or 'home').upper()
# ADR: po_box;extended;street;city;region;postal_code;country
parts = [_escape(a.get('po_box', '')), '', _escape(a.get('street', '')),
_escape(a.get('city', '')), _escape(a.get('region', '')),
_escape(a.get('postal_code', '')), _escape(a.get('country', ''))]
lines.append(f'ADR;TYPE={typ}:' + ';'.join(parts))
for w in (json.loads(contact.websites) if contact.websites else []):
typ = (w.get('type') or '').upper()
tag = f'URL;TYPE={typ}' if typ else 'URL'
lines.append(f'{tag}:{_escape(w.get("value", ""))}')
for i in (json.loads(contact.impp) if contact.impp else []):
proto = (i.get('protocol') or 'xmpp').lower()
lines.append(f'IMPP:{proto}:{_escape(i.get("value", ""))}')
if contact.birthday:
lines.append(f'BDAY:{contact.birthday}')
if contact.anniversary:
lines.append(f'ANNIVERSARY:{contact.anniversary}')
cats = json.loads(contact.categories) if contact.categories else []
if cats:
lines.append('CATEGORIES:' + ','.join(_escape(c) for c in cats))
if contact.notes:
lines.append(f'NOTE:{_escape(contact.notes)}')
if contact.photo:
# Photo can be a data: URL or http URL. In vCard 3.0 we use PHOTO;VALUE=uri.
lines.append(f'PHOTO;VALUE=uri:{contact.photo}')
lines.append(f'REV:{datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")}')
lines.append('END:VCARD')
return '\r\n'.join(lines)
def _unfold_vcard(raw: str):
"""Undo RFC 6350 line folding (continuation lines start with space/tab)."""
lines = []
for line in raw.replace('\r\n', '\n').split('\n'):
if line.startswith((' ', '\t')) and lines:
lines[-1] += line[1:]
else:
lines.append(line)
return lines
def parse_vcard(raw: str) -> dict:
"""Parse a VCARD text into a dict of fields usable by _apply_fields_to_contact.
Returns dict with keys matching contact fields + 'uid'."""
result = {
'emails': [], 'phones': [], 'addresses': [],
'websites': [], 'impp': [], 'categories': [],
}
for line in _unfold_vcard(raw):
if ':' not in line:
continue
key, _, value = line.partition(':')
parts = key.split(';')
name = parts[0].upper()
params = {}
for p in parts[1:]:
if '=' in p:
k, v = p.split('=', 1)
params[k.upper()] = v.upper()
if name == 'UID':
result['uid'] = value.strip()
elif name == 'FN':
result['display_name'] = _unescape(value)
elif name == 'N':
fields = value.split(';')
if len(fields) >= 5:
result['last_name'] = _unescape(fields[0]) or None
result['first_name'] = _unescape(fields[1]) or None
result['middle_name'] = _unescape(fields[2]) or None
result['prefix'] = _unescape(fields[3]) or None
result['suffix'] = _unescape(fields[4]) or None
elif name == 'NICKNAME':
result['nickname'] = _unescape(value)
elif name == 'ORG':
fields = value.split(';')
result['organization'] = _unescape(fields[0]) if fields else None
if len(fields) > 1:
result['department'] = _unescape(fields[1]) or None
elif name == 'TITLE':
result['job_title'] = _unescape(value)
elif name == 'EMAIL':
result['emails'].append({
'type': (params.get('TYPE') or 'home').lower(),
'value': _unescape(value),
})
elif name == 'TEL':
result['phones'].append({
'type': (params.get('TYPE') or 'cell').lower(),
'value': _unescape(value),
})
elif name == 'ADR':
fields = value.split(';')
pad = fields + [''] * (7 - len(fields))
result['addresses'].append({
'type': (params.get('TYPE') or 'home').lower(),
'po_box': _unescape(pad[0]),
'street': _unescape(pad[2]),
'city': _unescape(pad[3]),
'region': _unescape(pad[4]),
'postal_code': _unescape(pad[5]),
'country': _unescape(pad[6]),
})
elif name == 'URL':
result['websites'].append({
'type': (params.get('TYPE') or '').lower(),
'value': _unescape(value),
})
elif name == 'IMPP':
proto, _, addr = value.partition(':')
result['impp'].append({'protocol': proto.lower(), 'value': _unescape(addr or value)})
elif name == 'CATEGORIES':
result['categories'] = [_unescape(c).strip() for c in value.split(',') if c.strip()]
elif name == 'BDAY':
result['birthday'] = _normalise_date(value)
elif name == 'ANNIVERSARY':
result['anniversary'] = _normalise_date(value)
elif name == 'NOTE':
result['notes'] = _unescape(value)
elif name == 'PHOTO':
result['photo'] = value.strip() or None
return result
def _normalise_date(s: str):
s = s.strip()
m = re.match(r'^(\d{4})-?(\d{2})-?(\d{2})$', s[:10])
if m:
return f'{m.group(1)}-{m.group(2)}-{m.group(3)}'
return None
# ---------------------------------------------------------------------------
# Address books
# ---------------------------------------------------------------------------
@api_bp.route('/addressbooks', methods=['GET'])
@token_required
@@ -49,6 +285,9 @@ def list_addressbooks():
address_book_id=b.id, shared_with_id=user.id
).first()
d['permission'] = share.permission if share else 'read'
d['owner_color'] = d.get('color')
if share and share.color:
d['color'] = share.color
d['owner_name'] = b.owner.username
d['contact_count'] = b.contacts.count()
result.append(d)
@@ -61,13 +300,19 @@ def list_addressbooks():
def create_addressbook():
user = request.current_user
data = request.get_json()
name = data.get('name', '').strip()
name = (data.get('name') or '').strip()
if not name:
return jsonify({'error': 'Name erforderlich'}), 400
book = AddressBook(owner_id=user.id, name=name, description=data.get('description', ''))
book = AddressBook(
owner_id=user.id,
name=name,
color=data.get('color', '#3788d8'),
description=data.get('description') or None,
)
db.session.add(book)
db.session.commit()
_notify_addressbook(user.id, book.id, 'created')
return jsonify(book.to_dict()), 201
@@ -77,31 +322,66 @@ def update_addressbook(book_id):
user = request.current_user
book = db.session.get(AddressBook, book_id)
if not book or book.owner_id != user.id:
return jsonify({'error': 'Nicht gefunden'}), 404
return jsonify({'error': 'Nicht gefunden oder keine Berechtigung'}), 404
data = request.get_json()
if 'name' in data:
book.name = data['name'].strip()
if 'description' in data:
book.description = data['description']
book.description = data['description'] or None
if 'color' in data:
book.color = data['color']
db.session.commit()
_notify_addressbook(book.owner_id, book.id, 'updated',
shared_with=_book_recipients(book))
return jsonify(book.to_dict()), 200
@api_bp.route('/addressbooks/<int:book_id>/my-color', methods=['PUT'])
@token_required
def set_my_addressbook_color(book_id):
user = request.current_user
book = db.session.get(AddressBook, book_id)
if not book:
return jsonify({'error': 'Nicht gefunden'}), 404
color = ((request.get_json() or {}).get('color') or '').strip()
if book.owner_id == user.id:
if color:
book.color = color
db.session.commit()
return jsonify({'color': book.color}), 200
share = AddressBookShare.query.filter_by(
address_book_id=book_id, shared_with_id=user.id
).first()
if not share:
return jsonify({'error': 'Kein Zugriff'}), 403
share.color = color or None
db.session.commit()
return jsonify({'color': share.color or book.color}), 200
@api_bp.route('/addressbooks/<int:book_id>', methods=['DELETE'])
@token_required
def delete_addressbook(book_id):
user = request.current_user
book = db.session.get(AddressBook, book_id)
if not book or book.owner_id != user.id:
return jsonify({'error': 'Nicht gefunden'}), 404
return jsonify({'error': 'Nicht gefunden oder keine Berechtigung'}), 404
recipients = _book_recipients(book)
owner_id = book.owner_id
bid = book.id
db.session.delete(book)
db.session.commit()
_notify_addressbook(owner_id, bid, 'deleted', shared_with=recipients)
return jsonify({'message': 'Adressbuch geloescht'}), 200
# --- Contacts ---
# ---------------------------------------------------------------------------
# Contacts
# ---------------------------------------------------------------------------
@api_bp.route('/addressbooks/<int:book_id>/contacts', methods=['GET'])
@token_required
@@ -111,11 +391,16 @@ def list_contacts(book_id):
if err:
return err
search = request.args.get('search', '').strip()
query = Contact.query.filter_by(address_book_id=book_id)
search = (request.args.get('q') or '').strip()
q = Contact.query.filter_by(address_book_id=book_id)
if search:
query = query.filter(Contact.display_name.ilike(f'%{search}%'))
contacts = query.order_by(Contact.display_name).all()
like = f'%{search}%'
q = q.filter(
(Contact.display_name.ilike(like)) |
(Contact.primary_email.ilike(like)) |
(Contact.organization.ilike(like))
)
contacts = q.order_by(Contact.display_name).all()
return jsonify([c.to_dict() for c in contacts]), 200
@@ -127,29 +412,16 @@ def create_contact(book_id):
if err:
return err
data = request.get_json()
display_name = data.get('display_name', '').strip()
if not display_name:
return jsonify({'error': 'Name erforderlich'}), 400
contact_uid = str(uuid.uuid4())
email = data.get('email', '')
phone = data.get('phone', '')
org = data.get('organization', '')
notes = data.get('notes', '')
vcard = _build_vcard(contact_uid, display_name, email, phone, org, notes)
contact = Contact(
address_book_id=book_id,
uid=contact_uid,
vcard_data=vcard,
display_name=display_name,
email=email or None,
phone=phone or None,
)
data = request.get_json() or {}
contact = Contact(address_book_id=book_id, uid=str(uuid.uuid4()), vcard_data='')
_apply_fields_to_contact(contact, data)
if not contact.display_name:
return jsonify({'error': 'Name oder Firma erforderlich'}), 400
contact.vcard_data = _build_vcard(contact)
db.session.add(contact)
db.session.commit()
_notify_addressbook(book.owner_id, book.id, 'contact',
shared_with=_book_recipients(book))
return jsonify(contact.to_dict()), 201
@@ -160,11 +432,9 @@ def get_contact(contact_id):
contact = db.session.get(Contact, contact_id)
if not contact:
return jsonify({'error': 'Kontakt nicht gefunden'}), 404
book, err = _get_addressbook_or_err(contact.address_book_id, user)
if err:
return err
result = contact.to_dict()
result['vcard_data'] = contact.vcard_data
return jsonify(result), 200
@@ -177,29 +447,17 @@ def update_contact(contact_id):
contact = db.session.get(Contact, contact_id)
if not contact:
return jsonify({'error': 'Kontakt nicht gefunden'}), 404
book, err = _get_addressbook_or_err(contact.address_book_id, user, need_write=True)
if err:
return err
data = request.get_json()
if 'display_name' in data:
contact.display_name = data['display_name'].strip()
if 'email' in data:
contact.email = data['email'] or None
if 'phone' in data:
contact.phone = data['phone'] or None
contact.vcard_data = _build_vcard(
contact.uid,
contact.display_name,
data.get('email', contact.email or ''),
data.get('phone', contact.phone or ''),
data.get('organization', ''),
data.get('notes', ''),
)
data = request.get_json() or {}
_apply_fields_to_contact(contact, data)
contact.vcard_data = _build_vcard(contact)
contact.updated_at = datetime.now(timezone.utc)
db.session.commit()
_notify_addressbook(book.owner_id, book.id, 'contact',
shared_with=_book_recipients(book))
return jsonify(contact.to_dict()), 200
@@ -210,17 +468,19 @@ def delete_contact(contact_id):
contact = db.session.get(Contact, contact_id)
if not contact:
return jsonify({'error': 'Kontakt nicht gefunden'}), 404
book, err = _get_addressbook_or_err(contact.address_book_id, user, need_write=True)
if err:
return err
db.session.delete(contact)
db.session.commit()
_notify_addressbook(book.owner_id, book.id, 'contact',
shared_with=_book_recipients(book))
return jsonify({'message': 'Kontakt geloescht'}), 200
# --- Sharing ---
# ---------------------------------------------------------------------------
# Sharing
# ---------------------------------------------------------------------------
@api_bp.route('/addressbooks/<int:book_id>/share', methods=['POST'])
@token_required
@@ -230,10 +490,9 @@ def share_addressbook(book_id):
if not book or book.owner_id != user.id:
return jsonify({'error': 'Nur der Eigentuemer kann teilen'}), 403
data = request.get_json()
username = data.get('username', '').strip()
data = request.get_json() or {}
username = (data.get('username') or '').strip()
permission = data.get('permission', 'read')
if permission not in ('read', 'readwrite'):
return jsonify({'error': 'Ungueltige Berechtigung'}), 400
@@ -246,7 +505,6 @@ def share_addressbook(book_id):
existing = AddressBookShare.query.filter_by(
address_book_id=book_id, shared_with_id=target.id
).first()
is_new = not existing
if existing:
existing.permission = permission
else:
@@ -254,16 +512,9 @@ def share_addressbook(book_id):
address_book_id=book_id, shared_with_id=target.id, permission=permission
)
db.session.add(share)
db.session.commit()
if is_new:
try:
from app.services.system_mail import notify_contacts_shared
notify_contacts_shared(book.name, user.username, target, permission)
except Exception:
pass
_notify_addressbook(book.owner_id, book.id, 'share',
shared_with=[target.id, *_book_recipients(book)])
return jsonify({'message': f'Adressbuch mit {username} geteilt'}), 200
@@ -274,7 +525,6 @@ def list_addressbook_shares(book_id):
book = db.session.get(AddressBook, book_id)
if not book or book.owner_id != user.id:
return jsonify({'error': 'Nicht gefunden'}), 404
shares = AddressBookShare.query.filter_by(address_book_id=book_id).all()
return jsonify([{
'id': s.id,
@@ -291,17 +541,20 @@ def remove_addressbook_share(book_id, share_id):
book = db.session.get(AddressBook, book_id)
if not book or book.owner_id != user.id:
return jsonify({'error': 'Nicht gefunden'}), 404
share = db.session.get(AddressBookShare, share_id)
if not share or share.address_book_id != book_id:
return jsonify({'error': 'Freigabe nicht gefunden'}), 404
target_id = share.shared_with_id
db.session.delete(share)
db.session.commit()
_notify_addressbook(book.owner_id, book.id, 'share',
shared_with=[target_id, *_book_recipients(book)])
return jsonify({'message': 'Freigabe entfernt'}), 200
# --- Import/Export ---
# ---------------------------------------------------------------------------
# vCard export (all contacts of a book)
# ---------------------------------------------------------------------------
@api_bp.route('/addressbooks/<int:book_id>/export', methods=['GET'])
@token_required
@@ -310,40 +563,9 @@ def export_contacts(book_id):
book, err = _get_addressbook_or_err(book_id, user)
if err:
return err
contacts = Contact.query.filter_by(address_book_id=book_id).all()
vcards = '\r\n'.join(c.vcard_data for c in contacts)
from flask import Response
parts = [c.vcard_data for c in book.contacts]
return Response(
vcards,
mimetype='text/vcard',
'\r\n'.join(parts),
mimetype='text/vcard; charset=utf-8',
headers={'Content-Disposition': f'attachment; filename="{book.name}.vcf"'},
)
# --- Helpers ---
def _build_vcard(uid, display_name, email='', phone='', org='', notes=''):
parts = display_name.split(' ', 1)
first = parts[0]
last = parts[1] if len(parts) > 1 else ''
lines = [
'BEGIN:VCARD',
'VERSION:3.0',
f'UID:{uid}',
f'FN:{display_name}',
f'N:{last};{first};;;',
]
if email:
lines.append(f'EMAIL:{email}')
if phone:
lines.append(f'TEL:{phone}')
if org:
lines.append(f'ORG:{org}')
if notes:
lines.append(f'NOTE:{notes}')
lines.append(f'REV:{datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")}')
lines.append('END:VCARD')
return '\r\n'.join(lines)