| 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), |
| ] |
| ) |
|
|
|
|
|
|
|
|
