数据获取与 API 可视化实战

import csv

# 示例 CSV 内容（sitka_weather_2024.csv）：
# name,station,date,PRCP,TMAX,TMIN
# Sitka Airport,USW00094621,2024-01-01,0.00,7,-1
# Sitka Airport,USW00094621,2024-01-02,15.99,5,-3

filename = 'data/sitka_weather_2024.csv'
with open(filename) as f:
    reader = csv.reader(f)
    header_row = next(reader)  # 读取文件头

    print("表头字段：")
    for index, column_header in enumerate(header_row):
        print(index, column_header)

    # 输出：
    # 0 name
    # 1 station
    # 2 date
    # 3 PRCP（降水量）
    # 4 TMAX（最高温度）
    # 5 TMIN（最低温度）

import csv

filename = 'data/sitka_weather_2024.csv'

with open(filename) as f:
    reader = csv.DictReader(f)
    for row in reader:
        # 通过列名访问，更直观
        print(f"日期: {row['date']}, 最高温: {row['TMAX']}°C")
        break  # 只打印第一行演示

import csv

headers = ['name', 'age', 'city']
data = [
    ['张三', 28, '北京'],
    ['李四', 32, '上海'],
]

with open('output.csv', 'w', newline='', encoding='utf-8') as f:
    writer = csv.writer(f)
    writer.writerow(headers)       # 写入单行
    writer.writerows(data)         # 批量写入多行

from datetime import datetime

date_string = '2024-01-02'
converted_date = datetime.strptime(date_string, '%Y-%m-%d')
print(converted_date)   # 2024-01-02 00:00:00
print(type(converted_date))  # <class 'datetime.datetime'>

# 提取年月日
print(converted_date.year)   # 2024
print(converted_date.month)  # 1
print(converted_date.day)    # 2

"""
天气数据可视化：Sitka 与死亡谷温度对比
数据源：NOAA 天气数据（CSV格式）
"""
import csv
from datetime import datetime
import matplotlib.pyplot as plt

# ====== 配置中文显示 ======
plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS']
plt.rcParams['axes.unicode_minus'] = False


def get_weather_data(filename, date_index, tmax_index, tmin_index):
    """
    从 CSV 文件提取日期和温度数据

    Args:
        filename: CSV 文件路径
        date_index: 日期列索引
        tmax_index: 最高温列索引
        tmin_index: 最低温列索引

    Returns:
        (dates, highs, lows) 元组
    """
    dates, highs, lows = [], [], []

    with open(filename) as f:
        reader = csv.reader(f)
        header_row = next(reader)  # 跳过表头

        for row in reader:
            try:
                current_date = datetime.strptime(row[date_index], '%Y-%m-%d')
                high = int(row[tmax_index])
                low = int(row[tmin_index])
            except ValueError:
                # 跳过缺失或格式异常的数据行
                print(f"⚠️ 缺失数据: {row}")
                continue

            dates.append(current_date)
            highs.append(high)
            lows.append(low)

    return dates, highs, lows


def plot_temperature_comparison(sitka_dates, sitka_highs, sitka_lows,
                                 dv_dates, dv_highs, dv_lows):
    """绑制双城市温度对比图"""

    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(16, 6))
    fig.suptitle('Sitka vs 死亡谷 2024 年温度对比', fontsize=16, fontweight='bold')

    # ---- 左图：Sitka ----
    ax1.plot(sitka_dates, sitka_highs, c='red', alpha=0.6, label='最高温')
    ax1.plot(sitka_dates, sitka_lows, c='blue', alpha=0.6, label='最低温')
    ax1.fill_between(sitka_dates, sitka_highs, sitka_lows,
                     facecolor='blue', alpha=0.15, label='温差区间')
    ax1.set_title('Sitka（阿拉斯加）', fontsize=13, fontweight='bold')
    ax1.set_xlabel('日期', fontsize=11)
    ax1.set_ylabel('温度 (°C)', fontsize=11)
    ax1.legend(loc='upper right')
    ax1.grid(True, alpha=0.3)

    # ---- 右图：死亡谷 ----
    ax2.plot(dv_dates, dv_highs, c='red', alpha=0.6, label='最高温')
    ax2.plot(dv_dates, dv_lows, c='blue', alpha=0.6, label='最低温')
    ax2.fill_between(dv_dates, dv_highs, dv_lows,
                     facecolor='orange', alpha=0.15, label='温差区间')
    ax2.set_title('死亡谷（加利福尼亚）', fontsize=13, fontweight='bold')
    ax2.set_xlabel('日期', fontsize=11)
    ax2.set_ylabel('温度 (°C)', fontsize=11)
    ax2.legend(loc='upper right')
    ax2.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.savefig('temperature_comparison.png', dpi=150, bbox_inches='tight')
    plt.show()
    print("✅ 温度对比图已生成")


if __name__ == '__main__':
    # 注意：请确保 data 目录下有对应的 CSV 文件
    # 这里使用模拟数据演示完整流程

    # Sitka 数据（模拟）
    sitka_dates, sitka_highs, sitka_lows = [], [], []
    from datetime import timedelta
    base_date = datetime(2024, 1, 1)
    for i in range(365):
        sitka_dates.append(base_date + timedelta(days=i))
        sitka_highs.append(5 + (i % 180 - 90) * 0.05 + (i // 30))  # 模拟季节变化
        sitka_lows.append(-2 + (i % 180 - 90) * 0.03 + (i // 30))

    # 死亡谷数据（模拟）
    dv_dates, dv_highs, dv_lows = [], [], []
    for i in range(365):
        dv_dates.append(base_date + timedelta(days=i))
        dv_highs.append(35 + (i % 180 - 90) * 0.08)
        dv_lows.append(18 + (i % 180 - 90) * 0.05)

    plot_temperature_comparison(
        sitka_dates, sitka_highs, sitka_lows,
        dv_dates, dv_highs, dv_lows
    )

# 方案一：原生 Python try/except（适合小规模数据）
for row in reader:
    try:
        high = int(row[4])  # TMAX 列
        low = int(row[5])   # TMIN 列
    except ValueError:
        print(f"跳过异常行: {row}")
        continue
    # 正常处理...

# 方案二：pandas 自动处理（适合大规模数据）
import pandas as pd
df = pd.read_csv('data/weather.csv')
df['TMAX'] = pd.to_numeric(df['TMAX'], errors='coerce')  # 无法转换的变为 NaN
df['TMAX'] = df['TMAX'].fillna(df['TMAX'].mean())         # 用均值填充

import json

# === loads：字符串 → Python 对象 ===
json_str = '{"name": "python", "version": "3.12", "features": ["动态类型", "丰富生态"]}'
data = json.loads(json_str)
print(data['name'])          # python
print(data['features'][0])  # 动态类型

# === dumps：Python 对象 → 格式化字符串 ===
pretty_json = json.dumps(data, indent=2, ensure_ascii=False)
print(pretty_json)
# {
#   "name": "python",
#   "version": "3.12",
#   "features": ["动态类型", "丰富生态"]
# }

# === load/dump：文件读写 ===
with open('data.json', 'w', encoding='utf-8') as f:
    json.dump(data, f, indent=2, ensure_ascii=False)

with open('data.json', 'r', encoding='utf-8') as f:
    loaded = json.load(f)
    print(loaded)  # {'name': 'python', ...}

"""
GeoJSON 地震数据解析与 Plotly 地图可视化
数据源：USGS Earthquake API (https://earthquake.usgs.gov/)
"""
import json
import plotly.express as px


def extract_earthquake_data(json_file):
    """
    从 GeoJSON 文件提取地震数据

    Args:
        json_file: GeoJSON 文件路径

    Returns:
        list[dict]: 包含经度、纬度、震级、标题的字典列表
    """
    with open(json_file, 'r', encoding='utf-8') as f:
        all_eq_data = json.load(f)

    all_eq_features = all_eq_data['features']

    eq_data = []
    for eq_dict in all_eq_features:
        eq = {
            'longitude': eq_dict['geometry']['coordinates'][0],
            'latitude': eq_dict['geometry']['coordinates'][1],
            'magnitude': eq_dict['properties']['mag'],
            'title': eq_dict['properties']['title'],
            'depth': eq_dict['geometry']['coordinates'][2],  # 深度(km)
        }
        eq_data.append(eq)

    return eq_data


def plot_earthquake_map(eq_data, output_file='earthquake_map.html'):
    """
    使用 Plotly Express 创建交互式地震散点图

    Args:
        eq_data: 地震数据列表
        output_file: 输出 HTML 文件名
    """
    # 提取各维度数据
    lons = [eq['longitude'] for eq in eq_data]
    lats = [eq['latitude'] for eq in eq_data]
    mags = [eq['magnitude'] for eq in eq_data]
    titles = [eq['title'] for eq in eq_data]
    depths = [eq['depth'] for eq in eq_data]

    fig = px.scatter_geo(
        lat=lats,
        lon=lons,
        size=mags,              # 点大小映射到震级
        color=mags,             # 颜色映射到震级
        hover_name=titles,      # 悬停显示标题
        title='全球地震分布图',
        color_continuous_color='YlOrRd',  # 黄-橙-红渐变
        projection='natural earth',       # 地球投影方式
        size_max=15,                      # 最大点尺寸
        opacity=0.7,
    )

    fig.update_layout(
        title_font_size=18,
        title_x=0.5,
        coloraxis_colorbar=dict(title="震级"),
    )

    fig.write_html(output_file)
    print(f"✅ 地震地图已保存: {output_file}")
    return fig


if __name__ == '__main__':
    # 构建模拟 GeoJSON 数据用于演示
    mock_geojson = {
        "type": "FeatureCollection",
        "metadata": {"count": 5, "generated": 1717728000000},
        "features": [
            {
                "type": "Feature",
                "properties": {
                    "mag": 7.2,
                    "place": "95km W of Port-Olry, Vanuatu",
                    "time": 1717728000000,
                    "title": "M 7.2 - 95km W of Port-Olry, Vanuatu"
                },
                "geometry": {
                    "type": "Point",
                    "coordinates": [167.3, -15.5, 135.0]  # [经度, 纬度, 深度]
                }
            },
            {
                "type": "Feature",
                "properties": {
                    "mag": 5.8,
                    "place": "12km ENE of Hualien City, Taiwan",
                    "time": 1717641600000,
                    "title": "M 5.8 - 12km ENE of Hualien City, Taiwan"
                },
                "geometry": {
                    "type": "Point",
                    "coordinates": [121.6, 24.0, 10.5]
                }
            },
            {
                "type": "Feature",
                "properties": {
                    "mag": 6.5,
                    "place": "78km SSE of Padangsidimpuan, Indonesia",
                    "time": 1717555200000,
                    "title": "M 6.5 - 78km SSE of Padangsidimpuan, Indonesia"
                },
                "geometry": {
                    "type": "Point",
                    "coordinates": [98.9, 1.2, 89.0]
                }
            },
            {
                "type": "Feature",
                "properties": {
                    "mag": 4.2,
                    "place": "5km NE of Ridgecrest, CA",
                    "time": 1717468800000,
                    "title": "M 4.2 - 5km NE of Ridgecrest, CA"
                },
                "geometry": {
                    "type": "Point",
                    "coordinates": [-117.67, 35.62, 2.1]
                }
            },
            {
                "type": "Feature",
                "properties": {
                    "mag": 5.1,
                    "place": "42km WSW of Ovalle, Chile",
                    "time": 1717382400000,
                    "title": "M 5.1 - 42km WSW of Ovalle, Chile"
                },
                "geometry": {
                    "type": "Point",
                    "coordinates": [-71.5, -30.8, 25.0]
                }
            }
        ]
    }

    # 先保存模拟数据
    with open('data/earthquakes.json', 'w', encoding='utf-8') as f:
        json.dump(mock_geojson, f, indent=2, ensure_ascii=False)

    # 解析并可视化
    eq_data = extract_earthquake_data('data/earthquakes.json')
    print(f"\n共解析 {len(eq_data)} 条地震记录:")
    for eq in eq_data[:3]:
        print(f"  📍 {eq['title']} | 震级: {eq['magnitude']}")

    fig = plot_earthquake_map(eq_data)
    # fig.show()  # 取消注释可在浏览器中打开

pip install requests

import requests

# === 1. 基本 GET 请求 ===
response = requests.get('https://httpbin.org/get')
print(response.status_code)     # 200
print(response.text[:100])      # 响应文本前100字符

# === 2. 带查询参数 ===
params = {'q': 'python programming', 'lang': 'en'}
response = requests.get('https://httpbin.org/get', params=params)
print(response.url)  # https://httpbin.org/get?q=python+programming&lang=en

# === 3. 自定义请求头 ===
headers = {
    'User-Agent': 'Mozilla/5.0 (DataAnalysisBot/1.0)',
    'Accept': 'application/json',
}
response = requests.get('https://api.github.com', headers=headers)

# === 4. 解析 JSON 响应 ===
data = response.json()
print(type(data))  # <class 'dict'>

# === 5. POST 请求（发送 JSON）===
payload = {'username': 'alice', 'role': 'analyst'}
response = requests.post(
    'https://httpbin.org/post',
    json=payload,
    headers={'Content-Type': 'application/json'}
)
result = response.json()['json']
print(result)  # {'username': 'alice', 'role': 'analyst'}

# === 6. 超时设置（重要！）===
try:
    response = requests.get('https://api.example.com/data', timeout=5)
except requests.exceptions.Timeout:
    print("❌ 请求超时！")

# === 7. Session 复用连接（高性能推荐）===
session = requests.Session()
session.headers.update({'User-Agent': 'MyDataApp/1.0'})

# 多次请求复用同一 TCP 连接
r1 = session.get('https://api.example.com/users')
r2 = session.get('https://api.example.com/posts')
session.close()

# API Key 认证（最常用）
headers = {'X-API-Key': 'your_api_key_here'}
response = requests.get('https://api.example.com/v1/data', headers=headers)

# Bearer Token 认证
headers = {'Authorization': 'Bearer eyJhbGciOiJIUzI1NiJ9...'}
response = requests.get('https://api.example.com/v1/user', headers=headers)

# Basic Auth
from requests.auth import HTTPBasicAuth
response = requests.get(
    'https://api.example.com/v1/resource',
    auth=HTTPBasicAuth('username', 'password')
)

import time
import requests


def api_request_with_retry(url, max_retries=3, initial_delay=1):
    """
    带重试机制的 API 请求

    Args:
        url: 请求地址
        max_retries: 最大重试次数
        initial_delay: 初始等待秒数

    Returns:
        Response 对象或 None
    """
    delay = initial_delay

    for attempt in range(max_retries + 1):
        try:
            response = requests.get(
                url,
                headers={'User-Agent': 'DataVizBot/1.0'},
                timeout=10
            )

            if response.status_code == 200:
                return response

            elif response.status_code == 429:
                # 从响应头获取限流信息
                retry_after = int(response.headers.get('Retry-After', delay))
                print(f"⏳ 触发限流，等待 {retry_after} 秒后重试... (第{attempt+1}次)")
                time.sleep(retry_after)
                delay *= 2  # 指数退避

            elif response.status_code >= 500:
                print(f"⚠️ 服务器错误 {response.status_code}，重试中...")
                time.sleep(delay)
                delay *= 2

            else:
                print(f"❌ 请求失败: {response.status_code} - {response.reason}")
                return None

        except requests.exceptions.RequestException as e:
            print(f"❌ 网络异常: {e}")
            time.sleep(delay)
            delay *= 2

    print(f"❌ 达到最大重试次数 ({max_retries})")
    return None

"""
GitHub API 实战：搜索热门 Python 项目并用 Plotly 可视化
API 文档：https://docs.github.com/en/rest/search
"""
import requests
import plotly.graph_objects as go
from plotly.subplots import make_subplots


def search_github_repositories(query, sort='stars', per_page=30):
    """
    搜索 GitHub 仓库

    Args:
        query: 搜索关键词（如 language:python）
        sort: 排序方式 (stars/forks/updated)
        per_page: 返回数量

    Returns:
        dict: API 响应数据
    """
    url = 'https://api.github.com/search/repositories'
    params = {
        'q': query,
        'sort': sort,
        'per_page': per_page,
    }
    headers = {
        'Accept': 'application/vnd.github.v3+json',
        'User-Agent': 'Python-DataViz/1.0',
    }

    response = requests.get(url, params=params, headers=headers, timeout=15)

    if response.status_code != 200:
        raise Exception(f"API 请求失败: {response.status_code}")

    return response.json()


def extract_repo_data(response_json):
    """
    从 API 响应中提取关键字段

    GitHub API 响应结构:
    {
        "total_count": 1234567,
        "items": [{
            "name": "awesome-python",
            "full_name": "vinta/awesome-python",
            "owner": {"login": "vinta"},
            "stargazers_count": 150000,
            "forks_count": 25000,
            "html_url": "https://github.com/...",
            "description": "Curated list...",
            "language": "Python",
            "created_at": "2014-06-27T..."
        }, ...]
    }
    """
    items = response_json.get('items', [])

    repos = []
    for item in items:
        repo = {
            'name': item['name'],
            'owner': item['owner']['login'],
            'stars': item['stargazers_count'],
            'forks': item['forks_count'],
            'url': item['html_url'],
            'description': item.get('description', 'N/A') or 'N/A',
            'language': item.get('language', 'Unknown'),
        }
        repos.append(repo)

    return repos


def plot_top_repos(repos, top_n=15, output_file='github_top_repos.html'):
    """
    使用 Plotly 创建交互式水平条形图

    Args:
        repos: 仓库数据列表
        top_n: 显示前 N 个
        output_file: 输出文件名
    """
    # 按 star 数降序排列，取前 top_n
    sorted_repos = sorted(repos, key=lambda x: x['stars'], reverse=True)[:top_n]

    names = [f"{r['owner']}/{r['name']}" for r in sorted_repos]
    stars = [r['stars'] for r in sorted_repos]
    forks = [r['forks'] for r in sorted_repos]
    descriptions = [r['description'][:60] + '...' if len(r['description']) > 60
                    else r['description'] for r in sorted_repos]
    urls = [r['url'] for r in sorted_repos]

    # 创建子图：条形图 + 信息表格
    fig = make_subplots(
        rows=1, cols=2,
        column_widths=[0.7, 0.3],
        specs=[[{"type": "bar"}, {"type": "table"}]],
        subplot_titles=['Star 数排名', '项目详情']
    )

    # 条形图
    fig.add_trace(
        go.Bar(
            y=names,
            x=stars,
            orientation='h',
            marker_color='#0366d6',
            text=stars,
            textposition='outside',
            hovertemplate=(
                '<b>%{y}</b><br>'
                '⭐ Stars: %{x:,}<br>'
                '🍴 Forks: %{customdata[0]:,}<br>'
                '<extra></extra>'
            ),
            customdata=list(zip(forks, descriptions)),
        ),
        row=1, col=1
    )

    # 信息表格
    table_data = [
        [r['name'] for r in sorted_repos],
        [r['language'] for r in sorted_repos],
        [f"{r['stars']:,}" for r in sorted_repos],
        [f"{r['forks']:,}" for r in sorted_repos],
    ]
    fig.add_trace(
        go.Table(
            header=dict(
                values=['项目', '语言', 'Stars', 'Forks'],
                fill_color='#0366d6',
                font=dict(color='white', size=12),
                align='center'
            ),
            cells=dict(
                values=table_data,
                fill_color=[['#f6f8fa', 'white'] * top_n],
                align='left',
                font_size=11,
                height=25
            )
        ),
        row=1, col=2
    )

    fig.update_layout(
        title={
            'text': f'🔥 GitHub 热门 Python 项目 Top {top_n}',
            'x': 0.5,
            'font_size': 20
        },
        height=700,
        showlegend=False,
        bargap=0.15,
        yaxis=dict(autorange='reversed'),  # 第一名在最上面
    )

    fig.update_xaxes(title_text='Stars 数量', row=1, col=1)

    fig.write_html(output_file)
    print(f"✅ GitHub 项目排行图已保存: {output_file}")
    return fig


if __name__ == '__main__':
    # 搜索语言为 Python 的热门项目
    print("🔍 正在搜索 GitHub 热门 Python 项目...")
    response = search_github_repositories(
        query='language:python',
        sort='stars',
        per_page=30
    )

    total = response.get('total_count', 0)
    print(f"📊 共找到 {total:,} 个相关仓库")

    repos = extract_repo_data(response)
    print(f"\n🏆 Top 5 项目:")
    for i, repo in enumerate(repos[:5], 1):
        print(f"  {i}. {repo['owner']}/{repo['name']}"
              f" ⭐{repo['stars']:,} 🍴{repo['forks']:,}")

    # 可视化
    fig = plot_top_repos(repos, top_n=15)
    # fig.show()  # 取消注释可在浏览器中查看

"""
Hacker News API 实战：获取热门文章并可视化
API 地址：https://hacker-news.firebaseio.com/v0/
"""
import requests
import time
from operator import itemgetter
import matplotlib.pyplot as plt
import numpy as np

plt.rcParams['font.sans-serif'] = ['SimHei', 'Arial Unicode MS']
plt.rcParams['axes.unicode_minus'] = False


BASE_URL = 'https://hacker-news.firebaseio.com/v0'


def get_top_story_ids(limit=20):
    """获取热门文章 ID 列表"""
    url = f'{BASE_URL}/topstories.json'
    response = requests.get(url, timeout=10)
    return response.json()[:limit]


def get_item(item_id):
    """获取单篇文章详情"""
    url = f'{BASE_URL}/item/{item_id}.json'
    response = requests.get(url, timeout=10)
    return response.json()


def fetch_top_stories_with_sequential(ids):
    """
    顺序请求模式（简单但慢）
    每次请求一个，完成后才请求下一个
    """
    stories = []
    start_time = time.time()

    for sid in ids:
        item = get_item(sid)
        if item and item.get('type') == 'story':
            stories.append({
                'title': item.get('title', 'N/A'),
                'score': item.get('score', 0),
                'descendants': item.get('descendants', 0),
                'url': item.get('url', ''),
                'by': item.get('by', 'unknown'),
            })
        time.sleep(0.1)  # 礼貌延迟，避免被限流

    elapsed = time.time() - start_time
    print(f"⏱️ 顺序模式耗时: {elapsed:.2f}s ({len(stories)} 篇文章)")
    return stories


def fetch_top_stories_with_concurrent(ids, batch_size=5):
    """
    批量并发模式（快但需控制并发数）
    使用 requests.Session + 批量请求
    """
    stories = []
    session = requests.Session()
    start_time = time.time()

    for i in range(0, len(ids), batch_size):
        batch_ids = ids[i:i + batch_size]
        urls = [f'{BASE_URL}/item/{sid}.json' for sid in batch_ids]

        # 批量请求（Session 复用连接）
        responses = [session.get(url, timeout=10) for url in urls]

        for resp in responses:
            item = resp.json()
            if item and item.get('type') == 'story':
                stories.append({
                    'title': item.get('title', 'N/A'),
                    'score': item.get('score', 0),
                    'descendants': item.get('descendants', 0),
                    'url': item.get('url', ''),
                    'by': item.get('by', 'unknown'),
                })

        time.sleep(0.2)  # 批次间延迟

    session.close()
    elapsed = time.time() - start_time
    print(f"⏱️ 并发模式耗时: {elapsed:.2f}s ({len(stories)} 篇文章)")
    return stories


def plot_hn_rankings(stories, output_file='hn_rankings.png'):
    """绑制 HN 热门文章排行榜"""

    # 按 score 降序排列
    sorted_stories = sorted(stories, key=itemgetter('score'), reverse=True)[:15]

    titles = [s['title'][:30] + '...' if len(s['title']) > 30
              else s['title'] for s in sorted_stories]
    scores = [s['score'] for s in sorted_stories]
    comments = [s['descendants'] for s in sorted_stories]

    fig, ax = plt.subplots(figsize=(12, 8))

    y_pos = np.arange(len(titles))

    # 水平条形图
    bars = ax.barh(y_pos, scores, color='#ff6600', alpha=0.8, edgecolor='white')

    # 在柱子上添加评论数标注
    for i, (bar, comment_count) in enumerate(zip(bars, comments)):
        ax.text(bar.get_width() + 5, bar.get_y() + bar.get_height()/2,
                f'💬 {comment_count}', va='center', fontsize=9, color='#666')

    ax.set_yticks(y_pos)
    ax.set_yticklabels(titles, fontsize=10)
    ax.invert_yaxis()  # 第一名在最上面
    ax.set_xlabel('投票分数 (Points)', fontsize=12)
    ax.set_title('🔥 Hacker News 当前热门文章 TOP 15',
                 fontsize=14, fontweight='bold', pad=15)
    ax.grid(axis='x', alpha=0.3)
    ax.set_xlim(0, max(scores) * 1.2)

    # 添加统计信息文本框
    stats_text = (
        f"平均分数: {np.mean(scores):.0f}\n"
        f"平均评论: {np.mean(comments):.0f}\n"
        f"最高分: {max(scores)}"
    )
    ax.text(0.98, 0.02, stats_text,
            transform=ax.transAxes,
            fontsize=10,
            verticalalignment='bottom',
            horizontalalignment='right',
            bbox=dict(boxstyle='round', facecolor='#fff3cd',
                     edgecolor='#ffc107', alpha=0.9))

    plt.tight_layout()
    plt.savefig(output_file, dpi=150, bbox_inches='tight')
    plt.show()
    print(f"✅ HN 排行榜已保存: {output_file}")


if __name__ == '__main__':
    print("📰 获取 Hacker News 热门文章...\n")

    # 获取热门文章 ID
    story_ids = get_top_story_ids(limit=20)
    print(f"获取到 {len(story_ids)} 个热门文章 ID")

    # 使用批量并发模式获取详情
    stories = fetch_top_stories_with_concurrent(story_ids, batch_size=5)

    # 打印前 5 名
    print("\n🏆 HN 热门文章 Top 5:")
    for i, s in enumerate(sorted(stories, key=lambda x: x['score'], reverse=True)[:5], 1):
        print(f"  {i}. [{s['score']}pts] {s['title']}"
              f" (💬{s['descendants']}) by {s['by']}")

    # 可视化
    plot_hn_rankings(stories)

import json
import os
import hashlib
import time
from pathlib import Path


class SimpleCache:
    """简单的文件缓存系统"""

    def __init__(self, cache_dir='cache', ttl_seconds=3600):
        self.cache_dir = Path(cache_dir)
        self.ttl = ttl_seconds
        self.cache_dir.mkdir(exist_ok=True)

    def _get_cache_path(self, url):
        """根据 URL 生成缓存文件路径"""
        url_hash = hashlib.md5(url.encode()).hexdigest()
        return self.cache_dir / f"{url_hash}.json"

    def get(self, url):
        """尝试从缓存获取数据"""
        cache_path = self._get_cache_path(url)

        if cache_path.exists():
            age = time.time() - cache_path.stat().st_mtime
            if age < self.ttl:
                print(f"📦 命中缓存: {url[:50]}...")
                with open(cache_path, 'r', encoding='utf-8') as f:
                    return json.load(f)
            else:
                print(f"⏰ 缓存已过期: {url[:50]}...")
        return None

    def set(self, url, data):
        """写入缓存"""
        cache_path = self._get_cache_path(url)
        with open(cache_path, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False)
        print(f"💾 已缓存: {url[:50]}...")


# 使用示例
cache = SimpleCache(cache_dir='api_cache', ttl_seconds=1800)  # 30分钟过期

url = 'https://api.github.com/search/repositories?q=language:python&sort=stars'

# 先查缓存
cached_data = cache.get(url)
if cached_data is None:
    # 缓存未命中，发起请求
    response = requests.get(url, headers={'User-Agent': 'MyApp/1.0'})
    data = response.json()
    cache.set(url, data)  # 存入缓存
else:
    data = cached_data

import csv

# 方式一：逐行迭代（内存友好）
chunk_size = 10000
rows_processed = 0

with open('large_dataset.csv', 'r', encoding='utf-8') as f:
    reader = csv.reader(f)
    header = next(reader)

    chunk = []
    for row in reader:
        chunk.append(row)
        rows_processed += 1

        if len(chunk) >= chunk_size:
            process_chunk(chunk)  # 处理当前批次
            chunk.clear()          # 释放内存
            print(f"已处理 {rows_processed:,} 行")

    # 处理剩余不足一个 chunk 的数据
    if chunk:
        process_chunk(chunk)

# 方式二：pandas 分块读取（推荐用于数据分析）
import pandas as pd

chunk_iter = pd.read_csv(
    'large_dataset.csv',
    chunksize=50000,           # 每块 5 万行
    encoding='utf-8',
    usecols=['date', 'value']  # 只读需要的列
)

results = []
for i, chunk in enumerate(chunk_iter):
    aggregated = chunk.groupby('date')['value'].sum()
    results.append(aggregated)
    print(f"已处理第 {i+1} 个数据块")

final_result = pd.concat(results).groupby(level=0).sum()

# 使用 tenacity 库实现优雅重试
# pip install tenacity
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests.exceptions

@retry(
    stop=stop_after_attempt(3),                    # 最多重试 3 次
    wait=wait_exponential(multiplier=1, min=1, max=10),  # 指数退避
    retry=retry_if_exception_type((               # 仅对以下异常重试
        requests.exceptions.ConnectionError,
        requests.exceptions.Timeout,
    )),
    reraise=True                                   # 最终失败时抛出原异常
)
def robust_api_call(url):
    """带自动重试的 API 调用"""
    return requests.get(url, timeout=10)