如何统计复杂JSON数据中过去6个月的release数量？

核心问题在于清理带HTML标签的日期字段+精确的日期范围筛选，结合批量URL处理的效率优化，以下是可落地的实现步骤：

1. 单个JSON处理流程 #

步骤1：清理releaseDate字段的HTML标签 #

如果releaseDate字段包含HTML标签（比如<span>2024-01-01</span>），先提取纯文本：

• 用正则快速清理：re.sub(r'<.*?>', '', release_date_str)
• 或者用BeautifulSoup更稳妥（避免正则漏处理复杂标签）：BeautifulSoup(release_date_str, "html.parser").get_text(strip=True)

步骤2：日期解析与范围筛选 #

把清理后的日期字符串转成可比较的datetime对象，再筛选过去6个月的记录：

• 使用dateutil.parser自动识别多种日期格式（不需要预先指定格式）
• 用relativedelta计算6个月前的时间点（比timedelta更准确，适配不同月份天数）

示例代码：

import datetime
from dateutil.parser import parse
from dateutil.relativedelta import relativedelta
import re
import json

# 模拟单个JSON数据（实际是从URL请求得到的）
sample_json = '''{
  "releases": [
    {"id": 1, "version": "v1.0", "releaseDate": "<div>2024-03-15</div>"},
    {"id": 2, "version": "v1.1", "releaseDate": "<span>2023-10-01</span>"},
    {"id": 3, "version": "v1.2", "releaseDate": "2024-05-20"}
  ],
  "user": {"name": "test"}
}'''

json_data = json.loads(sample_json)
six_months_ago = datetime.datetime.now() - relativedelta(months=6)
valid_releases = []

for release in json_data['releases']:
    # 清理HTML标签
    raw_date = release['releaseDate']
    clean_date_str = re.sub(r'<.*?>', '', raw_date).strip()
    # 解析日期
    try:
        release_date = parse(clean_date_str).replace(tzinfo=None)  # 统一移除时区，避免比较报错
        if release_date >= six_months_ago:
            valid_releases.append(release)
    except ValueError:
        # 处理日期格式无效的情况，可根据需求跳过或记录日志
        continue

# 统计符合条件的数量
print(f"过去6个月的release数量：{len(valid_releases)}")

2. 批量处理100个URL #

优化点： #

• 用requests.Session复用连接，减少TCP握手开销
• 可选：用多线程（concurrent.futures.ThreadPoolExecutor）或异步（aiohttp）加速批量请求，避免串行等待

示例代码（多线程版本）：

import requests
from concurrent.futures import ThreadPoolExecutor
import datetime
from dateutil.parser import parse
from dateutil.relativedelta import relativedelta
import re

# 你的100个URL列表
url_list = ["https://example.com/api/release/1", ...]  # 替换为实际URL
six_months_ago = datetime.datetime.now() - relativedelta(months=6)
total_valid = 0

def process_single_url(url):
    global total_valid
    try:
        with requests.Session() as s:
            response = s.get(url, timeout=10)
            response.raise_for_status()  # 抛出HTTP错误
            json_data = response.json()
            count = 0
            for release in json_data.get('releases', []):
                raw_date = release.get('releaseDate', '')
                clean_date_str = re.sub(r'<.*?>', '', raw_date).strip()
                try:
                    release_date = parse(clean_date_str).replace(tzinfo=None)
                    if release_date >= six_months_ago:
                        count +=1
                except ValueError:
                    continue
            total_valid += count
            return count
    except Exception as e:
        # 处理请求失败或JSON解析错误，可记录日志
        print(f"处理URL {url}失败：{str(e)}")
        return 0

# 多线程处理，建议线程数设为10-20（避免服务器限流）
with ThreadPoolExecutor(max_workers=15) as executor:
    executor.map(process_single_url, url_list)

print(f"所有URL中过去6个月的release总数量：{total_valid}")

注意事项 #

• 如果日期字段的HTML标签结构复杂（比如嵌套标签），建议改用BeautifulSoup清理，避免正则匹配不全
• 若服务器有请求频率限制，可在请求中加入延迟（time.sleep(0.1)）或调整线程数
• 可添加日志记录（用logging模块），方便排查无效日期或请求失败的情况

内容的提问来源于stack exchange，提问作者xlxdxf