import json import logging from pathlib import Path from models import User logger = logging.getLogger(__name__) class Radicale: data_path: Path collection_path: Path ROOT_USER = "root" def __init__(self, data_path: str) -> None: self.data_path = Path(data_path) self.collection_root = self.data_path / Path("collections", "collection-root") self.collection_path = self.collection_root / Path( Radicale.ROOT_USER, "contacts" ) self.init_radicale_dir() def init_radicale_dir(self): logger.info("Initializing Radicale directory..") self.collection_path.mkdir(parents=True, exist_ok=True) with open(self.collection_path / Path(".Radicale.props"), "w") as f: meta = { "D:displayname": "contacts", "tag": "VADDRESSBOOK", "{http://inf-it.com/ns/ab/}addressbook-color": "#ff0000ff", } f.write(json.dumps(meta)) # ensure radicale data is in the correct state for user in User.all_viewable_users(): self.create_or_update_vcf(user) # delete and recreate user symlinks for dir in self.collection_root.iterdir(): if dir.is_symlink(): dir.unlink() for user in User.all_users(): self.ensure_user_exists(user) self.create_or_update_authfile() @staticmethod def user_to_vcf(user: User): fname = f"{user.uuid}.vcf" content = [ "BEGIN:VCARD", "VERSION:2.1", f"UID:{user.uuid}", f"REV:{user.last_updated.strftime('%Y%m%dT%H%M%SZ')}", ] fn = " ".join(filter(None, [user.first_name, user.last_name])) # RFC6350 says FN has cardinality 1*, so must have at least one # software may not support multiple FNs, so output exactly one if fn: content.append(f"FN:{fn}") elif user.display_name: content.append(f"FN:{user.display_name}") else: content.append(f"FN:{user.username}") # only include N if one of its fields is non-empty if user.last_name or user.first_name: content.append( f"N:" f"{user.last_name if user.last_name else ''};" f"{user.first_name if user.first_name else ''};;;" ) if user.phone: content.append(f"TEL:{user.phone}") if user.email: content.append(f"EMAIL:{user.email}") if user.birthdate: content.append(f"BDAY:{user.birthdate.strftime('%Y%m%d')}") if user.lat and user.lon: content.append(f"GEO:geo:{user.lat},{user.lon}") if any([user.street, user.number, user.postal, user.city, user.country]): street = " ".join(filter(None, [user.street, user.number])) content.append( f"ADR:" ";" # po-box ";" # ext addr f"{street};" f"{user.city if user.city else ''};" ";" # region f"{user.postal if user.postal else ''};" f"{user.country if user.country else ''};" ) content.append("END:VCARD") return Path(fname), "\n".join(content) def create_or_update_vcf(self, user: User): vcf_fname, vcf_content = Radicale.user_to_vcf(user) with open(self.collection_path / vcf_fname, "w") as f: f.write(vcf_content) def ensure_user_exists(self, user: User): if user.username == Radicale.ROOT_USER: raise Exception( f"Got user object with reserved Radicale username '{Radicale.ROOT_USER}'. Panicking!" ) userpath = self.collection_root / Path(user.username) if userpath.exists(): userpath.unlink() userpath.symlink_to(Path(Radicale.ROOT_USER)) def create_or_update_authfile(self): with open(self.data_path / Path("users"), "w") as f: f.write( "\n".join( f"{user.username}:{user.carddav_key}" for user in User.all_users() ) )