Alexa 小组件安装问题诊断与 DataStore API 实践指南

def _post_put_namespace(datastore_namespace, access_token=None):
    if access_token == None:
        access_token = get_access_token()
    # 问题：'type' 和 'namespace' 被错误地作为 HTTP headers 发送
    requests.post(
        url='https://api.eu.amazonalexa.com/v1/datastore/commands',
        headers={'access_token': get_access_token(), # 问题：access_token 也应该在 Authorization 头中
            'type': 'PUT_NAMESPACE',
            'namespace': datastore_namespace
            }
        )

def _post_put_object(namespace, key, content, access_token=None):
    if access_token == None:
        access_token = get_access_token()
    # 问题：'type', 'namespace', 'key', 'content' 被错误地作为 HTTP headers 发送
    # 问题：content 是字符串，但 API 期望 JSON 对象
    requests.post(
        url='https://api.eu.amazonalexa.com/v1/datastore/commands',
        headers={"access_token": get_access_token(), # 问题：access_token 也应该在 Authorization 头中
                'type': 'PUT_OBJECT',
                'namespace': namespace,
                'key': key,
                'content': content
            }
        )

import requests
import json # 导入 json 模块

# 假设 get_access_token() 函数能够正确获取访问令牌
def get_access_token():
    # ... (您的获取 access_token 的逻辑)
    return "YOUR_ACCESS_TOKEN" # 替换为实际获取到的令牌

def _post_datastore_command(command_body, access_token=None):
    """
    通用函数，用于向 DataStore API 发送命令。
    """
    if access_token is None:
        access_token = get_access_token()

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }

    response = requests.post(
        url='https://api.eu.amazonalexa.com/v1/datastore/commands',
        headers=headers,
        data=json.dumps(command_body) # 将命令体序列化为 JSON 字符串
    )
    response.raise_for_status() # 如果请求失败，抛出 HTTPError
    return response.json()

def put_namespace_to_datastore(datastore_namespace, access_token=None):
    """
    向 DataStore 创建或更新命名空间。
    """
    command_body = {
        "type": "PUT_NAMESPACE",
        "namespace": datastore_namespace
    }
    return _post_datastore_command(command_body, access_token)

def put_object_to_datastore(namespace, key, content_object, access_token=None):
    """
    向 DataStore 存储一个对象。
    content_object 应该是一个 Python 字典或列表，将被序列化为 JSON。
    """
    command_body = {
        "type": "PUT_OBJECT",
        "namespace": namespace,
        "key": key,
        "content": content_object # 直接传递 Python 对象
    }
    return _post_datastore_command(command_body, access_token)

def put_bin_data_to_datastore(postcode, bin_colours):
    access_token = get_access_token() # 仅获取一次 access_token

    # 首先确保命名空间存在
    put_namespace_to_datastore('widget_bin_datastore', access_token=access_token)

    # 准备要存储的数据对象
    data_to_store = {
        "postcode": postcode,
        "bin_text": ", ".join(bin_colours)
    }

    # 存储数据对象
    put_object_to_datastore(
        namespace='widget_bin_datastore', 
        key='binData', 
        content_object=data_to_store, # 传递 Python 字典
        access_token=access_token
    )

import datetime
from ask_sdk_core.skill_builder import SkillBuilder
from ask_sdk_core.handler_input import HandlerInput
from ask_sdk_core.utils import is_request_type, is_intent_name
from ask_sdk_model import Response
from ask_sdk_core.dispatch_components import AbstractRequestHandler

# 假设上述修正后的 DataStore API 辅助函数已定义
# from .datastore_utils import put_bin_data_to_datastore, get_access_token

class WidgetRefreshHandler(AbstractRequestHandler):
    def can_handle(self, handler_input):
        # type: (HandlerInput) -> bool
        return _check_user_events(
            handler_input,
            [
                ['widgetRefresh'],
                ['foreground'],
                ['background']
            ]
            ) or (
                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)
                )

    def handle(self, handler_input):
        # type: (HandlerInput) -> Response
        attr = handler_input.attributes_manager.persistent_attributes
        postcode = attr.get('postcode')
        saved_collections = attr.get('saved_collections')

        bin_colours = [] # 初始化为空列表，以防未找到匹配日期
        timestamp_now = datetime.datetime.now().timestamp()
        for saved_collection_day in saved_collections:
            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

        # 调用修正后的 DataStore 数据推送函数
        put_bin_data_to_datastore(postcode=postcode, bin_colours=bin_colours)

        return handler_input.response_builder.response