benya/AnabasisChatRemove
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 18 Wiki Activity

feat(app): улучшения UX, логирование и безопасность

Browse Source
This commit is contained in:
Alex
2026-02-03 22:26:41 +03:00
parent ad24cb6fca
commit 4629037890
1 changed files with 377 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

434
main.py
View File

@@ -1,4 +1,7 @@
import sys
import base64
import ctypes
import shutil
from vk_api import VkApi
import json
import time
@@ -14,11 +17,86 @@ from PySide6.QtWebEngineCore import QWebEnginePage, QWebEngineProfile
from urllib.parse import urlparse, parse_qs, unquote
from vk_api.exceptions import VkApiError
from PySide6.QtCore import QStandardPaths
from ctypes import wintypes
# --- Управление токенами и настройками ---
APP_DATA_DIR = os.path.join(QStandardPaths.writableLocation(QStandardPaths.AppDataLocation), "AnabasisVKChatManager")
TOKEN_FILE = os.path.join(APP_DATA_DIR, "token.json")
WEB_ENGINE_CACHE_DIR = os.path.join(APP_DATA_DIR, "web_engine_cache")
CACHE_CLEANUP_MARKER = os.path.join(APP_DATA_DIR, "pending_cache_cleanup")
LOG_FILE = os.path.join(APP_DATA_DIR, "app.log")
LOG_MAX_BYTES = 1024 * 1024 # 1 MB
LOG_BACKUP_FILE = os.path.join(APP_DATA_DIR, "app.log.1")
class _DataBlob(ctypes.Structure):
_fields_ = [("cbData", wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_byte))]
_crypt32 = None
_kernel32 = None
if os.name == "nt":
_crypt32 = ctypes.WinDLL("crypt32", use_last_error=True)
_kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
_crypt32.CryptProtectData.argtypes = [
ctypes.POINTER(_DataBlob),
wintypes.LPCWSTR,
ctypes.POINTER(_DataBlob),
ctypes.c_void_p,
ctypes.c_void_p,
wintypes.DWORD,
ctypes.POINTER(_DataBlob),
]
_crypt32.CryptProtectData.restype = wintypes.BOOL
_crypt32.CryptUnprotectData.argtypes = [
ctypes.POINTER(_DataBlob),
ctypes.POINTER(wintypes.LPWSTR),
ctypes.POINTER(_DataBlob),
ctypes.c_void_p,
ctypes.c_void_p,
wintypes.DWORD,
ctypes.POINTER(_DataBlob),
]
_crypt32.CryptUnprotectData.restype = wintypes.BOOL
def _crypt_protect_data(data, description=""):
buffer = ctypes.create_string_buffer(data)
data_in = _DataBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_byte)))
data_out = _DataBlob()
if not _crypt32.CryptProtectData(ctypes.byref(data_in), description, None, None, None, 0, ctypes.byref(data_out)):
raise ctypes.WinError(ctypes.get_last_error())
try:
return ctypes.string_at(data_out.pbData, data_out.cbData)
finally:
_kernel32.LocalFree(data_out.pbData)
def _crypt_unprotect_data(data):
buffer = ctypes.create_string_buffer(data)
data_in = _DataBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_byte)))
data_out = _DataBlob()
if not _crypt32.CryptUnprotectData(ctypes.byref(data_in), None, None, None, None, 0, ctypes.byref(data_out)):
raise ctypes.WinError(ctypes.get_last_error())
try:
return ctypes.string_at(data_out.pbData, data_out.cbData)
finally:
_kernel32.LocalFree(data_out.pbData)
def _encrypt_token(token):
if os.name != "nt":
raise RuntimeError("DPAPI is available only on Windows.")
encrypted_bytes = _crypt_protect_data(token.encode("utf-8"))
return base64.b64encode(encrypted_bytes).decode("ascii")
def _decrypt_token(token_data):
if os.name != "nt":
raise RuntimeError("DPAPI is available only on Windows.")
encrypted_bytes = base64.b64decode(token_data.encode("ascii"))
decrypted_bytes = _crypt_unprotect_data(encrypted_bytes)
return decrypted_bytes.decode("utf-8")
def get_resource_path(relative_path):
""" Получает абсолютный путь к ресурсу, работает для скрипта и для .exe """
@@ -39,9 +117,19 @@ def save_token(token, expires_in=0):
# ИСПРАВЛЕНИЕ: если expires_in равно 0, то и время истечения ставим 0
expiration_time = (time.time() + expires_in) if expires_in > 0 else 0
stored_token = token
encrypted = False
if os.name == "nt":
try:
stored_token = _encrypt_token(token)
encrypted = True
except Exception as e:
print(f"Ошибка шифрования токена: {e}")
data = {
"token": token,
"expiration_time": expiration_time
"token": stored_token,
"expiration_time": expiration_time,
"encrypted": encrypted
}
try:
@@ -66,6 +154,15 @@ def load_token():
data = json.load(f)
token = data.get("token")
encrypted = data.get("encrypted", False)
if token and encrypted:
try:
token = _decrypt_token(token)
except Exception as e:
print(f"Ошибка расшифровки токена: {e}")
if os.path.exists(TOKEN_FILE):
os.remove(TOKEN_FILE)
return None, None
expiration_time = data.get("expiration_time")
# ИСПРАВЛЕНИЕ: токен валиден, если время истечения 0 ИЛИ оно больше текущего
@@ -253,12 +350,16 @@ class VkChatManager(QMainWindow):
self.vk_session = None
self.vk = None
self.user_ids_to_process = []
self._busy = False
self.suppress_resolve = False
self.resolve_timer = QTimer(self)
self.resolve_timer.setSingleShot(True)
self.resolve_timer.setInterval(750)
self.resolve_timer.timeout.connect(self.resolve_single_user_id_from_input)
self._cleanup_cache_if_needed()
self._ensure_log_dir()
self.init_ui()
self.load_saved_token_on_startup()
self.setup_token_timer()
@@ -356,6 +457,8 @@ class VkChatManager(QMainWindow):
self.set_ui_state(False)
def on_vk_url_input_changed(self, text):
if self.suppress_resolve:
return
if self.vk_url_input.hasFocus():
self.resolve_timer.start()
@@ -364,7 +467,7 @@ class VkChatManager(QMainWindow):
if dialog.exec():
links = dialog.get_links()
if links:
self.vk_url_input.clear()
self._set_vk_url_input_text("")
self._process_links_list(links)
else:
QMessageBox.information(self, "Информация", "Список ссылок пуст.")
@@ -387,33 +490,37 @@ class VkChatManager(QMainWindow):
resolved_ids = []
failed_links = []
self.status_label.setText("Статус: Определяю ID...")
QApplication.processEvents()
self._set_busy(True, "Статус: Определяю ID...")
try:
for link in links_list:
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]
for link in links_list:
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)
continue
if not screen_name:
resolved_object = self._vk_call_with_retry(self.vk.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)
except VkApiError as e:
self._log_error("resolveScreenName", e)
failed_links.append(f"{link} ({self._format_vk_error(e)})")
except Exception:
failed_links.append(link)
continue
resolved_object = self.vk.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)
except Exception:
failed_links.append(link)
finally:
self._set_busy(False)
self.user_ids_to_process = resolved_ids
status_message = f"Статус: Готово к работе с {len(resolved_ids)} пользовател(ем/ями)."
if len(links_list) > 1:
self.vk_url_input.setText(f"Загружено {len(resolved_ids)}/{len(links_list)} из списка")
self._set_vk_url_input_text(f"Загружено {len(resolved_ids)}/{len(links_list)} из списка")
if failed_links:
QMessageBox.warning(self, "Ошибка получения ID",
@@ -435,6 +542,11 @@ class VkChatManager(QMainWindow):
make_admin_action.triggered.connect(self.set_user_admin)
tools_menu.addAction(make_admin_action)
logout_action = QAction("Выйти и очистить", self)
logout_action.setStatusTip("Выйти, удалить токен и кэш")
logout_action.triggered.connect(self.logout_and_clear)
tools_menu.addAction(logout_action)
def create_chat_tab(self):
# This implementation correctly creates a scrollable area for chat lists.
tab_content_widget = QWidget()
@@ -511,7 +623,150 @@ class VkChatManager(QMainWindow):
if not authorized:
self.user_ids_to_process.clear()
self.vk_url_input.clear()
self._set_vk_url_input_text("")
self._clear_chat_tabs()
def _set_busy(self, busy, status_text=None):
if status_text:
self.status_label.setText(status_text)
if busy:
self._busy = True
QApplication.setOverrideCursor(Qt.WaitCursor)
for widget in [
self.refresh_chats_btn, self.select_all_btn, self.deselect_all_btn,
self.vk_url_input, self.multi_link_btn, self.visible_messages_checkbox,
self.remove_user_btn, self.add_user_btn
]:
widget.setEnabled(False)
else:
self._busy = False
QApplication.restoreOverrideCursor()
if self.token is None:
self.set_ui_state(False)
else:
self.set_ui_state(True)
def _ensure_log_dir(self):
os.makedirs(APP_DATA_DIR, exist_ok=True)
def _log_error(self, context, exc):
try:
os.makedirs(APP_DATA_DIR, exist_ok=True)
self._rotate_log_if_needed()
timestamp = QDateTime.currentDateTime().toString("yyyy-MM-dd HH:mm:ss")
message = self._format_vk_error(exc)
with open(LOG_FILE, "a", encoding="utf-8") as f:
f.write(f"[{timestamp}] {context}: {message}\n")
except Exception:
pass
def _rotate_log_if_needed(self):
try:
if not os.path.exists(LOG_FILE):
return
if os.path.getsize(LOG_FILE) < LOG_MAX_BYTES:
return
if os.path.exists(LOG_BACKUP_FILE):
os.remove(LOG_BACKUP_FILE)
os.replace(LOG_FILE, LOG_BACKUP_FILE)
except Exception:
pass
def _format_vk_error(self, exc):
error = getattr(exc, "error", None)
code = None
message = str(exc)
if isinstance(error, dict):
code = error.get("error_code")
message = error.get("error_msg") or message
hints = {
5: "Ошибка авторизации. Проверьте токен.",
6: "Слишком много запросов. Подождите и повторите.",
7: "Недостаточно прав.",
9: "Слишком много однотипных действий.",
10: "Внутренняя ошибка VK. Повторите позже.",
15: "Доступ запрещен.",
100: "Некорректный параметр запроса.",
113: "Неверный идентификатор пользователя.",
200: "Доступ к чату запрещен.",
}
if code in hints:
message = f"{message} ({hints[code]})"
if code is not None:
return f"[{code}] {message}"
return message
def _set_vk_url_input_text(self, text):
self.suppress_resolve = True
try:
self.vk_url_input.blockSignals(True)
self.vk_url_input.setText(text)
finally:
self.vk_url_input.blockSignals(False)
self.suppress_resolve = False
def logout_and_clear(self):
confirm = QMessageBox.question(
self,
"Подтверждение выхода",
"Вы уверены, что хотите выйти и удалить сохраненный токен и кэш?",
QMessageBox.Yes | QMessageBox.No
)
if confirm != QMessageBox.Yes:
return
self.token = None
self.token_expiration_time = None
self.vk_session = None
self.vk = None
self.user_ids_to_process.clear()
self._set_vk_url_input_text("")
self.token_input.clear()
self.token_timer_label.setText("Срок действия токена: Н/Д")
self.status_label.setText("Статус: не авторизован")
if self.token_countdown_timer.isActive():
self.token_countdown_timer.stop()
self._clear_chat_tabs()
self.set_ui_state(False)
try:
if os.path.exists(TOKEN_FILE):
os.remove(TOKEN_FILE)
except Exception as e:
print(f"Ошибка удаления токена: {e}")
try:
self._try_remove_web_cache()
except Exception as e:
print(f"Ошибка удаления кэша: {e}")
def _cleanup_cache_if_needed(self):
if os.path.exists(CACHE_CLEANUP_MARKER):
try:
self._try_remove_web_cache()
if os.path.exists(CACHE_CLEANUP_MARKER):
os.remove(CACHE_CLEANUP_MARKER)
except Exception as e:
print(f"Ошибка отложенной очистки кэша: {e}")
def _try_remove_web_cache(self):
if not os.path.exists(WEB_ENGINE_CACHE_DIR):
return
attempts = 5
last_error = None
for _ in range(attempts):
try:
shutil.rmtree(WEB_ENGINE_CACHE_DIR)
last_error = None
break
except Exception as e:
last_error = e
time.sleep(0.2)
if last_error:
os.makedirs(APP_DATA_DIR, exist_ok=True)
with open(CACHE_CLEANUP_MARKER, "w") as f:
f.write("pending")
raise last_error
def load_saved_token_on_startup(self):
loaded_token, expiration_time = load_token()
@@ -570,6 +825,26 @@ class VkChatManager(QMainWindow):
checkbox.deleteLater()
chk_list.clear()
def _vk_error_code(self, exc):
error = getattr(exc, "error", None)
if isinstance(error, dict):
return error.get("error_code")
return getattr(exc, "code", None)
def _vk_call_with_retry(self, func, *args, **kwargs):
max_attempts = 5
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except VkApiError as e:
code = self._vk_error_code(e)
if code not in (6, 9, 10) or attempt == max_attempts:
raise
delay = min(2.0, 0.35 * (2 ** (attempt - 1)))
if code == 9:
delay = max(delay, 1.0)
time.sleep(delay)
def load_chats(self):
self._clear_chat_tabs()
@@ -583,7 +858,27 @@ class VkChatManager(QMainWindow):
]
try:
conversations = self.vk.messages.getConversations(count=200, filter="all")['items']
self._set_busy(True, "Статус: загрузка чатов...")
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 = self._vk_call_with_retry(self.vk.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
for conv in conversations:
if conv['conversation']['peer']['type'] == 'chat':
chat_id = conv['conversation']['peer']['local_id']
@@ -616,8 +911,11 @@ class VkChatManager(QMainWindow):
self.chat_tabs.setTabText(3, f"AG Кофейни ({len(self.coffee_chat_checkboxes)})")
self.chat_tabs.setTabText(4, f"Прочие ({len(self.other_chat_checkboxes)})")
except VkApiError as e:
QMessageBox.critical(self, "Ошибка", f"Не удалось загрузить чаты: {e}")
self._log_error("load_chats", e)
QMessageBox.critical(self, "Ошибка", f"Не удалось загрузить чаты: {self._format_vk_error(e)}")
self.set_ui_state(False)
finally:
self._set_busy(False)
def get_user_info_by_id(self, user_id):
try:
@@ -694,20 +992,31 @@ class VkChatManager(QMainWindow):
return
results = []
for chat in selected_chats:
for user_id, user_info in user_infos.items():
try:
if action_type == "remove":
self.vk.messages.removeChatUser(chat_id=chat['id'], member_id=user_id)
results.append(f"✓ '{user_info}' исключен из '{chat['title']}'.")
else:
params = {'chat_id': chat['id'], 'user_id': user_id}
if self.visible_messages_checkbox.isChecked():
params['visible_messages_count'] = 250
self.vk.messages.addChatUser(**params)
results.append(f"✓ '{user_info}' приглашен в '{chat['title']}'.")
except VkApiError as e:
results.append(f"✗ Ошибка для '{user_info}' в '{chat['title']}': {e}")
total = len(selected_chats) * len(user_infos)
processed = 0
try:
action_label = "исключение" if action_type == "remove" else "приглашение"
self._set_busy(True, f"Статус: выполняется {action_label} (0/{total})...")
for chat in selected_chats:
for user_id, user_info in user_infos.items():
try:
if action_type == "remove":
self._vk_call_with_retry(self.vk.messages.removeChatUser, chat_id=chat['id'], member_id=user_id)
results.append(f"✓ '{user_info}' исключен из '{chat['title']}'.")
else:
params = {'chat_id': chat['id'], 'user_id': user_id}
if self.visible_messages_checkbox.isChecked():
params['visible_messages_count'] = 250
self._vk_call_with_retry(self.vk.messages.addChatUser, **params)
results.append(f"✓ '{user_info}' приглашен в '{chat['title']}'.")
except VkApiError as e:
self._log_error("execute_user_action", e)
results.append(f"✗ Ошибка для '{user_info}' в '{chat['title']}': {self._format_vk_error(e)}")
finally:
processed += 1
self.status_label.setText(f"Статус: выполняется {action_label} ({processed}/{total})...")
finally:
self._set_busy(False)
QMessageBox.information(self, "Результаты", "\n".join(results))
self.vk_url_input.clear()
@@ -758,24 +1067,35 @@ class VkChatManager(QMainWindow):
# 4. Выполнение API запросов
results = []
for chat in selected_chats:
# VK API требует peer_id. Для чатов это 2000000000 + local_id
try:
peer_id = 2000000000 + int(chat['id'])
except ValueError:
results.append(f"✗ Ошибка ID чата: {chat['id']}")
continue
for user_id, user_info in user_infos.items():
total = len(selected_chats) * len(user_infos)
processed = 0
try:
self._set_busy(True, f"Статус: назначение админов (0/{total})...")
for chat in selected_chats:
# VK API требует peer_id. Для чатов это 2000000000 + local_id
try:
self.vk.messages.setMemberRole(
peer_id=peer_id,
member_id=user_id,
role="admin"
)
results.append(f"✓ '{user_info}' назначен админом в '{chat['title']}'.")
except VkApiError as e:
results.append(f"✗ Ошибка для '{user_info}' в '{chat['title']}': {e}")
peer_id = 2000000000 + int(chat['id'])
except ValueError:
results.append(f"✗ Ошибка ID чата: {chat['id']}")
continue
for user_id, user_info in user_infos.items():
try:
self._vk_call_with_retry(
self.vk.messages.setMemberRole,
peer_id=peer_id,
member_id=user_id,
role="admin"
)
results.append(f"✓ '{user_info}' назначен админом в '{chat['title']}'.")
except VkApiError as e:
self._log_error("set_user_admin", e)
results.append(f"✗ Ошибка для '{user_info}' в '{chat['title']}': {self._format_vk_error(e)}")
finally:
processed += 1
self.status_label.setText(f"Статус: назначение админов ({processed}/{total})...")
finally:
self._set_busy(False)
# 5. Вывод результата
QMessageBox.information(self, "Результаты назначения", "\n".join(results))
@@ -798,4 +1118,4 @@ if __name__ == "__main__":
window = VkChatManager()
window.show()
sys.exit(app.exec())
sys.exit(app.exec())