114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
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()
|
|
)
|
|
)
|