38 lines
1.2 KiB
Python
38 lines
1.2 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from pathlib import Path
|
|
|
|
from vk_parser_app.config import settings
|
|
from vk_parser_app.vk_api import VKAPIClient, post_vk_url
|
|
|
|
|
|
async def main() -> None:
|
|
if not settings.vk_access_token:
|
|
raise RuntimeError("VK_ACCESS_TOKEN is empty")
|
|
if not settings.vk_storage_group_id:
|
|
raise RuntimeError("VK_STORAGE_GROUP_ID is empty")
|
|
|
|
async with VKAPIClient() as client:
|
|
attachments: list[str] = []
|
|
sample = Path("sample_photo.jpg")
|
|
if sample.exists():
|
|
saved = await client.upload_wall_photo_bytes(
|
|
abs(settings.vk_storage_group_id),
|
|
sample.read_bytes(),
|
|
filename="sample_photo.jpg",
|
|
)
|
|
attachments.append(f"photo{saved['owner_id']}_{saved['id']}")
|
|
|
|
post_id = await client.create_wall_post(
|
|
owner_id=settings.vk_storage_owner_id,
|
|
message="VK storage spike: test raw copy post",
|
|
attachments=attachments,
|
|
from_group=True,
|
|
)
|
|
print(post_vk_url(settings.vk_storage_owner_id, post_id))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|