Baseline md2vk: docs, audit log, docker 8420, openspec, deploy

app/vk_client.py

@@ -0,0 +1,108 @@
+"""VK API клиент — вызов wall.post с format_data."""
+
+from __future__ import annotations
+
+import json
+from typing import Optional
+
+import httpx
+
+from app.config import settings
+from app.converters.markdown_to_vk import FormatItem, format_data_json
+
+
+class VkApiError(Exception):
+    """Ошибка VK API. Содержит код ошибки и описание."""
+    def __init__(self, error_code: int, error_msg: str):
+        self.error_code = error_code
+        self.error_msg = error_msg
+        super().__init__(f"VK API error #{error_code}: {error_msg}")
+
+
+class VkClient:
+    """HTTP-клиент для VK API."""
+
+    BASE_URL = "https://api.vk.com/method"
+
+    def __init__(self, access_token: str):
+        self.access_token = access_token
+        self._client = httpx.AsyncClient(timeout=30.0)
+
+    async def close(self):
+        await self._client.aclose()
+
+    async def wall_post(
+        self,
+        message: str,
+        owner_id: Optional[int] = None,
+        from_group: bool = False,
+        friends_only: bool = False,
+        publish_date: Optional[int] = None,
+        attachments: Optional[str] = None,
+        signed: Optional[bool] = None,
+        format_data: Optional[list[FormatItem]] = None,
+    ) -> dict:
+        """Публикует запись на стене через wall.post.
+        
+        Возвращает ответ VK API с полями post_id, owner_id.
+        """
+        params = {
+            "access_token": self.access_token,
+            "v": settings.vk_api_version,
+            "message": message,
+        }
+
+        if owner_id is not None:
+            params["owner_id"] = owner_id
+        if from_group:
+            params["from_group"] = 1
+        if friends_only:
+            params["friends_only"] = 1
+        if publish_date is not None:
+            params["publish_date"] = publish_date
+        if attachments is not None:
+            params["attachments"] = attachments
+        if signed is not None:
+            params["signed"] = 1 if signed else 0
+        if format_data:
+            params["format_data"] = format_data_json(format_data)
+
+        response = await self._client.post(f"{self.BASE_URL}/wall.post", data=params)
+        data = response.json()
+
+        if "error" in data:
+            err = data["error"]
+            raise VkApiError(
+                error_code=err.get("error_code", 0),
+                error_msg=err.get("error_msg", "Unknown error"),
+            )
+
+        return data.get("response", {})
+
+    async def users_get(self, user_ids: str) -> list[dict]:
+        """Получает информацию о пользователе (для проверки токена/имени)."""
+        params = {
+            "access_token": self.access_token,
+            "v": settings.vk_api_version,
+            "user_ids": user_ids,
+        }
+        response = await self._client.post(f"{self.BASE_URL}/users.get", data=params)
+        data = response.json()
+        if "error" in data:
+            err = data["error"]
+            raise VkApiError(
+                error_code=err.get("error_code", 0),
+                error_msg=err.get("error_msg", "Unknown error"),
+            )
+        return data.get("response", [])
+
+    async def check_token(self) -> tuple[bool, str]:
+        """Проверяет валидность токена. Возвращает (is_valid, display_name)."""
+        try:
+            users = await self.users_get("")
+            if users:
+                name = f"{users[0].get('first_name', '')} {users[0].get('last_name', '')}".strip()
+                return True, name or "Unknown"
+            return False, "Token invalid"
+        except VkApiError:
+            return False, "Token invalid or expired"