Captured at (local ISO): 2026-05-18 05:16:58

Preface

A Raspberry Pi inspection rover needed local weather features—temperature, humidity, rain, light, “is it raining,” etc. (what you actually get depends on the provider; the site I picked exposes far more fields.) Below is the build-out.

I. Picking a site and URL pattern

1. Site choice

Target: 全国历史天气查询平台

Hourly last-24-hours history: weather code, temp, precip, wind dir/speed, pressure, humidity, and more—fits the bill.

2. Page anatomy

Two county pages share the prefix https://datashareclub.com/weather/ plus 福建/福州/, but the trailing code differs (01230103 vs 01230102)—that tail is the district key.
Split the URL into three parts:

Swap the green and yellow segments to bulk-scrape other districts. For this note we only need 闽侯 under Fuzhou, Fujian.

II. Implementation

1. Inspect the response

Method A: print HTML from requests (simple pages)

Start with a small fetch helper fetch_url_content:

import requests
from requests.exceptions import RequestException
target_url="https://datashareclub.com/weather/%E7%A6%8F%E5%BB%BA/%E7%A6%8F%E5%B7%9E/101230102.html"

def fetch_url_content(url, headers=None, timeout=10):

    """
    使用requests访问目标URL并返回响应内容

    参数:
        url (str): 要访问的目标URL
        headers (dict, 可选): 请求头，默认使用模拟浏览器的User-Agent
        timeout (int, 可选): 请求超时时间（秒），默认10秒

    返回:
        str: 响应文本内容（成功时）；None（失败时）
    """
    # 默认请求头（模拟浏览器，避免被简单反爬拦截）
    default_headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36'
    }
    # 如果用户传入自定义headers，合并到默认headers中（用户传入的优先）
    if headers:
        default_headers.update(headers)

    try:
        # 发送GET请求
        response = requests.get(
            url=url,
            headers=default_headers,
            timeout=timeout  # 超时保护
        )
        # 检查响应状态码（200表示成功）
        response.raise_for_status()

        # 打印成功信息（可选，可删除）
        print(f"✅ 访问成功：{url}（状态码：{response.status_code}）")
        # 返回响应文本（如果需要二进制内容，可改为 return response.content）
        return response.text

    except RequestException as e:
        # 捕获所有requests相关异常（网络错误、超时、404/500等状态码）
        print(f"❌ 访问失败：{url}，错误：{str(e)}")
        return None
    except Exception as e:
        # 捕获其他未知异常
        print(f"❌ 未知错误：{url}，错误：{str(e)}")
        return None


def crawler(url):
    content = fetch_url_content(url)
    if content:
        print("\n--- 响应内容---")
        print(content)
        print('成功响应')

crwaler(target_url)

Run output:

Once you eyeball where the series live in HTML/JS, move on to structured parsing.

Method B: DevTools Network tab (messy pages)

F12 → Network → reload—inspect the document for 101230102.html.

Search for a distinctive value like 25.7°C:

Yellow: response belongs to 101230102.html; Green: substring positions—guides regex/string extraction.

2. Parsing

import re
import ast
def extract_weather_data(html_content):
    """
    从HTML内容中提取天气数据（针对提供的JS变量格式）

    参数:
        html_content (str): 包含天气数据的HTML/JS文本内容

    返回:
        dict: 结构化的天气数据字典，包含各类气象指标
    """
    # 定义需要提取的JS变量及其对应的数据名称
    data_vars = {
        'aqi_data': '空气质量AQI',
        'wd_data': '温度(℃)',
        'sd_data': '湿度(%)',
        'rain_data': '降水量(mm)',
        'qy_data': '气压(hPa)',
        'njd_data': '能见度(km)',
        'data': '风向风力',
        'solar_data': '太阳辐射(W/m²)',
        'dewpoint_data': '露点温度(℃)'
    }

    # 存储提取后的所有数据
    result = {}

    # 遍历所有需要提取的变量
    for var_name, desc in data_vars.items():
        # 正则表达式：匹配 "var 变量名 = [数组内容];" 的结构
        # 匹配规则：忽略空格，捕获数组部分
        pattern = re.compile(
            rf'var\s+{var_name}\s*=\s*(\[.*?\]);',
            re.DOTALL  # 让 .*? 可以匹配换行
        )

        # 查找匹配结果
        match = pattern.search(html_content)
        if not match:
            result[desc] = []
            continue

        # 提取数组字符串并转换为Python列表
        try:
            # 去除可能的注释和多余字符，确保是标准数组格式
            data_str = match.group(1).strip().rstrip(';')
            # 使用ast.literal_eval安全转换字符串为Python对象（避免eval的安全问题）
            data_list = ast.literal_eval(data_str)
            result[desc] = data_list
        except (SyntaxError, ValueError) as e:
            print(f"提取 {desc} 失败：{str(e)}")
            result[desc] = []

    return result

Parser is site-specific—notes on data_vars and the regex:

data_vars: keys are JS identifiers (aqi_data, …); values are human labels.

        pattern = re.compile(
            rf'var\s+{var_name}\s*=\s*(\[.*?\]);',
            re.DOTALL  # 让 .*? 可以匹配换行
        )

Matches var name = [...];, re.DOTALL lets . cross lines.

Parsed dump:

3. Saving CSV

Two helpers: per-series files vs one wide CSV. Snippets:

get_csv_row_count (per-file row count helper):

import pandas as pd
import csv

def get_csv_row_count(file_path):
    """
    读取CSV文件并返回行数（不包含标题行）

    参数:
    file_path (str): CSV文件路径

    返回:
    int: CSV文件的行数（不包含标题行）

    异常:
    FileNotFoundError: 如果文件不存在
    pd.errors.ParserError: 如果文件格式错误
    """
    try:
        # 使用chunksize分块读取，避免加载整个文件到内存
        chunk_size = 10000
        row_count = 0

        for chunk in pd.read_csv(file_path, chunksize=chunk_size):
            row_count += len(chunk)

        return row_count
    except FileNotFoundError:
        raise FileNotFoundError(f"文件不存在: {file_path}")
    except pd.errors.ParserError as e:
        raise ValueError(f"CSV解析错误: {str(e)}")
    except Exception as e:
        raise Exception(f"读取文件时发生未知错误: {str(e)}")

Wide merge:

def merge_to_single_csv(weather_data, output_file='2.csv'):
    """
    将所有天气数据合并到一个CSV文件中

    参数:
        weather_data (dict): 提取的天气数据字典（由extract_weather_data函数返回）
        output_file (str): 输出的CSV文件路径

    返回:
        str: 生成的CSV文件路径
    """
    # 1. 收集所有时间点并去重、排序
    all_timestamps = set()
    for data_type, data_list in weather_data.items():
        for item in data_list:
            timestamp = item[0]
            # 统一转换为datetime对象
            dt = parse_time(timestamp)
            # 转换回标准格式字符串
            all_timestamps.add(format_time(dt))

    # 排序时间点，使用智能解析函数
    sorted_timestamps = sorted(all_timestamps, key=lambda x: parse_time(x))

    # 2. 构建表头
    headers = ['时间']
    # 数据类型映射（用于简化表头）
    data_type_mapping = {
        '空气质量AQI': 'AQI',
        '温度(℃)': '温度(℃)',
        '湿度(%)': '湿度(%)',
        '降水量(mm)': '降水量(mm)',
        '气压(hPa)': '气压(hPa)',
        '能见度(km)': '能见度(km)',
        '风向风力': '风向风力',
        '太阳辐射(W/m²)': '太阳辐射',
        '露点温度(℃)': '露点温度(℃)'
    }

    # 为每种数据类型添加对应的列
    for data_type in weather_data.keys():
        if data_type == '风向风力':
            headers.extend(['风力等级', '风向代码', '风速', '风向名称'])
        elif data_type == '太阳辐射(W/m²)':
            headers.extend(['短波辐射(W/m²)', '直接辐射(W/m²)', '散射辐射(W/m²)', '直接正常辐照度(W/m²)'])
        else:
            headers.append(data_type_mapping.get(data_type, data_type))

    # 3. 构建数据行
    rows = []
    for timestamp in sorted_timestamps:
        row = [timestamp]

        # 为每个时间点收集所有数据
        for data_type, data_list in weather_data.items():
            # 查找该时间点对应的数据
            data_point = None
            for item in data_list:
                item_timestamp = item[0]
                # 统一格式后比较
                if format_time(parse_time(item_timestamp)) == timestamp:
                    data_point = item
                    break

            # 根据数据类型添加对应的值
            if data_point:
                if data_type == '风向风力':
                    row.extend(data_point[1:5])  # 风力等级、风向代码、风速、风向名称
                elif data_type == '太阳辐射(W/m²)':
                    row.extend(data_point[1:5])  # 所有四个辐射指标
                else:
                    row.append(data_point[1])  # 普通数据类型（只有一个值）
            else:
                # 无数据时填充空值
                if data_type == '风向风力' or data_type == '太阳辐射(W/m²)':
                    row.extend([''] * 4)  # 这两种数据类型各有4列
                else:
                    row.append('')  # 其他数据类型只有1列

        rows.append(row)

    # 4. 写入CSV文件
    with open(output_file, 'w', newline='', encoding='utf-8-sig') as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(headers)
        writer.writerows(rows)

    print(f"已生成综合CSV文件: {output_file} ({len(rows)} 条记录)")
    return output_file

4. Helper utilities

Time normalizers:

def parse_time(time_str):
    """智能解析不同格式的时间字符串"""
    if isinstance(time_str, datetime):
        return time_str
    try:
        # 尝试解析带秒的格式
        return datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S')
    except ValueError:
        try:
            # 尝试解析不带秒的格式
            return datetime.strptime(time_str, '%Y-%m-%d %H:%M')
        except ValueError:
            # 如果是时间戳（整数或浮点数），转换为datetime
            if isinstance(time_str, (int, float)):
                timestamp = time_str / 1000.0 if time_str > 1000000000 else time_str
                return datetime.fromtimestamp(timestamp)
            # 无法解析的格式，返回原始值
            print(f"警告: 无法解析时间格式: {time_str}")
            return time_str

def format_time(dt):
    """将datetime对象格式化为标准字符串"""
    if isinstance(dt, datetime):
        return dt.strftime('%Y-%m-%d %H:%M:%S')
    return str(dt)

parse_time handles heterogeneous stamps; format_time canonicalizes to a string key.

5. Full script

import requests
from requests.exceptions import RequestException
import re
import ast
import csv
from datetime import datetime, timedelta
target_url="https://datashareclub.com/weather/%E7%A6%8F%E5%BB%BA/%E7%A6%8F%E5%B7%9E/101230102.html"

def parse_time(time_str):
    """智能解析不同格式的时间字符串"""
    if isinstance(time_str, datetime):
        return time_str
    try:
        # 尝试解析带秒的格式
        return datetime.strptime(time_str, '%Y-%m-%d %H:%M:%S')
    except ValueError:
        try:
            # 尝试解析不带秒的格式
            return datetime.strptime(time_str, '%Y-%m-%d %H:%M')
        except ValueError:
            # 如果是时间戳（整数或浮点数），转换为datetime
            if isinstance(time_str, (int, float)):
                timestamp = time_str / 1000.0 if time_str > 1000000000 else time_str
                return datetime.fromtimestamp(timestamp)
            # 无法解析的格式，返回原始值
            print(f"警告: 无法解析时间格式: {time_str}")
            return time_str

def format_time(dt):
    """将datetime对象格式化为标准字符串"""
    if isinstance(dt, datetime):
        return dt.strftime('%Y-%m-%d %H:%M:%S')
    return str(dt)


def fetch_url_content(url, headers=None, timeout=10):

    """
    使用requests访问目标URL并返回响应内容

    参数:
        url (str): 要访问的目标URL
        headers (dict, 可选): 请求头，默认使用模拟浏览器的User-Agent
        timeout (int, 可选): 请求超时时间（秒），默认10秒

    返回:
        str: 响应文本内容（成功时）；None（失败时）
    """
    # 默认请求头（模拟浏览器，避免被简单反爬拦截）
    default_headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36'
    }
    # 如果用户传入自定义headers，合并到默认headers中（用户传入的优先）
    if headers:
        default_headers.update(headers)

    try:
        # 发送GET请求
        response = requests.get(
            url=url,
            headers=default_headers,
            timeout=timeout  # 超时保护
        )
        # 检查响应状态码（200表示成功）
        response.raise_for_status()

        # 打印成功信息（可选，可删除）
        print(f"✅ 访问成功：{url}（状态码：{response.status_code}）")
        # 返回响应文本（如果需要二进制内容，可改为 return response.content）
        return response.text

    except RequestException as e:
        # 捕获所有requests相关异常（网络错误、超时、404/500等状态码）
        print(f"❌ 访问失败：{url}，错误：{str(e)}")
        return None
    except Exception as e:
        # 捕获其他未知异常
        print(f"❌ 未知错误：{url}，错误：{str(e)}")
        return None

def extract_weather_data(html_content):
    """
    从HTML内容中提取天气数据（针对提供的JS变量格式）

    参数:
        html_content (str): 包含天气数据的HTML/JS文本内容

    返回:
        dict: 结构化的天气数据字典，包含各类气象指标
    """
    # 定义需要提取的JS变量及其对应的数据名称
    data_vars = {
        'aqi_data': '空气质量AQI',
        'wd_data': '温度(℃)',
        'sd_data': '湿度(%)',
        'rain_data': '降水量(mm)',
        'qy_data': '气压(hPa)',
        'njd_data': '能见度(km)',
        'data': '风向风力',
        'solar_data': '太阳辐射(W/m²)',
        'dewpoint_data': '露点温度(℃)'
    }

    # 存储提取后的所有数据
    result = {}

    # 遍历所有需要提取的变量
    for var_name, desc in data_vars.items():
        # 正则表达式：匹配 "var 变量名 = [数组内容];" 的结构
        # 匹配规则：忽略空格，捕获数组部分
        pattern = re.compile(
            rf'var\s+{var_name}\s*=\s*(\[.*?\]);',
            re.DOTALL  # 让 .*? 可以匹配换行
        )

        # 查找匹配结果
        match = pattern.search(html_content)
        if not match:
            result[desc] = []
            continue

        # 提取数组字符串并转换为Python列表
        try:
            # 去除可能的注释和多余字符，确保是标准数组格式
            data_str = match.group(1).strip().rstrip(';')
            # 使用ast.literal_eval安全转换字符串为Python对象（避免eval的安全问题）
            data_list = ast.literal_eval(data_str)
            result[desc] = data_list
        except (SyntaxError, ValueError) as e:
            print(f"提取 {desc} 失败：{str(e)}")
            result[desc] = []

    return result

def merge_to_single_csv(weather_data, output_file='2.csv'):
    """
    将所有天气数据合并到一个CSV文件中

    参数:
        weather_data (dict): 提取的天气数据字典（由extract_weather_data函数返回）
        output_file (str): 输出的CSV文件路径

    返回:
        str: 生成的CSV文件路径
    """
    # 1. 收集所有时间点并去重、排序
    all_timestamps = set()
    for data_type, data_list in weather_data.items():
        for item in data_list:
            timestamp = item[0]
            # 统一转换为datetime对象
            dt = parse_time(timestamp)
            # 转换回标准格式字符串
            all_timestamps.add(format_time(dt))

    # 排序时间点，使用智能解析函数
    sorted_timestamps = sorted(all_timestamps, key=lambda x: parse_time(x))

    # 2. 构建表头
    headers = ['时间']
    # 数据类型映射（用于简化表头）
    data_type_mapping = {
        '空气质量AQI': 'AQI',
        '温度(℃)': '温度(℃)',
        '湿度(%)': '湿度(%)',
        '降水量(mm)': '降水量(mm)',
        '气压(hPa)': '气压(hPa)',
        '能见度(km)': '能见度(km)',
        '风向风力': '风向风力',
        '太阳辐射(W/m²)': '太阳辐射',
        '露点温度(℃)': '露点温度(℃)'
    }

    # 为每种数据类型添加对应的列
    for data_type in weather_data.keys():
        if data_type == '风向风力':
            headers.extend(['风力等级', '风向代码', '风速', '风向名称'])
        elif data_type == '太阳辐射(W/m²)':
            headers.extend(['短波辐射(W/m²)', '直接辐射(W/m²)', '散射辐射(W/m²)', '直接正常辐照度(W/m²)'])
        else:
            headers.append(data_type_mapping.get(data_type, data_type))

    # 3. 构建数据行
    rows = []
    for timestamp in sorted_timestamps:
        row = [timestamp]

        # 为每个时间点收集所有数据
        for data_type, data_list in weather_data.items():
            # 查找该时间点对应的数据
            data_point = None
            for item in data_list:
                item_timestamp = item[0]
                # 统一格式后比较
                if format_time(parse_time(item_timestamp)) == timestamp:
                    data_point = item
                    break

            # 根据数据类型添加对应的值
            if data_point:
                if data_type == '风向风力':
                    row.extend(data_point[1:5])  # 风力等级、风向代码、风速、风向名称
                elif data_type == '太阳辐射(W/m²)':
                    row.extend(data_point[1:5])  # 所有四个辐射指标
                else:
                    row.append(data_point[1])  # 普通数据类型（只有一个值）
            else:
                # 无数据时填充空值
                if data_type == '风向风力' or data_type == '太阳辐射(W/m²)':
                    row.extend([''] * 4)  # 这两种数据类型各有4列
                else:
                    row.append('')  # 其他数据类型只有1列

        rows.append(row)

    # 4. 写入CSV文件
    with open(output_file, 'w', newline='', encoding='utf-8-sig') as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(headers)
        writer.writerows(rows)

    print(f"已生成综合CSV文件: {output_file} ({len(rows)} 条记录)")
    return output_file


def crawler(url):
    content = fetch_url_content(url)
    if content:
        print("\n--- 响应内容---")
        print(content)
        print('成功响应')
        merge_to_single_csv(extract_weather_data(content),'10.csv')




crawler(target_url)

III. Summary

Without heavy anti-bot crypto, scrapers follow a steady recipe—like this outline: pick a site, study the HTML/JS, fetch, parse, store. The piece that always changes is parsing, because each backend shapes data differently—usually tweak that function only. Often you can paste the raw response into an LLM, ask for a first-pass parser, then hand-edit—good enough for many jobs.