中國環境監測總站（原環保部）發布的 全國城市空氣質量實時發布平台：
https://datacenter.mep.gov.cn/（現遷移至 www.mee.gov.cn）目標頁面 https://datacenter.mep.gov.cn/aqi/Default.jsp 為 JavaScript 動態渲染（FusionCharts + jQuery），數據嵌入在 HTML/JS 中：
<table>
<tr class="c3"><th>城市</th><th>AQI</th><th>級別</th><th>首要污染物</th></tr>
<tr class="c2">
<td onclick="window.open('...airCityMain.jsp?city=北京市')">北京市</td>
<td>68</td><td>良</td><td>PM10</td>
</tr>
<tr class="c3">
<td onclick="...city=天津市">天津市</td>
<td>108</td><td>輕度污染</td><td>PM10</td>
</tr>
...
</table>
CSS 類名含義：c1=優、c2=良、c3=輕度污染、c4=中度污染、c5=重度污染、c6=嚴重污染
chart1.setDataXML("
<?xml version='1.0' encoding='utf-8'?>
<graph ...>
<set name='優' value='5.6' color='00FF01' />
<set name='良' value='65.8' color='00B8E6' />
<set name='輕度污染' value='21.1' color='838134' />
...
</graph>
");
| 方案 | 優點 | 缺點 | 適用場景 |
|---|---|---|---|
| 靜態 HTML 解析 | 無需瀏覽器、速度快、資源少 | 無法獲取 JS 渲染數據、反爬蟲風險 | 日報歷史數據、結構穩定頁面 |
| Headless 瀏覽器 | 完整執行 JS、獲取實時報、繞過簡單反爬 | 資源消耗大、啟動慢、維護複雜 | 實時報、動態加載、需交互頁面 |
| API 逆向 | 結構化、穩定、高效 | 需逆向分析、可能加密簽名、法律風險 | 高頻采集、生產系統 |
| 官方開放接口 | 合法、穩定、權威 | 需申請、有配額、文檔可能舊 | 企業級應用、政府項目 |
#!/usr/bin/env python3
"""
PM2.5 數據采集器 - 靜態頁面解析版
依賴: pip install requests beautifulsoup4 lxml pandas
"""
import requests
from bs4 import BeautifulSoup
import pandas as pd
import re
from datetime import datetime
from typing import List, Dict, Optional
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class PM25Collector:
BASE_URL = "https://datacenter.mep.gov.cn/aqi/Default.jsp"
HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}
# 級別映射
LEVEL_MAP = {
'c1': ('優', 0, 50),
'c2': ('良', 51, 100),
'c3': ('輕度污染', 101, 150),
'c4': ('中度污染', 151, 200),
'c5': ('重度污染', 201, 300),
'c6': ('嚴重污染', 301, 500),
}
def fetch_page(self) -> Optional[str]:
try:
resp = requests.get(self.BASE_URL, headers=self.HEADERS, timeout=30)
resp.raise_for_status()
resp.encoding = 'utf-8'
return resp.text
except Exception as e:
logger.error(f"請求失敗: {e}")
return None
def parse_daily_table(self, html: str) -> List[Dict]:
soup = BeautifulSoup(html, 'lxml')
records = []
# 定位日報表格 (class="tj" 容器下的 table)
table = soup.select_one('.tj table')
if not table:
logger.warning("未找到日報表格")
return records
for row in table.select('tr')[1:]: # 跳過表頭
cells = row.find_all('td')
if len(cells) < 4:
continue
city = cells[0].get_text(strip=True)
aqi_text = cells[1].get_text(strip=True)
level_class = cells[0].get('class', [''])[0] # c1-c6
pollutant = cells[3].get_text(strip=True)
# 解析 AQI
try:
aqi = int(aqi_text)
except ValueError:
aqi = None
# 解析級別
level_name, level_min, level_max = self.LEVEL_MAP.get(level_class, ('未知', 0, 0))
# 提取城市詳情頁鏈接
onclick = cells[0].get('onclick', '')
detail_url = None
if onclick:
match = re.search(r"window\.open\('([^']+)'\)", onclick)
if match:
detail_url = match.group(1)
records.append({
'city': city,
'aqi': aqi,
'level': level_name,
'level_code': level_class,
'primary_pollutant': pollutant if pollutant else None,
'detail_url': detail_url,
'collected_at': datetime.now().isoformat(),
'data_type': 'daily',
})
return records
def parse_hourly_table(self, html: str) -> List[Dict]:
soup = BeautifulSoup(html, 'lxml')
records = []
# 實時報在 .ova 容器下
table = soup.select_one('.ova .table_double')
if not table:
return records
for row in table.select('tr'):
cells = row.find_all('td')
# 雙列佈局：每行 2 個城市
for i in range(0, len(cells), 5): # 每城市 5 列 (含間隔列)
if i + 3 >= len(cells): break
city_cell = cells[i]
aqi_cell = cells[i+1]
level_cell = cells[i+2]
pollutant_cell = cells[i+3]
city = city_cell.get_text(strip=True)
if not city: continue
try:
aqi = int(aqi_cell.get_text(strip=True))
except ValueError:
aqi = None
level = level_cell.get_text(strip=True)
pollutant = pollutant_cell.get_text(strip=True)
onclick = city_cell.get('onclick', '')
detail_url = None
if onclick:
match = re.search(r"window\.open\('([^']+)'\)", onclick)
if match:
detail_url = match.group(1)
records.append({
'city': city,
'aqi': aqi,
'level': level,
'primary_pollutant': pollutant if pollutant else None,
'detail_url': detail_url,
'collected_at': datetime.now().isoformat(),
'data_type': 'hourly',
})
return records
def collect(self) -> Dict[str, List[Dict]]:
html = self.fetch_page()
if not html:
return {'daily': [], 'hourly': []}
daily = self.parse_daily_table(html)
hourly = self.parse_hourly_table(html)
logger.info(f"采集完成: 日報 {len(daily)} 條, 實時報 {len(hourly)} 條")
return {'daily': daily, 'hourly': hourly}
def save_csv(self, data: Dict[str, List[Dict]], prefix: str = 'pm25'):
for dtype, records in data.items():
if not records: continue
df = pd.DataFrame(records)
filename = f"{prefix}_{dtype}_{datetime.now().strftime('%Y%m%d_%H%M')}.csv"
df.to_csv(filename, index=False, encoding='utf-8-sig')
logger.info(f"已保存: {filename} ({len(records)} 條)")
if __name__ == '__main__':
collector = PM25Collector()
data = collector.collect()
collector.save_csv(data)
# 入庫示例
# for record in data['daily'] + data['hourly']:
# db.insert('pm25_records', record)
package main
import (
"encoding/csv"
"fmt"
"log"
"net/http"
"os"
"regexp"
"strconv"
"strings"
"sync"
"time"
"github.com/PuerkitoBio/goquery"
)
type Record struct {
City string
AQI *int
Level string
PrimaryPollutant string
DetailURL string
CollectedAt time.Time
DataType string
}
var levelMap = map[string]string{
"c1": "優", "c2": "良", "c3": "輕度污染",
"c4": "中度污染", "c5": "重度污染", "c6": "嚴重污染",
}
func main() {
client := &http.Client{Timeout: 30 * time.Second}
resp, err := client.Get("https://datacenter.mep.gov.cn/aqi/Default.jsp")
if err != nil {
log.Fatal(err)
}
defer resp.Body.Close()
doc, err := goquery.NewDocumentFromReader(resp.Body)
if err != nil {
log.Fatal(err)
}
var wg sync.WaitGroup
recordsChan := make(chan []Record, 2)
// 並發解析日報和實時報
wg.Add(2)
go func() {
defer wg.Done()
recordsChan <- parseDaily(doc)
}()
go func() {
defer wg.Done()
recordsChan <- parseHourly(doc)
}()
go func() {
wg.Wait()
close(recordsChan)
}()
allRecords := make([]Record, 0)
for r := range recordsChan {
allRecords = append(allRecords, r...)
}
// 寫入 CSV
file, _ := os.Create(fmt.Sprintf("pm25_%s.csv", time.Now().Format("20060102_1504")))
defer file.Close()
writer := csv.NewWriter(file)
defer writer.Flush()
writer.Write([]string{"city", "aqi", "level", "primary_pollutant", "detail_url", "collected_at", "data_type"})
for _, r := range allRecords {
aqi := ""
if r.AQI != nil {
aqi = strconv.Itoa(*r.AQI)
}
writer.Write([]string{
r.City, aqi, r.Level, r.PrimaryPollutant, r.DetailURL,
r.CollectedAt.Format(time.RFC3339), r.DataType,
})
}
log.Printf("采集完成，共 %d 條記錄", len(allRecords))
}
func parseDaily(doc *goquery.Document) []Record {
var records []Record
doc.Find(".tj table tr").Each(func(i int, s *goquery.Selection) {
if i == 0 { return } // 跳過表頭
cells := s.Find("td")
if cells.Length() < 4 { return }
city := strings.TrimSpace(cells.Eq(0).Text())
aqiStr := strings.TrimSpace(cells.Eq(1).Text())
levelClass := strings.TrimSpace(cells.Eq(0).AttrOr("class", ""))
pollutant := strings.TrimSpace(cells.Eq(3).Text())
aqi, _ := strconv.Atoi(aqiStr)
level := levelMap[levelClass]
onclick, _ := cells.Eq(0).Attr("onclick")
detailURL := extractDetailURL(onclick)
records = append(records, Record{
City: city,
AQI: &aqi,
Level: level,
PrimaryPollutant: pollutant,
DetailURL: detailURL,
CollectedAt: time.Now(),
DataType: "daily",
})
})
return records
}
func parseHourly(doc *goquery.Document) []Record {
var records []Record
doc.Find(".ova .table_double tr").Each(func(i int, s *goquery.Selection) {
cells := s.Find("td")
for j := 0; j < cells.Length(); j += 5 {
if j+3 >= cells.Length() { break }
cityCell := cells.Eq(j)
city := strings.TrimSpace(cityCell.Text())
if city == "" { continue }
aqiStr := strings.TrimSpace(cells.Eq(j+1).Text())
level := strings.TrimSpace(cells.Eq(j+2).Text())
pollutant := strings.TrimSpace(cells.Eq(j+3).Text())
aqi, _ := strconv.Atoi(aqiStr)
onclick, _ := cityCell.Attr("onclick")
detailURL := extractDetailURL(onclick)
records = append(records, Record{
City: city,
AQI: &aqi,
Level: level,
PrimaryPollutant: pollutant,
DetailURL: detailURL,
CollectedAt: time.Now(),
DataType: "hourly",
})
}
})
return records
}
var detailURLRe = regexp.MustCompile(`window\.open\('([^']+)'\)`)
func extractDetailURL(onclick string) string {
matches := detailURLRe.FindStringSubmatch(onclick)
if len(matches) > 1 {
return matches[1]
}
return ""
}
-- 事實表：逐小時/每日城市級聚合
CREATE TABLE fact_city_aqi_hourly (
id BIGSERIAL PRIMARY KEY,
city_code VARCHAR(20) NOT NULL, -- 行政區劃代碼
city_name VARCHAR(50) NOT NULL,
province VARCHAR(20) NOT NULL,
aqi SMALLINT, -- 空氣質量指數
level VARCHAR(10), -- 優/良/輕度污染...
primary_pollutant VARCHAR(20), -- PM2.5/PM10/O3...
pm2_5 DECIMAL(6,1), -- 細顆粒物 μg/m³
pm10 DECIMAL(6,1), -- 可吸入顆粒物
so2 DECIMAL(6,1), -- 二氧化硫
no2 DECIMAL(6,1), -- 二氧化氮
co DECIMAL(6,2), -- 一氧化碳 mg/m³
o3 DECIMAL(6,1), -- 臭氧
recorded_at TIMESTAMPTZ NOT NULL, -- 數據時間
collected_at TIMESTAMPTZ DEFAULT NOW(), -- 采集時間
UNIQUE (city_code, recorded_at)
) PARTITION BY RANGE (recorded_at);
-- 按月分區
CREATE TABLE fact_city_aqi_hourly_2024_01 PARTITION OF fact_city_aqi_hourly
FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
-- 維度表：城市信息
CREATE TABLE dim_city (
code VARCHAR(20) PRIMARY KEY,
name VARCHAR(50) NOT NULL,
province VARCHAR(20) NOT NULL,
lat DECIMAL(10,6),
lng DECIMAL(10,6),
station_count SMALLINT
);
-- 索引優化
CREATE INDEX idx_fact_city_time ON fact_city_aqi_hourly (city_code, recorded_at DESC);
CREATE INDEX idx_fact_level_time ON fact_city_aqi_hourly (level, recorded_at DESC);
robots.txt（該站點通常允許爬取公開數據）# Prometheus 指標
pm25_collect_total{status="success"} 12450
pm25_collect_total{status="failed"} 12
pm25_collect_duration_seconds_bucket{le="5"} 12000
pm25_records_collected{dtype="daily"} 367
pm25_records_collected{dtype="hourly"} 734
# 告警規則
- alert: PM25CollectFailed
expr: increase(pm25_collect_total{status="failed"}[5m]) > 5
labels:
severity: warning
annotations:
summary: "PM2.5 采集失敗率異常"
- alert: PM25DataStale
expr: time() - pm25_last_success_timestamp > 3600
labels:
severity: critical
annotations:
summary: "PM2.5 數據超過 1 小時未更新"
環保部空氣質量數據采集關鍵點：
https://opendata.mee.gov.cn/ 開放 API，爬蟲作為補充/備用掌握此采集體系，可構建覆蓋全國、歷史完整、分鐘級更新的空氣質量數據平台，支撐環保決策、健康預警、學術研究等多元場景。
