Chat/group infos

Author: dyakovri

migrations/versions/4a3336b87036_group_customization.py

@@ -0,0 +1,33 @@
+"""Group customization
+
+Revision ID: 4a3336b87036
+Revises: 27dda7e6236a
+Create Date: 2024-04-27 18:42:55.905145
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = '4a3336b87036'
+down_revision = '27dda7e6236a'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column('group', sa.Column('name', sa.String(), nullable=True))
+    op.add_column('group', sa.Column('description', sa.String(), nullable=True))
+    op.add_column('group', sa.Column('invite_link', sa.String(), nullable=True))
+    op.add_column('group', sa.Column('hidden', sa.Boolean(), nullable=True))
+    op.execute('UPDATE "group" SET hidden = false;')
+    op.alter_column('group', 'hidden', nullable=False)
+
+
+def downgrade():
+    op.drop_column('group', 'hidden')
+    op.drop_column('group', 'invite_link')
+    op.drop_column('group', 'description')
+    op.drop_column('group', 'name')

social/exceptions.py

@@ -9,3 +9,12 @@ def __init__(self, user_id: int, secret_key: str, *args) -> None:
         self.user_id = user_id
         self.secret_key = secret_key
         super().__init__(*args)
+
+
+class GroupNotFound(SocialApiError):
+    """Запрошенная группа не найдена"""
+
+    def __init__(self, user_id: int | None, group_id: int, *args) -> None:
+        self.user_id = user_id
+        self.group_id = group_id
+        super().__init__(*args)

social/handlers_telegram/base.py

@@ -49,6 +49,7 @@ async def send_help(update: Update, context: CustomContext):
 
 
 async def validate_group(update: Update, context: CustomContext):
+    """Если получено сообщение команды /validate, то за группой закрепляется владелец"""
     logger.info("Validation message received")
     with db():
         approve_telegram_group(update)

social/handlers_vk/base.py

@@ -32,4 +32,5 @@ def process_event(event: dict):
     object=lambda i: i.get("message", {}).get("text", "").startswith("/validate"),
 )
 def validate_group(event: dict):
+    """Если получено сообщение команды /validate, то за группой закрепляется владелец"""
     approve_vk_chat(event)

social/models/group.py

@@ -11,6 +11,11 @@ class Group(Base):
     type: Mapped[str]
     owner_id: Mapped[int | None]
 
+    name: Mapped[str | None]
+    description: Mapped[str | None]
+    invite_link: Mapped[str | None]
+    hidden: Mapped[bool] = mapped_column(default=True)
+
     is_deleted: Mapped[bool] = mapped_column(default=False)
     last_active_ts: Mapped[datetime] = mapped_column(default=lambda: datetime.now(UTC))
 

social/routes/exceptions.py

@@ -1,7 +1,7 @@
 from fastapi import Request
 from fastapi.responses import JSONResponse
 
-from social.exceptions import GroupRequestNotFound
+from social.exceptions import GroupNotFound, GroupRequestNotFound
 
 from .base import app
 
@@ -17,3 +17,15 @@ def group_request_not_found(request: Request, exc: GroupRequestNotFound) -> JSON
             'secret_key': exc.secret_key,
         },
     )
+
+
+@app.exception_handler(GroupNotFound)
+def group_not_found(request: Request, exc: GroupNotFound) -> JSONResponse:
+    return JSONResponse(
+        status_code=404,
+        content={
+            'details': 'Group not found',
+            'ru': 'Группа не найдена',
+            'group_id': exc.group_id,
+        },
+    )

social/routes/group.py

@@ -6,9 +6,12 @@
 from fastapi_sqlalchemy import db
 from pydantic import BaseModel
 
-from social.exceptions import GroupRequestNotFound
+from social.exceptions import GroupNotFound, GroupRequestNotFound
 from social.models.create_group_request import CreateGroupRequest
+from social.models.group import Group
 from social.settings import get_settings
+from social.utils.telegram_groups import update_tg_chat
+from social.utils.vk_groups import update_vk_chat
 
 
 router = APIRouter(prefix="/group", tags=['User defined groups'])
@@ -23,6 +26,21 @@ class GroupRequestGet(BaseModel):
 
 class GroupGet(BaseModel):
     id: int
+    owner_id: int | None = None
+    name: str | None = None
+    description: str | None = None
+    invite_link: str | None = None
+
+
+class GroupGetMany(BaseModel):
+    items: list[GroupGet]
+
+
+class GroupPatch(BaseModel):
+    update_from_source: bool | None = False
+    name: str | None = None
+    description: str | None = None
+    invite_link: str | None = None
 
 
 @router.post('')
@@ -35,11 +53,15 @@ def create_group_request(
     return obj
 
 
-@router.get('')
+@router.get('/validation')
 def validate_group_request(
     secret_key: str,
     user: dict[str] = Depends(UnionAuth(["social.group.create"])),
 ) -> GroupGet | GroupRequestGet:
+    """Получение состояния валидации группы по коду валидации
+
+    Трубуются права: `social.group.create`
+    """
     obj = (
         db.session.query(CreateGroupRequest)
         .where(CreateGroupRequest.secret_key == secret_key, CreateGroupRequest.owner_id == user.get("id"))
@@ -52,3 +74,61 @@ def validate_group_request(
         return GroupGet.model_validate(obj.mapped_group, from_attributes=True)
 
     return GroupRequestGet.model_validate(obj, from_attributes=True)
+
+
+@router.get('')
+def get_all_groups(
+    my: bool = True,
+    user: dict[str] = Depends(UnionAuth(allow_none=True, auto_error=False)),
+) -> GroupGetMany:
+    """Получение списка групп
+
+    Трубуются права:
+    - Для получения списка своих групп права не требуются (`my=True`)
+    - `social.group.read` для чтения списка всех групп, подключенных к приложению
+    """
+    if not user:
+        # Возвращаем список видимых всем групп
+        return {"items": db.session.query(Group).where(Group.hidden == False).all()}
+
+    if user and my:
+        # Возвращаем только свои группы
+        return {"items": db.session.query(Group).where(Group.owner_id == user.get("id")).all()}
+
+    # Если у пользователя есть права на просмотр всех групп – показываем все неудаленные группы
+    for scope in user.get("session_scopes", []):
+        if scope.get("name") == "social.group.read":
+            return {"items": db.session.query(Group).where(Group.is_deleted == False).all()}
+
+    # Возвращаем пустный список если не прошли ни по одному условию
+    logger.debug("User %s has no rights to get groups", user.get("id") if user else None)
+    return {"items": []}
+
+
+@router.patch('/{group_id}')
+def update_group_info(
+    group_id: int,
+    patch_info: GroupPatch,
+    user: dict[str] = Depends(UnionAuth()),
+):
+    group = db.session.get(Group, group_id)
+    if group.owner_id != user.get("id"):
+        raise GroupNotFound(user_id=user.get("id"), group_id=group_id)
+
+    # Пытаемся получить данные из источника (получение название чата ВК/Telegram)
+    if patch_info.update_from_source:
+        if group.type == "vk_chat":
+            update_vk_chat(group)
+        elif group.type == "tg_chat" or group.type == "tg_channel":
+            update_tg_chat(group)
+
+    # Ручное обновление данных
+    if patch_info.name:
+        group.name = patch_info.name
+    if patch_info.description:
+        group.description = patch_info.description
+    if patch_info.invite_link:
+        group.invite_link = patch_info.invite_link
+
+    db.session.commit()
+    return GroupGet.model_validate(group, from_attributes=True)

social/utils/telegram_groups.py

@@ -1,13 +1,23 @@
 import logging
 from datetime import UTC, datetime
 
+import requests
 from fastapi_sqlalchemy import db
 from telegram import Update
 
 from social.models import CreateGroupRequest, TelegramChannel, TelegramChat
+from social.settings import get_settings
 
 
 logger = logging.getLogger(__name__)
+settings = get_settings()
+
+
+def get_chat_info(id: int) -> dict:
+    return requests.post(
+        f'https://api.telegram.org/bot{settings.TELEGRAM_BOT_TOKEN}/getChat',
+        json={'chat_id': id},
+    ).json()
 
 
 def create_telegram_group(update: Update):
@@ -47,3 +57,10 @@ def approve_telegram_group(update: Update):
     group.owner_id = request.owner_id
     db.session.commit()
     logger.info("Telegram group %d validated (secret=%s)", group.id, text)
+
+
+def update_tg_chat(group: TelegramChat):
+    chat_info = get_chat_info(group.chat_id)
+    group.name = chat_info.get("title")
+    group.description = chat_info.get("description")
+    group.invite_link = chat_info.get("invite_link")

social/utils/vk_groups.py

@@ -12,7 +12,7 @@
 settings = get_settings()
 
 
-def get_chat_name(peer_id):
+def get_chat_info(peer_id):
     """Получить название чата ВК"""
     conversation = requests.post(
         "https://api.vk.com/method/messages.getConversationsById",
@@ -24,7 +24,25 @@ def get_chat_name(peer_id):
         },
     )
     try:
-        return conversation["response"]["items"][0]["chat_settings"]["title"]
+        return conversation["response"]["items"][0]["chat_settings"]
+    except Exception as exc:
+        logger.exception(exc)
+        return None
+
+
+def get_chat_invite_link(peer_id):
+    """Получить название чата ВК"""
+    conversation = requests.post(
+        "https://api.vk.com/method/messages.getInviteLink",
+        json={
+            "peer_ids": peer_id,
+            "group_id": settings.VK_BOT_GROUP_ID,
+            "access_token": settings.VK_BOT_TOKEN,
+            "v": 5.199,
+        },
+    )
+    try:
+        return conversation["response"]["link"]
     except Exception as exc:
         logger.exception(exc)
         return None
@@ -57,10 +75,19 @@ def approve_vk_chat(request_data: dict[str]):
     group = create_vk_chat(request_data)
     text = request_data.get("object", {}).get("message", {}).get("text", "").removeprefix("/validate").strip()
     if not text or not group or group.owner_id is not None:
-        logger.error("Telegram group not validated (secret=%s, group=%s)", text, group)
+        logger.error("VK group not validated (secret=%s, group=%s)", text, group)
         return
     request = db.session.query(CreateGroupRequest).where(CreateGroupRequest.secret_key == text).one_or_none()
     request.mapped_group_id = group.id
     group.owner_id = request.owner_id
     db.session.commit()
     logger.info("VK group %d validated (secret=%s)", group.id, text)
+
+
+def update_vk_chat(group: VkChat):
+    """Обновляет информацию о группе ВК"""
+    chat_info = get_chat_info(group.peer_id)
+    group.name = chat_info.get("title")
+    group.description = chat_info.get("description")
+    group.invite_link = get_chat_invite_link(group.peer_id)
+    return group