Xiaomi Cloud Tokens Extractor从设备令牌提取到智能家居生态扩展【免费下载链接】Xiaomi-cloud-tokens-extractorThis tool retrieves tokens for all devices connected to Xiaomi cloud and encryption keys for BLE devices.项目地址: https://gitcode.com/gh_mirrors/xia/Xiaomi-cloud-tokens-extractor快速导航开发痛点为什么需要自定义扩展架构解密模块化设计的智慧认证扩展突破传统登录限制数据处理性能优化实战输出定制灵活适配各种场景进阶技巧打造企业级应用部署集成从开发到生产开发痛点为什么需要自定义扩展在智能家居开发领域小米设备占据了重要地位但开发者常常面临这样的困境官方API限制严格、文档不全、设备通信协议复杂。当你需要批量管理设备、构建自定义控制面板或集成到第三方系统时标准的SDK往往无法满足需求。典型痛点包括认证方式单一无法适配企业级SSO或OAuth流程设备数据获取效率低下无法支持大规模设备管理输出格式固定难以与现有系统集成缺乏错误重试和容错机制注意小米云服务有每日请求限制3-5次/天在设计扩展功能时需要考虑这一点。架构解密模块化设计的智慧Xiaomi Cloud Tokens Extractor采用清晰的抽象层设计将核心功能解耦为可独立扩展的模块。让我们通过代码结构来理解这一设计理念# 核心抽象类定义 class XiaomiCloudConnector(ABC): def __init__(self): self._agent self.generate_agent() self._device_id self.generate_device_id() self._session requests.session() self._ssecurity None self.userId None self._serviceToken None abstractmethod def login(self) - bool: pass def get_api_url(self, country: str) - str: return fhttps://{country}.api.io.mi.com项目采用策略模式实现认证模块通过抽象基类定义接口具体实现类提供不同认证方式。这种设计使得添加新的认证方式变得异常简单。模块关系图认证层 (XiaomiCloudConnector) ├── 密码认证 (PasswordXiaomiCloudConnector) ├── QR码认证 (QrCodeXiaomiCloudConnector) └── [可扩展] OAuth认证、API密钥认证等 数据处理层 ├── 设备信息获取 ├── 令牌解析 └── BLE密钥提取 输出层 ├── 控制台输出 ├── JSON文件保存 └── [可扩展] 数据库存储、API推送等认证扩展突破传统登录限制扩展点1自定义认证方式假设你需要集成企业SSO系统可以创建新的认证类class SSOXiaomiCloudConnector(XiaomiCloudConnector): def __init__(self, sso_token: str): super().__init__() self.sso_token sso_token def login(self) - bool: 使用SSO令牌进行认证 try: # 验证SSO令牌有效性 sso_valid self._validate_sso_token() if not sso_valid: return False # 将SSO令牌转换为小米云会话 return self._exchange_sso_for_xiaomi_session() except Exception as e: _LOGGER.error(fSSO认证失败: {e}) return False def _validate_sso_token(self) - bool: # 实现SSO令牌验证逻辑 # 可以调用企业内部认证服务 pass def _exchange_sso_for_xiaomi_session(self) - bool: # 将SSO会话转换为小米云服务令牌 pass扩展点2多因素认证支持对于安全性要求更高的场景可以增强现有认证流程class MFAXiaomiCloudConnector(PasswordXiaomiCloudConnector): def __init__(self, username: str, password: str, mfa_provider: str totp): super().__init__(username, password) self.mfa_provider mfa_provider def login(self) - bool: # 先进行基础密码认证 if not super().login(): return False # 触发多因素认证 if self._requires_mfa(): return self._perform_mfa_verification() return True def _perform_mfa_verification(self) - bool: if self.mfa_provider totp: totp_code input(请输入TOTP验证码: ) return self._verify_totp(totp_code) elif self.mfa_provider sms: return self._request_sms_verification() # 支持更多MFA方式扩展思路可以考虑集成硬件安全密钥如YubiKey、生物识别认证或基于时间的访问令牌。数据处理性能优化实战批量设备信息获取优化默认实现逐个获取设备信息对于大量设备效率较低。我们可以实现并行处理import concurrent.futures from typing import List, Dict class ParallelDeviceFetcher: def __init__(self, connector: XiaomiCloudConnector, max_workers: int 5): self.connector connector self.max_workers max_workers def get_all_devices_parallel(self, countries: List[str]) - Dict[str, List[Dict]]: 并行获取所有国家服务器的设备信息 results {} with concurrent.futures.ThreadPoolExecutor(max_workersself.max_workers) as executor: # 为每个国家创建任务 future_to_country { executor.submit(self.connector.get_devices, country): country for country in countries } # 收集结果 for future in concurrent.futures.as_completed(future_to_country): country future_to_country[future] try: devices future.result() results[country] devices except Exception as e: _LOGGER.error(f获取{country}服务器设备失败: {e}) results[country] [] return results缓存机制实现为减少API调用频率可以添加缓存层import pickle from datetime import datetime, timedelta from pathlib import Path class DeviceCache: def __init__(self, cache_dir: str .cache, ttl_hours: int 24): self.cache_dir Path(cache_dir) self.cache_dir.mkdir(exist_okTrue) self.ttl timedelta(hoursttl_hours) def get_cached_devices(self, username: str, server: str) - Optional[List[Dict]]: 获取缓存的设备信息 cache_file self.cache_dir / f{username}_{server}.pkl if not cache_file.exists(): return None # 检查缓存是否过期 mtime datetime.fromtimestamp(cache_file.stat().st_mtime) if datetime.now() - mtime self.ttl: return None try: with open(cache_file, rb) as f: return pickle.load(f) except: return None def save_devices(self, username: str, server: str, devices: List[Dict]): 保存设备信息到缓存 cache_file self.cache_dir / f{username}_{server}.pkl with open(cache_file, wb) as f: pickle.dump(devices, f)兼容性说明缓存实现使用pickle序列化确保Python版本一致。对于跨版本兼容性建议使用JSON格式。输出定制灵活适配各种场景扩展点多格式输出支持默认输出为JSON格式我们可以扩展支持更多格式from abc import ABC, abstractmethod import csv import xml.etree.ElementTree as ET from typing import List, Dict class OutputFormatter(ABC): abstractmethod def format(self, devices: List[Dict]) - str: pass class JSONFormatter(OutputFormatter): def format(self, devices: List[Dict]) - str: import json return json.dumps(devices, indent2, ensure_asciiFalse) class CSVFormatter(OutputFormatter): def format(self, devices: List[Dict]) - str: import io if not devices: return # 提取所有可能的字段 all_fields set() for device in devices: all_fields.update(device.keys()) output io.StringIO() writer csv.DictWriter(output, fieldnamessorted(all_fields)) writer.writeheader() writer.writerows(devices) return output.getvalue() class XMLFormatter(OutputFormatter): def format(self, devices: List[Dict]) - str: import xml.etree.ElementTree as ET root ET.Element(devices) for device in devices: device_elem ET.SubElement(root, device) for key, value in device.items(): field ET.SubElement(device_elem, key) field.text str(value) return ET.tostring(root, encodingunicode, methodxml) class OutputManager: def __init__(self): self.formatters { json: JSONFormatter(), csv: CSVFormatter(), xml: XMLFormatter() } def format_output(self, devices: List[Dict], format_type: str json) - str: formatter self.formatters.get(format_type.lower()) if not formatter: raise ValueError(f不支持的格式: {format_type}) return formatter.format(devices)实时推送扩展除了文件输出还可以实现实时数据推送class WebhookPusher: def __init__(self, webhook_url: str, auth_token: str None): self.webhook_url webhook_url self.auth_token auth_token def push_devices(self, devices: List[Dict]): 将设备信息推送到Webhook import requests headers {} if self.auth_token: headers[Authorization] fBearer {self.auth_token} try: response requests.post( self.webhook_url, json{devices: devices}, headersheaders, timeout10 ) response.raise_for_status() return True except requests.RequestException as e: _LOGGER.error(fWebhook推送失败: {e}) return False class MQTTPublisher: def __init__(self, broker: str, topic_prefix: str xiaomi/devices): import paho.mqtt.client as mqtt self.client mqtt.Client() self.broker broker self.topic_prefix topic_prefix def publish_device(self, device: Dict): 通过MQTT发布设备信息 device_id device.get(did, unknown) topic f{self.topic_prefix}/{device_id} self.client.connect(self.broker) self.client.publish(topic, json.dumps(device)) self.client.disconnect()进阶技巧打造企业级应用异步处理实现对于需要处理大量设备或实时更新的场景异步处理可以显著提升性能import asyncio import aiohttp from typing import List, Dict class AsyncXiaomiCloudConnector: def __init__(self, session: aiohttp.ClientSession): self.session session self.base_headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept: application/json, } async def login_async(self, username: str, password: str) - bool: 异步登录实现 # 实现异步登录逻辑 pass async def get_devices_async(self, country: str) - List[Dict]: 异步获取设备信息 url fhttps://{country}.api.io.mi.com/api/home/device_list async with self.session.get(url, headersself.base_headers) as response: if response.status 200: data await response.json() return self._parse_devices(data) return [] async def get_all_devices_async(self, countries: List[str]) - Dict[str, List[Dict]]: 并行获取所有服务器的设备 tasks [self.get_devices_async(country) for country in countries] results await asyncio.gather(*tasks, return_exceptionsTrue) devices_by_country {} for country, result in zip(countries, results): if isinstance(result, Exception): _LOGGER.error(f获取{country}设备失败: {result}) devices_by_country[country] [] else: devices_by_country[country] result return devices_by_country错误处理与重试机制健壮的企业应用需要完善的错误处理from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import requests.exceptions class ResilientDeviceFetcher: def __init__(self, connector: XiaomiCloudConnector, max_retries: int 3): self.connector connector self.max_retries max_retries retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10), retryretry_if_exception_type(( requests.exceptions.ConnectionError, requests.exceptions.Timeout, requests.exceptions.HTTPError )) ) def get_devices_with_retry(self, country: str) - List[Dict]: 带重试机制的设备获取 try: return self.connector.get_devices(country) except requests.exceptions.HTTPError as e: if e.response.status_code 429: # 请求过多 _LOGGER.warning(f请求过多等待重试...) raise elif e.response.status_code 500: # 服务器错误 _LOGGER.error(f服务器错误: {e.response.status_code}) raise else: # 客户端错误不重试 _LOGGER.error(f客户端错误: {e}) return []测试策略建议为确保扩展功能的稳定性建议采用分层测试策略# 单元测试示例 import pytest from unittest.mock import Mock, patch class TestSSOXiaomiCloudConnector: def test_sso_token_validation(self): 测试SSO令牌验证 connector SSOXiaomiCloudConnector(valid_sso_token) with patch.object(connector, _validate_sso_token, return_valueTrue): assert connector._validate_sso_token() is True def test_sso_exchange_failure(self): 测试SSO令牌交换失败场景 connector SSOXiaomiCloudConnector(invalid_token) with patch.object(connector, _validate_sso_token, return_valueFalse): result connector.login() assert result is False # 集成测试示例 class TestIntegration: pytest.fixture def mock_cloud_connector(self): 创建模拟的云连接器 connector Mock(specXiaomiCloudConnector) connector.get_devices.return_value [ {did: device1, name: Test Device, token: abc123} ] return connector def test_parallel_fetcher_integration(self, mock_cloud_connector): 测试并行获取器集成 fetcher ParallelDeviceFetcher(mock_cloud_connector) devices fetcher.get_all_devices_parallel([cn, de]) assert cn in devices assert de in devices assert len(devices[cn]) 1部署集成从开发到生产Docker容器化扩展基于现有的Dockerfile我们可以创建支持扩展功能的镜像# 基础镜像 FROM python:3.9-slim # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ g \ rm -rf /var/lib/apt/lists/* # 创建工作目录 WORKDIR /app # 复制依赖文件 COPY requirements.txt . # 安装Python依赖 RUN pip install --no-cache-dir -r requirements.txt # 安装扩展依赖 RUN pip install --no-cache-dir \ aiohttp \ paho-mqtt \ tenacity # 复制源代码 COPY token_extractor.py . COPY extensions/ ./extensions/ # 设置环境变量 ENV PYTHONUNBUFFERED1 # 创建非root用户 RUN useradd -m -u 1000 appuser chown -R appuser:appuser /app USER appuser # 启动命令 ENTRYPOINT [python, token_extractor.py]CI/CD流水线配置使用GitHub Actions实现自动化测试和部署name: Build and Test on: push: branches: [ main, develop ] pull_request: branches: [ main ] jobs: test: runs-on: ubuntu-latest strategy: matrix: python-version: [3.8, 3.9, 3.10] steps: - uses: actions/checkoutv2 - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-pythonv2 with: python-version: ${{ matrix.python-version }} - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt pip install pytest pytest-cov - name: Run tests run: | pytest tests/ -v --covextensions --cov-reportxml - name: Upload coverage uses: codecov/codecov-actionv2 with: file: ./coverage.xml build-docker: needs: test runs-on: ubuntu-latest steps: - uses: actions/checkoutv2 - name: Set up Docker Buildx uses: docker/setup-buildx-actionv1 - name: Login to DockerHub uses: docker/login-actionv1 with: username: ${{ secrets.DOCKER_USERNAME }} password: ${{ secrets.DOCKER_PASSWORD }} - name: Build and push uses: docker/build-push-actionv2 with: context: . push: true tags: | yourusername/xiaomi-tokens-extractor:latest yourusername/xiaomi-tokens-extractor:${{ github.sha }}监控与日志配置为生产环境添加监控和日志功能import logging import structlog from prometheus_client import Counter, Gauge, start_http_server # 配置结构化日志 structlog.configure( processors[ structlog.stdlib.filter_by_level, structlog.stdlib.add_logger_name, structlog.stdlib.add_log_level, structlog.stdlib.PositionalArgumentsFormatter(), structlog.processors.TimeStamper(fmtiso), structlog.processors.StackInfoRenderer(), structlog.processors.format_exc_info, structlog.processors.UnicodeDecoder(), structlog.processors.JSONRenderer() ], context_classdict, logger_factorystructlog.stdlib.LoggerFactory(), wrapper_classstructlog.stdlib.BoundLogger, cache_logger_on_first_useTrue, ) # Prometheus指标 DEVICES_FETCHED Counter(xiaomi_devices_fetched_total, Total devices fetched) FETCH_ERRORS Counter(xiaomi_fetch_errors_total, Total fetch errors) ACTIVE_CONNECTIONS Gauge(xiaomi_active_connections, Active connections to Xiaomi cloud) class MonitoredXiaomiCloudConnector(XiaomiCloudConnector): def get_devices(self, country: str) - List[Dict]: 监控增强的设备获取方法 ACTIVE_CONNECTIONS.inc() try: devices super().get_devices(country) DEVICES_FETCHED.inc(len(devices)) return devices except Exception as e: FETCH_ERRORS.inc() raise finally: ACTIVE_CONNECTIONS.dec() # 启动监控服务器 def start_monitoring(port: int 8000): start_http_server(port) logging.info(f监控服务器已启动端口: {port})配置管理最佳实践使用环境变量和配置文件管理不同环境的设置import os from dataclasses import dataclass from typing import Optional import yaml dataclass class AppConfig: # 认证配置 username: Optional[str] None password: Optional[str] None use_qr_code: bool False # 服务器配置 servers: list None preferred_server: Optional[str] None # 输出配置 output_format: str json output_file: Optional[str] None enable_webhook: bool False webhook_url: Optional[str] None # 性能配置 max_workers: int 5 cache_enabled: bool True cache_ttl_hours: int 24 # 监控配置 enable_monitoring: bool False monitoring_port: int 8000 classmethod def from_env(cls) - AppConfig: 从环境变量加载配置 return cls( usernameos.getenv(XIAOMI_USERNAME), passwordos.getenv(XIAOMI_PASSWORD), use_qr_codeos.getenv(USE_QR_CODE, false).lower() true, serversos.getenv(SERVERS, cn,de,us).split(,), output_formatos.getenv(OUTPUT_FORMAT, json), enable_webhookos.getenv(ENABLE_WEBHOOK, false).lower() true, webhook_urlos.getenv(WEBHOOK_URL), max_workersint(os.getenv(MAX_WORKERS, 5)), cache_enabledos.getenv(CACHE_ENABLED, true).lower() true, enable_monitoringos.getenv(ENABLE_MONITORING, false).lower() true, ) classmethod def from_yaml(cls, config_path: str) - AppConfig: 从YAML文件加载配置 with open(config_path, r) as f: config_data yaml.safe_load(f) return cls(**config_data)扩展思路可以进一步集成配置中心如Consul、Etcd实现动态配置更新和热重载。总结通过本文的扩展指南你已经掌握了如何将Xiaomi Cloud Tokens Extractor从一个简单的令牌提取工具转变为功能强大的智能家居管理平台。关键在于理解项目的模块化架构并在适当的扩展点上添加自定义功能。记住这些核心原则保持向后兼容扩展时尽量不破坏现有接口模块化设计每个扩展功能应该是独立的、可测试的模块错误处理为所有扩展功能添加适当的错误处理和日志记录配置驱动通过配置文件或环境变量管理扩展功能随着智能家居生态的不断发展这样的扩展能力将帮助你快速适应新的需求和挑战。无论是构建企业级设备管理平台还是开发个性化的智能家居应用这个工具都能成为你强大的技术基础。最后提醒在使用任何扩展功能时请确保遵守小米云服务的使用条款并尊重用户隐私和数据安全。【免费下载链接】Xiaomi-cloud-tokens-extractorThis tool retrieves tokens for all devices connected to Xiaomi cloud and encryption keys for BLE devices.项目地址: https://gitcode.com/gh_mirrors/xia/Xiaomi-cloud-tokens-extractor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
