Untitled
4ae4d
plain_text
05/06/2026 8:49 AM
54.6 KB
8
Indexable
f=services/bot/core/services/base.py
f=services/bot/core/services/event.py
f=services/bot/core/services/file.py
f=services/bot/core/services/__init__.py
f=services/bot/core/services/__pycache__
f=services/bot/core/services/registry
f=services/bot/core/services/report.py
f=services/bot/core/services/requests.py
f=services/bot/core/services/tag.py
f=services/bot/core/services/terbank.py
--- services/bot/core/services/base.py ---
from core.services.registry import BaseServiceRegistry
class BaseService: # legacy
def __init__(self, registry: BaseServiceRegistry): # legacy
self.registry = registry
--- services/bot/core/services/event.py ---
from datetime import datetime, timezone
from typing import Any
from requests import HTTPError
from core.clients.event import EventClient
from core.schemas import EventCardSchema, ParticipationSchema
from core.services.base import BaseService
from core.services.registry import BaseServiceRegistry
from core.utils.logger import logger
class EventService(BaseService): # legacy
"""
сервис мероприятий на стороне бота.
на данный момент:
- ходит в backend через EventClient
- умеет получать список мероприятий по городу
- умеет получать одно мероприятие по id
- создает мероприятие через клиент используя мастера создания
"""
def __init__(self, registry: BaseServiceRegistry, client: EventClient): # legacy
# DI может вызвать клиент без аргументов.
super().__init__(registry)
self.client = client
def get_event_cards_page(
self,
*,
requester_messenger_id: int,
page: int = 1,
limit: int = 10,
title_str: str | None = None,
tag_ids: list[str] | None = None,
mine: bool = False,
) -> tuple[list[EventCardSchema], int, int]: # меняем тип возвращаемого списка
resp = (
self.client.get_my_event_cards_page(
requester_messenger_id=requester_messenger_id,
page=page,
limit=limit,
title_str=title_str,
tag_ids=tag_ids,
)
if mine
else self.client.get_event_cards_page(
requester_messenger_id=requester_messenger_id,
page=page,
limit=limit,
title_str=title_str,
tag_ids=tag_ids,
)
)
match resp.status_code:
case 200:
pass
case 400:
raise HTTPError("Error while getting event cards list: invalid parameters")
case 401:
raise HTTPError("Error while getting event cards list: unauthorized")
case 500:
raise HTTPError("Error while getting event cards list: internal server error")
case _:
raise HTTPError(
f"Error while getting event cards list: unexpected status {resp.status_code}"
)
body = resp.json() or {}
items = body.get("event_cards") or []
page_meta = body.get("page_meta") or {}
result: list[EventCardSchema] = []
for item in items:
if not isinstance(item, dict):
continue
result.append(self._map_event_card(item))
current_page = int(page_meta.get("current_page") or page)
total_pages = int(page_meta.get("total_pages") or current_page)
return result, current_page, total_pages
def sign_up(self, event_id: str, messenger_id: int) -> bool: # legacy
"""messenger_id = peer.id (как на бэкенде: user.get_by_messenger_id)"""
resp = self.client.sign_up(event_id=event_id, messenger_id=messenger_id)
return resp.status_code == 200
def get_event_by_id(
self,
event_id: str,
requester_messenger_id: int,
*,
mine: bool = False,
) -> EventCardSchema | None:
logger.debug(
"[EventService] get_event_by_id: event_id=%s requester=%s mine=%s",
event_id,
requester_messenger_id,
mine,
)
resp = (
self.client.get_my_event_info(
event_id=event_id,
requester_messenger_id=requester_messenger_id,
)
if mine
else self.client.get_event_info(
event_id=event_id,
requester_messenger_id=requester_messenger_id,
)
)
match resp.status_code:
case 200:
pass
case 400:
raise HTTPError("Error while getting event card: invalid parameters")
case 401:
raise HTTPError("Error while getting event card: unauthorized")
case 404:
return None
case 500:
raise HTTPError("Error while getting event card: internal server error")
case _:
raise HTTPError(
f"Error while getting event card: unexpected status {resp.status_code}"
)
logger.debug(
"[EventService] get_event_by_id: body=%s",
resp.json(),
)
return self._map_event_card(resp.json() or {})
def sign_out_by_event(self, event_id: str, messenger_id: int) -> bool:
resp = self.client.sign_out(event_id=event_id, messenger_id=messenger_id)
return resp.status_code == 200
def enter_code_by_event(
self,
event_id: str,
messenger_id: int,
code: str,
) -> ParticipationSchema | None:
resp = self.client.confirm_participation(
event_id=event_id,
messenger_id=messenger_id,
code=code,
)
match resp.status_code:
case 200:
pass
case 400:
return None
case 401:
raise HTTPError("Error while confirming participation: unauthorized")
case 404:
return None
case 500:
raise HTTPError("Error while confirming participation: internal server error")
case _:
raise HTTPError(
f"Error while confirming participation: unexpected status {resp.status_code}"
)
return self.get_participation_status(
event_id=event_id,
requester_messenger_id=messenger_id,
)
def get_participation_status(self, event_id: str, requester_messenger_id: int):
resp = self.client.get_event_info(event_id, requester_messenger_id)
card = self._map_event_card(resp.json() or {})
return card.participation
def create_event(
self,
*,
title: str,
description: str,
date_obj: datetime,
gosp_id: str,
project_id: str,
verification_code: str,
hours: int,
requester_messenger_id: int,
tags: list[str],
):
"""
создает мероприятие через backend и возвращает dict из backend'a
- ожидает, что все данные уже провалидированы и разобраны
- метод не мапит ответ backend'a в EventSchema и не делает никакой обработки, а возвращает ровно тот json, который вернул backend (dect)
- может бросить Exception, если status_code не 201/200
"""
date_iso = date_obj.isoformat()
resp = self.client.create_event(
title=title,
description=description,
date_time=date_iso,
gosp_id=gosp_id,
project_id=project_id,
verification_code=verification_code,
hours=hours,
requester_messenger_id=requester_messenger_id,
tags=tags,
)
if resp.status_code not in [200, 201]:
raise Exception(f"Error while creating event: {resp.status_code}\n{resp.text}")
return resp.json()
def create_from_wizard(self, data: dict[str, Any], peer) -> dict[str, Any]:
title = data.get("event_name", "")
description = data.get("event_description", "")
date_obj = data.get("event_date", "")
time_obj = data.get("event_time", "")
gosp_id = data.get("gosp_id", "")
project_id = data.get("project_id", "")
hours = data.get("hours", "")
verification_code = data.get("event_bonus_code", "")
# Теги: массив ID, сформированный в хендлере
tags: list[str] = data.get("event_app_tag_ids") or []
# === Формирование RFC3339-совместимой строки ===
event_dt = datetime.combine(date_obj, time_obj).replace(tzinfo=timezone.utc)
date_time = event_dt.isoformat().replace("+00:00", "Z")
resp = self.client.create_event(
title=title,
description=description,
date_time=date_time,
gosp_id=gosp_id,
project_id=project_id,
verification_code=verification_code,
hours=hours,
requester_messenger_id=peer.id,
tags=tags if tags else None,
)
match resp.status_code:
case 200:
pass
case 401:
raise HTTPError("Error while creating event: unauthorized")
case 500:
raise HTTPError("Error while creating event: internal server error")
case _:
raise HTTPError(f"Error while creating event: unexpected status {resp.status_code}")
return {"TYPE": resp.json().get("code")} # показывать карточку мероприятия
def _map_event_card(self, body: dict[str, Any]) -> EventCardSchema:
"""преобразует json event card в EventCardSchema"""
return EventCardSchema.from_dict(body)
--- services/bot/core/services/file.py ---
from __future__ import annotations
from concurrent.futures import TimeoutError
from core.config import bot
from core.utils.logger import logger
class FileService: # legacy
@staticmethod
def send_file( # legacy
peer,
*,
file_bytes,
filename: str,
uid: str,
text: str,
reply_mid=None,
interactive_media_groups=None,
) -> bool:
logger.info("FileService:")
try:
logger.info(f"sending file {uid}, name={filename}, size={len(file_bytes)} bytes")
task = (
bot.messaging.send_file(
peer=peer,
file=file_bytes,
name=filename,
text=text,
)
if interactive_media_groups is None
else bot.messaging.send_file(
peer=peer,
file=file_bytes,
name=filename,
text=text,
interactive_media_groups=interactive_media_groups,
)
)
result = task.wait()
if isinstance(result, Exception):
logger.error(f"Sending file failed. uid={uid}, name={filename}: {result!r}")
raise result
logger.info(f"file sent successfully: {result!r}")
return True
except TimeoutError as e:
logger.error(f"Timeout sending file {uid}, name={filename}: {e}")
return False
except Exception as e:
logger.error(f"Error sending file {uid}, name={filename}: {e}")
return False
--- services/bot/core/services/__init__.py ---
from .base import BaseService
from .event import EventService
from .file import FileService
from .report import ReportService
from .requests import RequestService
from .tag import TagService
from .terbank import TerbankService
from .user import UserService
__all__ = [
"BaseService",
"EventService",
"FileService",
"ReportService",
"RequestService",
"TagService",
"TerbankService",
"UserService",
]
Error: 'services/bot/core/services/__pycache__' is not a valid file
Error: 'services/bot/core/services/registry' is not a valid file
--- services/bot/core/services/report.py ---
from __future__ import annotations
from urllib.parse import unquote
from requests import HTTPError
from core.clients.report import ReportClient
from core.services.base import BaseService
from core.services.registry import BaseServiceRegistry
class ReportService(BaseService): # legacy
def __init__(self, registry: BaseServiceRegistry, client: ReportClient): # legacy
super().__init__(registry)
self.client = client
def create_report(
self,
*,
event_id: str,
requester_messenger_id: int,
description: str,
beneficiaries: int,
) -> None:
resp = self.client.create_report(
event_id=event_id,
requester_messenger_id=requester_messenger_id,
description=description,
beneficiaries=beneficiaries,
)
match resp.status_code:
case 200:
pass
case 401:
raise HTTPError("Error while creating report: unauthorized")
case 403:
raise HTTPError("Error while creating report: forbidden")
case 404:
raise HTTPError("Error while creating report: event not found")
case 500:
raise HTTPError("Error while creating report: internal server error")
case _:
raise HTTPError(
f"Error while creating report: unexpected status {resp.status_code}"
)
def get_report(
self,
*,
event_id: str,
requester_messenger_id: int,
) -> tuple[bytes, str]:
resp = self.client.get_report(
event_id=event_id,
requester_messenger_id=requester_messenger_id,
)
match resp.status_code:
case 200:
pass
case 401:
raise HTTPError("Error while getting report: unauthorized")
case 403:
raise HTTPError("Error while getting report: forbidden")
case 404:
raise HTTPError("Error while getting report: event not found")
case 500:
raise HTTPError("Error while getting report: internal server error")
case _:
raise HTTPError(f"Error while getting report: unexpected status {resp.status_code}")
filename = "event_report.xlsx"
cd = resp.headers.get("Content-Disposition")
if cd and "filename=" in cd:
try:
raw = cd.split("filename=", 1)[1].strip().strip('"')
filename = unquote(raw) or filename
except Exception:
pass
return resp.content, filename
def report_exists( # legacy
self,
*,
event_id: str,
requester_messenger_id: int,
) -> bool:
resp = self.client.get_report(
event_id=event_id,
requester_messenger_id=requester_messenger_id,
)
if resp.status_code == 200:
return True
if resp.status_code in (401, 403, 404):
return False
self._raise_http_error("Error while checking report existance", resp)
return False
--- services/bot/core/services/requests.py ---
from core.clients.request import RequestClient
from core.schemas import RequestCardSchema, RequestCardsPageSchema
from core.services.base import BaseService
from core.services.registry import BaseServiceRegistry
from core.utils.logger import logger
class RequestService(BaseService):
"""Сервис для работы с заявками"""
def __init__(self, registry: BaseServiceRegistry, client: RequestClient):
super().__init__(registry)
self.client = client
def get_request_cards_page(
self,
*,
requester_messenger_id: int,
page: int = 1,
limit: int = 10,
status: str | None = None,
) -> RequestCardsPageSchema:
"""Получает страницу с заявками"""
resp = self.client.get_request_cards(
requester_messenger_id=requester_messenger_id,
page=page,
limit=limit,
status=status,
)
if resp.status_code != 200:
logger.error(f"[RequestService] get_request_cards_page failed: {resp.status_code}")
return RequestCardsPageSchema(request_cards=[], current_page=1, total_pages=1)
try:
return RequestCardsPageSchema.from_response(resp.json())
except Exception as e:
logger.error(f"[RequestService] Failed to parse response: {e}")
return RequestCardsPageSchema(request_cards=[], current_page=1, total_pages=1)
def get_request_by_id(
self, request_id: str, requester_messenger_id: int
) -> RequestCardSchema | None:
"""Получает заявку по ID"""
resp = self.client.get_request(request_id, requester_messenger_id)
if resp.status_code == 404:
return None
if resp.status_code != 200:
logger.error(f"[RequestService] get_request_by_id failed: {resp.status_code}")
return None
try:
return RequestCardSchema.from_dict(resp.json())
except Exception as e:
logger.error(f"[RequestService] Failed to parse request: {e}")
return None
def approve_request(self, request_id: str, requester_messenger_id: int) -> bool:
"""Одобряет заявку"""
resp = self.client.approve_request(request_id, requester_messenger_id)
return resp.status_code == 200
def reject_request(self, request_id: str, requester_messenger_id: int, reason: str) -> bool:
"""Отклоняет заявку с причиной"""
resp = self.client.reject_request(request_id, requester_messenger_id, reason)
return resp.status_code == 200
--- services/bot/core/services/tag.py ---
from requests import HTTPError
from core.clients import TagClient
from core.schemas import TagSchema
from core.services.base import BaseService
from core.services.registry import BaseServiceRegistry
from core.utils.logger import logger
class TagService(BaseService):
def __init__(self, registry: BaseServiceRegistry, client: TagClient):
super().__init__(registry)
self.client = client
def get_all_tags(self) -> list[TagSchema]:
resp = self.client.get_all_tags()
match resp.status_code:
case 200:
pass
case 401:
raise HTTPError("Error while getting tags list: unauthorized")
case 500:
raise HTTPError("Error while getting tags list: internal server error")
case _:
raise HTTPError(
f"Error while getting tags list: unexpected status {resp.status_code}"
)
return [TagSchema(**item) for item in resp.json()]
def sign_up(self, tag_id: str, messenger_id: int) -> bool:
resp = self.client.sign_up(tag_id, messenger_id)
if resp.status_code != 200:
try:
logger.error(
"[TagService]: sign_up failed tag_id=%s messenger_id=%s status=%s body=%s",
tag_id,
messenger_id,
resp.status_code,
resp.text,
)
except Exception as e:
logger.error("[TagService]: sign up failed")
logger.error("[TagService]: logger failed while logging an error: %s", e)
return resp.status_code == 200
def sign_out(self, tag_id: str, messenger_id: int) -> bool:
resp = self.client.sign_out(tag_id, messenger_id)
if resp.status_code != 200:
try:
logger.error(
"[TagService]: sign_out failed tag_id=%s messenger_id=%s status=%s body=%s",
tag_id,
messenger_id,
resp.status_code,
resp.text,
)
except Exception as e:
logger.error("[TagService]: sign up failed")
logger.error("[TagService]: logger failed while logging an error: %s", e)
return resp.status_code == 200
--- services/bot/core/services/terbank.py ---
from requests import HTTPError
from core.clients import TerbankClient
from core.schemas import GosbSchema, TerbankSchema
from core.services.base import BaseService
from core.services.registry import BaseServiceRegistry
class TerbankService(BaseService):
"""
сервис для заголовка "terbanks" бэкенда.
здесь обрабатывается всё что связано со структурой, включая ГОСБы
(и наверное потом модераторы)
"""
def __init__(self, registry: BaseServiceRegistry, client: TerbankClient):
super().__init__(registry)
self.client = client
def get_all_tbs(self, messenger_id: int) -> list[TerbankSchema]:
resp = self.client.get_all_tbs(messenger_id)
match resp.status_code:
case 200:
pass
case 401:
raise HTTPError("Error while getting TBs list: unauthorized")
case 500:
raise HTTPError("Error while getting TBs list: internal server error")
case _:
raise HTTPError(
f"Error while getting TBs list: unexpected status {resp.status_code}"
)
return [TerbankSchema(**item) for item in resp.json()]
def get_gosbs_for_tb(self, tb_id: str, messenger_id: int) -> list[GosbSchema]:
resp = self.client.get_gosbs_for_tb(tb_id, messenger_id)
match resp.status_code:
case 200:
pass
case 401:
raise HTTPError("Error while getting GOSBs list: unauthorized")
case 404:
raise HTTPError("Error while getting GOSBs list: terbank not found")
case 500:
raise HTTPError("Error while getting GOSBs list: internal server error")
case _:
raise HTTPError(
f"Error while getting GOSBs list: unexpected status {resp.status_code}"
)
return [GosbSchema(**item) for item in resp.json()]
--- services/bot/core/services/user.py ---
from requests import HTTPError
from core.clients.user import UserClient
from core.schemas import TagSchema, UserSchema
from core.services.base import BaseService
from core.services.registry import BaseServiceRegistry
from core.utils.logger import logger
class UserService(BaseService): # legacy
def __init__(
self, registry: BaseServiceRegistry, messenger_id: int, client: UserClient
): # legacy
super().__init__(registry)
self.messenger_id = messenger_id
self.client = client
def get(self) -> UserSchema | None: # legacy
resp = self.client.get_by_messenger_id(messenger_id=self.messenger_id)
if resp.status_code == 200:
body = resp.json() or {}
payload = {
**(body.get("volunteer") or {}),
"tags": body.get("tags") or [],
}
return self._map_user(payload)
if resp.status_code == 404:
return None
raise Exception(f"Error while getting user: {resp.status_code}.\n{resp.text}")
def ensure_user_exists(self, name: str, messenger_id) -> None:
logger.debug("[UserService]: ensure_user_exists is called")
self.client.ensure_user_exists(name, messenger_id)
def get_user_by_user_data(self, messenger_id: int) -> UserSchema | None:
logger.debug("[UserService]: get_user_by_user_data is called")
resp = self.client.get_by_messenger_id(messenger_id)
match resp.status_code:
case 200:
pass
case 401:
raise HTTPError("Error while getting user: unauthorized")
case 404:
return None
case 500:
raise HTTPError("Error while getting user: internal server error")
case _:
raise HTTPError(f"Error while getting user: unexpected status {resp.status_code}")
resp_json = resp.json() or {}
payload = {
**(resp_json.get("volunteer") or {}),
"tags": resp_json.get("tags") or [],
}
return UserSchema(**payload)
def _map_user(self, body: dict) -> UserSchema: # legacy
tags_data = body.get("tags") or []
tags = []
for tag_data in tags_data:
if isinstance(tag_data, dict):
try:
tag = TagSchema(
id=str(tag_data.get("id", "")),
title=str(tag_data.get("title", "")),
description=str(tag_data.get("description", "")),
depricated=bool(tag_data.get("depricated", False)),
created_at=str(tag_data.get("created_at") or ""),
updated_at=str(tag_data.get("updated_at") or ""),
)
tags.append(tag)
except Exception as e:
logger.error(f"[UserService] failed to map tag: {e}")
return UserSchema(
id=str(body.get("id") or ""),
leader=bool(body.get("leader") or False),
admin=bool(body.get("admin") or False),
messenger_id=int(body.get("messenger_id") or 0),
name=str(body.get("full_name") or body.get("name") or ""),
hours=int(body.get("hours") or 0),
created_at=str(body.get("created_at") or ""),
updated_at=str(body.get("updated_at") or ""),
tags=tags,
)
def func_or_none(func, arg): # legacy
if arg is not None:
return func(arg)
return None
f=services/bot/core/schemas/event.py
f=services/bot/core/schemas/gosb.py
f=services/bot/core/schemas/__init__.py
f=services/bot/core/schemas/participation.py
f=services/bot/core/schemas/__pycache__
f=services/bot/core/schemas/request.py
f=services/bot/core/schemas/tag.py
f=services/bot/core/schemas/terbank.py
--- services/bot/core/schemas/event.py ---
from dataclasses import field
from datetime import datetime
from typing import Any
from pydantic import BaseModel
from core.schemas.gosb import GosbSchema
from core.schemas.participation import ParticipationSchema
from core.schemas.tag import TagSchema
from core.schemas.terbank import TerbankSchema
from core.utils.date_time import parse_datetime
class EventCardSchema(BaseModel):
"""Схема карточки мероприятия из списка"""
event: "EventSchema"
tags: list[TagSchema] = field(default_factory=list)
gosb: GosbSchema | None = None
terbank: TerbankSchema | None = None
participation: ParticipationSchema | None = None
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "EventCardSchema":
event_data = data.get("event") or {}
tags_data = data.get("tags") or []
gosb = GosbSchema.from_dict(data.get("gosb"))
terbank = TerbankSchema.from_dict(data.get("terbank"))
participation = ParticipationSchema.from_dict(data.get("participation"))
event = EventSchema(
id=str(event_data.get("id", "")),
title=str(event_data.get("title", "")),
description=str(event_data.get("description", "")),
date_time=cls._parse_datetime(event_data.get("date_time")),
creator_id=str(event_data.get("creator_id", "")),
gosp_id=str(event_data.get("gosp_id", "")),
project_id=str(event_data.get("project_id", "")),
verification_code=(
event_data.get("verify_code") # TODO remove legacy
or event_data.get("verification_code")
),
archived_at=cls._parse_datetime(event_data.get("archived_at"))
if event_data.get("archived_at")
else None,
deleted=bool(event_data.get("deleted", False)),
active=bool(event_data.get("active", True)),
created_at=cls._parse_datetime(event_data.get("created_at"))
if event_data.get("created_at")
else None,
updated_at=cls._parse_datetime(event_data.get("updated_at"))
if event_data.get("updated_at")
else None,
)
tags = []
for tag_data in tags_data:
tag = TagSchema.from_dict(tag_data)
tags.append(tag)
return cls(
event=event,
tags=tags,
gosb=gosb,
terbank=terbank,
participation=participation,
)
@staticmethod
def _parse_datetime(value: Any) -> str:
return parse_datetime(value)
class EventSchema(BaseModel):
"""Полная схема мероприятия"""
id: str
title: str
description: str
date_time: datetime
creator_id: str
gosp_id: str
project_id: str
verification_code: str = ""
archived_at: datetime | None = None
deleted: bool = False
active: bool = True
created_at: datetime | None = None
updated_at: datetime | None = None
class EventCardsPageSchema(BaseModel):
"""Схема страницы с карточками мероприятий"""
event_cards: list[EventCardSchema] # добавляем поле для карточек
current_page: int
total_pages: int
@classmethod
def from_response(cls, response_data: dict[str, Any]) -> "EventCardsPageSchema":
"""Создаёт схему из ответа бэкенда"""
cards_data = response_data.get("event_cards") or []
page_meta = response_data.get("page_meta") or {}
event_cards = []
for card_data in cards_data:
card = EventCardSchema.from_dict(card_data)
event_cards.append(card)
return cls(
event_cards=event_cards,
current_page=int(page_meta.get("current_page", 1)),
total_pages=int(page_meta.get("total_pages", 1)),
)
@property
def has_prev(self) -> bool:
return self.current_page > 1
@property
def has_next(self) -> bool:
return self.current_page < self.total_pages
--- services/bot/core/schemas/gosb.py ---
from datetime import datetime
from typing import Any
from pydantic import BaseModel
from core.schemas.terbank import TerbankSchema
class GosbSchema(BaseModel):
"""Схема головного отделения Сбербанка (ГОСБ)"""
id: str
name: str
description: str
terbank_id: str
depricated: bool = False
created_at: datetime | None = None
updated_at: datetime | None = None
terbank: TerbankSchema | None = None
@classmethod
def from_dict(
cls, data: dict[str, Any] | None, include_terbank: bool = True
) -> "GosbSchema | None":
if not data:
return None
terbank = None
if include_terbank and "terbank" in data:
terbank = TerbankSchema.from_dict(data.get("terbank"))
elif include_terbank and "terbank_id" in data:
pass
return cls(
id=str(data.get("id", "")),
name=str(data.get("name", "")),
description=str(data.get("description", "")),
terbank_id=str(data.get("terbank_id", "")),
depricated=bool(data.get("depricated", False)),
created_at=cls._parse_datetime(data.get("created_at")),
updated_at=cls._parse_datetime(data.get("updated_at")),
terbank=terbank,
)
@staticmethod
def _parse_datetime(value: Any) -> datetime | None:
if value is None:
return None
if isinstance(value, datetime):
return value
if isinstance(value, str):
value = value.replace("Z", "+00:00")
try:
return datetime.fromisoformat(value)
except Exception:
return None
return None
def __str__(self) -> str:
return self.name
--- services/bot/core/schemas/__init__.py ---
from .event import EventCardSchema, EventCardsPageSchema, EventSchema
from .gosb import GosbSchema
from .participation import ParticipationSchema
from .request import RequestCardSchema, RequestCardsPageSchema, RequestSchema
from .tag import TagSchema
from .terbank import TerbankSchema
from .user import UserSchema
__all__ = [
"EventCardSchema",
"EventCardsPageSchema",
"EventSchema",
"GosbSchema",
"ParticipationSchema",
"RequestCardSchema",
"RequestCardsPageSchema",
"RequestSchema",
"TagSchema",
"TerbankSchema",
"UserSchema",
]
--- services/bot/core/schemas/participation.py ---
from datetime import datetime
from typing import Any, Optional
from pydantic import BaseModel
from core.utils.date_time import parse_datetime
class ParticipationSchema(BaseModel):
"""Схема участия пользователя в мероприятии"""
messenger_id: str
event_id: str
state: str
created_at: datetime
updated_at: datetime
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> Optional["ParticipationSchema"]:
if not data:
return None
return cls(
messenger_id=str(data.get("messenger_id", "")),
event_id=str(data.get("event_id", "")),
state=str(data.get("state", "")),
created_at=cls._parse_datetime(data.get("created_at")),
updated_at=cls._parse_datetime(data.get("updated_at")),
)
@staticmethod
def _parse_datetime(value: Any) -> str:
return parse_datetime(value)
Error: 'services/bot/core/schemas/__pycache__' is not a valid file
--- services/bot/core/schemas/request.py ---
from datetime import datetime
from typing import Any
from pydantic import BaseModel
from core.schemas.event import EventCardSchema
from core.utils.date_time import parse_datetime
class RequestSchema(BaseModel):
"""Схема заявки"""
id: str
messenger_id: str
request_type: str
reason: str
source_id: str
status: str
reject_reason: str
reviewed_by: str
reviewed_at: datetime | None
created_at: datetime
updated_at: datetime
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "RequestSchema":
reviewed_at = data.get("reviewed_at")
if reviewed_at and reviewed_at.startswith("0001-01-01"):
reviewed_at = None
return cls(
id=str(data.get("id", "")),
messenger_id=str(data.get("messenger_id", "")),
request_type=str(data.get("request_type", "")),
reason=str(data.get("reason", "")),
source_id=str(data.get("source_id", "")),
status=str(data.get("status", "")),
reject_reason=str(data.get("reject_reason", "")),
reviewed_by=str(data.get("reviewed_by", "")),
reviewed_at=parse_datetime(reviewed_at) if reviewed_at else None,
created_at=parse_datetime(data.get("created_at")),
updated_at=parse_datetime(data.get("updated_at")),
)
class RequestCardSchema(BaseModel):
"""Схема карточки заявки"""
request: RequestSchema
resource: EventCardSchema # | ProjectCardSchema
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "RequestCardSchema":
request_data = data.get("request") or {}
resource_data = data.get("resource") or {}
request = RequestSchema.from_dict(request_data)
# TODO: Сделать обработку карточки проекта здесь
resource = EventCardSchema.from_dict(resource_data)
return cls(
request=request,
resource=resource,
)
class RequestCardsPageSchema(BaseModel):
"""Схема страницы с карточками заявок"""
request_cards: list[RequestCardSchema]
current_page: int
total_pages: int
@classmethod
def from_response(cls, response_data: dict[str, Any]) -> "RequestCardsPageSchema":
"""Создаёт схему из ответа бэкенда"""
cards_data = response_data.get("request_cards") or []
page_meta = response_data.get("page_meta") or {}
request_cards = []
for card_data in cards_data:
card = RequestCardSchema.from_dict(card_data)
request_cards.append(card)
return cls(
request_cards=request_cards,
current_page=int(page_meta.get("current_page", 1)),
total_pages=int(page_meta.get("total_pages", 1)),
)
@property
def has_prev(self) -> bool:
return self.current_page > 1
@property
def has_next(self) -> bool:
return self.current_page < self.total_pages
--- services/bot/core/schemas/tag.py ---
from typing import Any
from pydantic import BaseModel
from core.utils.date_time import parse_datetime
class TagSchema(BaseModel):
id: str
title: str
description: str
depricated: bool # TODO rename to deprecated
created_at: str
updated_at: str
def __str__(self) -> str:
return self.title
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "TagSchema":
return cls(
id=str(data.get("id", "")),
title=str(data.get("title", "")),
description=str(data.get("description", "")),
depricated=bool(data.get("depricated", False)),
created_at=parse_datetime(data.get("created_at")),
updated_at=parse_datetime(data.get("updated_at")),
)
--- services/bot/core/schemas/terbank.py ---
from datetime import datetime
from typing import Any
from pydantic import BaseModel
class TerbankSchema(BaseModel):
"""Схема территориального банка"""
id: str
name: str
description: str
depricated: bool = False
created_at: datetime | None = None
updated_at: datetime | None = None
@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> "TerbankSchema | None":
if not data:
return None
return cls(
id=str(data.get("id", "")),
name=str(data.get("name", "")),
description=str(data.get("description", "")),
depricated=bool(data.get("depricated", False)),
created_at=cls._parse_datetime(data.get("created_at")),
updated_at=cls._parse_datetime(data.get("updated_at")),
)
@staticmethod
def _parse_datetime(value: Any) -> datetime | None:
if value is None:
return None
if isinstance(value, datetime):
return value
if isinstance(value, str):
value = value.replace("Z", "+00:00")
try:
return datetime.fromisoformat(value)
except Exception:
return None
return None
def __str__(self) -> str:
return self.name
--- services/bot/core/schemas/user.py ---
from pydantic import BaseModel
from core.schemas.tag import TagSchema
class UserSchema(BaseModel): # TODO переделать под новую ролевую модель
id: str
leader: bool
admin: bool
messenger_id: int
name: str
hours: int
created_at: str
updated_at: str
tags: list[TagSchema]
@staticmethod
def _display(value):
if isinstance(value, list):
return ", ".join(str(tag) for tag in value) if value != [] else "—"
return value
def __str__(self):
# Собираем дополнительные статусы
statuses = []
if self.admin:
statuses.append("👑 *Администратор*")
if self.leader:
statuses.append("⭐ *Лидер проекта*")
# Формируем строку со статусами
status_str = "\n".join(statuses) + "\n" if statuses else ""
# Используем str(tag) - он вернёт tag.title
tags_str = ", ".join(str(tag) for tag in self.tags) if self.tags else "—"
return (
f"😊 *Имя*: {self.name or '—'}\n"
f"🏆 *Часы*: {self.hours if self.hours is not None else '—'}\n"
f"{status_str}"
f"🎯 *Интересы*: {tags_str}\n"
)
def __repr__(self):
return (
f"UserSchema("
f"id = {self.id}, "
f"admin = {self.admin}, "
f"leader = {self.leader}, "
f"messenger_id = {self.messenger_id}, "
f"name = {self.name}, "
f"hours = {self.hours}, "
f"created_at = {self.created_at}, "
f"updated_at = {self.updated_at}, "
f"tags = [{', '.join(map(str, self.tags))}]"
f")"
)
Editor is loading...
Leave a Comment