每天早上九点,钉钉群里准时出现一张精美的天气图文卡片——城市地标微缩模型插画搭配实时天气信息,既有实用价值,又赏心悦目。

支撑这一切的,没有服务器、没有域名、没有数据库,也没有任何常驻进程。背后仅有一段运行在云函数计算(Function Compute,简称FC)中的 Python 代码,每天被定时任务唤醒一次,执行十几秒后便重新进入休眠状态。

项目简介

本项目基于云函数计算 FC (Serverless) 实现零服务器运维自动化天气推送,无需购买服务器、域名、数据库,依靠事件驱动模式,定时唤醒执行代码,执行完毕自动休眠。

程序定时从 Open‑Meteo 免费天气 API 拉取目标城市实时气象数据;调用通义 Qwen‑Image‑2.1‑Pro 大模型,结合城市地标、实时天气自动生成等距微缩风格天气插画;生成图片上传至 OSS 对象存储做持久化保存;最后通过钉钉自定义机器人,将天气图文卡片推送至指定钉钉群聊。

技术栈

组件

用途

费用

云函数计算 FC

函数计算,承载代码逻辑

可以直接免费领3个月的函数计算额度

Open‑Meteo API

获取实时天气数据

免费

通义百炼(Qwen-xxx)

生成简短出行穿衣提示

消耗极少 token,近乎免费

qwen-image-2.1-pro

根据天气自动生成插画

0.25元/张

OSS

暂存生成的图片

几分钱/月

钉钉自定义机器人

推送图文卡片到群聊

免费

一、架构与数据流

1.1整体架构

触发器层:云函数计算 FC 定时触发器(Cron 定时)作为入口,到设定时间自动触发函数执行;时区配置为 Asia/Shanghai 北京时间。

数据获取层:函数调用 Open‑Meteo 开放天气 API,传入目标城市经纬度,拿到实时天气编码、温度、风速风向原始 JSON 数据;内部完成天气编码映射为中文天气描述。

AI 生成层:结合城市地标库、实时天气,组装 AI 绘画 Prompt,内置敏感词替换逻辑规避内容审核拦截;调用通义百炼 Qwen‑Image‑2.1‑Pro 接口生成 16:9 横版微缩城市插画。

资源存储层:AI 返回临时有效期图片,下载到 FC 本地临时目录,做图片压缩处理,上传OSS,设置对象公共读 ACL,生成永久公网访问图片链接。

消息推送层:使用钉钉机器人 Webhook+HMAC‑SHA256 加签,组装 Markdown 消息体,把 OSS 图片链接 + 天气文本推送到钉钉群。

1.2 数据流流程图
FC定时触发器(09:00)
        ↓
读取环境变量CITY获取目标城市 → CITY_COORDS字典读取城市经纬度
        ↓
调用Open‑Meteo API获取原始天气数据
        ↓
WEATHER_CODE_MAP解析为中文天气、温度、风向风力
        ↓
LANDMARKS_DB读取城市地标 + WEATHER_EFFECT_MAP匹配天气画面效果
        ↓
敏感词过滤sanitize_prompt()生成最终AI绘画Prompt
        ↓
调用qwen‑image‑2.1‑pro生成插画,下载图片至FC /tmp临时目录
        ↓
图片压缩 → 上传OSS,设置public‑read权限,获取公网图片URL
        ↓
钉钉机器人加签签名,组装markdown消息
        ↓
POST请求推送钉钉群聊
        ↓
执行完成,FC实例休眠;发生异常输出错误日志返回500错误

二、前置准备

2.1 天气 API

方案 :Open-Meteo

如果你连注册都不想注册,Open-Meteo 是完全免费、无需 Key 的天气 API,支持全球经纬度查询,返回 JSON 格式。

2.2 通义百炼 API Key

登录云账号,进入 云百炼控制台:https://bailian.console.aliyun.com

右上角确认地域,建议选 华北2(北京);新用户如果没有开通,点“立即开通”即可。

左侧菜单进入 API-Key / 密钥管理。

点击 创建 API Key。

弹窗里:归属账号 主账号,归属业务空间 默认业务空间,权限 全部,描述:填 天气卡片 方便识别

点击 确定 后,会弹出完整 API Key,立刻复制保存。

关闭弹窗后就不能再看到明文了,只能重新创建。

回到 FC 函数,把环境变量 DASHSCOPE_API_KEY 填成这个 sk- 开头的值。

2.3 qwen-image-2.1-pro图片生成

登录https://bailian.console.aliyun.com/cn-beijing/model/market/detail/qwen-image-2.0

左侧导航栏进入 模型广场 或 应用中心,搜索"qwen-image-2.1-pro"。

点击模型卡片,确认已开通服务;如未开通,点击"立即开通"即可(与百炼 API Key 共用同一个账号,无需额外注册)。

充值开通付费(推荐)

进入「模型广场」→ 找到 qwen-image-2.1-pro

点击开通付费,给账户充点钱(按量计费,每次生图几毛钱)

或者在控制台里找到「配额管理」,把「仅使用免费额度」关掉

2.4 OSS Bucket

由于通义千问(Qwen-Image)生成的图片临时链接有效期较短(通常只有 24 小时),我们需要将这些生成的天气插画卡片保存到一个永久且可安全访问的云存储中。OSS(对象存储)是承载这一任务的最佳选择。

创建 Bucket(存储空间)

登录 OSS 控制台https://oss.console.aliyun.com/

点击左上角的 Bucket 列表 → 创建 Bucket。

Bucket 名称:建议起一个易记的名称,例如 weather--oss。

地域(Region):这是最关键的一步。请务必选择与你之前创建的 FC(函数计算)相同的地域(例如 FC 选的是杭州,OSS 也必须选杭州)。同地域内网传输不仅免收流量费,而且速度最快、极其稳定。

设置访问权限(ACL)

在创建 Bucket 的设置页面,将 读写权限(ACL) 设置为 公共读。

记录关键连接信息

创建成功后,进入该 Bucket 的 概览 页面,你需要记下以下两个核心参数(后续配置 FC 环境变量时会用到):

外网访问:复制以 .aliyuncs.com 结尾的外网访问域名(例如 oss-cn-hangzhou.aliyuncs.com)。

Bucket 名称:即你刚刚创建的名称(例如 weather--oss)

2.5 钉钉自定义机器人

1. 群聊 → 群设置 → 智能群助手 → 添加机器人 → 自定义(通过 Webhook 接入)

2. 安全设置建议同时开启两项:

◦ 加签:拿到以 SEC 开头的 secret,代码里用 HMAC-SHA256 算签名

◦ 自定义关键词:填一个词(如 天气),此后每条消息正文里必须包含这个词,否则接口返回成功但群里什么都不出现——这是最常见的「假成功」

3. 记下 Webhook URL 和 secret

三、核心亮点:天气图片 Prompt 自动匹配

3.1 Prompt 模板

为了让 AI 生成的文字准确,可以对图片里的原模板做微调。

以清晰的45度俯视角度,展示一个等距微缩模型场景。
画面包含:[地标A] 和 [地标B],构图居中。
场景中巧妙融合 [天气状态] 效果,氛围柔和,呈现城市与天气的自然互动感。
使用基于物理的真实渲染(PBR)和逼真的光照效果,搭配纯色背景,整体画面清晰简洁。
图片正上方展示文字:“{city} {weather} {temp}℃”
3.2 完整代码

整合好的完整 index.py,已写死贵阳,并配置了你指定的 6 个地标,直接复制替换即可。


import subprocess
subprocess.call(
    'pip install dashscope oss2 requests -t /tmp -q '
    '-i https://mirrors.ctyun.cn/pypi/simple/ '
    '--trusted-host mirrors.ctyun.cn',
    shell=True
)
import sys
sys.path.append('/tmp')
import dashscope

import requests
import oss2
import time
import hmac
import hashlib
import base64
import urllib.parse
import os
from datetime import date, datetime

# ==========================================
# 0. 依赖库自动安装
# ==========================================
try:
    from PIL import Image, ImageDraw, ImageFont
except ImportError:
    print("[INFO] 正在安装图像处理依赖库 (Pillow)...")
    os.system(f"{sys.executable} -m pip install Pillow")
    from PIL import Image, ImageDraw, ImageFont

# ==========================================
# 1. 核心配置与地标字典
# ==========================================
LANDMARKS_DB = {
    # 一线城市
    "北京": ["故宫", "天安广场", "天坛", "颐和园", "鸟巢"],
    "上海": ["东方明珠", "外滩", "豫园", "上海中心大厦", "迪士尼城堡", "静安寺"],
    "广州": ["广州塔", "陈家祠", "镇海楼", "珠江新城", "越秀公园五羊石像"],
    "深圳": ["平安金融中心", "市民中心", "世界之窗", "大梅沙", "地王大厦"],
    # 新一线
    "成都": ["天府熊猫塔", "宽窄巷子", "锦里", "太古里", "杜甫草堂"],
    "重庆": ["洪崖洞", "解放碑", "来福士", "长江索道", "李子坝轻轨"],
    "杭州": ["西湖雷峰塔", "钱江新城", "灵隐寺", "大运河拱宸桥"],
    "武汉": ["黄鹤楼", "东湖", "长江大桥", "楚河汉街", "龟山电视塔"],
    "西安": ["大雁塔", "城墙", "钟楼", "大唐不夜城", "大明宫"],
    "南京": ["夫子庙", "中山陵", "鸡鸣寺", "紫峰大厦", "明城墙"],
    "天津": ["天津之眼", "五大道", "世纪钟", "津湾广场"],
    "苏州": ["东方之门", "拙政园", "平江路", "寒山寺"],
    "郑州": ["二七纪念塔", "大玉米楼", "中原福塔", "少林寺"],
    "长沙": ["橘子洲伟人雕像", "岳麓山", "IFS国金中心", "太平老街"],
    "东莞": ["虎门炮台", "民盈国贸中心", "松山湖"],
    "青岛": ["栈桥", "五四广场", "八大关", "奥帆中心"],
    "沈阳": ["沈阳故宫", "彩电塔", "大帅府", "盛京大剧院"],
    "合肥": ["大蜀山", "天鹅湖", "渡江纪念馆", "合肥大剧院"],
    "佛山": ["祖庙", "岭南天地", "千灯湖"],
    # 二线城市
    "宁波": ["天一阁", "南塘老街", "宁波财富中心", "老外滩"],
    "昆明": ["滇池", "西山龙门", "金马碧鸡坊", "双塔"],
    "福州": ["三坊七巷", "镇海楼", "海峡明珠广场"],
    "厦门": ["鼓浪屿日光岩", "双子塔", "南普陀寺", "环岛路"],
    "哈尔滨": ["圣索菲亚教堂", "中央大街", "冰雪大世界"],
    "长春": ["伪满皇宫博物院", "净月潭", "长影世纪城"],
    "石家庄": ["河北博物院", "电视塔", "正定古城"],
    "太原": ["晋祠", "双塔寺", "五一广场", "蒙山大佛"],
    "贵阳": ["黔灵山", "甲秀楼", "双子塔", "花果园白宫", "文昌阁", "东山寺"],
    "南宁": ["东盟商务区", "青秀山", "三街两巷"],
    "常州": ["环球港", "天宁寺", "西太湖"],
    "南通": ["狼山", "中南城", "南通大剧院"],
    "泉州": ["开元寺", "西街", "洛阳桥", "蟳埔"],
    "潍坊": ["风筝博物馆", "十笏园", "富华游乐园"],
    "烟台": ["烟台山灯塔", "蓬莱阁", "养马岛"],
    "绍兴": ["鲁迅故里", "沈园", "兰亭", "迪荡湖"],
    "嘉兴": ["南湖红船", "乌镇", "月河历史街区"],
    "徐州": ["云龙湖", "楚王陵", "户部山"],
    "惠州": ["惠州西湖", "双月湾", "合江楼"],
    "珠海": ["渔女雕像", "日月贝大剧院", "横琴长隆"],
    "中山": ["中山纪念公园", "岐江桥", "兴中广场"],
    "保定": ["古莲花池", "直隶古署", "保定博物馆"],
    "兰州": ["中山桥", "白塔山", "黄河母亲雕像"],
    "台州": ["天台山", "江南长城", "东辉阁"],
    "温州": ["江心屿", "五马街", "雁荡山"],
    "乌鲁木齐": ["国际大巴扎", "红山公园", "中天广场"],
    "洛阳": ["龙门石窟", "应天门", "明堂天堂", "白马寺"],
    "襄阳": ["唐城", "古隆中", "襄阳古城墙"],
    "宜昌": ["三峡景区", "三峡人家", "西陵峡"],
    "沧州": ["铁狮子", "大运河博物馆"],
    "邯郸": ["丛台公园", "广府古城"],
}

# ==========================================
# 敏感词过滤映射 (AI 生图 API 内容审核白名单)
# ==========================================
SENSITIVE_WORD_MAP = {
    "天安门": "天安广场",
    "渡江战役": "渡江",
    "直隶总督署": "直隶古署",
    "三峡大坝": "三峡景区",
}

def sanitize_prompt(text):
    """过滤敏感词,避免 AI 生图 API 内容审核拦截"""
    for sensitive, safe in SENSITIVE_WORD_MAP.items():
        text = text.replace(sensitive, safe)
    return text

# ==========================================
# 城市经纬度字典 (用于 Open-Meteo API)
# ==========================================
CITY_COORDS = {
    "北京": (39.9042, 116.4074),
    "上海": (31.2304, 121.4737),
    "广州": (23.1291, 113.2644),
    "深圳": (22.5431, 114.0579),
    "成都": (30.5728, 104.0668),
    "重庆": (29.5630, 106.5516),
    "杭州": (30.2741, 120.1551),
    "武汉": (30.5928, 114.3055),
    "西安": (34.3416, 108.9398),
    "南京": (32.0603, 118.7969),
    "天津": (39.3434, 117.3616),
    "苏州": (31.2989, 120.5853),
    "郑州": (34.7466, 113.6253),
    "长沙": (28.2282, 112.9388),
    "东莞": (23.0205, 113.7518),
    "青岛": (36.0671, 120.3826),
    "沈阳": (41.8057, 123.4315),
    "合肥": (31.8206, 117.2272),
    "佛山": (23.0218, 113.1219),
    "宁波": (29.8683, 121.5440),
    "昆明": (25.0389, 102.7183),
    "福州": (26.0745, 119.2965),
    "厦门": (24.4798, 118.0894),
    "哈尔滨": (45.8038, 126.5340),
    "长春": (43.8171, 125.3235),
    "石家庄": (38.0428, 114.5149),
    "太原": (37.8706, 112.5489),
    "贵阳": (26.6470, 106.8426),
    "南宁": (22.8170, 108.3665),
    "常州": (31.8122, 119.9692),
    "南通": (31.9932, 120.8985),
    "泉州": (24.8740, 118.6757),
    "潍坊": (36.7093, 119.1015),
    "烟台": (37.4638, 121.4478),
    "绍兴": (30.0003, 120.5820),
    "嘉兴": (30.7467, 120.7507),
    "徐州": (34.2044, 117.2808),
    "惠州": (23.1115, 114.4152),
    "珠海": (22.2711, 113.5767),
    "中山": (22.5228, 113.3826),
    "保定": (38.8738, 115.4645),
    "兰州": (36.0586, 103.8348),
    "台州": (28.6569, 121.4209),
    "温州": (28.0006, 120.6718),
    "乌鲁木齐": (43.8256, 87.6168),
    "洛阳": (34.6197, 112.4539),
    "襄阳": (32.0400, 112.1352),
    "宜昌": (30.7099, 111.2810),
    "沧州": (38.3122, 116.8485),
    "邯郸": (36.6142, 114.5396),
}

# WMO 天气代码转中文描述
WEATHER_CODE_MAP = {
    0: "晴", 1: "晴", 2: "多云", 3: "阴",
    45: "阴", 48: "阴",
    51: "小雨", 53: "小雨", 55: "小雨", 56: "小雨", 57: "小雨",
    61: "小雨", 63: "中雨", 65: "大雨", 66: "小雨", 67: "小雨",
    71: "小雪", 73: "中雪", 75: "大雪", 77: "小雪",
    80: "小雨", 81: "中雨", 82: "大雨",
    85: "小雪", 86: "中雪",
    95: "阴", 96: "小雨", 99: "小雨",
}

# 风向角度转中文方向
WIND_DIR_MAP = [
    (11.25, "北"), (33.75, "东北"), (56.25, "东"), (78.75, "东南"),
    (101.25, "南"), (123.75, "西南"), (146.25, "西"), (168.75, "西北"),
    (180, "北"),
]

def deg_to_wind_dir(deg):
    for limit, direction in reversed(WIND_DIR_MAP):
        if deg >= limit:
            return direction
    return "北"

# ==========================================
# WEATHER_EFFECT_MAP (AI绘画天气效果)
# ==========================================
WEATHER_EFFECT_MAP = {
    "晴": "晴朗微缩城市,微距镜头,阳光斜射,强烈的光影对比,模型材质呈现温暖的哑光塑料质感",
    "多云": "多云微缩城市,微距镜头,柔和的散射光,天空带有细腻的模型云朵,色彩清新淡雅",
    "阴": "阴天微缩城市,微距镜头,漫反射自然光,冷色调模型材质,带有一点湿润的哑光感",
    "小雨": "微缩城市下着小雨,微距镜头,模型路面有细腻的水渍反光,朦胧的雨雾氛围",
    "大雨": "微缩城市暴雨,微距镜头,雨滴清晰可见,模型地面积水反光强烈,湿润的空气感",
    "雪": "微缩城市雪景,微距镜头,模型屋顶和树枝覆盖细腻薄雪,柔和的冷色光照,唯美宁静",
}

# ==========================================
# 2. 获取实时天气 (Open-Meteo 零配置版)
# ==========================================
def get_weather(city="贵阳"):
    print(f"[INFO] 正在通过 Open-Meteo 获取 {city} 的天气信息...")

    # 查找城市经纬度
    coords = CITY_COORDS.get(city)
    if not coords:
        print(f"[WARNING] 未找到城市 {city} 的经纬度,尝试使用默认坐标")
        coords = (26.6470, 106.8426)  # 贵阳默认

    latitude, longitude = coords

    weather_url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": latitude,
        "longitude": longitude,
        "current_weather": "true",
        "timezone": "Asia/Shanghai",
    }

    try:
        w_resp = requests.get(weather_url, params=params, timeout=10)
        w_resp.raise_for_status()
        data = w_resp.json()

        cw = data.get("current_weather", {})
        temp = str(int(round(cw.get("temperature", 18))))
        windspeed = cw.get("windspeed", 0)
        winddir = cw.get("winddirection", 0)
        weathercode = cw.get("weathercode", 3)

        # 天气描述
        weather_text = WEATHER_CODE_MAP.get(weathercode, "阴")
        # 风向
        wind_dir = deg_to_wind_dir(winddir)
        wind_text = f"{wind_dir}风 {windspeed}km/h"

        return {
            "city": city,
            "weather": weather_text,
            "temp": temp,
            "wind": wind_text,
        }
    except Exception as e:
        print(f"[ERROR] Open-Meteo 天气获取失败: {e}")

    # 降级返回默认值
    return {
        "city": city,
        "weather": "多云",
        "temp": "18",
        "wind": "无持续风 2级",
    }

# ==========================================
# 3. 动态生成 AI 绘画 Prompt
# ==========================================
def build_ai_image_prompt(weather_data):
    city = weather_data["city"]
    weather_cond = weather_data["weather"]

    if city in LANDMARKS_DB:
        landmarks_list = LANDMARKS_DB[city]
        landmarks_text = "、".join(landmarks_list)
    else:
        landmarks_text = f"{city}城市地标"

    weather_effect = WEATHER_EFFECT_MAP.get(
        weather_cond, "微距镜头下的城市微缩模型"
    )

    prompt = (
        f"以清晰的45度俯视角度,展示一个等距微缩模型场景。"
        f"画面比例必须是16:9横屏宽幅。"
        f"画面包含:[{landmarks_text}],构图居中。"
        f"场景中巧妙融合 [{weather_effect}] 效果,氛围柔和,呈现城市与天气的自然互动感。"
        f"使用基于物理的真实渲染(PBR)和逼真的光照效果,搭配纯色背景,整体画面清晰简洁。"
        f"画面中不要出现任何文字。"
    )
    # 过滤敏感词,避免 AI 生图 API 内容审核拦截
    prompt = sanitize_prompt(prompt)

    # 【调试用】打印实际发送给AI的prompt,方便排查绿网拦截问题
    print(f"[DEBUG] 实际发送的AI绘画Prompt: {prompt}")

    return prompt

# ==========================================
# 4. 调用 Qwen-Image 生成图片 (适配百炼API)
# ==========================================
def gen_image(prompt):
    # 获取环境变量
    api_key = os.environ.get("DASHSCOPE_API_KEY", "")
    base_url = os.environ.get("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com")

    if not api_key:
        print("[ERROR] 缺少 DASHSCOPE_API_KEY 环境变量")
        return None

    # 使用 multimodal-generation 接口
    url = f"{base_url}/api/v1/services/aigc/multimodal-generation/generation"

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

    payload = {
        "model": "qwen-image-2.1-pro",
        "input": {
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {
                            "text": prompt
                        }
                    ]
                }
            ]
        },
        "parameters": {
            "prompt_extend": True,
            "size": "1280*720"
        }
    }

    print(f"[INFO] 正在调用 qwen-image-2.1-pro 生成微缩模型插画...")
    print(f"[INFO] API地址: {url}")

    try:
        resp = requests.post(url, headers=headers, json=payload, timeout=120)
        print(f"[INFO] 响应状态码: {resp.status_code}")
        print(f"[INFO] 响应内容: {resp.text[:800]}")

        result = resp.json()

        # 提取图片URL
        img_url = None

        # 格式: output.choices[0].message.content[0].image
        try:
            choices = result.get("output", {}).get("choices", [])
            if choices:
                content_list = choices[0].get("message", {}).get("content", [])
                for item in content_list:
                    if item.get("image"):
                        img_url = item["image"]
                        break
        except Exception:
            pass

        # 备用格式
        if not img_url:
            try:
                choices = result.get("output", {}).get("choices", [])
                if choices:
                    content_list = choices[0].get("message", {}).get("content", [])
                    for item in content_list:
                        if item.get("image_url"):
                            img_url = item["image_url"]
                            break
            except Exception:
                pass

        if img_url:
            print(f"[INFO] 获取到图片URL: {img_url[:80]}...")
            r = requests.get(img_url, timeout=30)
            local_path = "/tmp/weather_bg.png"
            with open(local_path, "wb") as f:
                f.write(r.content)
            print(f"[SUCCESS] 图片已生成并保存: {local_path}")
            return local_path
        else:
            print(f"[ERROR] 未能从响应中提取图片URL")
            print(f"[ERROR] 完整响应: {result}")

    except requests.exceptions.Timeout:
        print("[ERROR] 请求超时,AI生成可能需要较长时间")
    except Exception as e:
        print(f"[ERROR] AI生图异常: {e}")

    return None

# ==========================================
# 5. Python 自动叠加天气文字排版 (透明无背景版)
# ==========================================
def add_text_to_image(image_path, weather_data):
    # 不在图片上叠加文字,直接返回原图
    print("[INFO] 跳过图片文字叠加,使用纯AI画面")
    return image_path

# ==========================================
# 6. 上传 OSS & 推送钉钉
# ==========================================
def send_to_dingtalk(local_path, weather_data):
    try:
        # 1. 压缩图片大小
        img = Image.open(local_path)
        w, h = img.size
        max_width = 1200
        if w > max_width:
            ratio = max_width / float(w)
            new_w = max_width
            new_h = int(h * ratio)
            img = img.resize((new_w, new_h), Image.LANCZOS)
            print(f"[INFO] 图片已压缩至 {new_w}x{new_h}")
        # 保存压缩后的图片
        small_path = "/tmp/weather_final_small.png"
        img.save(small_path, optimize=True)
        print(f"[INFO] 压缩后图片大小: {os.path.getsize(small_path) / 1024:.0f} KB")
    except Exception as e:
        print(f"[WARNING] 图片压缩失败,尝试上传原图: {e}")
        small_path = local_path

    # 2. 上传到 OSS
    oss_auth = oss2.Auth(
        os.environ.get("OSS_ACCESS_KEY_ID", "YOUR_OSS_KEY_ID"),
        os.environ.get("OSS_ACCESS_KEY_SECRET", "YOUR_OSS_KEY_SECRET"),
    )
    bucket_name = os.environ.get("OSS_BUCKET", "your-bucket-name")
    endpoint = os.environ.get("OSS_ENDPOINT", "oss-cn-xxx-internal.aliyuncs.com")
    public_endpoint = endpoint.replace("-internal", "")

    bucket = oss2.Bucket(
        oss_auth,
        public_endpoint,
        bucket_name
    )

    oss_key = f"weather{date.today().strftime('%Y%m%d')}_{int(time.time())}.png"

    # 【关键修改】上传时设置该文件为"公共读"权限
    bucket.put_object_from_file(
        oss_key, small_path,
        headers={'x-oss-object-acl': 'public-read'}
    )

    # 3. 拼接公网访问 URL
    img_url = f"https://{bucket_name}.{public_endpoint}/{oss_key}"
    print(f"[INFO] OSS 公网地址: {img_url}")

    # 4. 生成钉钉签名
    webhook = os.environ.get("DING_WEBHOOK", "YOUR_DINGTALK_WEBHOOK")
    secret = os.environ.get("DING_WEBHOOK_SECRET", "YOUR_DINGTALK_SECRET")

    ts = str(round(time.time() * 1000))
    hmac_code = hmac.HMAC(
        secret.encode("utf-8"),
        (ts + "\n" + secret).encode("utf-8"),
        digestmod=hashlib.sha256,
    ).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    send_url = f"{webhook}&timestamp={ts}&sign={sign}"

    # 5. 使用 "markdown" 格式发送消息(图片直接在群里展示)
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": f"{weather_data['city']} 实时天气",
            "text": f"![]({img_url})" + "\n\n"
                    f"{weather_data['city']} {weather_data['weather']} {weather_data['temp']}℃  {weather_data['wind']}"
        },
    }

    # 【调试】打印关键参数
    print(f"[DEBUG] 钉钉webhook前缀: {webhook[:50]}...")
    print(f"[DEBUG] 钉钉secret长度: {len(secret)}")
    print(f"[DEBUG] 签名timestamp: {ts}")
    print(f"[DEBUG] 签名sign前30字符: {sign[:30]}...")
    print(f"[DEBUG] 完整发送URL: {send_url[:80]}...")
    print(f"[DEBUG] 推送payload: {payload}")

    r = requests.post(send_url, json=payload, timeout=10)
    result = r.json()
    print(
        "[SUCCESS] 钉钉推送成功"
        if result.get("errcode") == 0
        else f"[ERROR] 钉钉推送失败: {result}"
    )

# ==========================================
# 7. FC 入口函数 (主流程)
# ==========================================
def handler(event, context):
    try:
        target_city = os.environ.get("CITY", "贵阳")

        if isinstance(event, dict) and event.get("city"):
            target_city = event["city"]

        weather_data = get_weather(target_city)
        print(f"[INFO] 正在为 {weather_data['city']} 生成横屏天气海报...")

        prompt = build_ai_image_prompt(weather_data)

        bg_path = gen_image(prompt)
        if not bg_path:
            raise Exception("AI生图失败")

        final_path = add_text_to_image(bg_path, weather_data)
        send_to_dingtalk(final_path, weather_data)

        return {
            "code": 200,
            "msg": f"已成功推送 {weather_data['city']} 的横屏天气海报",
        }
    except Exception as e:
        print(f"[FATAL ERROR]: {e}")
        return {"code": 500, "msg": str(e)}
3.4 更换地标字典
LANDMARKS_DB = {
    "贵阳": ["黔灵山", "甲秀楼", "双子塔", "花果园白宫", "文昌阁", "东山寺"],
}

四、FC 控制台

4.1 怎么操作

地址:https://free.aliyun.com/?spm=5176.30385154.J_RIoiIhIPOvZIGUchbo3zM.2.24c76381i3MmYr&productCode=fc

打开 FC 控制台地址:https://fcnext.console.aliyun.com/cn-hangzhou/sandbox-instances

「云函数」 标签页

点击页面左上角的那个蓝色的 「创建函数」 按钮。

点进去之后,选择 “创建事件函数”。

把“函数名称”随便填一个,比如叫 weather-push。

运行环境选择 Python 3.10。

码上传方式:默认选“使用示例代码”就行,我们先创建一个空的函数,代码回头再粘贴进去。

请求处理程序:index.handler 这个不用动,正好对应我们 Python 代码的入口。

执行超时时间: 300秒

点击底部的 “创建”。

找到代码编辑器上方的标签页(就在文件列表 index.py 旁边),点击打开 index.py 文件。

全选 index.py 里的所有代码,并删除掉。

把之前准备好的那段完整的天气推送代码,复制并粘贴到这个空白区域里。

粘贴完后,点击代码编辑器上方(大概在“上传代码”按钮旁边)的 「部署代码」 按钮。

找到页面的右上角,点击那个配置,高级配置 - -> 「编辑层」

点击添加(官方公共层)

Python 3.10 OSS
Pandas 2

Python 3.10 OSS:OSS 是“对象存储服务”,Python 3.10 OSS 这个工具包里面,已经装好了让 Python 语言(3.10版本)去连接和读写这个 U盘的必要库(比如我们代码里的 oss2 库)。

Pandas 2 是因为 Pandas 包比较大,里面顺带塞了很多其他的系统基础依赖库。

4.2 FC 环境变量配置表

找到页面的右上角,点击那个配置,高级配置 - -> 「编辑环境变量」

在「环境变量」区域,点击 「+ 添加变量」

FC 环境变量,直接在 FC 控制台的环境变量页面配置就行。

环境变量 Key

说明

DASHSCOPE_API_KEY

百炼 API Key

OSS_ACCESS_KEY_ID

OSS AccessKey ID

OSS_ACCESS_KEY_SECRET

OSS AccessKey Secret

OSS_ENDPOINT

OSS Endpoint(如 oss-cn-hangzhou-internal.aliyuncs.com

)

OSS_BUCKET

Bucket 名称

DING_WEBHOOK

钉钉机器人 Webhook

DING_WEBHOOK_SECRET

钉钉机器人加签 Secret(没开签名验证可留空)

CITY

查询城市(广州)等

4.3 更换城市方式

推荐方式:FC 环境变量 — 在云函数计算控制台,找到你的函数 → 配置 → 环境变量,把 CITY 的值改成你想查的城市(比如 广州、上海)。这样每次定时触发就自动查新城市的天气了。

4.4 获取OSS AccessKey ID 与 OSS AccessKey Secret

进入管理页面:在网页右上角,把你的鼠标移到你的头像上,在弹出的菜单里点击 「AccessKey 管理」。

确认创建:系统可能会弹出一个提示框告诉你这很危险,点击「继续使用 AccessKey」(或者勾选“我确认知晓安全风险”后继续)。

创建新钥匙:点击页面上的 「创建 AccessKey」 蓝色按钮,然后根据提示完成一下手机或邮箱的短信验证。

复制并保存:验证通过后,屏幕上会直接显示你的 AccessKey ID 和 AccessKey Secret。

把 AccessKey ID 复制,填入配置页面的“值”里面。

把 AccessKey Secret 复制,填入对应的“值”里面。

4.5 定时触发器

记得在触发器设置里把时区选为 Asia/Shanghai

名称:随便起个好听的名字,比如叫 DailyWeatherPush

版本或别名:默认保持 LATEST 即可。

触发方式:保持选择定时触发器

指定时间:9:00

触发消息:这个不用填,留空就好。

4.6 效果测试

继续点击右上角的 「部署代码」 按钮,等页面左上角提示“部署成功”或绿色对勾。

点击左下角的 「测试函数」 按钮。

结语

这个项目最有价值的地方,从来不在于「能看天气」——手机自带的气候应用已经做得足够出色。它真正的价值在于,完整跑通了一条生产级的 Serverless 最小闭环:定时触发、第三方 API 调用、大模型调用、AI 生图、对象存储、Webhook 推送、错误告警、成本可控——每一个环节都是现成的云服务,零运维、按需付费、弹性伸缩。

当这条管线被验证稳定运行之后,替换数据源便成了轻而易举的事:将天气 API 换成股票行情接口,就能变成每日财经简报推送;换成 RSS 订阅源,就能做成资讯聚合机器人;换成服务器健康检查接口,就能搭建轻量级的监控告警系统。架构骨架无需改动,只需替换数据层数十行代码,即可适配全新的业务场景。

从更宏观的视角来看,本项目也是一个极佳的云原生开发学习模板。它展示了函数计算、托管服务与 AI 大模型 API 如何被灵活编排组合,在不触碰任何服务器管理事务的前提下,快速构建出有价值的自动化应用。对于希望探索云函数、AI 集成或工作流自动化的开发者而言,这个项目提供了一份从理论到生产落地的端到端参考实现,弥合了「知道概念」与「真正跑起来」之间的鸿沟。

Logo

DAMO开发者矩阵,由阿里巴巴达摩院和中国互联网协会联合发起,致力于探讨最前沿的技术趋势与应用成果,搭建高质量的交流与分享平台,推动技术创新与产业应用链接,围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐