AnabasisChatRemove/auth_webview.py

import os
import sys
import json
import threading
from urllib.parse import urlparse, parse_qs, unquote

import webview


def extract_token(url_string):
    token = None
    expires_in = 3600
    parsed = urlparse(url_string)

    if parsed.fragment:
        params = parse_qs(parsed.fragment)
    else:
        params = parse_qs(parsed.query)

    if 'access_token' in params:
        token = params['access_token'][0]
    if 'expires_in' in params:
        try:
            expires_in = int(params['expires_in'][0])
        except ValueError:
            pass

    if not token:
        start_marker = "access_token%253D"
        end_marker = "%25"

        start_index = url_string.find(start_marker)
        if start_index != -1:
            token_start_index = start_index + len(start_marker)
            remaining_url = url_string[token_start_index:]
            end_index = remaining_url.find(end_marker)

            if end_index != -1:
                raw_token = remaining_url[:end_index]
            else:
                amp_index = remaining_url.find('&')
                if amp_index != -1:
                    raw_token = remaining_url[:amp_index]
                else:
                    raw_token = remaining_url

            token = unquote(raw_token)

    return token, expires_in


def extract_auth_error(url_string):
    parsed = urlparse(url_string)
    params = parse_qs(parsed.fragment or parsed.query)
    return (params.get("error") or [None])[0], (params.get("error_description") or [None])[0]

def extract_auth_error_from_page(page_text):
    try:
        data = json.loads(page_text.strip())
    except (AttributeError, TypeError, ValueError):
        return None, None
    if not isinstance(data, dict):
        return None, None
    return data.get("error"), data.get("error_description")


def extract_redirect_result(url):
    """Read an OAuth result from VK's redirect URL before the page loads."""
    parsed = urlparse(url)
    if parsed.scheme != "https" or parsed.hostname not in ("oauth.vk.com", "oauth.vk.ru"):
        return None
    if parsed.path != "/blank.html":
        return None
    token, expires_in = extract_token(url)
    if token:
        return {"token": token, "expires_in": expires_in}
    error, description = extract_auth_error(url)
    if error:
        return {"error": error, "error_description": description}
    return None


def login_continue_script(auth_url):
    """Add an in-window control for starting OAuth after VK sign-in."""
    return """(function() {
        if (document.getElementById('anabasis-vk-continue')) return;
        const button = document.createElement('button');
        button.id = 'anabasis-vk-continue';
        button.textContent = 'Я вошёл в VK — получить токен';
        button.style.cssText = 'position:fixed;top:12px;right:12px;z-index:2147483647;'
            + 'padding:12px 16px;border:0;border-radius:8px;background:#0077ff;'
            + 'color:white;font:600 14px sans-serif;cursor:pointer;'
            + 'box-shadow:0 2px 12px #0006';
        button.addEventListener('click', function() {
            window.location.assign(AUTH_URL);
        });
        document.body.appendChild(button);
    })();""".replace("AUTH_URL", json.dumps(auth_url))


def main_auth(auth_url, output_path, login_first=False):
    stop_polling = threading.Event()
    polling_started = threading.Event()
    redirect_result = []
    redirect_lock = threading.Lock()

    def on_native_navigation_start(_sender, args):
        # WebView2 exposes the target URL even if blank.html later returns 405.
        result = extract_redirect_result(str(args.Uri))
        if result:
            with redirect_lock:
                redirect_result.append(result)
            args.Cancel = True

    def attach_navigation_hook():
        native = getattr(window, "native", None)
        native_webview = getattr(native, "webview", None)
        if native_webview is not None and hasattr(native_webview, "NavigationStarting"):
            native_webview.NavigationStarting += on_native_navigation_start

    def poll_url():
        while not stop_polling.wait(0.75):
            with redirect_lock:
                result = redirect_result[0] if redirect_result else None
            if result:
                stop_polling.set()
                with open(output_path, "w", encoding="utf-8") as f:
                    json.dump(result, f)
                try:
                    window.destroy()
                except Exception:
                    pass
                return
            try:
                url = str(window.get_current_url() or "")
            except Exception:
                continue
            if not url:
                continue

            token, expires_in = extract_token(url)
            error, error_description = extract_auth_error(url)
            if not token and not error and urlparse(url).hostname in ("oauth.vk.com", "oauth.vk.ru"):
                try:
                    # A 405 response can still leave the token in location.href.
                    page_url = window.evaluate_js("window.location.href")
                    if page_url:
                        token, expires_in = extract_token(str(page_url))
                        error, error_description = extract_auth_error(str(page_url))
                except Exception:
                    pass
            if not token and not error and urlparse(url).hostname in ("oauth.vk.com", "oauth.vk.ru"):
                try:
                    page_text = window.evaluate_js("document.body.innerText")
                    error, error_description = extract_auth_error_from_page(page_text)
                except Exception:
                    pass
            if not token and not error:
                continue

            stop_polling.set()
            data = {"token": token, "expires_in": expires_in} if token else {
                "error": error, "error_description": error_description,
            }
            with open(output_path, "w", encoding="utf-8") as f:
                json.dump(data, f)
            try:
                window.destroy()
            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():
        start_polling()
        if login_first:
            try:
                current_url = urlparse(str(window.get_current_url() or ""))
                if current_url.scheme == "https" and current_url.hostname in (
                    "vk.ru", "www.vk.ru", "vk.com", "www.vk.com"
                ):
                    window.evaluate_js(login_continue_script(auth_url))
            except Exception:
                pass

    def on_closed():
        stop_polling.set()

    window = webview.create_window("VK Авторизация", "https://vk.ru/" if login_first else auth_url)
    window.events.before_show += attach_navigation_hook
    window.events.loaded += on_loaded
    try:
        window.events.closed += on_closed
    except Exception:
        pass
    storage_path = os.path.join(os.path.dirname(output_path), "webview_profile")
    webview.start(start_polling, private_mode=False, storage_path=storage_path)
    stop_polling.set()


def main():
    # Supports both: `python auth_webview.py <auth_url> <output_path>`
    # and: `python auth_webview.py --auth <auth_url> <output_path>`
    args = sys.argv[1:]
    login_first = "--login-first" in args
    args = [arg for arg in args if arg != "--login-first"]
    if len(args) == 3 and args[0] == "--auth":
        auth_url, output_path = args[1], args[2]
    elif len(args) == 2:
        auth_url, output_path = args[0], args[1]
    else:
        print("Usage: auth_webview.py [--auth] <auth_url> <output_path>")
        return 1

    main_auth(auth_url, output_path, login_first=login_first)
    return 0


if __name__ == "__main__":
    sys.exit(main())