File size: 5,127 Bytes
69f2337 a08f988 69f2337 a08f988 |
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 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 |
from datetime import datetime
from typing import List, Optional
from uuid import UUID, uuid4
from fastapi import HTTPException, status
from app.api.messages.parser import MessageLineParser
from app.api.messages.schemas import Message, MessageCreate
from app.api.participants.parser import ParticipantLineParser
from app.api.participants.schemas import Participant
from app.api.topics.schemas import Topic
from app.core.config import (
FORBIDDEN_WORDS,
FORBIDDEN_WORDS_MESSAGE,
PARTICIPANT_NOT_FOUND_MESSAGE,
MESSAGE_NOT_FOUND_MESSAGE,
TABLE_PARTICIPANTS,
TABLE_TOPICS,
TOPIC_NOT_FOUND_MESSAGE,
)
from app.core.file_manager import FileManager
from app.core.link import Link
class MessageService:
def __init__(
self,
file_manager: FileManager,
topic_file_manager: FileManager,
participant_file_manager: FileManager,
) -> None:
self.file_manager = file_manager
self.topic_file_manager = topic_file_manager
self.participant_file_manager = participant_file_manager
def _check_forbidden_words(self, content: str) -> Optional[str]:
content_lower = content.lower()
for word in FORBIDDEN_WORDS:
if word in content_lower:
return word
return None
def _load_topics(self) -> List[Topic]:
from app.api.topics.parser import TopicLineParser
topics: List[Topic] = []
for line in self.topic_file_manager.read_lines():
try:
topics.append(TopicLineParser.parse(line))
except ValueError:
continue
return topics
def _load_participants(self) -> List[Participant]:
participants: List[Participant] = []
for line in self.participant_file_manager.read_lines():
try:
participants.append(ParticipantLineParser.parse(line))
except ValueError:
continue
return participants
def _get_topic_by_id(self, topic_id: UUID) -> Topic:
for topic in self._load_topics():
if topic.id == topic_id:
return topic
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=TOPIC_NOT_FOUND_MESSAGE,
)
def _get_participant_by_id(self, participant_id: UUID) -> Participant:
for participant in self._load_participants():
if participant.id == participant_id:
return participant
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=PARTICIPANT_NOT_FOUND_MESSAGE,
)
def _resolve_id(self, link: Link, expected_table: str) -> UUID:
return link.resolve({expected_table: lambda value: value})
def get_topic(self, topic_link: Link) -> Topic:
topic_id = self._resolve_id(topic_link, TABLE_TOPICS)
return self._get_topic_by_id(topic_id)
def get_participant(self, participant_link: Link) -> Participant:
participant_id = self._resolve_id(participant_link, TABLE_PARTICIPANTS)
return self._get_participant_by_id(participant_id)
def list_messages(self) -> List[Message]:
messages: List[Message] = []
for line in self.file_manager.read_lines():
try:
messages.append(MessageLineParser.parse(line))
except ValueError:
continue
return messages
def list_messages_by_topic(self, topic_id: Link) -> List[Message]:
topic_value = self._resolve_id(topic_id, TABLE_TOPICS)
messages = [message for message in self.list_messages() if message.topic_id.value == topic_value]
return sorted(messages, key=lambda message: message.order_in_topic)
def create_message(self, payload: MessageCreate) -> Message:
forbidden_word = self._check_forbidden_words(payload.content)
if forbidden_word:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"{FORBIDDEN_WORDS_MESSAGE}: '{forbidden_word}'",
)
self.get_topic(payload.topic_id)
self.get_participant(payload.participant_id)
next_order = len(self.list_messages_by_topic(payload.topic_id)) + 1
message = Message(
id=uuid4(),
topic_id=payload.topic_id,
participant_id=payload.participant_id,
content=payload.content,
order_in_topic=next_order,
created_at=datetime.utcnow(),
)
serialized = MessageLineParser.serialize(message)
self.file_manager.append_line(serialized)
return message
def get_message(self, message_id: UUID) -> Message:
for message in self.list_messages():
if message.id == message_id:
return message
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=MESSAGE_NOT_FOUND_MESSAGE,
)
def export_message(self, message_id: UUID) -> str:
message = self.get_message(message_id)
return MessageLineParser.serialize(message)
|