import base64
import hashlib
import hmac
import json
import os
import random
import time
from sys import platform
from Crypto.Cipher import ARC4import requestsif platform != "win32":import readlineclass XiaomiCloudConnector:def __init__(self, username, password):self._username = usernameself._password = passwordself._agent = self.generate_agent()self._device_id = self.generate_device_id()self._session = requests.session()self._sign = Noneself._ssecurity = Noneself._userId = Noneself._cUserId = Noneself._passToken = Noneself._location = Noneself._code = Noneself._serviceToken = Nonedef login_step_1(self):url = "https://account.xiaomi.com/pass/serviceLogin?sid=xiaomiio&_json=true"headers = {"User-Agent": self._agent,"Content-Type": "application/x-www-form-urlencoded"}cookies = {"userId": self._username}response = self._session.get(url, headers=headers, cookies=cookies)valid = response.status_code == 200 and "_sign" in self.to_json(response.text)if valid:self._sign = self.to_json(response.text)["_sign"]return validdef login_step_2(self):url = "https://account.xiaomi.com/pass/serviceLoginAuth2"headers = {"User-Agent": self._agent,"Content-Type": "application/x-www-form-urlencoded"}fields = {"sid": "xiaomiio","hash": hashlib.md5(str.encode(self._password)).hexdigest().upper(),"callback": "https://sts.api.io.mi.com/sts","qs": "%3Fsid%3Dxiaomiio%26_json%3Dtrue","user": self._username,"_sign": self._sign,"_json": "true"}response = self._session.post(url, headers=headers, params=fields)valid = response is not None and response.status_code == 200if valid:json_resp = self.to_json(response.text)valid = "ssecurity" in json_resp and len(str(json_resp["ssecurity"])) > 4if valid:self._ssecurity = json_resp["ssecurity"]self._userId = json_resp["userId"]self._cUserId = json_resp["cUserId"]self._passToken = json_resp["passToken"]self._location = json_resp["location"]self._code = json_resp["code"]else:if "notificationUrl" in json_resp:print("Two factor authentication required, please use following url and restart extractor:")print(json_resp["notificationUrl"])print()return validdef login_step_3(self):headers = {"User-Agent": self._agent,"Content-Type": "application/x-www-form-urlencoded"}response = self._session.get(self._location, headers=headers)if response.status_code == 200:self._serviceToken = response.cookies.get("serviceToken")return response.status_code == 200def login(self):self._session.cookies.set("sdkVersion", "accountsdk-18.8.15", domain="mi.com")self._session.cookies.set("sdkVersion", "accountsdk-18.8.15", domain="xiaomi.com")self._session.cookies.set("deviceId", self._device_id, domain="mi.com")self._session.cookies.set("deviceId", self._device_id, domain="xiaomi.com")if self.login_step_1():if self.login_step_2():if self.login_step_3():return Trueelse:print("Unable to get service token.")else:print("Invalid login or password.")else:print("Invalid username.")return Falsedef get_devices(self, country):url = self.get_api_url(country) + "/home/device_list"params = {"data": '{"getVirtualModel":true,"getHuamiDevices":1,"get_split_device":false,"support_smart_home":true}'}return self.execute_api_call_encrypted(url, params)def get_beaconkey(self, country, did):url = self.get_api_url(country) + "/v2/device/blt_get_beaconkey"params = {"data": '{"did":"' + did + '","pdid":1}'}return self.execute_api_call_encrypted(url, params)def execute_api_call_encrypted(self, url, params):headers = {"Accept-Encoding": "identity","User-Agent": self._agent,"Content-Type": "application/x-www-form-urlencoded","x-xiaomi-protocal-flag-cli": "PROTOCAL-HTTP2","MIOT-ENCRYPT-ALGORITHM": "ENCRYPT-RC4",}cookies = {"userId": str(self._userId),"yetAnotherServiceToken": str(self._serviceToken),"serviceToken": str(self._serviceToken),"locale": "en_GB","timezone": "GMT+02:00","is_daylight": "1","dst_offset": "3600000","channel": "MI_APP_STORE"}millis = round(time.time() * 1000)nonce = self.generate_nonce(millis)signed_nonce = self.signed_nonce(nonce)fields = self.generate_enc_params(url, "POST", signed_nonce, nonce, params, self._ssecurity)response = self._session.post(url, headers=headers, cookies=cookies, params=fields)if response.status_code == 200:decoded = self.decrypt_rc4(self.signed_nonce(fields["_nonce"]), response.text)return json.loads(decoded)return Nonedef get_api_url(self, country):return "https://" + ("" if country == "cn" else (country + ".")) + "api.io.mi.com/app"def signed_nonce(self, nonce):hash_object = hashlib.sha256(base64.b64decode(self._ssecurity) + base64.b64decode(nonce))return base64.b64encode(hash_object.digest()).decode('utf-8')@staticmethoddef generate_nonce(millis):nonce_bytes = os.urandom(8) + (int(millis / 60000)).to_bytes(4, byteorder='big')return base64.b64encode(nonce_bytes).decode()@staticmethoddef generate_agent():agent_id = "".join(map(lambda i: chr(i), [random.randint(65, 69) for _ in range(13)]))return f"Android-7.1.1-1.0.0-ONEPLUS A3010-136-{agent_id} APP/xiaomi.smarthome APPV/62830"@staticmethoddef generate_device_id():return "".join(map(lambda i: chr(i), [random.randint(97, 122) for _ in range(6)]))@staticmethoddef generate_signature(url, signed_nonce, nonce, params):signature_params = [url.split("com")[1], signed_nonce, nonce]for k, v in params.items():signature_params.append(f"{k}={v}")signature_string = "&".join(signature_params)signature = hmac.new(base64.b64decode(signed_nonce), msg=signature_string.encode(), digestmod=hashlib.sha256)return base64.b64encode(signature.digest()).decode()@staticmethoddef generate_enc_signature(url, method, signed_nonce, params):signature_params = [str(method).upper(), url.split("com")[1].replace("/app/", "/")]for k, v in params.items():signature_params.append(f"{k}={v}")signature_params.append(signed_nonce)signature_string = "&".join(signature_params)return base64.b64encode(hashlib.sha1(signature_string.encode('utf-8')).digest()).decode()@staticmethoddef generate_enc_params(url, method, signed_nonce, nonce, params, ssecurity):params['rc4_hash__'] = XiaomiCloudConnector.generate_enc_signature(url, method, signed_nonce, params)for k, v in params.items():params[k] = XiaomiCloudConnector.encrypt_rc4(signed_nonce, v)params.update({'signature': XiaomiCloudConnector.generate_enc_signature(url, method, signed_nonce, params),'ssecurity': ssecurity,'_nonce': nonce,})return params@staticmethoddef to_json(response_text):return json.loads(response_text.replace("&&&START&&&", ""))@staticmethoddef encrypt_rc4(password, payload):r = ARC4.new(base64.b64decode(password))r.encrypt(bytes(1024))return base64.b64encode(r.encrypt(payload.encode())).decode()@staticmethoddef decrypt_rc4(password, payload):r = ARC4.new(base64.b64decode(password))r.encrypt(bytes(1024))return r.encrypt(base64.b64decode(payload))def print_tabbed(value, tab):print(" " * tab + value)def print_entry(key, value, tab):if value:print_tabbed(f'{key + ":": <10}{value}', tab)username = "小米ID"
password ="用户密码"
servers = ["cn"]connector = XiaomiCloudConnector(username, password)
print("Logging in...")
logged = connector.login()
if logged:print("Logged in.")print()for current_server in servers:devices = connector.get_devices(current_server)if devices is not None:if len(devices["result"]["list"]) == 0:print(f"No devices found for server \"{current_server}\".")continueprint(f"Devices found for server \"{current_server}\":")for device in devices["result"]["list"]:print_tabbed("---------", 3)if "name" in device:print_entry("NAME", device["name"], 3)if "did" in device:print_entry("ID", device["did"], 3)if "blt" in device["did"]:beaconkey = connector.get_beaconkey(current_server, device["did"])if beaconkey and "result" in beaconkey and "beaconkey" in beaconkey["result"]:print_entry("BLE KEY", beaconkey["result"]["beaconkey"], 3)if "mac" in device:print_entry("MAC", device["mac"], 3)if "localip" in device:print_entry("IP", device["localip"], 3)if "token" in device:print_entry("TOKEN", device["token"], 3)if "model" in device:print_entry("MODEL", device["model"], 3)print_tabbed("---------", 3)print()else:print(f"Unable to get devices from server {current_server}.")
else:print("Unable to log in.")

pip install python-miio

import miio
import timeip='192.168.1.100'
token='...................................' s = miio.device.Device(ip=ip, token=token)#设置为关闭状态:
ret=s.raw_command('set_properties' ,[{'did': 'Switch', 'siid': 2, 'piid': 1, 'value':False}])
print(ret)#获取开关状态:
ret=s.raw_command('get_properties' ,[{'did': 'Switch', 'siid': 2, 'piid': 1}])
print(ret)#打开:
ret=s.raw_command('set_properties' ,[{'did': 'Switch', 'siid': 2, 'piid': 1, 'value':True}])
print(ret)#获取功耗
for _ in range(10):        ret=s.raw_command('get_properties' ,[{'did': 'power_consumption', 'piid':6, 'siid': 5}])watt=ret[0]['value']/100.0print("{}.W".format(watt))time.sleep(3)#关闭:
ret=s.raw_command('set_properties' ,[{'did': 'Switch', 'siid': 2, 'piid': 1, 'value':False}])
print(ret) 

import miio
import timeip='192.168.1.100'
token='xxxxxxxxxxxxxxxxxxxxxxxxx' def set_switch_status(value):s = miio.device.Device(ip=ip, token=token)ret=s.raw_command('set_properties' ,[{'did': 'Switch', 'siid': 2, 'piid': 1, 'value':value}])print("set_switch_status:",ret)def get_power_consumption():s = miio.device.Device(ip=ip, token=token)ret=s.raw_command('get_properties' ,[{'did': 'power_consumption', 'piid': int(6), 'siid': 5}])return ret[0]['value']/100.0err=0
while True:p=get_power_consumption()print(p)if p<20:err+=1else:err=0if err>3 and p>0:set_switch_status(False)time.sleep(10)

cat <<EOF | sudo tee /etc/systemd/system/mictrl.service
[Unit]
Description=mictrl.service[Service]
Type=simple
ExecStart=python3 /home/rock/mictrl.py
Restart=always
RestartSec=10[Install]
WantedBy=multi-user.target
EOFsystemctl enable mictrl
systemctl start mictrl
systemctl status mictrl