|
|
from datetime import datetime |
|
|
from typing import List, Optional |
|
|
from uuid import UUID, uuid4 |
|
|
|
|
|
from fastapi import HTTPException, status |
|
|
|
|
|
from app.api.messages.schemas import MessageCreate |
|
|
from app.api.messages.services import MessageService |
|
|
from app.api.topics.parser import TopicLineParser |
|
|
from app.api.topics.schemas import AddMessageRequest, Topic, TopicCreate, TopicMessage |
|
|
from app.core.config import ( |
|
|
FORBIDDEN_WORDS, |
|
|
FORBIDDEN_WORDS_MESSAGE, |
|
|
TABLE_TOPICS, |
|
|
TOPIC_NOT_FOUND_MESSAGE, |
|
|
) |
|
|
from app.core.file_manager import FileManager |
|
|
from app.core.link import Link, TopicLink, ParticipantLink |
|
|
|
|
|
|
|
|
class TopicService: |
|
|
def __init__(self, file_manager: FileManager, message_service: MessageService | None = None) -> None: |
|
|
self.file_manager = file_manager |
|
|
self.message_service = message_service |
|
|
|
|
|
@staticmethod |
|
|
def _resolve_topic_id(value: UUID | Link) -> UUID: |
|
|
if isinstance(value, UUID): |
|
|
return value |
|
|
return value.resolve({TABLE_TOPICS: lambda link_value: link_value}) |
|
|
|
|
|
def list_topics(self) -> List[Topic]: |
|
|
topics: List[Topic] = [] |
|
|
for line in self.file_manager.read_lines(): |
|
|
try: |
|
|
topics.append(TopicLineParser.parse(line)) |
|
|
except ValueError: |
|
|
continue |
|
|
return topics |
|
|
|
|
|
def get_topic(self, topic_id: UUID | Link) -> Topic: |
|
|
resolved_id = self._resolve_topic_id(topic_id) |
|
|
for topic in self.list_topics(): |
|
|
if topic.id == resolved_id: |
|
|
return topic |
|
|
raise HTTPException( |
|
|
status_code=status.HTTP_404_NOT_FOUND, |
|
|
detail=TOPIC_NOT_FOUND_MESSAGE, |
|
|
) |
|
|
|
|
|
def create_topic(self, payload: TopicCreate) -> Topic: |
|
|
topic_messages = ( |
|
|
[] |
|
|
if self.message_service |
|
|
else [ |
|
|
TopicMessage( |
|
|
participant_id=message.participant_id, |
|
|
content=message.content, |
|
|
) |
|
|
for message in payload.messages |
|
|
] |
|
|
) |
|
|
topic = Topic( |
|
|
id=uuid4(), |
|
|
title=payload.title, |
|
|
description=payload.description, |
|
|
created_at=datetime.utcnow(), |
|
|
participants=payload.participants, |
|
|
messages=topic_messages, |
|
|
) |
|
|
serialized = TopicLineParser.serialize(topic) |
|
|
self.file_manager.append_line(serialized) |
|
|
if self.message_service: |
|
|
for message in payload.messages: |
|
|
self.message_service.create_message( |
|
|
MessageCreate( |
|
|
topic_id=TopicLink(value=topic.id), |
|
|
participant_id=message.participant_id, |
|
|
content=message.content, |
|
|
) |
|
|
) |
|
|
return topic |
|
|
|
|
|
@staticmethod |
|
|
def check_forbidden_words(content: str) -> Optional[str]: |
|
|
content_lower = content.lower() |
|
|
for word in FORBIDDEN_WORDS: |
|
|
if word in content_lower: |
|
|
return word |
|
|
return None |
|
|
|
|
|
def add_message(self, topic_id: UUID | Link, payload: AddMessageRequest) -> Topic: |
|
|
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}'", |
|
|
) |
|
|
|
|
|
resolved_id = self._resolve_topic_id(topic_id) |
|
|
|
|
|
if self.message_service: |
|
|
self.message_service.create_message( |
|
|
MessageCreate( |
|
|
topic_id=TopicLink(value=resolved_id), |
|
|
participant_id=ParticipantLink(value=payload.participant_id.value), |
|
|
content=payload.content, |
|
|
) |
|
|
) |
|
|
return self.get_topic(resolved_id) |
|
|
|
|
|
topics = self.list_topics() |
|
|
topic_index = None |
|
|
for i, topic in enumerate(topics): |
|
|
if topic.id == resolved_id: |
|
|
topic_index = i |
|
|
break |
|
|
|
|
|
if topic_index is None: |
|
|
raise HTTPException( |
|
|
status_code=status.HTTP_404_NOT_FOUND, |
|
|
detail=TOPIC_NOT_FOUND_MESSAGE, |
|
|
) |
|
|
|
|
|
new_message = TopicMessage( |
|
|
participant_id=payload.participant_id, |
|
|
content=payload.content, |
|
|
) |
|
|
topics[topic_index].messages.append(new_message) |
|
|
|
|
|
lines = [TopicLineParser.serialize(t) for t in topics] |
|
|
self.file_manager.write_lines(lines) |
|
|
|
|
|
return topics[topic_index] |
|
|
|
|
|
def export_topic(self, topic_id: UUID | Link) -> str: |
|
|
topic = self.get_topic(topic_id) |
|
|
return TopicLineParser.serialize(topic) |
|
|
|
