350 lines
11 KiB
Python
350 lines
11 KiB
Python
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from flask import request, jsonify
|
|
|
|
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
|
|
|
|
|
|
def _get_addressbook_or_err(book_id, user, need_write=False):
|
|
book = db.session.get(AddressBook, book_id)
|
|
if not book:
|
|
return None, (jsonify({'error': 'Adressbuch nicht gefunden'}), 404)
|
|
if book.owner_id == user.id:
|
|
return book, None
|
|
share = AddressBookShare.query.filter_by(
|
|
address_book_id=book_id, shared_with_id=user.id
|
|
).first()
|
|
if not share:
|
|
return None, (jsonify({'error': 'Zugriff verweigert'}), 403)
|
|
if need_write and share.permission != 'readwrite':
|
|
return None, (jsonify({'error': 'Schreibzugriff verweigert'}), 403)
|
|
return book, None
|
|
|
|
|
|
# --- Address Books ---
|
|
|
|
@api_bp.route('/addressbooks', methods=['GET'])
|
|
@token_required
|
|
def list_addressbooks():
|
|
user = request.current_user
|
|
own = AddressBook.query.filter_by(owner_id=user.id).all()
|
|
shared_ids = [s.address_book_id for s in
|
|
AddressBookShare.query.filter_by(shared_with_id=user.id).all()]
|
|
shared = AddressBook.query.filter(AddressBook.id.in_(shared_ids)).all() if shared_ids else []
|
|
|
|
result = []
|
|
for b in own:
|
|
d = b.to_dict()
|
|
d['permission'] = 'owner'
|
|
d['contact_count'] = b.contacts.count()
|
|
result.append(d)
|
|
for b in shared:
|
|
d = b.to_dict()
|
|
share = AddressBookShare.query.filter_by(
|
|
address_book_id=b.id, shared_with_id=user.id
|
|
).first()
|
|
d['permission'] = share.permission if share else 'read'
|
|
d['owner_name'] = b.owner.username
|
|
d['contact_count'] = b.contacts.count()
|
|
result.append(d)
|
|
|
|
return jsonify(result), 200
|
|
|
|
|
|
@api_bp.route('/addressbooks', methods=['POST'])
|
|
@token_required
|
|
def create_addressbook():
|
|
user = request.current_user
|
|
data = request.get_json()
|
|
name = data.get('name', '').strip()
|
|
if not name:
|
|
return jsonify({'error': 'Name erforderlich'}), 400
|
|
|
|
book = AddressBook(owner_id=user.id, name=name, description=data.get('description', ''))
|
|
db.session.add(book)
|
|
db.session.commit()
|
|
return jsonify(book.to_dict()), 201
|
|
|
|
|
|
@api_bp.route('/addressbooks/<int:book_id>', methods=['PUT'])
|
|
@token_required
|
|
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
|
|
|
|
data = request.get_json()
|
|
if 'name' in data:
|
|
book.name = data['name'].strip()
|
|
if 'description' in data:
|
|
book.description = data['description']
|
|
db.session.commit()
|
|
return jsonify(book.to_dict()), 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
|
|
|
|
db.session.delete(book)
|
|
db.session.commit()
|
|
return jsonify({'message': 'Adressbuch geloescht'}), 200
|
|
|
|
|
|
# --- Contacts ---
|
|
|
|
@api_bp.route('/addressbooks/<int:book_id>/contacts', methods=['GET'])
|
|
@token_required
|
|
def list_contacts(book_id):
|
|
user = request.current_user
|
|
book, err = _get_addressbook_or_err(book_id, user)
|
|
if err:
|
|
return err
|
|
|
|
search = request.args.get('search', '').strip()
|
|
query = 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()
|
|
return jsonify([c.to_dict() for c in contacts]), 200
|
|
|
|
|
|
@api_bp.route('/addressbooks/<int:book_id>/contacts', methods=['POST'])
|
|
@token_required
|
|
def create_contact(book_id):
|
|
user = request.current_user
|
|
book, err = _get_addressbook_or_err(book_id, user, need_write=True)
|
|
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,
|
|
)
|
|
db.session.add(contact)
|
|
db.session.commit()
|
|
return jsonify(contact.to_dict()), 201
|
|
|
|
|
|
@api_bp.route('/contacts/<int:contact_id>', methods=['GET'])
|
|
@token_required
|
|
def get_contact(contact_id):
|
|
user = request.current_user
|
|
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
|
|
|
|
|
|
@api_bp.route('/contacts/<int:contact_id>', methods=['PUT'])
|
|
@token_required
|
|
def update_contact(contact_id):
|
|
user = request.current_user
|
|
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', ''),
|
|
)
|
|
contact.updated_at = datetime.now(timezone.utc)
|
|
db.session.commit()
|
|
return jsonify(contact.to_dict()), 200
|
|
|
|
|
|
@api_bp.route('/contacts/<int:contact_id>', methods=['DELETE'])
|
|
@token_required
|
|
def delete_contact(contact_id):
|
|
user = request.current_user
|
|
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()
|
|
return jsonify({'message': 'Kontakt geloescht'}), 200
|
|
|
|
|
|
# --- Sharing ---
|
|
|
|
@api_bp.route('/addressbooks/<int:book_id>/share', methods=['POST'])
|
|
@token_required
|
|
def share_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': 'Nur der Eigentuemer kann teilen'}), 403
|
|
|
|
data = request.get_json()
|
|
username = data.get('username', '').strip()
|
|
permission = data.get('permission', 'read')
|
|
|
|
if permission not in ('read', 'readwrite'):
|
|
return jsonify({'error': 'Ungueltige Berechtigung'}), 400
|
|
|
|
target = User.query.filter_by(username=username).first()
|
|
if not target:
|
|
return jsonify({'error': 'Benutzer nicht gefunden'}), 404
|
|
if target.id == user.id:
|
|
return jsonify({'error': 'Kann nicht mit sich selbst teilen'}), 400
|
|
|
|
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:
|
|
share = AddressBookShare(
|
|
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
|
|
|
|
return jsonify({'message': f'Adressbuch mit {username} geteilt'}), 200
|
|
|
|
|
|
@api_bp.route('/addressbooks/<int:book_id>/shares', methods=['GET'])
|
|
@token_required
|
|
def list_addressbook_shares(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
|
|
|
|
shares = AddressBookShare.query.filter_by(address_book_id=book_id).all()
|
|
return jsonify([{
|
|
'id': s.id,
|
|
'user_id': s.shared_with_id,
|
|
'username': s.shared_with.username,
|
|
'permission': s.permission,
|
|
} for s in shares]), 200
|
|
|
|
|
|
@api_bp.route('/addressbooks/<int:book_id>/shares/<int:share_id>', methods=['DELETE'])
|
|
@token_required
|
|
def remove_addressbook_share(book_id, share_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
|
|
|
|
share = db.session.get(AddressBookShare, share_id)
|
|
if not share or share.address_book_id != book_id:
|
|
return jsonify({'error': 'Freigabe nicht gefunden'}), 404
|
|
|
|
db.session.delete(share)
|
|
db.session.commit()
|
|
return jsonify({'message': 'Freigabe entfernt'}), 200
|
|
|
|
|
|
# --- Import/Export ---
|
|
|
|
@api_bp.route('/addressbooks/<int:book_id>/export', methods=['GET'])
|
|
@token_required
|
|
def export_contacts(book_id):
|
|
user = request.current_user
|
|
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
|
|
return Response(
|
|
vcards,
|
|
mimetype='text/vcard',
|
|
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)
|