feat: Import/Export fuer Kontakte und Kalender + Bulk-Loeschen Kontakte
Kontakte: - Mehrfachauswahl in der Liste (Checkbox-Spalte) mit Bulk-Loeschen - Export als Sammel-vCard (.vcf), als ZIP mit Einzel-vCards oder als CSV - Import aus vCard (mehrere im File moeglich) oder CSV; Match per UID, bestehende Kontakte werden aktualisiert Kalender: - Export als iCalendar (.ics) oder CSV - Import aus .ics oder CSV; bestehende Termine via UID aktualisiert Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Stefan Hacker
@@ -1,6 +1,9 @@
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
import uuid
|
||||
import zipfile
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from flask import request, jsonify, Response
|
||||
@@ -404,6 +407,161 @@ def list_contacts(book_id):
|
||||
return jsonify([c.to_dict() for c in contacts]), 200
|
||||
|
||||
|
||||
@api_bp.route('/addressbooks/<int:book_id>/export', methods=['GET'])
|
||||
@token_required
|
||||
def export_addressbook(book_id):
|
||||
"""Export contacts as a single .vcf, a .zip with one .vcf per contact, or .csv."""
|
||||
user = request.current_user
|
||||
book, err = _get_addressbook_or_err(book_id, user)
|
||||
if err:
|
||||
return err
|
||||
fmt = (request.args.get('format') or 'vcf').lower()
|
||||
contacts = Contact.query.filter_by(address_book_id=book_id).order_by(Contact.display_name).all()
|
||||
safe_name = re.sub(r'[^A-Za-z0-9._-]+', '_', book.name or 'kontakte') or 'kontakte'
|
||||
|
||||
if fmt == 'vcf':
|
||||
body = '\r\n'.join((c.vcard_data or _build_vcard(c)).strip() for c in contacts) + '\r\n'
|
||||
return Response(
|
||||
body, mimetype='text/vcard; charset=utf-8',
|
||||
headers={'Content-Disposition': f'attachment; filename="{safe_name}.vcf"'},
|
||||
)
|
||||
if fmt == 'vcf-zip':
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||||
seen = {}
|
||||
for c in contacts:
|
||||
base = re.sub(r'[^A-Za-z0-9._-]+', '_', c.display_name or c.uid) or c.uid
|
||||
seen[base] = seen.get(base, 0) + 1
|
||||
fname = f"{base}.vcf" if seen[base] == 1 else f"{base}_{seen[base]}.vcf"
|
||||
zf.writestr(fname, (c.vcard_data or _build_vcard(c)).strip() + '\r\n')
|
||||
buf.seek(0)
|
||||
return Response(
|
||||
buf.read(), mimetype='application/zip',
|
||||
headers={'Content-Disposition': f'attachment; filename="{safe_name}.zip"'},
|
||||
)
|
||||
if fmt == 'csv':
|
||||
out = io.StringIO()
|
||||
cols = ['display_name', 'prefix', 'first_name', 'middle_name', 'last_name', 'suffix',
|
||||
'nickname', 'organization', 'department', 'job_title',
|
||||
'primary_email', 'primary_phone', 'birthday', 'anniversary',
|
||||
'emails', 'phones', 'addresses', 'websites', 'categories', 'notes']
|
||||
w = csv.writer(out, delimiter=';', quoting=csv.QUOTE_ALL)
|
||||
w.writerow(cols)
|
||||
for c in contacts:
|
||||
d = c.to_dict()
|
||||
row = []
|
||||
for col in cols:
|
||||
v = d.get(col, '')
|
||||
if isinstance(v, list):
|
||||
if v and isinstance(v[0], dict):
|
||||
v = '; '.join(
|
||||
(x.get('value') or x.get('street') or '') +
|
||||
(f" ({x.get('type')})" if x.get('type') else '')
|
||||
for x in v if isinstance(x, dict)
|
||||
)
|
||||
else:
|
||||
v = ', '.join(str(x) for x in v)
|
||||
row.append('' if v is None else str(v))
|
||||
w.writerow(row)
|
||||
return Response(
|
||||
'\ufeff' + out.getvalue(), mimetype='text/csv; charset=utf-8',
|
||||
headers={'Content-Disposition': f'attachment; filename="{safe_name}.csv"'},
|
||||
)
|
||||
return jsonify({'error': 'Unbekanntes Format'}), 400
|
||||
|
||||
|
||||
@api_bp.route('/addressbooks/<int:book_id>/import', methods=['POST'])
|
||||
@token_required
|
||||
def import_addressbook(book_id):
|
||||
"""Import vCard (.vcf, single oder mehrere im File) oder CSV."""
|
||||
user = request.current_user
|
||||
book, err = _get_addressbook_or_err(book_id, user, need_write=True)
|
||||
if err:
|
||||
return err
|
||||
file = request.files.get('file')
|
||||
if not file:
|
||||
return jsonify({'error': 'Keine Datei'}), 400
|
||||
raw = file.read()
|
||||
name = (file.filename or '').lower()
|
||||
try:
|
||||
text = raw.decode('utf-8-sig')
|
||||
except UnicodeDecodeError:
|
||||
text = raw.decode('latin-1', errors='replace')
|
||||
|
||||
imported = 0
|
||||
skipped = 0
|
||||
|
||||
def _add_from_parsed(parsed: dict, raw_text: str | None = None) -> bool:
|
||||
nonlocal imported, skipped
|
||||
if not parsed.get('display_name') and not parsed.get('first_name') \
|
||||
and not parsed.get('last_name') and not parsed.get('organization'):
|
||||
skipped += 1
|
||||
return False
|
||||
uid = parsed.get('uid') or str(uuid.uuid4())
|
||||
existing = Contact.query.filter_by(address_book_id=book_id, uid=uid).first()
|
||||
contact = existing or Contact(address_book_id=book_id, uid=uid, vcard_data='')
|
||||
_apply_fields_to_contact(contact, parsed)
|
||||
contact.vcard_data = (raw_text or '').strip() or _build_vcard(contact)
|
||||
contact.updated_at = datetime.now(timezone.utc)
|
||||
if not existing:
|
||||
db.session.add(contact)
|
||||
imported += 1
|
||||
return True
|
||||
|
||||
if name.endswith('.csv') or (b',' in raw[:200] and b'BEGIN:VCARD' not in raw[:200]):
|
||||
# CSV import
|
||||
reader = csv.DictReader(io.StringIO(text), delimiter=';')
|
||||
if not reader.fieldnames or len(reader.fieldnames) < 2:
|
||||
# try comma
|
||||
reader = csv.DictReader(io.StringIO(text), delimiter=',')
|
||||
for row in reader:
|
||||
row = {k.strip().lower(): (v or '').strip() for k, v in row.items() if k}
|
||||
parsed = {
|
||||
'display_name': row.get('display_name') or row.get('name')
|
||||
or row.get('vollname') or row.get('full name'),
|
||||
'first_name': row.get('first_name') or row.get('vorname'),
|
||||
'last_name': row.get('last_name') or row.get('nachname'),
|
||||
'middle_name': row.get('middle_name'),
|
||||
'prefix': row.get('prefix') or row.get('anrede'),
|
||||
'suffix': row.get('suffix'),
|
||||
'nickname': row.get('nickname') or row.get('spitzname'),
|
||||
'organization': row.get('organization') or row.get('firma') or row.get('company'),
|
||||
'department': row.get('department') or row.get('abteilung'),
|
||||
'job_title': row.get('job_title') or row.get('position') or row.get('title'),
|
||||
'birthday': row.get('birthday') or row.get('geburtstag'),
|
||||
'notes': row.get('notes') or row.get('notizen'),
|
||||
'emails': [], 'phones': [], 'addresses': [], 'websites': [], 'categories': [],
|
||||
}
|
||||
email = row.get('primary_email') or row.get('email') or row.get('e-mail')
|
||||
if email:
|
||||
parsed['emails'].append({'type': 'home', 'value': email})
|
||||
phone = row.get('primary_phone') or row.get('phone') or row.get('telefon') or row.get('mobil')
|
||||
if phone:
|
||||
parsed['phones'].append({'type': 'cell', 'value': phone})
|
||||
cats = row.get('categories') or row.get('kategorien')
|
||||
if cats:
|
||||
parsed['categories'] = [c.strip() for c in cats.split(',') if c.strip()]
|
||||
_add_from_parsed(parsed)
|
||||
else:
|
||||
# vCard - eine oder mehrere im File
|
||||
parts = re.findall(r'BEGIN:VCARD.*?END:VCARD', text, flags=re.DOTALL | re.IGNORECASE)
|
||||
if not parts:
|
||||
return jsonify({'error': 'Keine VCARD-Daten gefunden'}), 400
|
||||
for vcf in parts:
|
||||
try:
|
||||
parsed = parse_vcard(vcf)
|
||||
except Exception:
|
||||
skipped += 1
|
||||
continue
|
||||
_add_from_parsed(parsed, raw_text=vcf)
|
||||
|
||||
db.session.commit()
|
||||
if imported:
|
||||
_notify_addressbook(book.owner_id, book.id, 'contact',
|
||||
shared_with=_book_recipients(book))
|
||||
return jsonify({'imported': imported, 'skipped': skipped}), 200
|
||||
|
||||
|
||||
@api_bp.route('/addressbooks/<int:book_id>/contacts', methods=['POST'])
|
||||
@token_required
|
||||
def create_contact(book_id):
|
||||
|
||||
Reference in New Issue
Block a user