Windows: 透過右鍵功能表快速執行 Python 程式

import os
import sys
import winreg as reg
from typing import Union


def get_key_or_create_if_not_exist(
    parent_key: Union[int, reg.HKEYType],
    key_path: str,
):
    assert parent_key is not None, "parent_key can not be None"
    assert key_path is not None, "key_path can not be None"
    try:
        return reg.OpenKey(parent_key, key_path)
    except FileNotFoundError:
        return reg.CreateKey(parent_key, key_path)


def set_ex_if_not_exist(key: reg.HKEYType, value_name: str, value: str):
    assert key is not None, "key can not be None"
    assert value_name is not None, "value_name can not be None"
    if value is None:
        return
    try:
        reg.QueryValueEx(key, value_name)
    except FileNotFoundError:
        reg.SetValueEx(key, value_name, None, reg.REG_EXPAND_SZ, value)


def register(
    menu_name: str,
    menu_icon_path: str,
    item_name: str,
    script_path: str,
    script_args: str = "",
    multi_select: bool = False,
):
    python_exe = sys.executable
    hidden_python_exe = os.path.join(
        os.path.dirname(python_exe), "pythonw.exe"
    )
    script_abs_path = os.path.abspath(script_path)
    script_cmd = (
        f'"{hidden_python_exe}" "{script_abs_path}" "%1" {script_args}'
        if multi_select
        else f'"{hidden_python_exe}" "{script_abs_path}" {script_args} "%1"'
    )

    # otter shell
    menu_shell_key = get_key_or_create_if_not_exist(
        parent_key=reg.HKEY_CURRENT_USER,
        key_path=r"Software\\Classes\\*\\shell\\",
    )
    menu_key = get_key_or_create_if_not_exist(
        parent_key=menu_shell_key, key_path=menu_name
    )
    set_ex_if_not_exist(menu_key, "Icon", menu_icon_path)
    set_ex_if_not_exist(menu_key, "MUIVerb", "")
    set_ex_if_not_exist(menu_key, "SubCommands", "")
    if multi_select:
        set_ex_if_not_exist(menu_key, "MultiSelectModel", "Player")

    # inner shell
    item_shell_key = get_key_or_create_if_not_exist(
        parent_key=menu_key, key_path="shell"
    )
    item_key = get_key_or_create_if_not_exist(
        parent_key=item_shell_key, key_path=item_name
    )
    set_ex_if_not_exist(item_key, "", "")

    # command
    command_key = get_key_or_create_if_not_exist(item_key, "command")
    set_ex_if_not_exist(command_key, "", script_cmd)

if __name__ == "__main__":
    register("Converter", None, "Image to PDF", "./convert_img.pyw", "-t pdf")

import win32event as evt
import win32api as api
import win32con as con
import subprocess
import time
import sys
from multiprocessing import shared_memory
from typing import List

ERROR_ALREADY_EXISTS = 183  # Not defined in pywin32
CLASS_NAME = "MENU_CALLER_WindowClass"
NAME = "MENU_CALLER_9d2d0175-497c-42e6-a89d-9b10912970f2"
MUTEX_NAME_EXIST = f"Global\\{NAME}_EXIST"
MUTEX_NAME_SHARED_MEMORY = f"Global\\{NAME}_SHARED_MEMORY"
SHARED_MEMORY_NAME = NAME

WAIT_TIMEOUT = 600  # ms
REQUIRE_LOCK_TIMEOUT = 1000  # ms
SHARED_MEMORY_SIZE_PER_STR = 512  # bytes
SHARED_MEMORY_SIZE_PER_STR_LEN = 2  # bytes
SHARED_MEMORY_STR_COUNT = 255
SHARED_MEMORY_STR_COUNT_LEN = 1  # bytes
SHARED_MEMORY_TOTAL_SIZE = (
    SHARED_MEMORY_SIZE_PER_STR + SHARED_MEMORY_SIZE_PER_STR_LEN
) * SHARED_MEMORY_STR_COUNT + SHARED_MEMORY_STR_COUNT_LEN  # bytes


class InstanceChecker:
    def __init__(self):
        self.mutex = evt.CreateMutex(None, True, MUTEX_NAME_EXIST)
        self.instance_exist = api.GetLastError() == ERROR_ALREADY_EXISTS

    def is_instance_exist(self):
        return self.instance_exist

    def release(self):
        if self.mutex is not None and not self.instance_exist:
            evt.ReleaseMutex(self.mutex)
            api.CloseHandle(self.mutex)
            self.mutex = None

    def __del__(self):
        self.release()


class SharedMemoryLock:
    def __init__(self, create: bool):
        self.owner = create
        if create:
            self.mutex = evt.CreateMutex(None, False, MUTEX_NAME_SHARED_MEMORY)
        else:
            self.mutex = evt.OpenMutex(
                con.SYNCHRONIZE, True, MUTEX_NAME_SHARED_MEMORY
            )
        if self.mutex is None or self.mutex == 0:
            print(f"Shared Memory Lock Error: {api.GetLastError()}")

    def require(self):
        result = evt.WaitForSingleObject(self.mutex, REQUIRE_LOCK_TIMEOUT)
        if result == evt.WAIT_TIMEOUT:
            print("Shared Memory Lock Require Timeout Error")

    def release(self):
        evt.ReleaseMutex(self.mutex)

    def __del__(self):
        if self.owner:
            api.CloseHandle(self.mutex)


class SharedMemoryManager:
    def __init__(self, create: bool) -> None:
        self.owner = create
        self.lock = SharedMemoryLock(create)
        self.lock.require()
        self.sm = shared_memory.SharedMemory(
            name=SHARED_MEMORY_NAME,
            create=create,
            size=SHARED_MEMORY_TOTAL_SIZE,
        )
        if create:
            self.set_str_count(0)
        self.lock.release()

    def set_str_count(self, val: int):
        self.lock.require()
        self.sm.buf[SHARED_MEMORY_TOTAL_SIZE - 1] = val.to_bytes(1)[0]
        self.lock.release()

    def get_str_count(self):
        return int.from_bytes(
            self.sm.buf[SHARED_MEMORY_TOTAL_SIZE - 1].to_bytes()
        )

    def write_str(self, val: str):
        self.lock.require()
        str_count = self.get_str_count()
        start_idx = str_count * (
            SHARED_MEMORY_SIZE_PER_STR + SHARED_MEMORY_SIZE_PER_STR_LEN
        )
        str_start_idx = start_idx + SHARED_MEMORY_SIZE_PER_STR_LEN

        str_bytes = val.encode()
        self.sm.buf[str_start_idx : str_start_idx + len(str_bytes)] = str_bytes
        self.sm.buf[start_idx : start_idx + SHARED_MEMORY_SIZE_PER_STR_LEN] = (
            len(str_bytes).to_bytes(2)
        )

        self.lock.release()
        self.set_str_count(str_count + 1)

    def read_strs(self) -> List[str]:
        self.lock.require()
        strings = []
        str_count = self.get_str_count()
        for i in range(str_count):
            start_idx = i * (
                SHARED_MEMORY_SIZE_PER_STR + SHARED_MEMORY_SIZE_PER_STR_LEN
            )
            str_start_idx = start_idx + SHARED_MEMORY_SIZE_PER_STR_LEN

            str_len = int.from_bytes(
                self.sm.buf[
                    start_idx : start_idx + SHARED_MEMORY_SIZE_PER_STR_LEN
                ].tobytes()
            )
            s = (
                self.sm.buf[str_start_idx : str_start_idx + str_len]
                .tobytes()
                .decode()
            )
            strings.append(s)
        self.lock.release()
        return strings

    def __del__(self):
        del self.lock
        if self.owner:
            self.sm.unlink()
        self.sm.close()


def replace_params_with_files_path(params: List[str], paths: List[str]):
    try:
        token_idx = params.index("$files")
        params.pop(token_idx)
        for path in reversed(paths):
            params.insert(token_idx, path)
    except ValueError:
        params.extend(paths)


def main():
    if len(sys.argv) < 3:
        return
    target_file = sys.argv[1]
    params = sys.argv[2:]

    checker = InstanceChecker()
    if checker.is_instance_exist():
        smm = SharedMemoryManager(create=False)
        smm.write_str(target_file)
        del smm
    else:
        smm = SharedMemoryManager(create=True)
        smm.write_str(target_file)
        prev_str_count = smm.get_str_count()

        # loop to check whether we receive all messages
        while True:
            time.sleep(WAIT_TIMEOUT / 1000)
            cur_str_count = smm.get_str_count()
            if prev_str_count == cur_str_count:
                break
            prev_str_count = cur_str_count

        # get all paths
        paths = smm.read_strs()
        del smm

        # run the process
        replace_params_with_files_path(params, paths)
        p = subprocess.Popen(params, shell=True)
        p.wait()


if __name__ == "__main__":
    main()