mirror of
https://gitverse.ru/kpa39l/md2vk.git
synced 2026-09-29 09:55:04 +00:00
estorozhenko
108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
"""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" |