File size: 3,854 Bytes
69f2337 a08f988 69f2337 a08f988 69f2337 a08f988 69f2337 a08f988 69f2337 a08f988 69f2337 a08f988 69f2337 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 |
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),
]
)
|