### Authenticate and Make API Calls with VkApi Source: https://context7.com/python273/vk_api/llms.txt Use VkApi for authentication with login/password or a direct token. Handles captcha and 2FA. Get the high-level API interface to make calls and chain methods. Supports user and group tokens. ```python import vk_api # Authentication with login/password vk_session = vk_api.VkApi('+71234567890', 'mypassword') try: vk_session.auth(token_only=True) # token_only=True for API-only access except vk_api.AuthError as e: print(f"Authentication failed: {e}") exit(1) # Get high-level API interface vk = vk_session.get_api() # Make API calls using method chaining user_info = vk.users.get(user_ids='1,2,3', fields='photo_100,city') print(user_info) # Output: [{'id': 1, 'first_name': 'Pavel', 'last_name': 'Durov', ...}] # Post to wall result = vk.wall.post(message='Hello from vk_api!') print(f"Post ID: {result['post_id']}") # Alternative: Direct token usage (for bots or service tokens) vk_session = vk_api.VkApi(token='your_access_token') vk = vk_session.get_api() # For group tokens with higher rate limits (20 req/sec) vk_group = vk_api.VkApiGroup(token='group_token') vk = vk_group.get_api() ``` -------------------------------- ### Manage VkStreaming Rules and Listen for Events Source: https://context7.com/python273/vk_api/llms.txt Demonstrates how to add, view, and delete streaming rules for keyword matching. It also shows how to listen for and process incoming events from the streaming API. Requires a service token for initialization. ```python import vk_api from vk_api.streaming import VkStreaming vk_session = vk_api.VkApi(token='service_token') # Requires service token streaming = VkStreaming(vk_session) # Manage streaming rules # Add rules (keyword matching) streaming.add_rule(value='python', tag='python_mentions') streaming.add_rule(value='vk_api -spam', tag='library_mentions') # Exclude 'spam' streaming.add_rule(value='python OR programming', tag='coding') # View current rules rules = streaming.get_rules() for rule in rules: print(f"Tag: {rule['tag']}, Value: {rule['value']}") # Delete specific rule streaming.delete_rule(tag='python_mentions') # Delete all rules streaming.delete_all_rules() # Listen for matching events for event in streaming.listen(): print(f"Event type: {event['event_type']}") print(f"Event URL: {event['event_url']}") print(f"Text: {event.get('text', 'N/A')}") print(f"Tags: {event['tags']}") print(f"Creation time: {event['creation_time']}") print("---") ``` -------------------------------- ### Fetch All Wall Posts with Pagination Source: https://context7.com/python273/vk_api/llms.txt Utilize VkTools.get_all to automatically handle pagination for fetching large datasets like all wall posts. Specify max_count for items per API call and optionally a limit and stop condition. ```python import vk_api from vk_api.tools import VkTools # Assuming vk_session is initialized # vk_session = vk_api.VkApi(token='your_token') # tools = VkTools(vk_session) # all_posts = tools.get_all( # method='wall.get', # max_count=100, # values={'owner_id': -1} # ) # print(f"Total posts: {all_posts['count']}") ``` ```python import vk_api from vk_api.tools import VkTools # Assuming vk_session is initialized # vk_session = vk_api.VkApi(token='your_token') # tools = VkTools(vk_session) # def should_stop(items): # return any(item.get('is_pinned') for item in items) # posts = tools.get_all( # method='wall.get', # max_count=100, # values={'owner_id': -1}, # limit=500, # stop_fn=should_stop # ) ``` -------------------------------- ### Create Standard and Inline Keyboards with VkKeyboard Source: https://context7.com/python273/vk_api/llms.txt Generate interactive keyboards for bot messages using VkKeyboard. Supports text, callback, link, location, and VK Pay buttons. Use `one_time=True` for keyboards that disappear after use. ```python import vk_api from vk_api.keyboard import VkKeyboard, VkKeyboardColor from vk_api.utils import get_random_id vk_session = vk_api.VkApi(token='group_token') vk = vk_session.get_api() # Create standard keyboard keyboard = VkKeyboard(one_time=True) # Disappears after button press # Add text buttons with colors keyboard.add_button('Option A', color=VkKeyboardColor.PRIMARY) # Blue keyboard.add_button('Option B', color=VkKeyboardColor.POSITIVE) # Green keyboard.add_line() # New row keyboard.add_button('Cancel', color=VkKeyboardColor.NEGATIVE) # Red keyboard.add_button('Info', color=VkKeyboardColor.SECONDARY) # White # Send message with keyboard vk.messages.send( peer_id=123456789, random_id=get_random_id(), message='Choose an option:', keyboard=keyboard.get_keyboard() ) # Create inline keyboard (appears under message) inline_kb = VkKeyboard(inline=True) inline_kb.add_button('Like', payload={'action': 'like'}) inline_kb.add_callback_button('Subscribe', color=VkKeyboardColor.POSITIVE, payload={'action': 'subscribe'}) inline_kb.add_line() inline_kb.add_openlink_button('Visit Website', link='https://example.com') vk.messages.send( peer_id=123456789, random_id=get_random_id(), message='Interactive content:', keyboard=inline_kb.get_keyboard() ) # Special button types special_kb = VkKeyboard() special_kb.add_location_button() # Request user location special_kb.add_line() special_kb.add_vkpay_button(hash="action=transfer-to-group&group_id=123&aid=456") special_kb.add_line() special_kb.add_vkapps_button(app_id=6979558, owner_id=-181108510, label="Open App", hash="start") # Remove keyboard empty_keyboard = VkKeyboard.get_empty_keyboard() ``` -------------------------------- ### General API Error Handling with vk_api Source: https://context7.com/python273/vk_api/llms.txt Illustrates how to catch and handle various vk_api exceptions, including general API errors, HTTP errors, and authentication issues. It shows how to retry requests upon specific error codes like 'Too many requests' or handle captchas and authentication failures. ```python import vk_api from vk_api.exceptions import ( VkApiError, # Base exception ApiError, # API method errors ApiHttpError, # HTTP-level errors AuthError, # Authentication errors LoginRequired, # Login not provided PasswordRequired, # Password not provided BadPassword, # Incorrect password AccountBlocked, # Account is blocked TwoFactorError, # 2FA issues SecurityCheck, # Phone verification required Captcha, # Captcha required AccessDenied, # Permission denied ) vk_session = vk_api.VkApi(token='your_token') vk = vk_session.get_api() try: result = vk.wall.post(message='Test') except vk_api.ApiError as e: print(f"API Error [{e.code}]: {e.error['error_msg']}") # Retry the request if e.code == 6: # Too many requests import time time.sleep(1) result = e.try_method() except vk_api.Captcha as captcha: # Handle captcha result = captcha.try_again(input("Enter captcha: ")) except vk_api.AuthError as e: print(f"Authentication failed: {e}") except vk_api.ApiHttpError as e: print(f"HTTP Error: {e.response.status_code}") # Retry result = e.try_method() ``` -------------------------------- ### Access User Audio Tracks with VkAudio Source: https://context7.com/python273/vk_api/llms.txt Use the VkAudio class for unofficial access to VK audio content, requiring login and password authentication. This method allows fetching user audio tracks, albums, and searching for music. ```python import vk_api from vk_api.audio import VkAudio # vk_session = vk_api.VkApi('+71234567890', 'password') # vk_session.auth() # audio = VkAudio(vk_session) # tracks = audio.get(owner_id=123456789) # for track in tracks: # print(f"{track['artist']} - {track['title']}") # print(f"URL: {track['url']}") # print(f"Duration: {track['duration']}s") ``` ```python import vk_api from vk_api.audio import VkAudio # Assuming vk_session is authenticated # audio = VkAudio(vk_session) # albums = audio.get_albums(owner_id=123456789) # for album in albums: # print(f"Album: {album['title']} by {album['artist']}") # album_tracks = audio.get( # owner_id=album['owner_id'], # album_id=album['id'], # access_hash=album['access_hash'] # ) ``` ```python import vk_api from vk_api.audio import VkAudio # Assuming vk_session is authenticated # audio = VkAudio(vk_session) # results = list(audio.search('artist name song title', count=50)) # for track in results: # print(f"{track['artist']} - {track['title']}") ``` ```python import vk_api from vk_api.audio import VkAudio # Assuming vk_session is authenticated # audio = VkAudio(vk_session) # user_results = audio.search_user(owner_id=123456789, q='search query') ``` ```python import vk_api from vk_api.audio import VkAudio # Assuming vk_session is authenticated # audio = VkAudio(vk_session) # track = audio.get_audio_by_id(owner_id=123456789, audio_id=456789) ``` ```python import vk_api from vk_api.audio import VkAudio # Assuming vk_session is authenticated # audio = VkAudio(vk_session) # post_tracks = list(audio.get_post_audio(owner_id=-123456, post_id=789)) ``` ```python import vk_api from vk_api.audio import VkAudio # Assuming vk_session is authenticated # audio = VkAudio(vk_session) # for track in audio.get_iter(owner_id=123456789): # print(f"{track['artist']} - {track['title']}") ``` -------------------------------- ### Upload Files with VkUpload Source: https://context7.com/python273/vk_api/llms.txt VkUpload handles various file uploads including photos, documents, videos, and audio messages. It automates the multi-step upload process. Specify album_id for albums or peer_id for messages. ```python import vk_api from vk_api.upload import VkUpload vk_session = vk_api.VkApi(token='your_token') upload = VkUpload(vk_session) # Upload photo to album photo = upload.photo( '/path/to/image.jpg', album_id=123456789, group_id=74030368, # Optional: for group albums caption='Photo description' ) print(f"Photo URL: https://vk.ru/photo{photo[0]['owner_id']}_{photo[0]['id']}") # Upload photo for messages photo_msg = upload.photo_messages('/path/to/image.jpg', peer_id=123456) attachment = f"photo{photo_msg[0]['owner_id']}_{photo_msg[0]['id']}" # Upload photo to wall wall_photo = upload.photo_wall(['/path/to/photo1.jpg', '/path/to/photo2.jpg']) # Upload document doc = upload.document('/path/to/file.pdf', title='My Document', tags='important') print(f"Doc: {doc['doc']['owner_id']}_{doc['doc']['id']}") # Upload video (from file or URL) video = upload.video( video_file='/path/to/video.mp4', name='My Video', description='Video description', is_private=True ) # Or from YouTube/other sources: video = upload.video(link='https://youtube.com/watch?v=xxx', name='YouTube Video') # Upload voice message voice = upload.audio_message('/path/to/voice.ogg', peer_id=123456) # Upload story (photo or video) story = upload.story('/path/to/story.jpg', file_type='photo', add_to_news=True) ``` -------------------------------- ### Memory-Efficient Iteration for Large Datasets Source: https://context7.com/python273/vk_api/llms.txt Use VkTools.get_all_iter for memory-efficient iteration over large datasets, processing items one by one without loading everything into memory. Suitable for large collections where processing each item is sufficient. ```python import vk_api from vk_api.tools import VkTools # Assuming vk_session is initialized # vk_session = vk_api.VkApi(token='your_token') # tools = VkTools(vk_session) # for post in tools.get_all_iter('wall.get', max_count=100, values={'owner_id': -1}): # print(f"Post: {post['text'][:50]}...") ``` -------------------------------- ### Batch API Requests with VkRequestsPool Source: https://context7.com/python273/vk_api/llms.txt Optimize performance by executing up to 25 API requests in a single network call using VkRequestsPool. Use the context manager for recommended usage. Check the `.ok` attribute to verify request success before accessing `.result`. ```python import vk_api vk_session = vk_api.VkApi(token='your_token') # Using context manager (recommended) with vk_api.VkRequestsPool(vk_session) as pool: # Queue multiple requests friends = pool.method('friends.get') status = pool.method('status.get') wall = pool.method('wall.get', {'count': 10}) # Requests with potential errors invalid_request = pool.method('wall.getById') # Missing required params # Requests execute when exiting the context # Access results after the context print(f"Friends: {friends.result}") print(f"Status: {status.result}") # Check for errors before accessing result if invalid_request.ok: print(invalid_request.result) else: print(f"Error: {invalid_request.error}") # Batch requests for multiple users user_data = {} with vk_api.VkRequestsPool(vk_session) as pool: for user_id in [1, 2, 3, 183433824]: user_data[user_id] = pool.method('users.get', { 'user_ids': user_id, 'fields': 'photo_100,city,verified' }) for uid, response in user_data.items(): if response.ok: print(f"User {uid}: {response.result}") ``` -------------------------------- ### Handle User Events with VkLongPoll Source: https://context7.com/python273/vk_api/llms.txt Use VkLongPoll for real-time event streaming for user accounts, such as new messages, typing indicators, and online status changes. Requires authentication with user credentials. ```python import vk_api from vk_api.longpoll import VkLongPoll, VkEventType vk_session = vk_api.VkApi('+71234567890', 'password') vk_session.auth(token_only=True) longpoll = VkLongPoll(vk_session) for event in longpoll.listen(): if event.type == VkEventType.MESSAGE_NEW: if event.to_me: print(f"New message from {event.user_id}: {event.text}") # Check message source if event.from_user: print(f"Direct message from user {event.user_id}") elif event.from_chat: print(f"Message in chat {event.chat_id} from {event.user_id}") elif event.from_group: print(f"Message from group {event.group_id}") elif event.type == VkEventType.USER_ONLINE: print(f"User {event.user_id} is online (platform: {event.platform})") elif event.type == VkEventType.USER_OFFLINE: print(f"User {event.user_id} went offline") elif event.type == VkEventType.USER_TYPING: print(f"User {event.user_id} is typing...") ``` -------------------------------- ### Custom Captcha and 2FA Handling Source: https://context7.com/python273/vk_api/llms.txt Provides custom handlers for captcha challenges and two-factor authentication during the vk_api session initialization. The captcha handler can fetch the image and prompt for user input, while the auth handler requests a 2FA code. ```python import vk_api def captcha_handler(captcha): """Custom captcha handler""" # Get captcha image URL url = captcha.get_url() print(f"Captcha URL: {url}") # Or get image bytes directly image_data = captcha.get_image() with open('captcha.jpg', 'wb') as f: f.write(image_data) # Get user input or use captcha solving service key = input("Enter captcha text: ") # Retry the failed request with captcha answer return captcha.try_again(key) def auth_handler(): """Two-factor authentication handler""" code = input("Enter 2FA code: ") remember_device = True return code, remember_device # Use custom handlers vk_session = vk_api.VkApi( '+71234567890', 'password', captcha_handler=captcha_handler, auth_handler=auth_handler ) try: vk_session.auth() except vk_api.Captcha as captcha: # If no handler provided, exception is raised captcha.try_again(input("Captcha: ")) except vk_api.AuthError as e: print(f"Auth error: {e}") ``` -------------------------------- ### Execute Custom VKScript Functions Source: https://context7.com/python273/vk_api/llms.txt Call custom VKFunction objects with a vk_api.VkApi instance and necessary arguments. The library handles the execution on VK servers and returns the result. ```python import vk_api # Assuming vk_session and VkFunction objects (vk_add, vk_get_wall_ids, etc.) are defined # vk_session = vk_api.VkApi(token='your_token') # vk = vk_session.get_api() # result = vk_add(vk, 10, 20) # print(result) # post_ids = vk_get_wall_ids(vk, {'domain': 'durov', 'count': 100}) # print(f"Post IDs: {post_ids}") # texts = vk_filter_items(vk, 'wall.get', {'domain': 'durov', 'count': 10}, 'text') # print(f"Post texts: {texts}") # mutual = vk_get_mutual_friends(vk, 1) # print(f"Mutual friends data: {mutual}") ``` -------------------------------- ### Simplified API Call with Single Parameter Pool Source: https://context7.com/python273/vk_api/llms.txt Use vk_request_one_param_pool for methods that accept a single parameter and require multiple values. This is useful for batch operations on a specific parameter. ```python import vk_api # Assuming vk_session is already initialized # friends, errors = vk_api.vk_request_one_param_pool( # vk_session, # method='friends.get', # key='user_id', # values=[1, 2, 3, 183433824], # default_values={'fields': 'photo_100', 'count': 5} # ) # print(f"Results: {friends}") # print(f"Errors: {errors}") ``` -------------------------------- ### Slow Iteration for Debugging with VkTools Source: https://context7.com/python273/vk_api/llms.txt Employ VkTools.get_all_slow_iter for scenarios requiring non-execute method fetching, such as debugging or specific use cases. This method iterates through items without using the execute method. ```python import vk_api from vk_api.tools import VkTools # Assuming vk_session is initialized # vk_session = vk_api.VkApi(token='your_token') # tools = VkTools(vk_session) # for item in tools.get_all_slow_iter('groups.get', max_count=1000, # values={'extended': 1}, key='items'): # print(f"Group: {item['name']}") ``` -------------------------------- ### Define Reusable VKScript Functions with VkFunction Source: https://context7.com/python273/vk_api/llms.txt Create custom VKScript functions using VkFunction for complex or repetitive server-side operations. This reduces network overhead by executing logic directly on VK servers. Use clean_args to prevent specific arguments from being JSON-encoded. ```python import vk_api from vk_api.execute import VkFunction # Define reusable VKScript functions vk_add = VkFunction( args=('x', 'y'), code='return %(x)s + %(y)s;' ) vk_get_wall_ids = VkFunction( args=('values',), code='return API.wall.get(%(values)s)("items")@.id;' ) vk_filter_items = VkFunction( args=('method', 'values', 'key'), clean_args=('method', 'key'), code='return API.%(method)s(%(values)s)("items")@.%(key)s;' ) vk_get_mutual_friends = VkFunction( args=('user_id',), code=''' var friends = API.friends.get({"user_id": %(user_id)s}); var mutual = []; var i = 0; while (i < friends.items.length && i < 25) { var m = API.friends.getMutual({"target_uid": friends.items[i]}); if (m.length > 0) { mutual.push({"user_id": friends.items[i], "count": m.length}); } i = i + 1; } return mutual; ''' ) ``` -------------------------------- ### Handle Bot Events with VkBotLongPoll Source: https://context7.com/python273/vk_api/llms.txt Use VkBotLongPoll to stream bot-specific events like new messages, callbacks, and group activity. Ensure you have a group access token and group ID. ```python import vk_api from vk_api.bot_longpoll import VkBotLongPoll, VkBotEventType vk_session = vk_api.VkApi(token='group_access_token') longpoll = VkBotLongPoll(vk_session, group_id=123456789) for event in longpoll.listen(): if event.type == VkBotEventType.MESSAGE_NEW: # Access message data message = event.message print(f"New message: {message.text}") print(f"From: {message.from_id}, Peer: {message.peer_id}") # Check message source if event.from_user: print("Direct message") elif event.from_chat: print(f"Chat message, chat_id: {event.chat_id}") # Access client capabilities if event.client_info: print(f"Client supports keyboards: {event.client_info.keyboard}") elif event.type == VkBotEventType.MESSAGE_EVENT: # Callback button pressed print(f"Callback: {event.object.payload}") elif event.type == VkBotEventType.GROUP_JOIN: print(f"User {event.object.user_id} joined the group") elif event.type == VkBotEventType.GROUP_LEAVE: print(f"User {event.object.user_id} left the group") elif event.type == VkBotEventType.WALL_POST_NEW: print(f"New wall post: {event.object.text}") ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.