基于 Amazon Lambda，获取实时天气预报推送

> **本文作者：韩旭** > > 某大型云厂商技术支持 > > 亚马逊云科技技领云博主 如何能让手机推送的实时天气更加准确？本文将介绍使用国内的 API 进行尝试。 以下是本文使用的服务： * 地图 SDK 或 APP 获取经纬度。 * 彩云天气 API 通过地理位置获取天气信息。 * Amazon Lambda 作为运行环境执行 Python 代码。 * Amazon APIGatway+Amazon Lambda 代理集成。 * Amazon Eventbridge 触发定时任务。 * Server 酱推送消息。 选择 Amazon Lambda 而没有选择 [Amazon EC2 ](https://aws.amazon.com/cn/ec2/?trk=cndc-detail)的原因是，每天的请求不大（10 个以内），而且无需管理服务器，这正是[无服务器](https://aws.amazon.com/cn/serverless/?trk=cndc-detail)的优势。 本文的主要思路如下： 1.使用地图 SDK 获取当前定位的经纬度，由于这次使用了 Server 酱推无法实时获取前端地址，所以直接使用地图拾取器获取固定的坐标。下图是某地图 APP 的示例：  如果使用另一地图 APP 的话，会发现精度会更高些。  2.使用彩云 API，本文使用的是 V2.6 的稳定版，通过彩云 API 获取实时以及小时级别的天气。彩云对开发者提供了 REST API 调用。 ``` curl "https://api.caiyunapp.com/v2.6/{token}/101.6656,39.2072/realtime" ``` 3.编写 Python 代码调用这个 API，然后把拿回来的数据发送给 server 酱的 Webhook，再把这个代码部署到 Amazon Lambda 上。 server 酱的调用方式如下： ``` POST https://sctapi.ftqq.com/<token>.send?title=标题&desp=内容 ``` 转换成 Python 代码为： ``` import requests url = "https://sctapi.ftqq.com/<token>.send?title=标题&desp=内容" response = requests.request("POST", url) print(response.text) ``` 4.提供 [Amazon EventBridge](https://aws.amazon.com/cn/eventbridge/?trk=cndc-detail) 定时调用和 Amazon APIGateway 实时调用两种方式。 ### **操作步骤** #### **准备 Amazon Lambda** 先创建一个新的 Amazon Lambda 函数，使用最新的 Python3.11，因为本演示的本地 MacOS 环境是 M2 Pro，所以为了软件包更好的兼容性选择了 arm64，如果您是 Intel 芯片的 PC，当然也可以选择 X86。  接下来更改默认执行角色和高级设置，如果默认的话 Amazon Lambda 会自动创建一个角色，您可以在配置——权限——执行角色找到。默认给了写日志到 cloudwatch 的权限，如果缺少这个权限，那么在函数的 print（）和 log.info（）后将看不见任何输出。 ``` {   "Version": "2012-10-17",   "Statement": [     {       "Effect": "Allow",       "Action": "logs:CreateLogGroup",       "Resource": "arn:aws:logs:us-east-1:123456789012:*"   },     {       "Effect": "Allow",       "Action": [       "logs:CreateLogStream",     "logs:PutLogEvents"           ],         "Resource": [     "arn:amazon web services:logs:us-east-1:123456789012:log-group:/aws/lambda/caiyun:*"         ]   }   ] } ``` 信任关系如下，如果您的账户中恰好有这样信任关系的 Role，那么也可使用这个 Role 而无需新建。 ``` {   "Version": "2012-10-17",   "Statement": [   {         "Effect": "Allow",         "Principal": {         "Service": "lambda.amazonaws.com"   },         "Action": "sts:AssumeRole"   }   ] } ``` 然后是网络，默认情况下 Amazon Lambda 是 Public 的，无法连接到 VPC 内的资源，比如数据库，或者下一层的 API。不过 Amazon Lambda 可以随时在 Public 和 VPC 模式之间切换。最佳实践是至少为 Amazon Lambda 选择 2 个子网以在高可用性模式下运行。如果将一个函数连接到您账户中的 VPC 时，要提供函数对互联网的访问权限，需要将出站流量路由到公有子网中的 NAT 网关。 如果在创建 Amazon Lambda 的时候选择 VPC 模式，那么会自动加上如下策略，这个策略是允许在 Amazon Lambda 所在子网新建网卡的。 ``` {   "Version": "2012-10-17", "Statement": [   {         "Effect": "Allow",       "Action": [         "ec2:CreateNetworkInterface",   "ec2:DeleteNetworkInterface",     "ec2:DescribeNetworkInterfaces"       ],       "Resource": "*"     }   ] } ``` 下面看两种模式的对比，由于本演示不需要访问 VPC 内部的资源，所以选择了 Public 模式。  #### **准备配置** 在 Amazon Lambda 中，可以使用 Amazon Systems Manager 和 Amazon Secrets Manager 来管理环境变量。两者虽然有一些相似之处，但它们的用途和功能有所不同，以下是它们的区别： 比较在 Amazon Lambda 中存储和访问数据的方式，并加入 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail)： 1.Amazon Lambda 环境变量： * 直接将敏感信息（例如 API 密钥）或配置数据作为键值对存储在 Amazon Lambda 函数的配置中。 * 优点：简单易用，无需额外配置。 * 缺点：安全性较低，环境变量会存储在 Amazon Lambda 函数代码包中，如果代码泄露，敏感信息也会随之暴露。不适合存储高度敏感的信息。 2.Amazon Systems Manager Parameter Store： * 将配置数据（例如数据库连接字符串、API 端点）存储在 Amazon Systems Manager Parameter Store 中，该服务提供安全存储和版本控制功能。 * 优点：安全性更高，Amazon Systems Manager Parameter Store 可以使用 KMS 加密参数值，并支持版本控制和访问控制。 * 缺点：需要额外配置 Amazon Lambda 函数以访问 Amazon Systems Manager Parameter Store。 3.Amazon Secrets Manager： * 将敏感信息（例如数据库密码、API 密钥）存储在 Amazon Secrets Manager 中，该服务提供高安全性的加密存储和访问控制。 * 优点：安全性最高，Amazon Secrets Manager 使用 KMS 加密敏感信息，并提供严格的访问控制机制，支持密钥旋转功能。 * 缺点：需要额外配置 Amazon Lambda 函数以访问 Amazon Secrets Manager。 4.Amazon S3： * 将数据（例如图片、日志文件、模型文件）存储在 Amazon S3 对象存储中，该服务提供高可扩展性、持久性和安全性。 * 优点：高可扩展性、低成本、高度可用性。可以配置访问控制策略来限制对数据的访问。 * 缺点：需要额外配置 Amazon Lambda 函数以访问 Amazon S3。Amazon S3 不是专门为存储敏感信息设计的，因此需要使用其他机制（例如 KMS 加密）来保护数据安全。 以下是关键区别的表格：  建议： * 对于非关键的配置数据，例如 API 端点，可以使用 Amazon Lambda 环境变量。 * 对于需要版本控制和访问控制的配置数据，例如数据库连接字符串，可以使用 Amazon Systems Manager Parameter Store。 * 对于高安全性的信息，例如数据库密码、API 密钥等，强烈建议使用 Amazon Secrets Manager。 * 对于大型文件或不需要频繁更新的数据，例如图片、日志文件、模型文件等，可以使用 Amazon S3 对象存储。 #### **实例代码** 以下是使用 Amazon SDK 获取 Amazon Systems Manager Parameter Store 参数、Amazon Secrets Manager 密钥以及设置环境变量的 Python 代码示例： 1.从 Amazon Systems Manager Parameter Store 获取参数： ``` import boto3 # 创建 Systems Manager 客户端 ssm_client = boto3.client('ssm') # 获取参数值 response = ssm_client.get_parameter( Name='your-parameter-name',  # 替换为您的参数名称   WithDecryption=True  # 如果参数已加密，请设置为 True ) parameter_value = response['Parameter']['Value'] print(f"Parameter value: {parameter_value}") ``` 解释： * 使用 boto3 库创建 Amazon Systems Manager 客户端。 * get_parameter（）函数用于获取指定参数的值。 * WithDecryption=True 参数确保加密的参数会被解密。 * response\['Parameter']\['Value'] 包含参数值，您可以将其打印或使用。 2.从 Amazon Secrets Manager 获取密钥： ``` import boto3 # 创建 Secrets Manager 客户端 secrets_client = boto3.client('secretsmanager') # 获取 Secret 值 response = secrets_client.get_secret_value(SecretId='your-secret-id') # 解密 Secret 字符串 secret_string = response['SecretString'] print(f"Secret value: {secret_string}") ``` 解释： * 使用 boto3 库创建 Amazon Secrets Manager 客户端。 * get_secret_value（）函数用于获取指定 Secret 的值。 * response\['SecretString'] 包含解密后的 Secret 值，您可以将其打印或使用。 3.设置环境变量： ``` import os # 设置环境变量 os.environ['MY_ENVIRONMENT_VARIABLE'] = 'value' ``` 解释： * 使用 os.environ 来设置环境变量。 * 以 MY_ENVIRONMENT_VARIABLE 为例，您需要将它替换为实际的环境变量名称，并用相应的参数值填充 value。 希望这些代码示例能够帮助您理解如何从 Amazon Systems Manager Parameter Store 和 Amazon Secrets Manager 获取数据，以及如何将其设置为环境变量。 3.使用 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail)： 使用 boto3 从 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) 下载文件非常简单，以下是一个完整的 Python 代码示例，展示了如何实现此功能： ``` import boto3 # 创建 S3 客户端对象 s3 = boto3.client('s3') # 设置您的 S3 Bucket 名称和要下载的文件路径 bucket_name = 'your-bucket-name' file_key = 'path/to/your/file.txt' # 设置要保存文件本地路径 download_path = '/local/path/to/save/file.txt' # 下载文件 s3.download_file(bucket_name, file_key, download_path) print('File downloaded successfully!') ``` 解释： 1. 导入 boto3 库：import boto3 将 boto3 库导入您的 Python 代码中，以便使用其功能。 2. 创建 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) 客户端对象：s3 = boto3.client（'s3'） 创建一个 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) 客户端对象，用于与 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) 服务进行交互。 3. 设置 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) Bucket 和文件路径： bucket_name：将此变量设置为您要下载文件所在的 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) Bucket 名称。 file_key：将此变量设置为要下载的文件在 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) Bucket 中的完整路径（包括文件夹）。 4. 设置本地保存路径：download_path 应该指向您想要将文件保存到本地机器上的位置。 5. 下载文件： s3.download_file（bucket_name,file_key,download_path）使用download_file（）方法从 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) Bucket 下载指定的文件，并将内容保存到本地路径中。 6. 打印成功消息：如果下载成功，代码将打印 “File downloaded successfully!”消息。 补充一下关于使用 boto3 从 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) 流式下载文件的内容： ``` import boto3 # 创建 S3 客户端对象 s3 = boto3.client('s3') # 设置您的 S3 Bucket 名称和要下载的文件路径 bucket_name = 'your-bucket-name' file_key = 'path/to/your/file.txt' # 下载文件流 response = s3.download_fileobj(bucket_name, file_key) # 处理文件流，例如将其写入本地文件 with open('/local/path/to/save/file.txt', 'wb') as f:   for chunk in response:     f.write(chunk) print('File downloaded successfully!') ``` 1. 使用 download_fileobj（）方法： 这将返回一个文件对象流，而不是直接下载文件内容到内存中。 2. 处理文件流： 使用 with open（）语句打开一个本地文件以写入模式（‘wb’）。 然后循环遍历 response 中的每个块（chunk），并将它们写入打开的文件中。 优势： * 内存效率：流式下载避免将整个文件加载到内存中，这对于处理非常大的文件尤其重要。 * 灵活性：您可以使用流式下载来直接处理文件数据，例如将其传递给其他程序或将其转换为不同的格式。 ``` import boto3 import os import requests import json # 从环境变量中加载 S3 桶名称 BUCKET_NAME = os.environ.get('S3_BUCKET_NAME') # 加载配置（API 密钥、位置和服务器密钥） def load_config():   secretsmanager = boto3.client('secretsmanager') try:       # 从 Secrets Manager 中获取 API 密钥 response = secretsmanager.get_secret_value(SecretId='apikey-secret')       apikey = response['SecretString']        # 从 Secrets Manager 中获取服务器密钥 response = secretsmanager.get_secret_value(SecretId='server_key-secret') server_key = response['SecretString']   except ClientError as e:     print(f"从 Secrets Manager 获取配置失败: {e}")   exit(1)    # 从 Parameter Store 中获取位置信息   ssm = boto3.client('ssm')   try:     response = ssm.get_parameter(Name='location', WithDecryption=False)     location = response['Parameter']['Value']   except ClientError as e:     print(f"从 Parameter Store 获取位置信息失败: {e}")     exit(1)    return apikey, location, server_key # 从 S3 中加载天气数据 def load_skycon_data():   s3 = boto3.client('s3')   file_key = 'path/to/skycon_data.json' # 将此替换为 S3 中天气数据的路径    try:     response = s3.get_object(Bucket=BUCKET_NAME, Key=file_key)   skycon_data = response['Body'].read().decode('utf-8')   return json.loads(skycon_data)   except ClientError as e:   print(f"从 S3 获取天气数据失败: {e}")         exit(1) # 获取实时天气数据（需要根据您的 API 和逻辑实现） def get_real_data(): # 使用 apikey 和 location 调用天气 API，获取实时数据 pass # 从天气数据中提取每小时预报信息（需要根据数据格式实现） def hourly_data(skycon_data): # 处理 skycon_data，提取每小时的温度、降雨概率等信息 pass # 将 JSON 数据转换为 Markdown 格式（需要根据您的需求实现） def json2markdown(data): # 将处理后的天气数据转换为 Markdown 文本格式 pass # 将 Markdown 数据发送到服务器（需要根据您的服务器设置实现） def request_server(markdown_data, server_key): # 使用 server_key 对请求进行身份验证，将 markdown_data 发送到您的服务器 pass if __name__ == '__main__':   apikey, location, server_key = load_config()   skycon_data = load_skycon_data()    real_data = get_real_data()  # 替换为您获取实时天气数据的逻辑   hourly_forecast = hourly_data(skycon_data)  # 从天气数据中提取每小时预报信息   markdown_output = json2markdown(hourly_forecast) # 将数据转换为 Markdown 格式 request_server(markdown_output, server_key) # 发送 Markdown 数据到服务器 ``` 1. 加载配置： load_config（）函数从 Amazon Secrets Manager 中获取 API 密钥（apikey）和服务器密钥（server_key），并从 Amazon Systems Manager Parameter Store 中获取位置信息（location）. 2. 加载天气数据： load_skycon_data（）从 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) 中读取天气数据文件，并将其转换为 Python 字典。 3. 获取实时天气数据： get_real_data（）是一个占位符函数，需要根据您的实际需求和天气 API 实现。它使用 API 密钥和位置信息来获取最新的天气数据。 4. 提取每小时预报： hourly_data（）也是一个占位符函数，需要根据您从 [Amazon S3](https://aws.amazon.com/cn/s3/?trk=cndc-detail) 中读取的天气数据格式进行实现。它负责解析天气数据并提取每小时的温度、降雨概率等信息。 5. 转换为 Markdown 格式： json2markdown（）是一个占位符函数，需要根据您的需求实现将处理后的天气数据转换为 Markdown 文本格式。 6. 发送数据到服务器： request_server（）也是一个占位符函数，需要根据您的服务器设置实现。它使用服务器密钥对请求进行身份验证，并将 Markdown 数据发送到您的服务器。 请注意，您需要根据实际需求和数据格式来实现 get_real_data（）、hourly_data（）、json2markdown（）和request_server（）这些函数。 执行之后就可以在方糖的公众号上看到推送了。  这里列出来了实时天气和每小时天气的具体信息。  #### **参考链接** > 1.https\://aws.amazon.com/cn/lambda/?trk=cndc-detail > >2.https\://lbs.amap.com/tools/picker?trk=cndc-detail > >3.https\://docs.caiyunapp.com/weather-api/v2/v2.6/1-realtime.html?trk=cndc-detail > >4.https\://sct.ftqq.com/?trk=cndc-detail