diff --git a/auth_webview.py b/auth_webview.py
index 7bcb69b..a68214f 100644
--- a/auth_webview.py
+++ b/auth_webview.py
@@ -1,6 +1,5 @@
import os
import sys
-import time
import json
import threading
from urllib.parse import urlparse, parse_qs, unquote
@@ -8,6 +7,13 @@ from urllib.parse import urlparse, parse_qs, unquote
import webview
+AUTH_USER_AGENT = (
+ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+ "AppleWebKit/537.36 (KHTML, like Gecko) "
+ "Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0"
+)
+
+
def extract_token(url_string):
token = None
expires_in = 3600
@@ -50,28 +56,59 @@ def extract_token(url_string):
def main_auth(auth_url, output_path):
+ stop_polling = threading.Event()
+ polling_started = threading.Event()
+
def poll_url():
- try:
- url = window.get_current_url()
- except Exception:
- url = None
- if url:
+ while not stop_polling.wait(0.75):
+ try:
+ url = window.get_current_url()
+ except Exception:
+ continue
+ if not url:
+ continue
+
token, expires_in = extract_token(url)
- if token:
- data = {"token": token, "expires_in": expires_in}
- with open(output_path, "w", encoding="utf-8") as f:
- json.dump(data, f)
+ if not token:
+ continue
+
+ stop_polling.set()
+ data = {"token": token, "expires_in": expires_in}
+ with open(output_path, "w", encoding="utf-8") as f:
+ json.dump(data, f)
+ try:
window.destroy()
- return
- threading.Timer(0.5, poll_url).start()
+ except Exception:
+ pass
+ return
+
+ def start_polling():
+ if polling_started.is_set():
+ return
+ polling_started.set()
+ thread = threading.Thread(target=poll_url, name="vk-auth-url-poller", daemon=True)
+ thread.start()
def on_loaded():
- threading.Timer(0.5, poll_url).start()
+ start_polling()
+
+ def on_closed():
+ stop_polling.set()
window = webview.create_window("VK Авторизация", auth_url)
window.events.loaded += on_loaded
- storage_path = os.path.join(os.path.dirname(output_path), "webview_profile")
- webview.start(private_mode=False, storage_path=storage_path)
+ try:
+ window.events.closed += on_closed
+ except Exception:
+ pass
+ webview.start(
+ start_polling,
+ gui="edgechromium" if os.name == "nt" else None,
+ private_mode=True,
+ storage_path=None,
+ user_agent=AUTH_USER_AGENT,
+ )
+ stop_polling.set()
def main():
diff --git a/build.py b/build.py
index c1345e1..ab606c2 100644
--- a/build.py
+++ b/build.py
@@ -1,3 +1,4 @@
+import hashlib
import os
import shutil
import subprocess
@@ -62,6 +63,24 @@ def ensure_project_root():
sys.exit(1)
+def write_sha256_file(file_path):
+ if not os.path.exists(file_path):
+ print(f"[ERROR] Cannot create SHA256, file not found: {file_path}")
+ sys.exit(1)
+
+ digest = hashlib.sha256()
+ with open(file_path, "rb") as f:
+ for chunk in iter(lambda: f.read(1024 * 1024), b""):
+ digest.update(chunk)
+
+ checksum_path = f"{file_path}.sha256"
+ file_name = os.path.basename(file_path)
+ with open(checksum_path, "w", encoding="utf-8", newline="\n") as f:
+ f.write(f"{digest.hexdigest()} {file_name}\n")
+ print(f"[OK] SHA256 created: {checksum_path}")
+ return checksum_path
+
+
def run_build():
print(f"--- 1. Running PyInstaller for {APP_NAME} v{VERSION} ---")
icon_abs_path = os.path.abspath(ICON_PATH)
@@ -149,8 +168,9 @@ def create_archive():
try:
# Создаем zip-архив из папки DIST_DIR
# base_name - имя файла без расширения, format - 'zip', root_dir - что упаковываем
- shutil.make_archive(os.path.join("dist", ARCHIVE_NAME), 'zip', DIST_DIR)
- print(f"[OK] Archive created: dist/{ARCHIVE_NAME}.zip")
+ archive_path = shutil.make_archive(os.path.join("dist", ARCHIVE_NAME), 'zip', DIST_DIR)
+ print(f"[OK] Archive created: {archive_path}")
+ write_sha256_file(archive_path)
except Exception as e:
print(f"[ERROR] Failed to create archive: {e}")
@@ -245,6 +265,7 @@ def build_installer():
print(f"[ERROR] Installer was not created: {installer_path}")
sys.exit(1)
print(f"[OK] Installer created: {installer_path}")
+ write_sha256_file(installer_path)
except Exception as e:
print(f"[ERROR] Failed to build installer: {e}")
sys.exit(1)
@@ -268,5 +289,7 @@ if __name__ == "__main__":
print("\n" + "=" * 30)
print("BUILD COMPLETED")
print(f"Release archive: dist/{ARCHIVE_NAME}.zip")
+ print(f"Release archive SHA256: dist/{ARCHIVE_NAME}.zip.sha256")
print(f"Installer: dist/{INSTALLER_NAME}")
+ print(f"Installer SHA256: dist/{INSTALLER_NAME}.sha256")
print("=" * 30)
diff --git a/deploy/nginx/vk.daemonlord.ru.conf b/deploy/nginx/vk.daemonlord.ru.conf
new file mode 100644
index 0000000..929ce12
--- /dev/null
+++ b/deploy/nginx/vk.daemonlord.ru.conf
@@ -0,0 +1,35 @@
+server {
+ listen 80;
+ server_name vk.daemonlord.ru;
+
+ return 301 https://$host$request_uri;
+}
+
+server {
+ listen 443 ssl http2;
+ server_name vk.daemonlord.ru;
+
+ ssl_certificate /etc/letsencrypt/live/vk.daemonlord.ru/fullchain.pem;
+ ssl_certificate_key /etc/letsencrypt/live/vk.daemonlord.ru/privkey.pem;
+
+ location = /vk/callback {
+ proxy_pass http://127.0.0.1:8787/vk/callback;
+ proxy_http_version 1.1;
+ proxy_set_header Host $host;
+ proxy_set_header X-Real-IP $remote_addr;
+ proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+ proxy_set_header X-Forwarded-Proto https;
+ proxy_buffering off;
+ }
+
+ location = /health {
+ proxy_pass http://127.0.0.1:8787/health;
+ proxy_http_version 1.1;
+ proxy_set_header Host $host;
+ proxy_set_header X-Forwarded-Proto https;
+ }
+
+ location / {
+ return 404;
+ }
+}
diff --git a/deploy/systemd/anabasis-vk-callback.service b/deploy/systemd/anabasis-vk-callback.service
new file mode 100644
index 0000000..820348f
--- /dev/null
+++ b/deploy/systemd/anabasis-vk-callback.service
@@ -0,0 +1,15 @@
+[Unit]
+Description=Anabasis VK OAuth callback host
+After=network.target
+
+[Service]
+Type=simple
+WorkingDirectory=/opt/anabasis-chat-remove
+ExecStart=/usr/bin/python3 /opt/anabasis-chat-remove/deploy/vk_callback_server.py --host 127.0.0.1 --port 8787 --quiet
+Restart=always
+RestartSec=3
+User=www-data
+Group=www-data
+
+[Install]
+WantedBy=multi-user.target
diff --git a/deploy/vk-auth-setup.md b/deploy/vk-auth-setup.md
new file mode 100644
index 0000000..daff470
--- /dev/null
+++ b/deploy/vk-auth-setup.md
@@ -0,0 +1,81 @@
+# VK OAuth callback setup
+
+The desktop app uses this redirect URI by default:
+
+```text
+https://vk.daemonlord.ru/vk/callback
+```
+
+The public HTTPS endpoint is expected to be handled by a reverse proxy. The
+backend callback host itself is plain HTTP on a non-standard local port:
+
+```text
+http://127.0.0.1:8787/vk/callback
+```
+
+## VK app settings
+
+1. Open the VK developer dashboard.
+2. Select the standalone app with ID `54454043`.
+3. Make sure the app type is `Standalone application`.
+4. Add the exact redirect URI:
+
+```text
+https://vk.daemonlord.ru/vk/callback
+```
+
+If the redirect URI in the OAuth request does not exactly match the app settings,
+VK can return:
+
+```json
+{"error":"invalid_request","error_description":"Security Error"}
+```
+
+## Backend callback host
+
+Run the local HTTP callback host:
+
+```bash
+python deploy/vk_callback_server.py --host 127.0.0.1 --port 8787
+```
+
+Health check:
+
+```text
+http://127.0.0.1:8787/health
+```
+
+Optional systemd unit:
+
+```text
+deploy/systemd/anabasis-vk-callback.service
+```
+
+Adjust `WorkingDirectory`, `ExecStart`, and `User` for the server path/user.
+
+The callback page does not process or store the token. With implicit OAuth, VK
+puts `access_token` in the URL fragment. The desktop webview reads that final URL
+directly from the embedded browser.
+
+## Reverse proxy
+
+Use the nginx example:
+
+```text
+deploy/nginx/vk.daemonlord.ru.conf
+```
+
+It proxies:
+
+```text
+https://vk.daemonlord.ru/vk/callback -> http://127.0.0.1:8787/vk/callback
+```
+
+## Desktop app override
+
+The app already defaults to `https://vk.daemonlord.ru/vk/callback`. To override it:
+
+```powershell
+$env:ANABASIS_VK_REDIRECT_URI = "https://vk.daemonlord.ru/vk/callback"
+python main.py
+```
diff --git a/deploy/vk-callback/index.html b/deploy/vk-callback/index.html
new file mode 100644
index 0000000..c63438d
--- /dev/null
+++ b/deploy/vk-callback/index.html
@@ -0,0 +1,46 @@
+
+
+
+
+
+ VK authorization
+
+
+
+
+ Авторизация VK завершена
+ Вернитесь в приложение. Это окно можно закрыть после завершения входа.
+
+
+
diff --git a/deploy/vk_callback_server.py b/deploy/vk_callback_server.py
new file mode 100644
index 0000000..9c18acc
--- /dev/null
+++ b/deploy/vk_callback_server.py
@@ -0,0 +1,78 @@
+import argparse
+import os
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+
+DEFAULT_HOST = "127.0.0.1"
+DEFAULT_PORT = 8787
+CALLBACK_PATH = "/vk/callback"
+HEALTH_PATH = "/health"
+
+
+def _read_callback_html():
+ html_path = Path(__file__).resolve().parent / "vk-callback" / "index.html"
+ return html_path.read_bytes()
+
+
+class VkCallbackHandler(BaseHTTPRequestHandler):
+ server_version = "AnabasisVkCallback/1.0"
+
+ def do_GET(self):
+ path = self.path.split("?", 1)[0]
+ if path == HEALTH_PATH:
+ self._send_text(200, "ok\n")
+ return
+ if path in ("/", CALLBACK_PATH):
+ self._send_html(200, _read_callback_html())
+ return
+ self._send_text(404, "not found\n")
+
+ def log_message(self, fmt, *args):
+ if getattr(self.server, "quiet", False):
+ return
+ super().log_message(fmt, *args)
+
+ def _send_html(self, status, body):
+ self.send_response(status)
+ self.send_header("Content-Type", "text/html; charset=utf-8")
+ self.send_header("Cache-Control", "no-store")
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+ def _send_text(self, status, text):
+ body = text.encode("utf-8")
+ self.send_response(status)
+ self.send_header("Content-Type", "text/plain; charset=utf-8")
+ self.send_header("Cache-Control", "no-store")
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+
+def parse_args():
+ parser = argparse.ArgumentParser(description="Anabasis VK OAuth callback host")
+ parser.add_argument("--host", default=os.getenv("ANABASIS_VK_CALLBACK_HOST", DEFAULT_HOST))
+ parser.add_argument("--port", type=int, default=int(os.getenv("ANABASIS_VK_CALLBACK_PORT", DEFAULT_PORT)))
+ parser.add_argument("--quiet", action="store_true")
+ return parser.parse_args()
+
+
+def main():
+ args = parse_args()
+ server = ThreadingHTTPServer((args.host, args.port), VkCallbackHandler)
+ server.quiet = args.quiet
+ print(f"VK callback server listening on http://{args.host}:{args.port}{CALLBACK_PATH}")
+ print(f"Health check: http://{args.host}:{args.port}{HEALTH_PATH}")
+ try:
+ server.serve_forever()
+ except KeyboardInterrupt:
+ pass
+ finally:
+ server.server_close()
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/main.py b/main.py
index 7e4dc5e..18326ce 100644
--- a/main.py
+++ b/main.py
@@ -3,6 +3,7 @@ import os
import shutil
import sys
import time
+from urllib.parse import urlencode
from PySide6.QtCore import QProcess
from PySide6.QtCore import QStandardPaths
@@ -38,18 +39,24 @@ APP_DATA_DIR = os.path.join(
TOKEN_FILE = os.path.join(APP_DATA_DIR, "token.json")
SETTINGS_FILE = os.path.join(APP_DATA_DIR, "settings.json")
WEB_ENGINE_CACHE_DIR = os.path.join(APP_DATA_DIR, "web_engine_cache")
+AUTH_WEBVIEW_PROFILE_DIR = os.path.join(APP_DATA_DIR, "webview_profile")
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")
AUTH_RELOGIN_BACKOFF_SECONDS = 5.0
+VK_APP_ID = "54454043"
+VK_AUTH_SCOPE = "1073737727"
+VK_API_VERSION = "5.131"
+VK_AUTH_REDIRECT_URI = os.getenv("ANABASIS_VK_REDIRECT_URI", "https://vk.daemonlord.ru/vk/callback").strip() or "https://vk.daemonlord.ru/vk/callback"
# Legacy owner/repo format for GitHub-only fallback.
UPDATE_REPOSITORY = ""
# Full repository URL is preferred (supports GitHub/Gitea).
UPDATE_REPOSITORY_URL = "https://git.daemonlord.ru/benya/AnabasisChatRemove"
UPDATE_CHANNEL_DEFAULT = "stable"
UPDATE_REQUEST_TIMEOUT = 8
-AUTH_ERROR_CONTEXTS = ("load_chats", "execute_user_action", "set_user_admin")
+AUTH_ERROR_CONTEXTS = ("load_chats", "execute_user_action", "set_user_admin", "unset_user_admin")
+VK_AUTH_URL_OVERRIDE = ""
def get_resource_path(relative_path):
@@ -80,8 +87,14 @@ class BulkActionWorker(QObject):
def _is_auth_error(exc):
return VkService.is_auth_error(exc, str(exc).lower())
+ def _role_context(self):
+ return "unset_user_admin" if self.action_type == "unadmin" else "set_user_admin"
+
+ def _role_action_name(self):
+ return "разжалования администраторов" if self.action_type == "unadmin" else "назначения администраторов"
+
def _emit_progress(self, processed):
- label = "admin" if self.action_type == "admin" else ("remove" if self.action_type == "remove" else "add")
+ label = self.action_type if self.action_type in ("remove", "add", "admin", "unadmin") else "add"
self.progress.emit(processed, self.total, label)
def run(self):
@@ -90,7 +103,7 @@ class BulkActionWorker(QObject):
try:
for chat in self.selected_chats:
peer_id = None
- if self.action_type == "admin":
+ if self.action_type in ("admin", "unadmin", "remove"):
try:
peer_id = 2000000000 + int(chat["id"])
except (ValueError, TypeError):
@@ -103,6 +116,19 @@ class BulkActionWorker(QObject):
for user_id, user_info in self.user_infos.items():
try:
if self.action_type == "remove":
+ try:
+ self.vk_call_with_retry(
+ self.vk.messages.setMemberRole,
+ peer_id=peer_id,
+ member_id=user_id,
+ role="member",
+ )
+ results.append(f"✓ '{user_info}' разжалован перед исключением из '{chat['title']}'.")
+ except VkApiError as demote_exc:
+ if self._is_auth_error(demote_exc):
+ self.auth_error.emit("unset_user_admin", demote_exc, "разжалования администратора перед исключением")
+ return
+ results.append(f"⚠ Не удалось разжаловать '{user_info}' в '{chat['title']}': {demote_exc}")
self.vk_call_with_retry(self.vk.messages.removeChatUser, chat_id=chat["id"], member_id=user_id)
results.append(f"✓ '{user_info}' исключен из '{chat['title']}'.")
elif self.action_type == "add":
@@ -119,12 +145,20 @@ class BulkActionWorker(QObject):
role="admin",
)
results.append(f"✓ '{user_info}' назначен админом в '{chat['title']}'.")
+ elif self.action_type == "unadmin":
+ self.vk_call_with_retry(
+ self.vk.messages.setMemberRole,
+ peer_id=peer_id,
+ member_id=user_id,
+ role="member",
+ )
+ results.append(f"✓ '{user_info}' разжалован в '{chat['title']}'.")
else:
raise RuntimeError(f"Unknown action: {self.action_type}")
except VkApiError as exc:
if self._is_auth_error(exc):
- context = "set_user_admin" if self.action_type == "admin" else "execute_user_action"
- action_name = "назначения администраторов" if self.action_type == "admin" else "выполнения операций с пользователями"
+ context = self._role_context() if self.action_type in ("admin", "unadmin") else "execute_user_action"
+ action_name = self._role_action_name() if self.action_type in ("admin", "unadmin") else "выполнения операций с пользователями"
self.auth_error.emit(context, exc, action_name)
return
results.append(f"✗ Ошибка для '{user_info}' в '{chat['title']}': {exc}")
@@ -327,6 +361,12 @@ class VkChatManager(QMainWindow):
tools_menu.addAction(make_admin_action)
self.make_admin_action = make_admin_action
+ unset_admin_action = QAction("Разжаловать администратора", self)
+ unset_admin_action.setStatusTip("Снять права администратора с выбранных пользователей в выбранных чатах")
+ unset_admin_action.triggered.connect(self.unset_user_admin)
+ tools_menu.addAction(unset_admin_action)
+ self.unset_admin_action = unset_admin_action
+
check_updates_action = QAction("Проверить обновления", self)
check_updates_action.setStatusTip("Проверить наличие новой версии приложения")
check_updates_action.triggered.connect(self.check_for_updates)
@@ -629,6 +669,8 @@ class VkChatManager(QMainWindow):
self._clear_chat_tabs()
if hasattr(self, "make_admin_action"):
self.make_admin_action.setEnabled(authorized and (not self._auth_ui_busy))
+ if hasattr(self, "unset_admin_action"):
+ self.unset_admin_action.setEnabled(authorized and (not self._auth_ui_busy))
if hasattr(self, "logout_action"):
self.logout_action.setEnabled(not self._auth_ui_busy)
@@ -639,6 +681,8 @@ class VkChatManager(QMainWindow):
self.logout_action.setEnabled(not in_progress)
if hasattr(self, "make_admin_action"):
self.make_admin_action.setEnabled(not in_progress and self.token is not None)
+ if hasattr(self, "unset_admin_action"):
+ self.unset_admin_action.setEnabled(not in_progress and self.token is not None)
self.auth_btn.setEnabled((self.token is None) and (not in_progress))
def _set_busy(self, busy, status_text=None):
@@ -702,7 +746,12 @@ class VkChatManager(QMainWindow):
clear_inputs_on_success=True,
):
total = max(1, len(selected_chats) * len(user_infos))
- self._bulk_action_context = "set_user_admin" if action_type == "admin" else "execute_user_action"
+ if action_type == "admin":
+ self._bulk_action_context = "set_user_admin"
+ elif action_type == "unadmin":
+ self._bulk_action_context = "unset_user_admin"
+ else:
+ self._bulk_action_context = "execute_user_action"
self._bulk_action_success_message_title = success_message_title
self._bulk_clear_inputs_on_success = clear_inputs_on_success
self._log_event(
@@ -737,7 +786,12 @@ class VkChatManager(QMainWindow):
thread.start()
def _on_bulk_action_progress(self, processed, total, label):
- label_text = {"remove": "исключение", "add": "приглашение", "admin": "назначение админов"}.get(label, label)
+ label_text = {
+ "remove": "исключение",
+ "add": "приглашение",
+ "admin": "назначение админов",
+ "unadmin": "разжалование админов",
+ }.get(label, label)
self._update_operation_progress(processed, total, label_text)
self.status_label.setText(f"Статус: выполняется {label_text} ({processed}/{max(1, total)})...")
@@ -855,18 +909,23 @@ class VkChatManager(QMainWindow):
print(f"Ошибка отложенной очистки кэша: {e}")
def _try_remove_web_cache(self):
- if not os.path.exists(WEB_ENGINE_CACHE_DIR):
+ cache_dirs = [WEB_ENGINE_CACHE_DIR, AUTH_WEBVIEW_PROFILE_DIR]
+ existing_dirs = [path for path in cache_dirs if os.path.exists(path)]
+ if not existing_dirs:
return
attempts = 5
last_error = None
- for _ in range(attempts):
- try:
- shutil.rmtree(WEB_ENGINE_CACHE_DIR)
- last_error = None
+ for path in existing_dirs:
+ for _ in range(attempts):
+ try:
+ shutil.rmtree(path)
+ last_error = None
+ break
+ except Exception as e:
+ last_error = e
+ time.sleep(0.2)
+ if last_error:
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:
@@ -890,6 +949,19 @@ class VkChatManager(QMainWindow):
def _build_auth_command(self, auth_url, output_path):
return self.vk_service.build_auth_command(auth_url, output_path, entry_script_path=os.path.abspath(__file__))
+ def _build_vk_auth_url(self, display="mobile"):
+ if VK_AUTH_URL_OVERRIDE:
+ return VK_AUTH_URL_OVERRIDE
+ params = {
+ "client_id": VK_APP_ID,
+ "display": display,
+ "redirect_uri": VK_AUTH_REDIRECT_URI,
+ "scope": VK_AUTH_SCOPE,
+ "response_type": "token",
+ "v": VK_API_VERSION,
+ }
+ return f"https://oauth.vk.com/authorize?{urlencode(params)}"
+
def _on_auth_process_error(self, process_error):
self._auth_process_error_text = f"Ошибка запуска авторизации: {process_error}"
# For failed starts Qt may not emit finished(), so release UI here.
@@ -965,15 +1037,7 @@ class VkChatManager(QMainWindow):
status_text = "Статус: ожидание авторизации..."
self.status_label.setText(status_text)
- auth_url = (
- "https://oauth.vk.com/authorize?"
- "client_id=2685278&"
- "display=page&"
- "redirect_uri=https://oauth.vk.com/blank.html&"
- "scope=1073737727&"
- "response_type=token&"
- "v=5.131"
- )
+ auth_url = self._build_vk_auth_url(display="mobile")
output_path = os.path.join(APP_DATA_DIR, "auth_result.json")
try:
if os.path.exists(output_path):
@@ -1269,6 +1333,50 @@ class VkChatManager(QMainWindow):
)
return
+ def unset_user_admin(self):
+ """Снимает права администратора в выбранных чатах."""
+ if not self.user_ids_to_process:
+ QMessageBox.warning(self, "Ошибка", "Нет ID пользователей для операции.")
+ return
+
+ selected_chats = self._get_selected_chats()
+ if not selected_chats:
+ QMessageBox.warning(self, "Ошибка", "Выберите хотя бы один чат.")
+ return
+
+ user_infos = {uid: self.get_user_info_by_id(uid) for uid in self.user_ids_to_process}
+ user_names_str = "\n".join([f"• {name}" for name in user_infos.values()])
+
+ msg = (
+ f"Вы уверены, что хотите разжаловать следующих администраторов:\n\n"
+ f"{user_names_str}\n\n"
+ f"в {len(selected_chats)} выбранных чатах?"
+ )
+
+ confirm_dialog = QMessageBox(self)
+ confirm_dialog.setWindowTitle("Подтверждение прав")
+ confirm_dialog.setText(msg)
+ confirm_dialog.setIcon(QMessageBox.Icon.Question)
+ yes_button = confirm_dialog.addButton("Да", QMessageBox.ButtonRole.YesRole)
+ no_button = confirm_dialog.addButton("Нет", QMessageBox.ButtonRole.NoRole)
+ confirm_dialog.setDefaultButton(no_button)
+
+ confirm_dialog.exec()
+
+ if confirm_dialog.clickedButton() != yes_button:
+ return
+
+ self._start_bulk_action_worker(
+ action_type="unadmin",
+ selected_chats=selected_chats,
+ user_infos=user_infos,
+ action_label="разжалование админов",
+ action_button=None,
+ success_message_title="Результаты разжалования",
+ clear_inputs_on_success=True,
+ )
+ return
+
# Refactor overrides: keep logic in service modules and thin UI orchestration here.
def _process_links_list(self, links_list):
if not self.vk:
diff --git a/tests/test_auto_update_service.py b/tests/test_auto_update_service.py
index 8d2f0f7..8e4a0f7 100644
--- a/tests/test_auto_update_service.py
+++ b/tests/test_auto_update_service.py
@@ -31,6 +31,15 @@ class AutoUpdateServiceTests(unittest.TestCase):
expected = hashlib.sha256(payload).hexdigest()
self.assertEqual(AutoUpdateService.sha256_file(str(path)), expected)
+ def test_extract_sha256_from_build_sidecar_format(self):
+ digest = "b" * 64
+ text = f"{digest} AnabasisManager-2.2.5.zip\n"
+ extracted = AutoUpdateService.extract_sha256_from_text(
+ text,
+ "AnabasisManager-2.2.5.zip",
+ )
+ self.assertEqual(extracted, digest)
+
def test_build_update_script_contains_core_vars(self):
script = AutoUpdateService.build_update_script(
app_dir=r"C:\Apps\AnabasisManager",
diff --git a/tests/test_main_contracts.py b/tests/test_main_contracts.py
index 0732c69..533a44e 100644
--- a/tests/test_main_contracts.py
+++ b/tests/test_main_contracts.py
@@ -27,7 +27,7 @@ class MainContractsTests(unittest.TestCase):
return ast.walk(node)
def test_auth_error_contexts_contains_only_supported_contexts(self):
- expected_contexts = {"load_chats", "execute_user_action", "set_user_admin"}
+ expected_contexts = {"load_chats", "execute_user_action", "set_user_admin", "unset_user_admin"}
for node in self.module.body:
if isinstance(node, ast.Assign):
for target in node.targets:
@@ -37,6 +37,52 @@ class MainContractsTests(unittest.TestCase):
return
self.fail("AUTH_ERROR_CONTEXTS assignment not found")
+ def test_unset_user_admin_method_exists(self):
+ self._find_method("unset_user_admin")
+
+ def test_uses_custom_standalone_vk_app(self):
+ constants = {}
+ for node in self.module.body:
+ if not isinstance(node, ast.Assign) or len(node.targets) != 1:
+ continue
+ target = node.targets[0]
+ if isinstance(target, ast.Name):
+ try:
+ constants[target.id] = ast.literal_eval(node.value)
+ except Exception:
+ continue
+
+ self.assertEqual(constants.get("VK_APP_ID"), "54454043")
+ self.assertIn("ANABASIS_VK_REDIRECT_URI", self.main_source)
+ self.assertIn("https://vk.daemonlord.ru/vk/callback", self.main_source)
+
+ def test_remove_action_demotes_member_before_removing(self):
+ bulk_worker = None
+ for node in self.module.body:
+ if isinstance(node, ast.ClassDef) and node.name == "BulkActionWorker":
+ bulk_worker = node
+ break
+ self.assertIsNotNone(bulk_worker, "BulkActionWorker class not found")
+
+ run_method = None
+ for node in bulk_worker.body:
+ if isinstance(node, ast.FunctionDef) and node.name == "run":
+ run_method = node
+ break
+ self.assertIsNotNone(run_method, "BulkActionWorker.run method not found")
+
+ has_member_role = False
+ has_remove_call = False
+ for node in ast.walk(run_method):
+ if isinstance(node, ast.keyword) and node.arg == "role":
+ if isinstance(node.value, ast.Constant) and node.value.value == "member":
+ has_member_role = True
+ if isinstance(node, ast.Attribute) and node.attr == "removeChatUser":
+ has_remove_call = True
+
+ self.assertTrue(has_member_role, "remove action must demote admins with role='member'")
+ self.assertTrue(has_remove_call, "remove action must still call removeChatUser")
+
def test_check_for_updates_has_reentry_guard(self):
method = self._find_method("check_for_updates")
has_guard = False