66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
import mimetypes
|
|
|
|
from sqlalchemy import Integer, String
|
|
|
|
from zvk.plugins.vk.api import VKApi
|
|
from zvk.util.db import DBBase, NNColumn, Database
|
|
from zvk.util.network import Network
|
|
from zvk.util.zlogging import logger
|
|
|
|
|
|
class CachedUpload(DBBase):
|
|
id = NNColumn(Integer, primary_key=True)
|
|
local_path = NNColumn(String)
|
|
vk_object = NNColumn(String)
|
|
|
|
|
|
def is_local_file_an_image(local_path):
|
|
mimetype, _ = mimetypes.guess_type(local_path)
|
|
return mimetype in ['image/jpeg', 'image/png', 'image/gif']
|
|
|
|
|
|
async def upload_image(db: Database,
|
|
net: Network,
|
|
api: VKApi,
|
|
local_path: str,
|
|
use_cache: bool = True) -> str:
|
|
logger.info(f'Uploading local image {local_path}')
|
|
|
|
if use_cache:
|
|
with db as session:
|
|
cached_upload = session \
|
|
.query(CachedUpload) \
|
|
.filter_by(local_path=local_path) \
|
|
.first()
|
|
if cached_upload:
|
|
logger.info(f'Found {local_path} in upload cache: {cached_upload.vk_object}')
|
|
return cached_upload.vk_object
|
|
|
|
if not is_local_file_an_image(local_path):
|
|
raise ValueError(f'{local_path} does not look like an image!')
|
|
|
|
response = await api.photos.getMessagesUploadServer()
|
|
upload_url = response['upload_url']
|
|
|
|
with open(local_path, 'rb') as f:
|
|
files = {'photo': f}
|
|
|
|
_, response = await net.post_json(upload_url, data=files)
|
|
|
|
if 'server' not in response or 'photo' not in response or 'hash' not in response:
|
|
raise RuntimeError(f'Could not upload {local_path}')
|
|
|
|
logger.info(f'Uploaded {local_path} as {response}')
|
|
|
|
response = await api.photos.saveMessagesPhoto(**response)
|
|
if len(response) < 1:
|
|
raise RuntimeError(f'Could not save image {local_path}')
|
|
|
|
vk_object = f'photo{response[0]["owner_id"]}_{response[0]["id"]}'
|
|
|
|
if use_cache:
|
|
with db as session:
|
|
session.add(CachedUpload(local_path=local_path, vk_object=vk_object))
|
|
|
|
return vk_object
|