Alexa Widget安装问题诊断与DataStore API交互指南

class WidgetRefreshHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        # ... 其他条件 ...
        return (
            is_request_type("Alexa.DataStore.PackageManager.UsagesInstalled")(handler_input)
        ) or (
            is_request_type("Alexa.DataStore.PackageManager.UpdateRequest")(handler_input)
        ) or (
            is_intent_name('WidgetRefreshIntent')(handler_input)
        )

import requests
import json
import logging # 引入日志模块，便于调试

# 配置日志
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

def get_access_token():
    """
    获取Alexa DataStore API所需的访问令牌。
    此函数应安全地处理客户端凭据。
    """
    try:
        response = requests.post("https://api.amazon.com/auth/o2/token", {
            "grant_type": "client_credentials",
            "client_id": "[您的Client ID]", # 请替换为实际的Client ID
            "client_secret": "[您的Client Secret]", # 请替换为实际的Client Secret
            "scope": "alexa::datastore"
        })
        response.raise_for_status() # 对HTTP错误抛出异常
        return response.json()['access_token']
    except requests.exceptions.RequestException as e:
        logger.error(f"获取访问令牌失败: {e}")
        raise

def _post_datastore_command(command_body, access_token):
    """
    向Alexa DataStore API发送通用命令。
    """
    headers = {
        'Authorization': f'Bearer {access_token}', # 正确的Authorization头
        'Content-Type': 'application/json'          # 必须指定Content-Type
    }
    api_url = 'https://api.eu.amazonalexa.com/v1/datastore/commands' # 根据您的技能区域调整URL

    try:
        response = requests.post(
            url=api_url,
            headers=headers,
            data=json.dumps(command_body) # 将命令体序列化为JSON字符串
        )
        response.raise_for_status() # 对HTTP错误（如4xx, 5xx）抛出异常
        logger.info(f"DataStore命令发送成功: {command_body}")
        return response
    except requests.exceptions.RequestException as e:
        logger.error(f"发送DataStore命令失败: {command_body}, 错误: {e}, 响应: {e.response.text if e.response else '无响应'}")
        raise

def _put_namespace(datastore_namespace, access_token):
    """
    在DataStore中创建或更新命名空间。
    """
    command_body = {
        "command": {
            "type": "PUT_NAMESPACE",
            "namespace": datastore_namespace
        }
    }
    _post_datastore_command(command_body, access_token)

def _put_object(namespace, key, content, access_token):
    """
    在DataStore中创建或更新对象。
    content 必须是一个Python字典或列表，将被序列化为JSON对象。
    """
    if not isinstance(content, (dict, list)):
        raise ValueError("Content for PUT_OBJECT must be a Python dictionary or list.")

    command_body = {
        "command": {
            "type": "PUT_OBJECT",
            "namespace": namespace,
            "key": key,
            "content": content # 直接传递Python对象，json.dumps会处理
        }
    }
    _post_datastore_command(command_body, access_token)

def put_bin_data_to_datastore(postcode, bin_colours):
    """
    将垃圾桶数据存入DataStore。
    """
    access_token = get_access_token() # 只获取一次访问令牌

    # 确保在后续调用中传递access_token，避免重复获取
    _put_namespace('widget_bin_datastore', access_token=access_token)
    _put_object(
        namespace='widget_bin_datastore',
        key='binData',
        content={ # 直接传递Python字典
            "postcode": postcode,
            "bin_text": ", ".join(bin_colours)
        },
        access_token=access_token # 传递已获取的令牌
    )
    logger.info(f"垃圾桶数据 {postcode} 已成功存入DataStore.")

import datetime
# ... 其他必要的导入 ...

class WidgetRefreshHandler(AbstractRequestHandler):
    # ... can_handle 方法不变 ...

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        logger.info("进入 WidgetRefreshHandler 的 handle 方法")
        attr = handler_input.attributes_manager.persistent_attributes
        postcode = attr.get('postcode')
        saved_collections = attr.get('saved_collections')

        # 确保 postcode 和 saved_collections 在安装时可用
        if not postcode or not saved_collections:
            logger.error("安装时缺少 postcode 或 saved_collections 数据。")
            # 可以选择抛出异常或返回错误响应
            return handler_input.response_builder.speak("初始化小部件数据失败，请稍后再试。").response

        bin_colours = []
        timestamp_now = datetime.datetime.now().timestamp()
        for saved_collection_day in saved_collections:
            try:
                collection_timestamp = datetime.datetime.strptime(saved_collection_day, "%A %d %B %Y  %H:%M:%S").timestamp()
                if timestamp_now < collection_timestamp:
                    bin_colours = saved_collections[saved_collection_day]
                    break
            except ValueError:
                logger.warning(f"无法解析日期字符串: {saved_collection_day}")
                continue

        try:
            put_bin_data_to_datastore(postcode=postcode, bin_colours=bin_colours)
            logger.info("DataStore 数据初始化成功。")
        except Exception as e:
            logger.error(f"在安装时向DataStore写入数据失败: {e}")
            # 根据需要处理错误，例如返回一个表示失败的响应
            return handler_input.response_builder.speak("初始化小部件数据时发生错误。").response

        return handler_input.response_builder.response