|
|
import base64 |
|
|
from datetime import datetime |
|
|
from typing import List |
|
|
from uuid import UUID |
|
|
|
|
|
from app.api.topics.schemas import Topic, TopicMessage |
|
|
from app.core.config import ( |
|
|
ENCODING_ASCII, |
|
|
ENCODING_UTF8, |
|
|
TABLE_PARTICIPANTS, |
|
|
TOPIC_EXPECTED_PARTS, |
|
|
TOPIC_INNER_SEPARATOR, |
|
|
TOPIC_LIST_SEPARATOR, |
|
|
TOPIC_MESSAGE_PART_SEPARATOR, |
|
|
TOPIC_PARSE_EMPTY_ERROR, |
|
|
TOPIC_PARSE_INVALID_ERROR, |
|
|
TOPIC_SEPARATOR, |
|
|
) |
|
|
from app.core.link import Link, ParticipantLink |
|
|
|
|
|
|
|
|
class TopicLineParser: |
|
|
SEPARATOR = TOPIC_SEPARATOR |
|
|
EXPECTED_PARTS = TOPIC_EXPECTED_PARTS |
|
|
LIST_SEPARATOR = TOPIC_LIST_SEPARATOR |
|
|
INNER_SEPARATOR = TOPIC_INNER_SEPARATOR |
|
|
MESSAGE_PART_SEPARATOR = TOPIC_MESSAGE_PART_SEPARATOR |
|
|
|
|
|
@staticmethod |
|
|
def _decode_message(encoded: str) -> str: |
|
|
return base64.b64decode(encoded.encode(ENCODING_ASCII)).decode(ENCODING_UTF8) |
|
|
|
|
|
@staticmethod |
|
|
def _encode_message(value: str) -> str: |
|
|
return base64.b64encode(value.encode(ENCODING_UTF8)).decode(ENCODING_ASCII) |
|
|
|
|
|
@classmethod |
|
|
def _parse_participants(cls, raw: str) -> List[ParticipantLink]: |
|
|
if not raw: |
|
|
return [] |
|
|
links: List[ParticipantLink] = [] |
|
|
for value in raw.split(cls.INNER_SEPARATOR): |
|
|
if not value: |
|
|
continue |
|
|
links.append(ParticipantLink.from_raw(value)) |
|
|
return links |
|
|
|
|
|
@classmethod |
|
|
def _parse_messages(cls, raw: str) -> List[TopicMessage]: |
|
|
if not raw: |
|
|
return [] |
|
|
messages: List[TopicMessage] = [] |
|
|
for chunk in raw.split(cls.LIST_SEPARATOR): |
|
|
if not chunk: |
|
|
continue |
|
|
try: |
|
|
participant_raw, encoded_content = chunk.split(cls.MESSAGE_PART_SEPARATOR, 1) |
|
|
except ValueError: |
|
|
continue |
|
|
messages.append( |
|
|
TopicMessage( |
|
|
participant_id=ParticipantLink.from_raw(participant_raw), |
|
|
content=cls._decode_message(encoded_content), |
|
|
) |
|
|
) |
|
|
return messages |
|
|
|
|
|
@classmethod |
|
|
def _serialize_participants(cls, participants: List[ParticipantLink]) -> str: |
|
|
return cls.INNER_SEPARATOR.join(participant.as_path() for participant in participants) |
|
|
|
|
|
@classmethod |
|
|
def _serialize_messages(cls, messages: List[TopicMessage]) -> str: |
|
|
return cls.LIST_SEPARATOR.join( |
|
|
cls.MESSAGE_PART_SEPARATOR.join( |
|
|
[message.participant_id.as_path(), cls._encode_message(message.content)] |
|
|
) |
|
|
for message in messages |
|
|
) |
|
|
|
|
|
@classmethod |
|
|
def parse(cls, line: str) -> Topic: |
|
|
raw = line.strip() |
|
|
if not raw: |
|
|
raise ValueError(TOPIC_PARSE_EMPTY_ERROR) |
|
|
|
|
|
parts = raw.split(cls.SEPARATOR) |
|
|
if len(parts) != cls.EXPECTED_PARTS: |
|
|
raise ValueError(TOPIC_PARSE_INVALID_ERROR) |
|
|
|
|
|
( |
|
|
topic_id, |
|
|
title, |
|
|
description, |
|
|
created_at_value, |
|
|
participants_raw, |
|
|
messages_raw, |
|
|
) = parts |
|
|
|
|
|
participants = cls._parse_participants(participants_raw) |
|
|
messages = cls._parse_messages(messages_raw) |
|
|
|
|
|
return Topic( |
|
|
id=UUID(topic_id), |
|
|
title=title, |
|
|
description=description, |
|
|
created_at=datetime.fromisoformat(created_at_value), |
|
|
participants=participants, |
|
|
messages=messages, |
|
|
) |
|
|
|
|
|
@classmethod |
|
|
def serialize(cls, topic: Topic) -> str: |
|
|
return cls.SEPARATOR.join( |
|
|
[ |
|
|
str(topic.id), |
|
|
topic.title, |
|
|
topic.description, |
|
|
topic.created_at.isoformat(), |
|
|
cls._serialize_participants(topic.participants), |
|
|
cls._serialize_messages(topic.messages), |
|
|
] |
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
|
