Python 自动化实战：API 自动化——调用50+免费API

pip install requests

import requests

# 1. GET 请求（拿数据）
resp = requests.get(
    "https://api.example.com/data",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    params={"page": 1, "limit": 20}  # 查询参数
)

# 2. POST 请求（提交数据）
resp = requests.post(
    "https://api.example.com/create",
    headers={"Authorization": "Bearer YOUR_API_KEY"},
    json={"name": "test", "value": 42}  # JSON body
)

# 3. 解析响应
if resp.status_code == 200:
    data = resp.json()  # 直接拿到 dict
    print(data["results"])
else:
    print(f"请求失败: {resp.status_code} - {resp.text}")

import requests

def get_weather(lat, lon):
    """获取指定坐标的7天天气预报"""
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": lat,
        "longitude": lon,
        "daily": ["temperature_2m_max", "temperature_2m_min",
                   "precipitation_sum", "weathercode"],
        "timezone": "Asia/Shanghai",
        "forecast_days": 7
    }
    resp = requests.get(url, params=params)
    data = resp.json()

    print(f"📍 坐标 ({lat}, {lon}) 未来7天预报：")
    for i, date in enumerate(data["daily"]["time"]):
        high = data["daily"]["temperature_2m_max"][i]
        low = data["daily"]["temperature_2m_min"][i]
        rain = data["daily"]["precipitation_sum"][i]
        print(f"  {date} | {low}°C ~ {high}°C | 降水 {rain}mm")

# 上海坐标
get_weather(31.23, 121.47)

import requests
from datetime import datetime

def convert_currency(amount, from_currency, to_currency):
    """实时汇率转换"""
    url = "https://api.frankfurter.app/latest"
    params = {
        "amount": amount,
        "from": from_currency,
        "to": to_currency
    }
    resp = requests.get(url, params=params)
    data = resp.json()

    rate = data["rates"][to_currency]
    result = data["rates"][to_currency]  # amount * rate 已在响应中
    print(f"💱 {amount} {from_currency} = {result} {to_currency}")
    print(f"   汇率: 1 {from_currency} = {rate} {to_currency}")
    return result

# 100 美元换人民币
convert_currency(100, "USD", "CNY")

# 查看所有支持币种
def list_currencies():
    resp = requests.get("https://api.frankfurter.app/currencies")
    currencies = resp.json()
    print(f"支持 {len(currencies)} 种货币:")
    for code, name in list(currencies.items())[:10]:
        print(f"  {code}: {name}")

list_currencies()

import requests

def get_trending_repos(language="python", per_page=5):
    """获取 GitHub 上某语言的热门仓库"""
    url = "https://api.github.com/search/repositories"
    params = {
        "q": f"language:{language}",
        "sort": "stars",
        "order": "desc",
        "per_page": per_page
    }
    headers = {"Accept": "application/vnd.github.v3+json"}

    resp = requests.get(url, headers=headers, params=params)
    data = resp.json()

    print(f"🔥 {language} 语言 Stars 最高的仓库：")
    for repo in data["items"]:
        print(f"  ⭐ {repo['stargazers_count']:,} | {repo['full_name']}")
        print(f"     {repo['description'][:80] if repo['description'] else '无描述'}")
        print(f"     {repo['html_url']}\n")

get_trending_repos("rust", 5)

def check_latest_release(owner, repo):
    """检查仓库最新 Release"""
    url = f"https://api.github.com/repos/{owner}/{repo}/releases/latest"
    resp = requests.get(url)
    if resp.status_code == 200:
        release = resp.json()
        print(f"📦 {owner}/{repo}")
        print(f"   最新版本: {release['tag_name']}")
        print(f"   发布时间: {release['published_at']}")
        print(f"   下载地址: {release['html_url']}")
    else:
        print(f"该仓库没有 Release")

check_latest_release("tauri-apps", "tauri")

import requests

def call_llm(api_key, base_url, model, prompt):
    """通用大模型调用函数——兼容 OpenAI 格式的所有平台"""
    url = f"{base_url}/chat/completions"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "model": model,
        "messages": [
            {"role": "system", "content": "你是 Python 技术专家，回答简洁"},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.7,
        "max_tokens": 500
    }

    resp = requests.post(url, headers=headers, json=payload)
    data = resp.json()
    return data["choices"][0]["message"]["content"]

# 示例：调用 DeepSeek（替换成你自己的 key）
# result = call_llm(
#     api_key="sk-your-deepseek-key",
#     base_url="https://api.deepseek.com/v1",
#     model="deepseek-chat",
#     prompt="用 3 句话解释什么是装饰器"
# )
# print(result)

import requests

def search_photos(query, per_page=5):
    """搜索 Unsplash 高质量图片"""
    url = "https://api.unsplash.com/search/photos"
    # 注册 https://unsplash.com/developers 获取 Access Key
    headers = {"Authorization": "Client-ID YOUR_ACCESS_KEY"}
    params = {"query": query, "per_page": per_page}

    resp = requests.get(url, headers=headers, params=params)
    data = resp.json()

    print(f"🖼️ 搜索 '{query}' 的结果：")
    for photo in data["results"]:
        print(f"  📷 {photo['description'][:60] if photo['description'] else '无描述'}")
        print(f"     作者: {photo['user']['name']}")
        print(f"     原图: {photo['urls']['regular']}")
        print(f"     缩略图: {photo['urls']['thumb']}\n")

# search_photos("coding setup")

import requests
import time
from functools import wraps
from typing import Callable

class APIClient:
    """通用 API 客户端——错误处理 + 重试 + 限流 + 超时"""

    def __init__(self, base_url: str, headers: dict = None,
                 timeout: int = 10, max_retries: int = 3):
        self.base_url = base_url.rstrip("/")
        self.headers = headers or {}
        self.timeout = timeout
        self.max_retries = max_retries
        self.session = requests.Session()
        self.session.headers.update(self.headers)

    def _request(self, method: str, path: str, **kwargs):
        """核心请求方法——自动重试 + 指数退避"""
        url = f"{self.base_url}{path}"
        kwargs.setdefault("timeout", self.timeout)

        for attempt in range(self.max_retries):
            try:
                resp = self.session.request(method, url, **kwargs)

                # 限流：等到 Retry-After 再重试
                if resp.status_code == 429:
                    wait = int(resp.headers.get("Retry-After", 5))
                    print(f"⚠️ 限流，{wait}s 后重试 (第 {attempt+1}/{self.max_retries} 次)")
                    time.sleep(wait)
                    continue

                # 服务端错误：指数退避
                if resp.status_code >= 500:
                    wait = 2 ** attempt
                    print(f"⚠️ 服务端错误 {resp.status_code}，{wait}s 后重试")
                    time.sleep(wait)
                    continue

                resp.raise_for_status()
                return resp.json()

            except requests.exceptions.Timeout:
                print(f"⏱️ 请求超时 (第 {attempt+1}/{self.max_retries} 次)")
                if attempt == self.max_retries - 1:
                    raise
                time.sleep(2 ** attempt)

            except requests.exceptions.RequestException as e:
                print(f"❌ 请求失败: {e}")
                if attempt == self.max_retries - 1:
                    raise
                time.sleep(2 ** attempt)

        raise Exception(f"超过最大重试次数 {self.max_retries}")

    def get(self, path: str, **kwargs):
        return self._request("GET", path, **kwargs)

    def post(self, path: str, **kwargs):
        return self._request("POST", path, **kwargs)


# 使用示例
weather_client = APIClient(
    base_url="https://api.open-meteo.com/v1",
    timeout=5
)

data = weather_client.get("/forecast", params={
    "latitude": 31.23,
    "longitude": 121.47,
    "daily": "temperature_2m_max",
    "forecast_days": 3
})
print(data["daily"])

from APIClient import APIClient  # 上面写的通用框架
import json
from datetime import datetime

# 1. 坐标转城市名（Nominatim）
geo = APIClient("https://nominatim.openstreetmap.org")
geo.session.headers["User-Agent"] = "WeatherBot/1.0"  # Nominatim 要求

# 2. 天气数据（Open-Meteo）
weather = APIClient("https://api.open-meteo.com/v1")

# 3. 推送通知（假设你有一个 webhook）
# notify = APIClient("https://your-webhook.com")


def daily_weather_report(lat, lon):
    """每日天气报告"""
    # 获取城市名
    location = geo.get("/reverse", params={
        "lat": lat, "lon": lon, "format": "json"
    })
    city = location.get("address", {}).get("city", "未知城市")

    # 获取天气
    data = weather.get("/forecast", params={
        "latitude": lat, "longitude": lon,
        "daily": ["temperature_2m_max", "temperature_2m_min",
                   "precipitation_sum", "weathercode"],
        "forecast_days": 1,
        "timezone": "Asia/Shanghai"
    })

    today = data["daily"]
    report = f"""
📅 {today['time'][0]} {city}天气预报
━━━━━━━━━━━━━━━━━━
🌡️ 温度: {today['temperature_2m_min'][0]}°C ~ {today['temperature_2m_max'][0]}°C
🌧️ 降水: {today['precipitation_sum'][0]}mm
💡 {'今天有雨，出门带伞' if today['precipitation_sum'][0] > 0 else '天气不错'}
"""

    print(report)
    # 推送到飞书/Telegram/邮件（具体代码见系列第 01 篇）
    return report

daily_weather_report(31.23, 121.47)

import os
api_key = os.environ["OPENAI_API_KEY"]  # 正确
# api_key = "sk-xxxxxxxx"  # 错误：一推 GitHub 就泄露

resp = requests.get(url)
remaining = resp.headers.get("X-RateLimit-Remaining")
if remaining and int(remaining) < 10:
    print("⚠️ 额度预警，暂停使用")
    time.sleep(60)

from concurrent.futures import ThreadPoolExecutor

def fetch_all(urls):
    with ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(
            lambda u: requests.get(u, timeout=5).json(),
            urls
        ))
    return results

import hashlib

# 记录响应 key 结构的哈希
resp = requests.get("https://api.example.com/data").json()
structure_hash = hashlib.md5(
    str(sorted(resp.keys())).encode()
).hexdigest()

# 存到文件，下次对比，变了就告警