代码收藏家技术教程 2024-09-03

小爱同学自定义指令执行详解

1.前言

之前买了小爱同学音响，一直想让其让我的生活变得更智能，编写一些程序来完成一些自动化任务，但是经过搜索发现，官方开发者平台不能用了，寻找api阶段浪费了我很长时间。最后在github 开源项目发现了俩个比较关键的项目：
https://github.com/Yonsm/MiService
https://github.com/yihong0618/xiaogpt
其实关键点在于，我需要俩个接口：

我说了什么
让音响答复我说了什么

只需要这俩个接口

2. 实现

对于第一个接口我找了很久，最后在xiaogpt这个项目中找到了这个api 的实现，xiaogpt也是miservice 这个项目实现的。

这里我简单做了个demo，相信能帮助大家节约很长时间：

#!/usr/bin/python3
from miservice.miaccount   import   MiAccount
from miservice.minaservice import   MiNAService
from aiohttp import ClientSession
from requests.utils import cookiejar_from_dict
import time
import os
import json
import asyncio

COMMAND = {
"打开主机":["wakeonlan xx:xx:xx:xx:xx:xx","已开启主机"],
"关闭主机":["ssh -o ConnectTimeout=1 root@192.168.123.60 shutdown -h now","已关闭主机"],
"打开桌面":["/root/tools/switch/switch.py on","已打开桌面电脑"],
"关闭桌面":["ssh -o ConnectTimeout=1 root@192.168.123.100 shutdown -h now; sleep 30; /root/tools/switch/switch.py off","已关闭桌面电脑"]
}

async def main():
    user_id = "小米账号" 
    password = "密码"
    hardware = "LX06" # 音响型号
    while True:
        try:
            misession = ClientSession()
            account = MiAccount(misession, user_id,password,"mi.token")
            await account.login("micoapi")
            service = MiNAService(account)
            device_list = await service.device_list()
            deviceID = device_list[0]['deviceID']
            LATEST_ASK_API=  "https://userprofile.mina.mi.com/device_profile/v2/conversation?source=dialogu&hardware={hardware}×tamp={timestamp}&limit=1"
            async with ClientSession() as session:
                cookies = dict(
                    deviceId=deviceID,
                )
                misession.cookie_jar.update_cookies(cookiejar_from_dict(cookies))
                session._cookie_jar = misession.cookie_jar
                # print(session._cookie_jar._cookies)
                while True:
                    r = await session.get(LATEST_ASK_API.format(hardware=hardware, timestamp=str(int(time.time()* 1000))))
                    data = await r.json()
                    query = json.loads(data["data"])['records'][0]['query']
                    qtime = int(json.loads(data["data"])['records'][0]['time']/1000)
                    # print(qtime,int(time.time()))
                    if abs(qtime - int(time.time())) < 5 and COMMAND.get(query):
                        print(f"{query} 已执行命令")
                        os.popen(COMMAND.get(query)[0])
                        await service.text_to_speech(deviceID, COMMAND.get(query)[1])
                        time.sleep(5)
                    time.sleep(1)
        except Exception as e:
            print(e)

要使用上述代码，首先要满足几个条件: