关于小爱同学自定义指令执行

wshuo

1.前言

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

1. 我说了什么
2. 让音响答复我说了什么  

只需要这俩个接口

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)

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

1. python支持异步（忘记是几点几版本开始支持了）
2. 安装miservice

   pip install miservice

其他把账号、密码、音响型号 配置进去 就可以在COMMAND 字典中自定义shell 命令了

这里简单介绍一些 COMMAND 字典，键 为语控命令，值 为一个列表，列表中第一个值 为 要执行的shell 命令，第二个值 为 音响答复。  

后面要增加命令可以直接修改 COMMAND字典。这里可以看到我对俩个电脑开关机进行控制：
对于桌面电脑，我是先 执行shutdown -h now,  等待30s后，再对电源进行关闭操作（switch为我写的控制电源的一个脚本），因为我总觉得直接下电对 电脑不好。  

对于主机电脑，我直接使用 网络唤醒实现的开机（连接网线），对于桌面电脑由于我没有网线接口了，所有用上电自动开机实现的开机控制。  

3.其他一些细节

3.1 让小爱知道这个指令

由于自定义的一些指令，直接问小爱，小爱会回答不知道这个指令，那么我们可以在 小爱训练中加入这些自定义指令：  

这里小爱，会先回答“好的”， 然后我们api响应的话语会打断说话，紧接着说“已开启主机”  

3.2 回答内容为自定义查询的内容

在COMMAND 里，我们可以用这种方式进行定义列表中的第二个字段：  

f"查询的信息是 {data}"

可以注意到，这里我执行shell, 使用 os.popen 来执行的（非阻塞），一方面也是为了可以读取返回内容，然后给与相应的赋值
data 直接赋值为 os.popen 返回的read() 即可
当然os.popen不是一种良好的执行 命令的方式，有更官方的写法，懒得优化了。  

4.效果展示

https://7.z.wiki/autoupload/20240707/Yfx4/1.mp4
终于可以在床上控制电脑开关机了

xiaomihzj

感谢楼主分享，跟我的需求不谋而合，代码也很简洁，调一下就可以用了。
在此也给出一些改动，可能是因为小米那边接口变了，cookie里面需要增加两个字段，不然会返回 code 999
（另外我这边的一个主要作用是控制投影仪，我的品牌是当贝，别的品牌不知道是否通用。只要有网页控制功能的应该都可以做到。如果有人有兴趣的话，可以回复我）
以下是修改之后的小爱音箱自定义指令的代码，供参考：

[Python] 纯文本查看 复制代码

?

01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79

#!/usr/bin/python3 import os import sys from miservice.miaccount import MiAccount from miservice.minaservice import MiNAService from aiohttp import ClientSession from requests.utils import cookiejar_from_dict import time import json import asyncio import logging   COMMAND = {     "投影仪对焦": ["/root/projector.py focus", "已对焦"],     "投影仪音量加": ["/root/projector.py volumn1", "已增加音量"],     "投影仪音量减": ["/root/projector.py volumn2", "已降低音量"],     "投影仪关机": ["/root/projector.py poweroff", "已关机"],     "打开主机": ["wakeonlan xx:xx:xx:xx:xx:xx", "已开启主机"],     "关闭主机": ["ssh -o ConnectTimeout=1 [email]root@192.168.123.60[/email] shutdown -h now", "已关闭主机"],     "打开桌面": ["/root/tools/switch/switch.py on", "已打开桌面电脑"],     "关闭桌面": [         "ssh -o ConnectTimeout=1 [email]root@192.168.123.100[/email] shutdown -h now; sleep 30; /root/tools/switch/switch.py off",         "已关闭桌面电脑"] }   LATEST_ASK_API = "https://userprofile.mina.mi.com/device_profile/v2/conversation?source=dialogu&hardware={hardware}&timestamp={timestamp}&limit=1"   # 创建一个logger logger = logging.getLogger(__name__) logger.setLevel(logging.INFO) handler = logging.StreamHandler(sys.stdout) handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')) logger.addHandler(handler)     async def main():     username = "小米账号手机号"     password = "小米账号密码"     hardware = "L06A"  # 音箱型号     while True:         try:             misession = ClientSession()             account = MiAccount(misession, username, password, "mi.token")             await account.login("micoapi")             service = MiNAService(account)             device_list = await service.device_list()             deviceID = device_list[0]['deviceID']               async with ClientSession() as session:                 cookies = dict(                     deviceId=deviceID,                     serviceToken=service.account.token['micoapi'][1],                     userId=service.account.token['userId']                 )                 misession.cookie_jar.update_cookies(cookiejar_from_dict(cookies))                 session._cookie_jar = misession.cookie_jar                 while True:                     r = await session.get(                         LATEST_ASK_API.format(hardware=hardware, timestamp=str(int(time.time() * 1000))))                     res = await r.json()                     data = json.loads(res["data"])                     record = data['records'][0]                     query = record['query']                     answer = ""                     if len(record['answers']) > 0:                         answer = record['answers'][0]['tts']['text']                     query_timestamp = record['time'] / 1000                     query_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(query_timestamp))                     logger.info("%s %s %s", query_time, query, answer)                     if abs(query_timestamp - time.time()) < 3 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(2)         except Exception as e:             logger.error(e)             time.sleep(5)   asyncio.run(main())

Arcticlyc

之前想让小爱播放特定音乐的时候发现没有，但是发现小爱训练可以播放录音但是有时长限制，于是把音乐截成几段后上传。现在好像不能播放录音了

LATEST_ASK_API 这个接口是用于获取历史记录的，请求频率太高了会不会被 ban 呀？

jslyj

厉害啊 谢谢

cjy2323

我的红米是否可以也这样操作

Yifan2007

看到别人有用树莓派自定义音乐源的，能不能直接刷小爱同学操作

BilboBaggins

小白表示膜拜，这也太厉害了

龍謹

楼主厉害，小米还可以这么玩

等到烟火也清凉

同步不行？

bluepeb

小米在使用过程中，还是感觉可定制化程度偏弱，有了这个可以试试。

lnxctz

谢谢，小爱音箱吃灰好几年了。不知道怎么利用