Python и API: мощные инструменты для работы с данными в интернете

pip install requests

import requests

# Отправляем GET-запрос к публичному API
response = requests.get('https://jsonplaceholder.typicode.com/posts/1')

# Проверяем статус ответа
if response.status_code == 200:
# Преобразуем ответ из JSON в Python-словарь
data = response.json()
print(f"Заголовок поста: {data['title']}")
print(f"Содержание: {data['body']}")
else:
print(f"Ошибка при запросе: {response.status_code}")

import requests

# Параметры запроса
params = {
'q': 'Москва',
'appid': 'ваш_api_ключ',
'units': 'metric'
}

# Отправка запроса с параметрами
response = requests.get('https://api.openweathermap.org/data/2.5/weather', params=params)

if response.status_code == 200:
weather_data = response.json()
print(f"Температура в Москве: {weather_data['main']['temp']}°C")
print(f"Ощущается как: {weather_data['main']['feels_like']}°C")
print(f"Погодные условия: {weather_data['weather'][0]['description']}")
else:
print(f"Ошибка: {response.status_code}")

import requests

headers = {
'Authorization': 'Bearer ваш_api_ключ',
'Content-Type': 'application/json'
}

response = requests.get('https://api.example.com/data', headers=headers)

if response.status_code == 200:
print("Успешная аутентификация")
print(response.json())
else:
print(f"Ошибка аутентификации: {response.status_code}")

import requests
import json

# Данные для отправки
new_post = {
'title': 'Мой первый пост через API',
'body': 'Это содержимое моего первого поста, отправленного через API',
'userId': 1
}

# Преобразуем Python-словарь в JSON
json_data = json.dumps(new_post)

# Устанавливаем заголовок Content-Type
headers = {'Content-Type': 'application/json'}

# Отправляем POST-запрос
response = requests.post(
'https://jsonplaceholder.typicode.com/posts',
data=json_data,
headers=headers
)

if response.status_code == 201: # 201 – Created
created_post = response.json()
print(f"Пост успешно создан с ID: {created_post['id']}")
print(f"Заголовок: {created_post['title']}")
else:
print(f"Ошибка при создании поста: {response.status_code}")

import requests

# Данные для обновления
updated_post = {
'id': 1,
'title': 'Обновленный заголовок',
'body': 'Обновленное содержимое поста',
'userId': 1
}

# Отправляем PUT-запрос
response = requests.put(
'https://jsonplaceholder.typicode.com/posts/1',
json=updated_post # Автоматическое преобразование в JSON
)

if response.status_code == 200:
updated_data = response.json()
print(f"Пост успешно обновлен: {updated_data}")
else:
print(f"Ошибка при обновлении: {response.status_code}")

import requests

# Только поля, которые нужно обновить
patch_data = {
'title': 'Только новый заголовок'
}

# Отправляем PATCH-запрос
response = requests.patch(
'https://jsonplaceholder.typicode.com/posts/1',
json=patch_data
)

if response.status_code == 200:
patched_post = response.json()
print(f"Частичное обновление выполнено: {patched_post}")
else:
print(f"Ошибка при частичном обновлении: {response.status_code}")

import requests

# Отправляем DELETE-запрос
response = requests.delete('https://jsonplaceholder.typicode.com/posts/1')

if response.status_code == 200 or response.status_code == 204:
print("Пост успешно удален")
else:
print(f"Ошибка при удалении: {response.status_code}")

import json

# Преобразование строки JSON в объект Python (десериализация)
json_string = '{"name": "Иван", "age": 30, "city": "Москва"}'
python_dict = json.loads(json_string)
print(f"Имя: {python_dict['name']}, Возраст: {python_dict['age']}")

# Преобразование объекта Python в строку JSON (сериализация)
user_data = {
'name': 'Анна',
'age': 25,
'languages': ['Python', 'JavaScript'],
'is_active': True
}
json_data = json.dumps(user_data, ensure_ascii=False, indent=4)
print(json_data)

import requests

response = requests.get('https://api.github.com/users/octocat')
user_data = response.json()

print(f"Имя пользователя: {user_data['login']}")
print(f"Количество репозиториев: {user_data['public_repos']}")

import requests
from json.decoder import JSONDecodeError

response = requests.get('https://example.com/api/data')

try:
data = response.json()
print(f"Получены данные: {data}")
except JSONDecodeError:
print(f"Ответ не содержит валидный JSON: {response.text}")

import requests

def get_user_data(user_id):
url = f'https://api.example.com/users/{user_id}'

try:
response = requests.get(url, timeout=5) # Устанавливаем таймаут 5 секунд

# Проверяем статус-код
if response.status_code == 200:
return response.json()
elif response.status_code == 404:
print(f"Пользователь с ID {user_id} не найден")
elif response.status_code == 401:
print("Требуется аутентификация")
elif response.status_code >= 500:
print(f"Ошибка сервера: {response.status_code}")
else:
print(f"Неизвестная ошибка: {response.status_code}")

return None

except requests.exceptions.ConnectionError:
print("Ошибка соединения с сервером")
except requests.exceptions.Timeout:
print("Превышено время ожидания ответа")
except requests.exceptions.RequestException as e:
print(f"Произошла ошибка при выполнении запроса: {e}")

return None

# Использование функции
user_data = get_user_data(123)
if user_data:
print(f"Имя пользователя: {user_data['name']}")

import requests
import jmespath

response = requests.get('https://api.github.com/users/octocat/repos')
repos = response.json()

# Извлекаем только имена репозиториев с количеством звезд > 10
popular_repos = jmespath.search("[?stargazers_count > `10`].name", repos)
print(f"Популярные репозитории: {popular_repos}")

import requests
import time
import os
from datetime import datetime

# API ключи (лучше хранить в переменных окружения)
TELEGRAM_TOKEN = os.environ.get('TELEGRAM_TOKEN')
WEATHER_API_KEY = os.environ.get('OPENWEATHER_API_KEY')
CHAT_ID = os.environ.get('TELEGRAM_CHAT_ID')

def get_weather(city):
"""Получает текущую погоду для указанного города"""
url = f"https://api.openweathermap.org/data/2.5/weather"
params = {
'q': city,
'appid': WEATHER_API_KEY,
'units': 'metric',
'lang': 'ru'
}

response = requests.get(url, params=params)

if response.status_code == 200:
data = response.json()
return {
'temperature': data['main']['temp'],
'feels_like': data['main']['feels_like'],
'description': data['weather'][0]['description'],
'humidity': data['main']['humidity'],
'wind_speed': data['wind']['speed']
}
else:
print(f"Ошибка получения погоды: {response.status_code}")
return None

def send_telegram_message(text):
"""Отправляет сообщение в Telegram"""
url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
payload = {
'chat_id': CHAT_ID,
'text': text,
'parse_mode': 'HTML'
}

response = requests.post(url, json=payload)

if response.status_code != 200:
print(f"Ошибка отправки сообщения: {response.status_code}")
print(response.text)

# Основная функция
def send_weather_report():
city = "Москва"
weather = get_weather(city)

if weather:
now = datetime.now().strftime('%d.%m.%Y %H:%M')
message = f"<b>Погода в {city} на {now}</b>\n\n"
message += f"🌡 Температура: {weather['temperature']}°C\n"
message += f"🤔 Ощущается как: {weather['feels_like']}°C\n"
message += f"☁️ Условия: {weather['description']}\n"
message += f"💧 Влажность: {weather['humidity']}%\n"
message += f"💨 Скорость ветра: {weather['wind_speed']} м/с"

send_telegram_message(message)

# Запуск отправки прогноза погоды
send_weather_report()

import requests
import csv
import time
from datetime import datetime

# Файл для записи результатов
OUTPUT_FILE = 'product_prices.csv'

def get_product_info(product_id):
"""Получает информацию о товаре по API магазина"""
url = f"https://api.example.com/products/{product_id}"

headers = {
'User-Agent': 'PriceMonitor/1.0',
'Authorization': 'Bearer YOUR_API_KEY'
}

try:
response = requests.get(url, headers=headers)

if response.status_code == 200:
data = response.json()
return {
'id': product_id,
'name': data['name'],
'price': data['price'],
'in_stock': data['availability'],
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
else:
print(f"Ошибка получения информации о товаре {product_id}: {response.status_code}")
except Exception as e:
print(f"Исключение при запросе товара {product_id}: {str(e)}")

return None

def save_to_csv(product_data):
"""Сохраняет информацию о товаре в CSV-файл"""
file_exists = os.path.isfile(OUTPUT_FILE)

with open(OUTPUT_FILE, 'a', newline='', encoding='utf-8') as csvfile:
fieldnames = ['id', 'name', 'price', 'in_stock', 'timestamp']
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)

if not file_exists:
writer.writeheader()

writer.writerow(product_data)

print(f"Информация о товаре {product_data['id']} сохранена")

def monitor_products(product_ids, interval=3600):
"""Мониторит цены на указанные товары с заданным интервалом"""
while True:
for product_id in product_ids:
product_info = get_product_info(product_id)

if product_info:
save_to_csv(product_info)

print(f"Следующая проверка через {interval} секунд")
time.sleep(interval)

# Список ID товаров для мониторинга
products_to_monitor = ['12345', '67890', '24680']

# Запуск мониторинга (проверка каждый час)
monitor_products(products_to_monitor)

import requests
import os
import json
from datetime import datetime
import time

# Аутентификация с помощью OAuth 1.0a
def get_twitter_auth():
return {
'consumer_key': os.environ.get('TWITTER_CONSUMER_KEY'),
'consumer_secret': os.environ.get('TWITTER_CONSUMER_SECRET'),
'access_token': os.environ.get('TWITTER_ACCESS_TOKEN'),
'access_token_secret': os.environ.get('TWITTER_ACCESS_SECRET')
}

def post_tweet(text):
"""Публикует твит через API Twitter"""
auth = get_twitter_auth()

# Для Twitter API v2
url = "https://api.twitter.com/2/tweets"

# Создаем OAuth 1.0a заголовок
import oauth2 as oauth
consumer = oauth.Consumer(key=auth['consumer_key'], secret=auth['consumer_secret'])
token = oauth.Token(key=auth['access_token'], secret=auth['access_token_secret'])
client = oauth.Client(consumer, token)

payload = json.dumps({"text": text})

headers = {
'Content-Type': 'application/json'
}

response, content = client.request(url, method="POST", body=payload, headers=headers)

if response.status == 201:
print("Твит успешно опубликован")
return json.loads(content)
else:
print(f"Ошибка публикации: {response.status}")
print(content)
return None

def schedule_tweets(tweets_data):
"""Планирует публикацию твитов на определенное время"""
for tweet in tweets_data:
# Получаем время публикации и текст твита
publish_time = datetime.strptime(tweet['time'], '%Y-%m-%d %H:%M')
tweet_text = tweet['text']

# Вычисляем, сколько нужно ждать
now = datetime.now()
if publish_time > now:
wait_seconds = (publish_time – now).total_seconds()

print(f"Твит '{tweet_text}' будет опубликован в {tweet['time']}")
print(f"Ожидание: {wait_seconds} секунд")

time.sleep(wait_seconds)
post_tweet(tweet_text)
else:
print(f"Пропуск твита '{tweet_text}' – запланированное время уже прошло")

# Пример списка твитов для публикации
tweets = [
{
'text': 'Изучаю работу с API в Python! #python #api #coding',
'time': '2023-08-10 15:00'
},
{
'text': 'Автоматизация – ключ к продуктивности. Мой скрипт сам публикует этот твит! #automation',
'time': '2023-08-10 18:30'
}
]

# Запуск планировщика
schedule_tweets(tweets)