intermittent commit
This commit is contained in:
500
src/govoplan_addresses/backend/vcard.py
Normal file
500
src/govoplan_addresses/backend/vcard.py
Normal file
@@ -0,0 +1,500 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Literal
|
||||
|
||||
from govoplan_addresses.backend.db.models import Contact
|
||||
from govoplan_addresses.backend.schemas import (
|
||||
ContactCreateRequest,
|
||||
ContactEmailPayload,
|
||||
ContactPhonePayload,
|
||||
ContactPostalAddressPayload,
|
||||
)
|
||||
|
||||
|
||||
class VCardError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ParsedVCard:
|
||||
payload: ContactCreateRequest
|
||||
raw: str
|
||||
source_ref: str | None = None
|
||||
source_revision: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ParsedVCardIssue:
|
||||
index: int
|
||||
message: str
|
||||
severity: Literal["warning", "error"] = "error"
|
||||
field: str | None = None
|
||||
line: int | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class VCardParseResult:
|
||||
cards: list[ParsedVCard]
|
||||
issues: list[ParsedVCardIssue]
|
||||
skipped: int
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class _VCardDraft:
|
||||
fn: str | None = None
|
||||
given_name: str | None = None
|
||||
family_name: str | None = None
|
||||
organization: str | None = None
|
||||
role_title: str | None = None
|
||||
note: str | None = None
|
||||
version: str | None = None
|
||||
uid: str | None = None
|
||||
revision: str | None = None
|
||||
tags: list[str] = field(default_factory=list)
|
||||
emails: list[ContactEmailPayload] = field(default_factory=list)
|
||||
phones: list[ContactPhonePayload] = field(default_factory=list)
|
||||
addresses: list[ContactPostalAddressPayload] = field(default_factory=list)
|
||||
urls: list[str] = field(default_factory=list)
|
||||
|
||||
|
||||
def _normalize_lines(content: str) -> list[str]:
|
||||
raw_lines = content.replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
||||
lines: list[str] = []
|
||||
for line in raw_lines:
|
||||
if line.startswith((" ", "\t")) and lines:
|
||||
lines[-1] += line[1:]
|
||||
elif line:
|
||||
lines.append(line)
|
||||
return lines
|
||||
|
||||
|
||||
def _split_unescaped(value: str, separator: str) -> list[str]:
|
||||
parts: list[str] = []
|
||||
current: list[str] = []
|
||||
escaped = False
|
||||
for char in value:
|
||||
if escaped:
|
||||
current.append(char)
|
||||
escaped = False
|
||||
elif char == "\\":
|
||||
current.append(char)
|
||||
escaped = True
|
||||
elif char == separator:
|
||||
parts.append("".join(current))
|
||||
current = []
|
||||
else:
|
||||
current.append(char)
|
||||
parts.append("".join(current))
|
||||
return parts
|
||||
|
||||
|
||||
def _unescape_text(value: str) -> str:
|
||||
return (
|
||||
value.replace("\\n", "\n")
|
||||
.replace("\\N", "\n")
|
||||
.replace("\\,", ",")
|
||||
.replace("\\;", ";")
|
||||
.replace("\\\\", "\\")
|
||||
.strip()
|
||||
)
|
||||
|
||||
|
||||
def _escape_text(value: str | None) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
return (
|
||||
value.replace("\\", "\\\\")
|
||||
.replace("\r\n", "\n")
|
||||
.replace("\r", "\n")
|
||||
.replace("\n", "\\n")
|
||||
.replace(";", "\\;")
|
||||
.replace(",", "\\,")
|
||||
)
|
||||
|
||||
|
||||
def _parse_head(head: str) -> tuple[str, dict[str, list[str]]]:
|
||||
parts = head.split(";")
|
||||
name = parts[0].split(".")[-1].upper()
|
||||
params: dict[str, list[str]] = {}
|
||||
for part in parts[1:]:
|
||||
if not part:
|
||||
continue
|
||||
if "=" in part:
|
||||
key, raw_value = part.split("=", 1)
|
||||
values = [item.strip().strip('"') for item in raw_value.split(",") if item.strip()]
|
||||
else:
|
||||
key = "TYPE"
|
||||
values = [part.strip().strip('"')]
|
||||
params.setdefault(key.upper(), []).extend(values)
|
||||
return name, params
|
||||
|
||||
|
||||
def _label_from_params(params: dict[str, list[str]]) -> str | None:
|
||||
ignored = {"INTERNET", "VOICE", "PREF"}
|
||||
for value in params.get("TYPE", []):
|
||||
normalized = value.strip().lower()
|
||||
if normalized and normalized.upper() not in ignored:
|
||||
return normalized
|
||||
return None
|
||||
|
||||
|
||||
def _line_value(line: str, *, card_index: int) -> tuple[str, dict[str, list[str]], str]:
|
||||
if ":" not in line:
|
||||
raise VCardError(f"vCard {card_index}: line is missing ':' separator.")
|
||||
head, value = line.split(":", 1)
|
||||
name, params = _parse_head(head)
|
||||
return name, params, value
|
||||
|
||||
|
||||
def _is_pref(params: dict[str, list[str]]) -> bool:
|
||||
values = [value.strip().upper() for value in params.get("TYPE", [])]
|
||||
values.extend(value.strip().upper() for value in params.get("PREF", []))
|
||||
return "PREF" in values or "1" in values
|
||||
|
||||
|
||||
def _card_blocks(content: str) -> list[list[str]]:
|
||||
result = _card_blocks_with_issues(content)
|
||||
if result.issues:
|
||||
raise VCardError(result.issues[0].message)
|
||||
return result.cards
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _CardBlockResult:
|
||||
cards: list[list[str]]
|
||||
issues: list[ParsedVCardIssue]
|
||||
|
||||
|
||||
def _card_blocks_with_issues(content: str) -> _CardBlockResult:
|
||||
lines = _normalize_lines(content)
|
||||
blocks: list[list[str]] = []
|
||||
issues: list[ParsedVCardIssue] = []
|
||||
current: list[str] | None = None
|
||||
for line in lines:
|
||||
card_index = len(blocks) + 1
|
||||
try:
|
||||
name, _params, value = _line_value(line, card_index=card_index)
|
||||
except VCardError as exc:
|
||||
issues.append(ParsedVCardIssue(index=card_index, message=str(exc), field="line"))
|
||||
continue
|
||||
if name == "BEGIN" and line.split(":", 1)[1].upper() == "VCARD":
|
||||
if current is not None:
|
||||
issues.append(ParsedVCardIssue(index=card_index, message="Nested vCard BEGIN is not supported."))
|
||||
current = None
|
||||
continue
|
||||
current = [line]
|
||||
elif name == "END" and value.upper() == "VCARD":
|
||||
if current is None:
|
||||
issues.append(ParsedVCardIssue(index=card_index, message="vCard END appears before BEGIN."))
|
||||
continue
|
||||
current.append(line)
|
||||
blocks.append(current)
|
||||
current = None
|
||||
elif current is not None:
|
||||
current.append(line)
|
||||
if current is not None:
|
||||
issues.append(ParsedVCardIssue(index=len(blocks) + 1, message="vCard BEGIN has no matching END."))
|
||||
if not blocks:
|
||||
issues.append(ParsedVCardIssue(index=0, message="No vCard entries found."))
|
||||
return _CardBlockResult(cards=blocks, issues=issues)
|
||||
|
||||
|
||||
def _parse_card(index: int, block: list[str]) -> tuple[ParsedVCard | None, list[ParsedVCardIssue]]:
|
||||
draft = _VCardDraft()
|
||||
issues: list[ParsedVCardIssue] = []
|
||||
|
||||
for line in block:
|
||||
parsed = _parse_card_line(index, line, issues)
|
||||
if parsed is not None:
|
||||
name, params, value = parsed
|
||||
_apply_card_property(index, draft, name, params, value, issues)
|
||||
|
||||
if not _draft_has_identity(draft):
|
||||
issues.append(ParsedVCardIssue(index=index, message=f"vCard {index}: contact has no name or email."))
|
||||
return None, issues
|
||||
|
||||
raw = "\n".join(block)
|
||||
payload = _draft_contact_payload(draft)
|
||||
return ParsedVCard(payload=payload, raw=raw, source_ref=draft.uid, source_revision=draft.revision), issues
|
||||
|
||||
|
||||
def _parse_card_line(
|
||||
index: int,
|
||||
line: str,
|
||||
issues: list[ParsedVCardIssue],
|
||||
) -> tuple[str, dict[str, list[str]], str] | None:
|
||||
try:
|
||||
return _line_value(line, card_index=index)
|
||||
except VCardError as exc:
|
||||
issues.append(ParsedVCardIssue(index=index, message=str(exc), field="line"))
|
||||
return None
|
||||
|
||||
|
||||
def _apply_card_property(
|
||||
index: int,
|
||||
draft: _VCardDraft,
|
||||
name: str,
|
||||
params: dict[str, list[str]],
|
||||
value: str,
|
||||
issues: list[ParsedVCardIssue],
|
||||
) -> None:
|
||||
if name in {"BEGIN", "END"}:
|
||||
return
|
||||
if _apply_card_metadata(index, draft, name, value, issues):
|
||||
return
|
||||
if _apply_card_identity(draft, name, value):
|
||||
return
|
||||
_apply_card_contact_detail(index, draft, name, params, value, issues)
|
||||
|
||||
|
||||
def _apply_card_metadata(
|
||||
index: int,
|
||||
draft: _VCardDraft,
|
||||
name: str,
|
||||
value: str,
|
||||
issues: list[ParsedVCardIssue],
|
||||
) -> bool:
|
||||
if name == "VERSION":
|
||||
draft.version = value.strip()
|
||||
if draft.version and draft.version not in {"3.0", "4.0"}:
|
||||
issues.append(ParsedVCardIssue(index=index, severity="warning", field="VERSION", message=f"vCard version {draft.version} is not fully supported."))
|
||||
return True
|
||||
if name == "UID":
|
||||
draft.uid = _unescape_text(value) or draft.uid
|
||||
return True
|
||||
if name == "REV":
|
||||
draft.revision = _unescape_text(value) or draft.revision
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _apply_card_identity(draft: _VCardDraft, name: str, value: str) -> bool:
|
||||
if name == "FN":
|
||||
draft.fn = _unescape_text(value)
|
||||
return True
|
||||
if name == "N":
|
||||
parts = [_unescape_text(item) for item in _split_unescaped(value, ";")]
|
||||
draft.family_name = parts[0] if len(parts) > 0 and parts[0] else draft.family_name
|
||||
draft.given_name = parts[1] if len(parts) > 1 and parts[1] else draft.given_name
|
||||
return True
|
||||
if name == "ORG":
|
||||
organization_parts = _unescaped_nonempty_values(value, ";")
|
||||
draft.organization = " / ".join(organization_parts) or draft.organization
|
||||
return True
|
||||
if name in {"TITLE", "ROLE"}:
|
||||
draft.role_title = _unescape_text(value) or draft.role_title
|
||||
return True
|
||||
if name == "NOTE":
|
||||
draft.note = _unescape_text(value) or draft.note
|
||||
return True
|
||||
if name == "CATEGORIES":
|
||||
draft.tags.extend(_unescaped_nonempty_values(value, ","))
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _apply_card_contact_detail(
|
||||
index: int,
|
||||
draft: _VCardDraft,
|
||||
name: str,
|
||||
params: dict[str, list[str]],
|
||||
value: str,
|
||||
issues: list[ParsedVCardIssue],
|
||||
) -> None:
|
||||
if name == "EMAIL":
|
||||
_append_card_email(index, draft, params, value, issues)
|
||||
elif name == "TEL":
|
||||
_append_card_phone(draft, params, value)
|
||||
elif name == "ADR":
|
||||
_append_card_address(draft, params, value)
|
||||
elif name == "URL":
|
||||
url = _unescape_text(value)
|
||||
if url:
|
||||
draft.urls.append(url)
|
||||
|
||||
|
||||
def _append_card_email(
|
||||
index: int,
|
||||
draft: _VCardDraft,
|
||||
params: dict[str, list[str]],
|
||||
value: str,
|
||||
issues: list[ParsedVCardIssue],
|
||||
) -> None:
|
||||
email = _unescape_text(value)
|
||||
if not email:
|
||||
return
|
||||
if "@" not in email:
|
||||
issues.append(ParsedVCardIssue(index=index, severity="warning", field="EMAIL", message=f"Skipped invalid email address: {email}"))
|
||||
return
|
||||
draft.emails.append(ContactEmailPayload(label=_label_from_params(params), email=email, is_primary=_is_pref(params) or not draft.emails))
|
||||
|
||||
|
||||
def _append_card_phone(draft: _VCardDraft, params: dict[str, list[str]], value: str) -> None:
|
||||
phone = _unescape_text(value)
|
||||
if phone:
|
||||
draft.phones.append(ContactPhonePayload(label=_label_from_params(params), phone=phone, is_primary=_is_pref(params) or not draft.phones))
|
||||
|
||||
|
||||
def _append_card_address(draft: _VCardDraft, params: dict[str, list[str]], value: str) -> None:
|
||||
parts = [_unescape_text(item) for item in _split_unescaped(value, ";")]
|
||||
while len(parts) < 7:
|
||||
parts.append("")
|
||||
if not any(parts):
|
||||
return
|
||||
street = "\n".join(part for part in (parts[1], parts[2]) if part)
|
||||
draft.addresses.append(
|
||||
ContactPostalAddressPayload(
|
||||
label=_label_from_params(params),
|
||||
street=street or None,
|
||||
locality=parts[3] or None,
|
||||
region=parts[4] or None,
|
||||
postal_code=parts[5] or None,
|
||||
country=parts[6] or None,
|
||||
is_primary=_is_pref(params) or not draft.addresses,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _unescaped_nonempty_values(value: str, separator: str) -> list[str]:
|
||||
return [text for text in (_unescape_text(item) for item in _split_unescaped(value, separator)) if text]
|
||||
|
||||
|
||||
def _draft_has_identity(draft: _VCardDraft) -> bool:
|
||||
return bool(draft.fn or draft.given_name or draft.family_name or draft.emails)
|
||||
|
||||
|
||||
def _draft_contact_payload(draft: _VCardDraft) -> ContactCreateRequest:
|
||||
payload = ContactCreateRequest(
|
||||
display_name=draft.fn,
|
||||
given_name=draft.given_name,
|
||||
family_name=draft.family_name,
|
||||
organization=draft.organization,
|
||||
role_title=draft.role_title,
|
||||
note=draft.note,
|
||||
tags=draft.tags,
|
||||
emails=draft.emails,
|
||||
phones=draft.phones,
|
||||
postal_addresses=draft.addresses,
|
||||
provenance={
|
||||
"vcard": {
|
||||
"version": draft.version,
|
||||
"uid": draft.uid,
|
||||
"revision": draft.revision,
|
||||
"urls": draft.urls,
|
||||
}
|
||||
},
|
||||
)
|
||||
return payload
|
||||
|
||||
|
||||
def parse_vcards_with_issues(content: str) -> VCardParseResult:
|
||||
blocks = _card_blocks_with_issues(content)
|
||||
parsed: list[ParsedVCard] = []
|
||||
issues = list(blocks.issues)
|
||||
skipped = 0
|
||||
for index, block in enumerate(blocks.cards, start=1):
|
||||
card, card_issues = _parse_card(index, block)
|
||||
issues.extend(card_issues)
|
||||
if card is None:
|
||||
skipped += 1
|
||||
else:
|
||||
parsed.append(card)
|
||||
return VCardParseResult(cards=parsed, issues=issues, skipped=skipped)
|
||||
|
||||
|
||||
def parse_vcards(content: str) -> list[ParsedVCard]:
|
||||
result = parse_vcards_with_issues(content)
|
||||
errors = [issue for issue in result.issues if issue.severity == "error"]
|
||||
if errors:
|
||||
raise VCardError(errors[0].message)
|
||||
return result.cards
|
||||
|
||||
|
||||
def contact_to_vcard(contact: Contact) -> str:
|
||||
lines = _contact_identity_lines(contact)
|
||||
lines.extend(_contact_email_lines(contact))
|
||||
lines.extend(_contact_phone_lines(contact))
|
||||
lines.extend(_contact_address_lines(contact))
|
||||
lines.extend(_contact_note_and_tag_lines(contact))
|
||||
lines.extend(_contact_url_lines(contact))
|
||||
lines.append("END:VCARD")
|
||||
return "\r\n".join(lines) + "\r\n"
|
||||
|
||||
|
||||
def _contact_identity_lines(contact: Contact) -> list[str]:
|
||||
lines = [
|
||||
"BEGIN:VCARD",
|
||||
"VERSION:4.0",
|
||||
f"FN:{_escape_text(contact.display_name)}",
|
||||
f"N:{_escape_text(contact.family_name)};{_escape_text(contact.given_name)};;;",
|
||||
]
|
||||
if contact.source_ref:
|
||||
lines.append(f"UID:{_escape_text(contact.source_ref)}")
|
||||
if contact.source_revision:
|
||||
lines.append(f"REV:{_escape_text(contact.source_revision)}")
|
||||
if contact.organization:
|
||||
lines.append(f"ORG:{_escape_text(contact.organization)}")
|
||||
if contact.role_title:
|
||||
lines.append(f"TITLE:{_escape_text(contact.role_title)}")
|
||||
return lines
|
||||
|
||||
|
||||
def _contact_email_lines(contact: Contact) -> list[str]:
|
||||
lines: list[str] = []
|
||||
for email in contact.emails:
|
||||
label = f";TYPE={_escape_text(email.label)}" if email.label else ""
|
||||
lines.append(f"EMAIL{label}:{_escape_text(email.email)}")
|
||||
return lines
|
||||
|
||||
|
||||
def _contact_phone_lines(contact: Contact) -> list[str]:
|
||||
lines: list[str] = []
|
||||
for phone in contact.phones:
|
||||
label = f";TYPE={_escape_text(phone.label)}" if phone.label else ""
|
||||
lines.append(f"TEL{label}:{_escape_text(phone.phone)}")
|
||||
return lines
|
||||
|
||||
|
||||
def _contact_address_lines(contact: Contact) -> list[str]:
|
||||
lines: list[str] = []
|
||||
for address in contact.postal_addresses:
|
||||
label = f";TYPE={_escape_text(address.label)}" if address.label else ""
|
||||
lines.append(
|
||||
"ADR"
|
||||
f"{label}:;;{_escape_text(address.street)};{_escape_text(address.locality)};"
|
||||
f"{_escape_text(address.region)};{_escape_text(address.postal_code)};{_escape_text(address.country)}"
|
||||
)
|
||||
return lines
|
||||
|
||||
|
||||
def _contact_note_and_tag_lines(contact: Contact) -> list[str]:
|
||||
lines: list[str] = []
|
||||
if contact.note:
|
||||
lines.append(f"NOTE:{_escape_text(contact.note)}")
|
||||
if contact.tags:
|
||||
lines.append(f"CATEGORIES:{','.join(_escape_text(tag) for tag in contact.tags)}")
|
||||
return lines
|
||||
|
||||
|
||||
def _contact_url_lines(contact: Contact) -> list[str]:
|
||||
lines: list[str] = []
|
||||
urls = _contact_vcard_urls(contact)
|
||||
if isinstance(urls, list):
|
||||
for url in urls:
|
||||
if isinstance(url, str) and url.strip():
|
||||
lines.append(f"URL:{_escape_text(url.strip())}")
|
||||
return lines
|
||||
|
||||
|
||||
def _contact_vcard_urls(contact: Contact) -> object:
|
||||
if not isinstance(contact.provenance, dict):
|
||||
return None
|
||||
vcard = contact.provenance.get("vcard")
|
||||
if not isinstance(vcard, dict):
|
||||
return None
|
||||
return vcard.get("urls")
|
||||
|
||||
|
||||
def contacts_to_vcard(contacts: list[Contact]) -> str:
|
||||
return "".join(contact_to_vcard(contact) for contact in contacts)
|
||||
Reference in New Issue
Block a user