企业官网建设流程全解析

这次我们来看一个偏工程向的话题：从零搭建天气系统初版。

先说清楚，这里说的“天气系统”不是气象部门里那种做数值预报的巨型系统，而是一个面向本地业务的天气数据采集、存储、接口和可视化服务。它能解决一类很常见的问题：项目里需要展示实时天气、预报数据、历史曲线，但不想被第三方 SaaS 平台的订阅费用和隐私限制绑死，或者需要把天气数据接入到自己已有的监控、农业、物流、能源管理系统里。

很多开发者一听到“天气系统”，下意识就会往爬虫、API 转发、图表展示这些方向想。实际动手之后才会发现坑不少：数据源选型不清晰、定时任务没有重试机制、数据库表设计撑不起多城市查询、前端图表刷新频率和接口频率不匹配、没有记录请求日志导致问题无法回溯。这篇文章会围绕这些问题，给出一套可落地的初版架构方案，并把每一步拆成可操作的验证流程。

文章不会只停留在“能跑”的程度。会覆盖天气系统的核心能力速览、适用范围、数据源选型、本地环境准备、后端接口设计、定时采集与批量任务、资源占用观察、容器化部署、常见问题排查和工程化建议。文章里所有代码都采用通用模板写法，具体域名、端口、Token、路径需要按实际项目替换。

1. 天气系统核心能力速览

先把一个典型天气系统初版的轮廓定下来。按最常见的本地部署需求，系统通常包含数据源拉取、本地存储、REST API、前端展示和定时刷新五个部分。

初版有两个关键设计目标：一是数据链路必须端到端打通，从数据源到数据库再到页面能跑通；二是任务调度要能观察到执行状态和失败日志。在此基础上再谈高并发、高可用和更复杂的可视化效果才有意义。

2. 适用场景与使用边界

天气系统的初版并不是一个通吃所有气象需求的产品。开发前先确认自己的项目属于哪一类，能避免后续做重复重构。

适合这套结构的场景包括：

• 企业自用可视化大屏，展示区域天气、风力、气压、预警状态。
• 自动化业务系统补充环境数据，例如室外设备巡检、冷链运输、农业灌溉控制。
• 需要将多个城市的天气数据聚合到一个内部页面里，并进行历史趋势比较。
• 做算法实验，需要长时间稳定保存天气样本数据，供回归分析使用。
• 教学和团队技术验证，目标是把数据采集、接口设计、图表展示一条链路跑明白。

不适合或需要额外处理的场景：

• 需要提供权威气象预报，给公共服务或防灾决策作依据。这类场景对数据源资质、数据延迟和准确率有极高要求，不能只依托第三方免费 API 转发。
• 需要分钟级雷达影像、卫星云图、精细格点预报。初版的数据模型如果只围绕城市级天气，结构上会不匹配。
• 需要商用发布第三方数据。不同天气服务商对数据再分发有明确的许可限制，商用前要检查授权范围。
• 涉及“人工影响天气”或任何气象干预方向的物理控制，这不是软件系统职责，相关操作在国内有严格法律与管理边界，技术开发者不要碰。

使用第三方天气数据时，还需要注意版权和隐私边界。天气数据不一定全部属于公有领域，如实况采集、预报产品、分钟级降水曲线往往有特定版权声明。内部展示可以走标准授权，公开商用或对外提供服务前，必须确认数据源协议和署名要求。对用户位置信息也要做最小化处理，尽量通过城市编码或 IP 粗粒度定位，不要存储不必要的精确坐标。

3. 环境准备与前置条件

初版环境不用很夸张。以一台 CPU-only 服务器为例，只要内存不低于 2GB，磁盘留出 20GB 以上空间，就能完成整套系统搭建。如果数据量少，甚至可以在一台开发笔记本上完成开发和验证。

推荐的系统与软件基线如下，具体版本请按自己项目的实际情况确定，不要在没确认的情况下直接照抄：

本地启动前，最好先做几个基础检查。

# 检查 Python 版本 python --version # 检查 Node 版本，如果选择 Node 后端则执行 node -v # 检查 Docker 版本 docker --version docker compose version # 检查端口占用 netstat -ano | grep 8000

这里有一个关键设计决策：尽量把“数据源凭证”从代码中剥离。天气服务的 API Key、站点 Token 不要硬编码进源码仓库，统一放到.env文件或部署环境变量里。后续不管是更换数据源还是多人协作，都会省很多事。.env 文件要加入.gitignore，避免把敏感信息提交到公共仓库。

4. 核心模块实现：数据采集与模型设计

天气系统初版最需要花时间的地方，不是前端图表漂不漂亮，而是“数据模型是否管得住查询需求”。

4.1 数据表设计

一个最小可用模型通常包含三张核心表：城市表、实时天气表、天气预报表。城市表负责维度和查询关键字；实时天气表保存最新观测数据；预报表存储服务商返回的逐日或逐小时预报。数据库表结构设计需要重点考虑查询方式，常见的查询方式是“城市 + 时间范围”。

下面是 Postgres 风格建表示例，如果使用 SQLite，把TIMESTAMPTZ换成TIMESTAMP即可。

CREATE TABLE city ( id SERIAL PRIMARY KEY, name VARCHAR(100) NOT NULL, code VARCHAR(64) UNIQUE NOT NULL, latitude DECIMAL(10, 6), longitude DECIMAL(10, 6), enabled BOOLEAN DEFAULT TRUE, created_at TIMESTAMPTZ DEFAULT NOW() ); CREATE TABLE weather_realtime ( id BIGSERIAL PRIMARY KEY, city_id INTEGER NOT NULL, temp DECIMAL(6, 2), feels_like DECIMAL(6, 2), humidity DECIMAL(6, 2), pressure DECIMAL(8, 2), wind_dir VARCHAR(32), wind_scale INTEGER, wind_speed DECIMAL(6, 2), weather_text VARCHAR(64), observed_at TIMESTAMPTZ, fetched_at TIMESTAMPTZ DEFAULT NOW(), CONSTRAINT fk_realtime_city FOREIGN KEY (city_id) REFERENCES city(id) ); CREATE TABLE weather_forecast ( id BIGSERIAL PRIMARY KEY, city_id INTEGER NOT NULL, forecast_date DATE NOT NULL, hour TEXT, temp_max DECIMAL(6, 2), temp_min DECIMAL(6, 2), humidity DECIMAL(6, 2), precip_prob DECIMAL(6, 2), weather_text VARCHAR(64), fetch_time TIMESTAMPTZ DEFAULT NOW(), CONSTRAINT fk_forecast_city FOREIGN KEY (city_id) REFERENCES city(id) ); CREATE INDEX idx_realtime_city_time ON weather_realtime (city_id, observed_at DESC); CREATE INDEX idx_forecast_city_date ON weather_forecast (city_id, forecast_date DESC);

不要把数据一股脑塞进一个没有任何索引的大表里。预报表和实时表按“城市 + 时间”建立联合索引，是初版里性价比最高的一步。

4.2 数据采集器示例

数据采集器的核心逻辑是“请求源 API -> 解析参数 -> 写入数据库”，不要把它和业务接口写在一个模块里，避免定时任务影响 Web 请求响应。

下面给出一个 Python 通用采集器模板。实际项目要按天气服务商文档替换 URL、字段名和鉴权方式。

import os import time import logging import requests import psycopg2 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger("weather_collector") CITY_LIST = [ {"name": "北京", "code": "101010100"}, {"name": "上海", "code": "101020100"}, ] API_URL = os.getenv("WEATHER_API_URL", "https://api.example.com/weather") API_KEY = os.getenv("WEATHER_API_KEY", "your-key") def fetch_weather(city_code: str) -> dict: params = {"city": city_code, "key": API_KEY} response = requests.get(API_URL, params=params, timeout=15) response.raise_for_status() return response.json() def save_realtime(conn, city_id: int, payload: dict) -> None: insert_sql = """ INSERT INTO weather_realtime (city_id, temp, feels_like, humidity, pressure, wind_dir, wind_scale, wind_speed, weather_text, observed_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ cursor = conn.cursor() try: cursor.execute(insert_sql, ( city_id, payload.get("temp"), payload.get("feels_like"), payload.get("humidity"), payload.get("pressure"), payload.get("wind_dir"), payload.get("wind_scale"), payload.get("wind_speed"), payload.get("weather_text"), payload.get("obs_time"), )) conn.commit() except Exception: conn.rollback() logger.exception("save realtime failed") finally: cursor.close() def batch_run() -> None: conn = psycopg2.connect( host=os.getenv("DB_HOST", "127.0.0.1"), port=os.getenv("DB_PORT", "5432"), database=os.getenv("DB_NAME", "weather"), user=os.getenv("DB_USER", "postgres"), password=os.getenv("DB_PASSWORD", "postgres"), ) for city in CITY_LIST: try: data = fetch_weather(city["code"]) # 这里只是示例，实际字段需要按数据源返回结构调整 save_realtime(conn, city["id"], data.get("realtime", {})) logger.info("city %s fetch success", city["name"]) except Exception as exc: logger.error("city %s fetch failed: %s", city["name"], exc) time.sleep(1) conn.close() if __name__ == "__main__": batch_run()

初版采集器建议先用手工执行的方式把返回值打出来核对字段，尤其是以下三处：

• 温度单位是摄氏度还是华氏度。
• 风向是角度数值还是中文风向。
• 降水概率在无雨时段是0还是直接不返回字段。

这三点极容易在联调阶段曝出脏数据。数据源字段返回不稳定，入库前可以先加一层清洗函数，把所有值统一成内部规范。

4.3 数据更新策略

天气实况和预报的更新频率不能照抄数据源服务商给的上限。比如第三方接口允许每秒请求 10 次，但你的业务场景每 10 分钟刷新一次就已经足够。初版建议把“上游可调用频率”和“业务实际消费频率”分开设置。

常用策略如下：

• 实况数据：每 10 到 30 分钟拉取一次。
• 多日预报：每小时重新拉取一次。
• 城市列表变更：人工触发全量刷新。
• 历史数据回补：单独写脚本处理，不走定时主流程。

初版可以先通过一个配置 JSON 控制城市列表，不要硬编码。

{ "cities": [ {"name": "北京", "code": "101010100"}, {"name": "上海", "code": "101020100"} ], "realtime_interval_sec": 600, "forecast_interval_sec": 3600, "timeout_sec": 15, "retry_times": 3 }

定时刷新要避免两个典型问题：一是上一次任务还没结束，下一次又开始执行；二是 API 连续失败后没有退避策略，导致数据源被短时间请求轰炸。初版最简单的做法是把任务调度器设为“单实例可重入保护”，或者用APScheduler的max_instances=1参数限制并发。

5. 数据接口服务实现

数据采集完成之后，对外接口是天气系统能否接入其他系统的关键。这里采用 FastAPI 作为示例后端框架，因为它自带 OpenAPI 文档，联调时不需要额外写接口说明。

5.1 最小接口服务

import os from datetime import datetime, timedelta import psycopg2 from psycopg2.extras import RealDictCursor from fastapi import FastAPI, HTTPException app = FastAPI(title="Local Weather API") DB_CONFIG = { "host": os.getenv("DB_HOST", "127.0.0.1"), "port": os.getenv("DB_PORT", "5432"), "database": os.getenv("DB_NAME", "weather"), "user": os.getenv("DB_USER", "postgres"), "password": os.getenv("DB_PASSWORD", "postgres"), } def get_connection(): return psycopg2.connect(**DB_CONFIG, cursor_factory=RealDictCursor) @app.get("/api/v1/weather/current") def get_current_weather(city: str = "北京"): conn = get_connection() cursor = conn.cursor() try: cursor.execute( """ SELECT ct.name, ct.code, wr.temp, wr.feels_like, wr.humidity, wr.pressure, wr.wind_dir, wr.wind_scale, wr.wind_speed, wr.weather_text, wr.observed_at FROM weather_realtime wr JOIN city ct ON ct.id = wr.city_id WHERE ct.name = %s ORDER BY wr.observed_at DESC LIMIT 1 """, (city,), ) row = cursor.fetchone() finally: cursor.close() conn.close() if row is None: raise HTTPException(status_code=404, detail="city data not found") return {"code": 0, "data": row}

启动命令：

uvicorn app:app --host 0.0.0.0 --port 8000

启动后访问http://127.0.0.1:8000/docs，可以直接看到 Swagger 文档。

5.2 查询最近历史曲线

很多天气页面需要展示“最近 24 小时温度变化”。对应接口一般写成：

@app.get("/api/v1/weather/history") def get_weather_history(city: str = "北京", hours: int = 24): if hours < 1 or hours > 168: raise HTTPException(status_code=400, detail="hours must be between 1 and 168") start_time = datetime.utcnow() - timedelta(hours=hours) conn = get_connection() cursor = conn.cursor() try: cursor.execute( """ SELECT observed_at, temp, humidity, pressure, weather_text FROM weather_realtime wr JOIN city ct ON ct.id = wr.city_id WHERE ct.name = %s AND wr.observed_at >= %s ORDER BY wr.observed_at ASC """, (city, start_time), ) rows = cursor.fetchall() finally: cursor.close() conn.close() return {"code": 0, "data": rows}

接口的关键点是ORDER BY ... ASC，按时间升序返回，前端画折线图时不需要重复排序。

5.3 前端页面接入数据

前端页面不需要太重。真实工程里，可以用 Vue 或 React，但初版验证数据链路，一个原生静态页加上 ECharts 就够了。

<!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="UTF-8"> <title>本地天气看板</title> <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script> </head> <body> <h2>天气看板</h2> <div id="current"></div> <div id="trend" style="width: 800px; height: 400px;"></div> <script> async function fetchCurrent() { const res = await fetch('/api/v1/weather/current?city=北京'); const json = await res.json(); document.getElementById('current').innerText = JSON.stringify(json, null, 2); } async function fetchHistory() { const res = await fetch('/api/v1/weather/history?city=北京&hours=24'); const json = await res.json(); const chart = echarts.init(document.getElementById('trend')); chart.setOption({ xAxis: { type: 'time' }, yAxis: { type: 'value', name: '温度(°C)' }, series: [{ data: json.data.map(item => [item.observed_at, item.temp]), type: 'line', smooth: true }] }); } fetchCurrent(); fetchHistory(); </script> </body> </html>

这一步跑通，说明从数据源到前端展示的最小链路已闭环。

6. 批量任务与定时调度设计

天气系统上线后最日常的任务是定时拉取数据。如果城市只有一两个，任务调度可以直接用系统的 crontab。但如果城市数量上升到几十个，还涉及不同数据源的统一管理，建议单独建一个任务调度模块。

6.1 APScheduler 方案

Python 项目直接使用APScheduler最简单，代码不需要额外起 worker 进程。

from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.interval import IntervalTrigger from collector import batch_run def realtime_job(): print("start realtime job") batch_run() def forecast_job(): print("start forecast job") # forecast_batch_run() if __name__ == "__main__": scheduler = BlockingScheduler(timezone="Asia/Shanghai") scheduler.add_job( realtime_job, IntervalTrigger(minutes=10), id="realtime_job", max_instances=1, coalesce=True, replace_existing=True, ) scheduler.add_job( forecast_job, IntervalTrigger(hours=1), id="forecast_job", max_instances=1, coalesce=True, replace_existing=True, ) scheduler.start()

带max_instances=1和coalesce=True能控制任务叠加。假如上一次任务网络超时卡了 20 分钟，原本 10 分钟一次的下一次任务也不会重复抢占。

6.2 失败重试与告警

批量任务跑起来之后，最怕的是“一直失败，但没任何人发现”。建议在采集器里加上失败计数器，如果连续失败超过阈值，就往企业微信、钉钉或邮件里发一条告警。这里不依赖具体插件，推送服务背后的实现可以自行封装。

FAIL_THRESHOLD = 3 fail_count = 0 for city in CITY_LIST: try: data = fetch_weather(city["code"]) save_realtime(conn, city["id"], data) fail_count = 0 except Exception as exc: fail_count += 1 if fail_count >= FAIL_THRESHOLD: send_alert(f"城市 {city['name']} 天气数据连续失败 {fail_count} 次: {exc}")

还需要在批量任务执行逻辑中隐藏一个容错细节：单个城市失败不应该中断整个城市列表的遍历。所以采集循环应该逐城市地包try/except，而不是在循环外层直接抛错。

6.3 多任务日志规范

批量任务建议按日期维度保留日志。最简单的方式是在每条日志里带任务 ID，如batch_run_20250601_0800。当数据库里出现某个时段的数据空洞，可以凭任务 ID 直接定位是“没采集”还是“采集失败”。日志字段至少包括：

• 任务开始时间/结束时间
• 每个城市的请求耗时
• 写入行数
• 返回的原始 JSON 状态字段
• 异常字符串

初版不要把日志全部打进控制台就结束了，至少要输出到文件或标准日志服务中。

7. 资源占用与性能观察

天气系统是典型的低算力需求项目，不会像 AI 模型那样吃显存。但很多用户第一次部署时仍然会看到内存持续上涨，这通常不是高并发导致，而是代码里有连接泄漏或日志积累。

7.1 数据库连接管理

常见问题是每个请求都新建数据库连接，不关闭。时间一长，数据库端连接数就会耗尽。上面的实现里，每次请求后主动调用cursor.close()和conn.close()，这是最基本要求。在实际项目里可以使用数据库连接池解决。

使用psycopg2.pool的简单做法：

from psycopg2.pool import SimpleConnectionPool from tenacity import retry, stop_after_attempt, wait_fixed POOL = SimpleConnectionPool(5, 20, **DB_CONFIG) def get_conn(): return POOL.getconn() def release_conn(conn): POOL.putconn(conn)

无论接口成功还是失败，都尽量把连接归还给连接池，不要只关闭游标后就不管连接。

7.2 采集服务的 CPU 与内存观察

在 Linux 环境，可以观察采集任务运行时系统的 CPU 和内存状态：

# 观察整体资源 top # 观察特定 Python 进程 ps aux | grep python3 # 观察端口监听状态 ss -lntp | grep 8000

如果内存持续上涨，优先怀疑数据集是否未清理。天气数据如果每 10 分钟全量存一次，一张表里会累积大量中间记录，应该保留保留策略。比较通用的做法是按保留天数删除历史数据，再对历史数据做按小时汇总表。

7.3 查询性能观察

天气系统初版的数据量并不大，绝大多数接口查询应该毫秒级返回。如果某个历史曲线接口明显变慢，可以查看数据库慢查询日志。初版最直接的手段是在建表时加上“城市 + 时间”的联合索引。还有一个隐藏问题：日期字段如果存的是字符串，那么WHERE observed_at >= ...的索引可能会失效。时间字段一律使用数据库原生时间类型。

7.4 前端刷新频率优化

如果前端页面是 10 分钟刷新一次实时数据，那么没有必要每次刷新都把整页历史曲线重新请求一遍。建议实时卡片每隔 10 分钟请求实时接口，历史曲线接口只在页面初始化时请求一次，或者每隔一小时刷新一次。这种改动可以把后端接口压力降低一个量级。

8. Docker 化部署与灰度验证

天气系统适合用 Docker Compose 做一键式部署。流程很直观：一个服务是数据库，一个服务是后端接口，一个服务是定时采集任务。如果连前端，再加一个 Nginx 静态站点服务。

如果项目本地还没写 Dockerfile，可以参考下面的通用 Compose 模板：

version: "3.8" services: db: image: postgres:16-alpine environment: POSTGRES_DB: weather POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres volumes: - db_data:/var/lib/postgresql/data ports: - "5432:5432" healthcheck: test: ["CMD-SHELL", "pg_isready -U postgres"] interval: 5s timeout: 5s retries: 10 api: build: ./app environment: DB_HOST: db DB_PORT: "5432" DB_NAME: weather DB_USER: postgres DB_PASSWORD: postgres ports: - "8000:8000" depends_on: db: condition: service_healthy scheduler: build: ./app command: ["python", "scheduler.py"] environment: DB_HOST: db DB_PORT: "5432" DB_NAME: weather DB_USER: postgres DB_PASSWORD: postgres depends_on: db: condition: service_healthy volumes: db_data:

接口服务和定时采集任务不应被设计成同一个进程。因为在 Docker Compose 环境里，如果 API 容器里挂着调度器，每次重启或扩容都会导致重复调度。二者分开，便于职责边界管理和后续横向扩展。

容器化部署后需要验证几个关键点：

• 数据库是否健康启动。
• API 容器是否能连通数据库。
• 调度器是否在独立启动后正常拉取数据。
• 数据卷是否持久化，容器重建后数据是否仍存在。

验证命令：

# 启动整套环境 docker compose up -d # 查看服务状态 docker compose ps # 查看 API 日志 docker compose logs api # 查看调度器日志 docker compose logs scheduler # 进入数据库查看数据量 docker compose exec db psql -U postgres -d weather \ -c "select city_id, count(*) from weather_realtime group by city_id;"

9. 天气系统常见问题与排查方法

把本地部署和运行过程中最常见的问题集中到一张排查表里，遇到问题先从这张表开始查。

这里专门展开三个高频排错过程。

第一，如何确认采集本身就成功。写代码时不要直接在入库后看结果，而是先在脚本里打印拉取到的原始 JSON 结构。很多第三方服务返回的字段名并不是直观的temperature，可能是嵌套对象里的t或tmp。

python -c "from collector import fetch_weather; print(fetch_weather('101010100'))"

第二，如何处理历史时间缺失。如果服务凌晨断电，凌晨的天气数据就永远补不回来了。可以在调度器里加一个“自动补齐最近 6 小时”的逻辑，每次启动先请求最近 6 小时数据，但不覆盖已有时间点。

第三，如何处理城市编码不一致。同一个城市的 ID，在不同数据源里可能完全不同。建议内部统一用一套“业务城市编码”，只在采集器适配层里做数据源编码映射。

10. 天气系统初版：可观测性与后续演进

很多项目做到“接口能返回数据”就停了，这会导致后续排错异常困难。对天气系统这种需要长期定时运行的项目，建议在初版阶段就加入最小可观测能力。

10.1 简易指标监控

至少记录 4 个数字指标：

• 每次任务采集的城市数。
• 成功写入条数。
• 失败城市数。
• 平均单次请求耗时。

这些数据在 Prometheus 里可以直接通过/metrics暴露，初版如果不想引入指标体系，也可以先写到一个 JSON 状态文件里，页面定期读取显示。

一个简化版本的 metrics 输出，可以只在/api/v1/health里返回全链路状态：

@app.get("/api/v1/health") def health_check(): conn = get_connection() cursor = conn.cursor() try: cursor.execute("SELECT 1") db_status = "ok" except Exception: db_status = "error" finally: cursor.close() conn.close() return { "db": db_status, "last_realtime_fetch": get_last_fetch_time(), }

get_last_fetch_time的实现就是从实时天气表里取最大fetched_at。只要这项时间不是一直卡在很旧的时刻，就说明调度链路还在正常运转。

10.2 最适合初版扩展的方向

天气系统初版容易在三个方向上继续演化。

第一个方向是多数据源切换。把采集器改成适配器模式，不同数据源只要实现统一的parse和save接口，就能在配置项里一键切换。这个方向对项目本身价值最大。

第二个方向是历史数据服务化。当前只提供查询接口，后续可以继续做统计输出，比如“最近 30 天平均温湿度”或“降雨日统计”，这会显著提升数据利用率。

第三个方向是前端可视化丰富化。天气看板从单条曲线向多城市地图、生活指数、云图叠加演进时，建议把前端拆成独立工程，并开始考虑后端输出 GeoJSON 格式的格点数据。

10.3 最容易踩的坑总结

这个项目看起来小，但上线后会遇到“数据没更新、没人知道”的尴尬。最有效的改进方式不是加更多面板，而是在任务调度器里把告警推给群里的机器人。告警阈值设为连续失败 3 次，比“每天凌晨看一次日志”可靠得多。

另一个容易被忽略的坑是数据保留策略。实时天气表如果每 10 分钟写入一条，一年就是五万多条记录。虽然现在数据库很能扛，但查询响应会随着时间推移逐渐变慢。初版就加上定期清理，例如只保留最近 30 天原始数据，其余时间分桶聚合到日统计表。

10.4 下一步建议

如果你的目标是快速做一个内部可以访问的天气系统，第一步建议先把“拉数据 -> 写库 -> 提供 API -> 静态页面展示”这条链路跑通，不要一上来就折腾 Docker、监控和图表主题。等基础链路稳定后，再依次加上调度器、告警、数据保留策略和指标监控。

按照这个顺序推进，项目节奏会非常清楚。

先验证城市列表和数据源连通性，重点观察接口返回字段和入库映射；再跑一个手动批量拉取脚本，确认所有城市都能写入数据库；接着启动 API 服务，在浏览器打开/docs和静态页面；最后部署定时任务，运行一个晚上后检查数据连续性和日志大小。

这套流程走完，天气系统初版就算真正有了可以迭代的地基。后续不管是接入更丰富的数据源，还是把页面给其他团队使用，都不会因为基础链路不完整而返工。建议先在自己的开发机或小服务器上完整跑一遍，再决定要不要上容器编排和监控系统。