benya
e1e2f8f0e8
- вынес token/chat/update логику в services - вынес диалог и текст инструкции в ui - добавил и обновил тесты для нового слоя
47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
from urllib.parse import urlparse
|
|
|
|
|
|
def resolve_user_ids(vk_call_with_retry, vk_api, links):
|
|
resolved_ids = []
|
|
failed_links = []
|
|
for link in links:
|
|
try:
|
|
path = urlparse(link).path
|
|
screen_name = path.split("/")[-1] if path else ""
|
|
if not screen_name and len(path.split("/")) > 1:
|
|
screen_name = path.split("/")[-2]
|
|
if not screen_name:
|
|
failed_links.append((link, None))
|
|
continue
|
|
resolved_object = vk_call_with_retry(vk_api.utils.resolveScreenName, screen_name=screen_name)
|
|
if resolved_object and resolved_object.get("type") == "user":
|
|
resolved_ids.append(resolved_object["object_id"])
|
|
else:
|
|
failed_links.append((link, None))
|
|
except Exception as e:
|
|
failed_links.append((link, e))
|
|
return resolved_ids, failed_links
|
|
|
|
|
|
def load_chat_conversations(vk_call_with_retry, vk_api):
|
|
conversations = []
|
|
start_from = None
|
|
seen_start_tokens = set()
|
|
while True:
|
|
params = {"count": 200, "filter": "all"}
|
|
if start_from:
|
|
if start_from in seen_start_tokens:
|
|
break
|
|
params["start_from"] = start_from
|
|
seen_start_tokens.add(start_from)
|
|
response = vk_call_with_retry(vk_api.messages.getConversations, **params)
|
|
page_items = response.get("items", [])
|
|
if not page_items:
|
|
break
|
|
conversations.extend(page_items)
|
|
start_from = response.get("next_from")
|
|
if not start_from:
|
|
break
|
|
return conversations
|
|
|