Skip to main content

编写互联配方

互联配方是 KnotLink 的亮点功能——用一个轻量脚本将不同程序串联起来，实现自动化协作。你不需要修改任何一个程序，只需写一个"胶水"脚本即可。

配方加载器（如 SignalTransfer）负责加载配方、订阅信号，信号触发时执行对应操作。

什么是互联配方​

想象这个场景：存档时光机 备份完成时发送信号，消息提醒 弹出通知——但两个程序彼此不认识，各自独立开发。

配方就是中间的翻译官：

存档时光机 ──signal: backup_success──▶  配方脚本  ──call: ShowMsg──▶  消息提醒
                                             │
                                     订阅信号 + 调用接口

配方做两件事：订阅信号 → 调用接口。配方本身不对外提供功能，只是一个"消费者"。

快速上手：写一个配方​

配方文件结构​

配方是一个 Python 文件，放在配方加载器的 MappingDataPacks/ 目录下，必须实现 run_mapping_plugin() 入口函数：

SignalTransfer/
└── MappingDataPacks/
    └── 你的配方.py    ← run_mapping_plugin()

真实示例：备份事件广播​

这是一个实际运行的配方——存档时光机 在备份时发送信号，配方把事件翻译成人类可读的消息，推送至 消息提醒：

# 备份事件广播.py
# 放在 MappingDataPacks/ 目录下，加载器自动运行

import threading
from KnotLinkClient_Py import SignalSubscriber, OpenSocketQuerier

def run_mapping_plugin():
    """入口函数，加载器启动时自动调用。在独立线程中运行。"""

    def worker():
        # 订阅存档时光机的备份事件信号
        subscriber = SignalSubscriber("0x00000020", "0x00000020")

        def on_signal(data: str):
            # 解析信号数据
            result = {}
            for pair in data.split(";"):
                if "=" not in pair:
                    continue
                key, value = pair.split("=", 1)
                result[key.strip()] = value.strip()

            event = result.get("event", "")
            msg = ""

            # 根据事件类型生成播报内容
            if event == "app_startup":
                msg = f"存档时光机 {result['version']}，启动！"
            elif event == "backup_started":
                msg = f"开始备份：{result['world']}"
            elif event == "backup_success":
                msg = f"备份成功：{result['world']}"
            elif event == "backup_failed":
                msg = f"备份失败：{result['world']}"
            elif event == "app_shutdown":
                msg = "存档时光机，关机"

            # 调用消息提醒弹出通知
            if msg:
                querier = OpenSocketQuerier("0x00000014", "0x00000010")
                querier.query_async(msg)
                print(f"[配方] 已推送：{msg}")

        subscriber.set_RecvFunc(on_signal)

    t = threading.Thread(target=worker, daemon=True)
    t.start()

def stop_mapping_plugin():
    """加载器关闭时调用"""
    pass

运行​

将上述文件放入 MappingDataPacks/，启动 SignalTransfer 加载器即可。不需要 funclist.json，不需要打包。

配方模式​

模式一：一对一桥接​

信号源 ──signal──▶ 配方 ──call──▶ 目标程序

最常见的模式。一个信号触发一个接口调用。

模式二：条件路由​

根据信号内容决定执行什么操作。上例中的 if event == "xxx" 就是条件路由。

模式三：数据转换​

信号格式和接口参数格式不同时，配方做中间翻译：

def on_signal(data: str):
    # archive_app 的信号格式: world=我的世界;event=backup_success
    # msg_notification 的接口格式: msgContext=消息内容

    kv = KLKVMap()
    kv.deserialize(data)

    notify_kv = KLKVMap()
    notify_kv["msgContext"] = f"备份成功：{kv.get('world')}"
    notify_kv["sender"] = "备份配方"

    querier.query_async(notify_kv.serialize())

配方 vs 普通程序​

提供功能调用功能互联配方
提供接口需要不需要不需要
funclist.json需要不需要不需要
发送信号可能需要不需要不需要
订阅信号不需要可能需要核心
调用接口不需要核心核心
打包分发需要不需要不需要
运行方式独立程序独立程序配方加载器托管

三种编写方式​

最灵活的方式。在 MappingDataPacks/ 下放 .py 文件，实现 run_mapping_plugin()。加载器自动扫描并运行。

import threading
from KnotLinkClient_Py import SignalSubscriber, OpenSocketQuerier

def run_mapping_plugin():
    def worker():
        sub = SignalSubscriber("com.a.app", "someSignal")

        def handler(data: str):
            querier = OpenSocketQuerier("com.b.app", "someFunc")
            querier.query_async(data)

        sub.set_RecvFunc(handler)

    threading.Thread(target=worker, daemon=True).start()

下一步​