Python 接入

准备工作

获取 dm_api.dll 和 dm_api.py 文件
将 DLL 文件放置在程序运行目录
将 dm_api.py 添加到项目中

依赖

安装加密库：

pip install cryptography

导入模块

from dm_api import DmApi

快速开始

from dm_api import DmApi

PUBLIC_KEY = """
-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
-----END PUBLIC KEY-----
"""

def main():
    # 创建 API 实例
    api = DmApi(PUBLIC_KEY, "dm_api.dll")

    # 1. 检查是否由 Launcher 启动
    if api.restart_app_if_necessary():
        # Launcher 已启动，退出当前进程
        return

    # 2. 连接到 Launcher
    if not api.connect(r"\\.\pipe\distromate_pipe", 5000):
        print("Connection failed")
        return

    try:
        # 3. 验证许可证
        data = api.verify()
        if data:
            print("License verified!")
            print(f"License: {data.get('license_key')}")
            print(f"Expires: {data.get('expires_at')}")
        else:
            print("Verification failed")
            return

        # 4. 通知初始化完成
        if not api.initiated():
            print("Failed to notify initiated")

        # 应用主逻辑...
        print("Application running...")

    finally:
        api.close()

if __name__ == "__main__":
    main()

API 参考

DmApi 类

构造函数

DmApi(public_key: str, dll_path: str = "dm_api.dll")

参数:

参数	类型	描述
`public_key`	`str`	PEM 格式的 RSA 公钥
`dll_path`	`str`	DLL 文件路径，默认为 `dm_api.dll`

restart_app_if_necessary

检查程序是否由 Launcher 启动，如果不是则通过 Launcher 重启。

restart_app_if_necessary() -> bool

返回值:

True: 已启动 Launcher，当前进程应退出
False: 程序由 Launcher 启动，无需重启

connect

连接到 Launcher。

connect(pipe: str, timeout: int = 5000) -> bool

参数:

参数	类型	描述
`pipe`	`str`	Pipe 名称，格式: `\\.\pipe\<name>`
`timeout`	`int`	超时时间（毫秒），默认 5000

返回值: 连接是否成功

close

关闭与 Launcher 的连接。

close() -> None

verify

验证许可证。自动生成 nonce 并验证响应签名。

verify() -> dict | None

返回值:

成功: 包含许可证数据的字典
失败: None

activate

激活许可证。自动生成 nonce 并验证响应签名。

activate() -> dict | None

返回值: 同 verify

initiated

通知 Launcher 程序初始化完成。

initiated() -> bool

返回值: 操作是否成功

完整示例

import sys
from dm_api import DmApi

PUBLIC_KEY = """
-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
-----END PUBLIC KEY-----
"""

def print_license_info(data: dict) -> None:
    """打印许可证信息"""
    if user_name := data.get("user_name"):
        print(f"User: {user_name}")

    if product_id := data.get("product_id"):
        print(f"Product: {product_id}")

    if expires_at := data.get("expires_at"):
        print(f"Expires: {expires_at}")

    if features := data.get("features"):
        print("Features:")
        for feature in features:
            print(f"  - {feature}")

def main() -> int:
    # 初始化 API
    try:
        api = DmApi(PUBLIC_KEY)
    except Exception as e:
        print(f"Failed to initialize: {e}")
        return 1

    # 检查启动方式
    if api.restart_app_if_necessary():
        print("Restarting via launcher...")
        return 0

    # 连接
    if not api.connect(r"\\.\pipe\distromate_pipe"):
        print("Connection failed")
        return 1

    print("Connected to launcher")

    try:
        # 验证
        data = api.verify()
        if not data:
            print("Verification failed")
            return 1

        # 打印许可证信息
        print("License verified!")
        print_license_info(data)

        # 通知完成
        if not api.initiated():
            print("Warning: failed to notify initiated")

        # 应用主逻辑...
        print("Application running...")

        return 0

    finally:
        api.close()

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

使用上下文管理器

可以创建一个上下文管理器来自动管理连接：

from contextlib import contextmanager
from dm_api import DmApi

@contextmanager
def dm_connection(public_key: str, pipe: str, timeout: int = 5000):
    """DistroMate 连接上下文管理器"""
    api = DmApi(public_key)
    
    if api.restart_app_if_necessary():
        raise SystemExit(0)
    
    if not api.connect(pipe, timeout):
        raise RuntimeError("Connection failed")
    
    try:
        yield api
    finally:
        api.close()

# 使用
with dm_connection(PUBLIC_KEY, r"\\.\pipe\distromate_pipe") as api:
    data = api.verify()
    if data:
        print("License verified!")
        api.initiated()

PyInstaller 打包

spec 文件配置

# myapp.spec
a = Analysis(
    ['main.py'],
    pathex=[],
    binaries=[('dm_api.dll', '.')],
    datas=[],
    ...
)

命令行打包

pyinstaller --onefile --add-binary "dm_api.dll;." main.py

运行时 DLL 路径

import sys
import os
from dm_api import DmApi

def get_dll_path() -> str:
    """获取 DLL 路径（支持 PyInstaller 打包）"""
    if getattr(sys, 'frozen', False):
        # PyInstaller 打包后
        return os.path.join(sys._MEIPASS, 'dm_api.dll')
    else:
        # 开发环境
        return os.path.join(os.path.dirname(__file__), 'dm_api.dll')

api = DmApi(PUBLIC_KEY, get_dll_path())

异步使用

如果需要在异步应用中使用，可以使用线程池：

import asyncio
from concurrent.futures import ThreadPoolExecutor
from dm_api import DmApi

executor = ThreadPoolExecutor(max_workers=1)

async def verify_license_async(api: DmApi) -> dict | None:
    """异步验证许可证"""
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, api.verify)

async def main():
    api = DmApi(PUBLIC_KEY)
    
    if api.restart_app_if_necessary():
        return
    
    if not api.connect(r"\\.\pipe\distromate_pipe"):
        print("Connection failed")
        return
    
    try:
        data = await verify_license_async(api)
        if data:
            print("License verified!")
            api.initiated()
    finally:
        api.close()

asyncio.run(main())

类型提示

完整类型提示版本：

from __future__ import annotations
from typing import Optional
from dm_api import DmApi

def verify_and_get_features(api: DmApi) -> list[str]:
    """验证许可证并返回功能列表"""
    data: Optional[dict] = api.verify()
    if data is None:
        raise RuntimeError("Verification failed")
    
    features: list[str] = data.get("features", [])
    return features

注意事项

Python SDK 内置了 RSA 签名验证功能
需要安装 cryptography 库
所有 API 调用都是同步的
Windows 平台使用原始字符串 r"\\.\pipe\..." 处理路径
PyInstaller 打包时需要包含 DLL 文件

Python 接入

On this page