'to_store', 'vote', 'more', 'book', 'order',

'enroll', 'fill', 'signup', 'buy', 'ticket',

'write', 'open', 'learn_more', 'view', 'go_to',

'contact', 'watch', 'play', 'install', 'read'

""" Загрузка файлов через API (https://vk.ru/dev/upload_files)

__slots__ = ('vk', 'http')

def __init__(self, vk):

if not isinstance(vk, (VkApi, VkApiMethod)):

raise TypeError(

'The arg should be VkApi or VkApiMethod instance'

)

self.vk = vk if isinstance(vk, VkApiMethod) else vk.get_api()

self.http = requests.Session()

self.http.headers.pop('user-agent')

def photo(self, photos, album_id,

latitude=None, longitude=None, caption=None, description=None,

group_id=None):

""" Загрузка изображений в альбом пользователя

values = {'album_id': album_id}

if group_id:

values['group_id'] = group_id

url = self.vk.photos.getUploadServer(**values)['upload_url']

with FilesOpener(photos) as photo_files:

response = self.http.post(url, files=photo_files).json()

if 'album_id' not in response:

response['album_id'] = response['aid']

response.update({

'latitude': latitude,

'longitude': longitude,

'caption': caption,

'description': description

})

values.update(response)

return self.vk.photos.save(**values)

def photo_messages(self, photos, peer_id=None):

""" Загрузка изображений в сообщения

url = self.vk.photos.getMessagesUploadServer(peer_id=peer_id)['upload_url']

with FilesOpener(photos) as photo_files:

response = self.http.post(url, files=photo_files)

return self.vk.photos.saveMessagesPhoto(**response.json())

def photo_group_widget(self, photo, image_type):

""" Загрузка изображений в коллекцию сообщества для виджетов приложений сообществ

url = self.vk.appWidgets.getGroupImageUploadServer(image_type=image_type)['upload_url']

with FilesOpener(photo, key_format='file') as photo_files:

response = self.http.post(url, files=photo_files)

return self.vk.appWidgets.saveGroupImage(**response.json())

def photo_profile(self, photo, owner_id=None, crop_x=None, crop_y=None,

crop_width=None):

""" Загрузка изображения профиля

values = {}

if owner_id:

values['owner_id'] = owner_id

crop_params = {}

if crop_x is not None and crop_y is not None and crop_width is not None:

crop_params['_square_crop'] = f'{crop_x},{crop_y},{crop_width}'

response = self.vk.photos.getOwnerPhotoUploadServer(**values)

url = response['upload_url']

with FilesOpener(photo, key_format='file') as photo_files:

response = self.http.post(

url,

data=crop_params,

files=photo_files

)

return self.vk.photos.saveOwnerPhoto(**response.json())

def photo_chat(self, photo, chat_id):

""" Загрузка и смена обложки в беседе

values = {'chat_id': chat_id}

url = self.vk.photos.getChatUploadServer(**values)['upload_url']

with FilesOpener(photo, key_format='file') as photo_file:

response = self.http.post(url, files=photo_file)

return self.vk.messages.setChatPhoto(

file=response.json()['response']

)

def photo_wall(self, photos, user_id=None, group_id=None, caption=None):

""" Загрузка изображений на стену пользователя или в группу

values = {}

if user_id:

values['user_id'] = user_id

elif group_id:

values['group_id'] = group_id

if caption:

values['caption'] = caption

response = self.vk.photos.getWallUploadServer(**values)

url = response['upload_url']

with FilesOpener(photos) as photos_files:

response = self.http.post(url, files=photos_files)

values.update(response.json())

return self.vk.photos.saveWallPhoto(**values)

def photo_market(self, photo, group_id, main_photo=False,

crop_x=None, crop_y=None, crop_width=None):

""" Загрузка изображений для товаров в магазине

if group_id < 0:

group_id = abs(group_id)

values = {

'main_photo': main_photo,

'group_id': group_id,

}

if crop_x is not None:

values['crop_x'] = crop_x

if crop_y is not None:

values['crop_y'] = crop_y

if crop_width is not None:

values['crop_width'] = crop_width

response = self.vk.photos.getMarketUploadServer(**values)

url = response['upload_url']

with FilesOpener(photo) as photos_files:

response = self.http.post(url, files=photos_files)

values.update(response.json())

return self.vk.photos.saveMarketPhoto(**values)

def photo_market_album(self, photo, group_id):

""" Загрузка фотографии для подборки товаров

if group_id < 0:

group_id = abs(group_id)

values = {

'group_id': group_id,

}

response = self.vk.photos.getMarketAlbumUploadServer(**values)

url = response['upload_url']

with FilesOpener(photo) as photos_files:

response = self.http.post(url, files=photos_files)

values.update(response.json())

return self.vk.photos.saveMarketAlbumPhoto(**values)

def audio(self, audio, artist, title):

""" Загрузка аудио

url = self.vk.audio.getUploadServer()['upload_url']

with FilesOpener(audio, key_format='file') as f:

response = self.http.post(url, files=f).json()

response.update({

'artist': artist,

'title': title

})

return self.vk.audio.save(**response)

def video(self, video_file=None, link=None, name=None, description=None,

is_private=None, wallpost=None, group_id=None,

album_id=None, privacy_view=None, privacy_comment=None,

no_comments=None, repeat=None):

""" Загрузка видео

if not link and not video_file:

raise ValueError('Either link or video_file param is required')

if link and video_file:

raise ValueError('Both params link and video_file aren\'t allowed')

values = {

'name': name,

'description': description,

'is_private': is_private,

'wallpost': wallpost,

'link': link,

'group_id': group_id,

'album_id': album_id,

'privacy_view': privacy_view,

'privacy_comment': privacy_comment,

'no_comments': no_comments,

'repeat': repeat

}

response = self.vk.video.save(**values)

url = response.pop('upload_url')

with FilesOpener(video_file or [], 'video_file') as f:

response.update(self.http.post(

url,

files=f or None

).json())

return response

def thumb_video(self, photo_path: str, owner_id: int, video_id: int):

"""

response = self.vk.video.getThumbUploadUrl(owner_id=owner_id)

upload_url = response.pop('upload_url')

with FilesOpener(photo_path, key_format='file') as file:

upload_response = self.http.post(upload_url, files=file)

response = self.vk.video.saveUploadedThumb(owner_id=owner_id,

thumb_json=upload_response.text,

video_id=video_id,

set_thumb=1)

return response

def document(self, doc, title=None, tags=None, group_id=None,

to_wall=False, message_peer_id=None, doc_type=None):

""" Загрузка документа

values = {

'group_id': group_id,

'peer_id': message_peer_id,

'type': doc_type

}

if to_wall:

method = self.vk.docs.getWallUploadServer

elif message_peer_id:

method = self.vk.docs.getMessagesUploadServer

else:

method = self.vk.docs.getUploadServer

url = method(**values)['upload_url']

with FilesOpener(doc, 'file') as files:

response = self.http.post(url, files=files).json()

response.update({

'title': title,

'tags': tags

})

return self.vk.docs.save(**response)

def document_wall(self, doc, title=None, tags=None, group_id=None):

""" Загрузка документа в папку Отправленные,

return self.document(doc, title, tags, group_id, to_wall=True)

def document_message(self, doc, title=None, tags=None, peer_id=None):

""" Загрузка документа для отправки личным сообщением.

return self.document(doc, title, tags, message_peer_id=peer_id)

def audio_message(self, audio, peer_id=None, group_id=None):

""" Загрузка аудио-сообщения.

return self.document(

audio,

doc_type='audio_message',

message_peer_id=peer_id,

group_id=group_id,

to_wall=group_id is not None

)

def graffiti(self, image, peer_id=None, group_id=None):

""" Загрузка граффити

return self.document(

image,

doc_type='graffiti',

message_peer_id=peer_id,

group_id=group_id,

to_wall=group_id is not None

)

def photo_cover(self, photo, group_id,

crop_x=None, crop_y=None,

crop_x2=None, crop_y2=None):

""" Загрузка изображения профиля

values = {

'group_id': group_id,

'crop_x': crop_x,

'crop_y': crop_y,

'crop_x2': crop_x2,

'crop_y2': crop_y2

}

url = self.vk.photos.getOwnerCoverPhotoUploadServer(**values)['upload_url']

with FilesOpener(photo, key_format='file') as photo_files:

response = self.http.post(url, files=photo_files)

return self.vk.photos.saveOwnerCoverPhoto(

**response.json()

)

def story(self, file, file_type, add_to_news=True, user_ids=None,

reply_to_story=None, link_text=None,

link_url=None, group_id=None):

""" Загрузка истории

if user_ids is None:

user_ids = []

if file_type == 'photo':

method = self.vk.stories.getPhotoUploadServer

elif file_type == 'video':

method = self.vk.stories.getVideoUploadServer

else:

raise ValueError('type should be either photo or video')

if not add_to_news and not user_ids:

raise ValueError(

'add_to_news and/or user_ids param is required'

)

if (link_text or link_url) and not group_id:

raise ValueError('Link params available only for communities')

if (not link_text) != (not link_url):

raise ValueError(

'Either both link_text and link_url or neither one are required'

)

if link_text and link_text not in STORY_ALLOWED_LINK_TEXTS:

raise ValueError('Invalid link_text')

if link_url and not link_url.startswith('https://vk.ru'):

raise ValueError(

'Only internal https://vk.ru links are allowed for link_url'

)

if link_url and len(link_url) > 2048:

raise ValueError('link_url is too long. Max length - 2048')

values = {

'add_to_news': int(add_to_news),

'user_ids': ','.join(map(str, user_ids)),

'reply_to_story': reply_to_story,

'link_text': link_text,

'link_url': link_url,

'group_id': group_id

}

url = method(**values)['upload_url']

with FilesOpener(file, key_format='file') as files:

def __init__(self, paths, key_format='file{}'):

if not isinstance(paths, list):

paths = [paths]

self.paths = paths

self.key_format = key_format

self.opened_files = []

def __enter__(self):

return self.open_files()

def __exit__(self, type, value, traceback):

self.close_files()

def open_files(self):

self.close_files()

files = []

for x, file in enumerate(self.paths):

if hasattr(file, 'read'):

f = file

filename = file.name if hasattr(file, 'name') else '.jpg'

else:

filename = file

f = open(filename, 'rb')

self.opened_files.append(f)

ext = filename.split('.')[-1]

files.append((self.key_format.format(x), (f'file{x}.{ext}', f)))

return files

def close_files(self):

for f in self.opened_files:

f.close()