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

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" # 音响型号

    polling_event = asyncio.Event()
    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}&timestamp={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:
            try:
                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)
            except Exception as e:
                print(e)
            time.sleep(1)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())