Andrey K. Choi
3050e084fa
All checks were successful
continuous-integration/drone/push Build is passing
248 lines
10 KiB
Python
Executable File
248 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Скрипт для тестирования API FatSecret
|
||
Выполняет тестовые запросы к API FatSecret с использованием ключей из конфигурации приложения
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import time
|
||
import base64
|
||
import asyncio
|
||
import httpx
|
||
import urllib.parse
|
||
import hmac
|
||
import hashlib
|
||
from datetime import datetime
|
||
from dotenv import load_dotenv
|
||
|
||
|
||
# Загружаем .env файл
|
||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||
env_path = os.path.join(current_dir, ".env")
|
||
load_dotenv(env_path)
|
||
print(f"✅ Loaded .env from: {env_path}")
|
||
|
||
# Получаем API ключи из переменных окружения
|
||
FATSECRET_CLIENT_ID = os.environ.get("FATSECRET_CLIENT_ID")
|
||
FATSECRET_CLIENT_SECRET = os.environ.get("FATSECRET_CLIENT_SECRET")
|
||
FATSECRET_CUSTOMER_KEY = os.environ.get("FATSECRET_CUSTOMER_KEY")
|
||
|
||
if not FATSECRET_CLIENT_ID or not FATSECRET_CLIENT_SECRET:
|
||
raise ValueError("FatSecret API keys not found in .env file")
|
||
|
||
print(f"🔑 Using FatSecret API keys: CLIENT_ID={FATSECRET_CLIENT_ID[:8]}...")
|
||
if FATSECRET_CUSTOMER_KEY:
|
||
print(f"🔑 Using CUSTOMER_KEY={FATSECRET_CUSTOMER_KEY[:8]}...")
|
||
|
||
|
||
class FatSecretClient:
|
||
"""Клиент для работы с API FatSecret"""
|
||
|
||
BASE_URL = "https://platform.fatsecret.com/rest/server.api"
|
||
|
||
def __init__(self, client_id, client_secret):
|
||
self.client_id = client_id
|
||
self.client_secret = client_secret
|
||
self.access_token = None
|
||
self.token_expires = 0
|
||
|
||
async def get_access_token(self):
|
||
"""Получение OAuth 2.0 токена для доступа к API"""
|
||
now = time.time()
|
||
|
||
# Если у нас уже есть токен и он не истек, используем его
|
||
if self.access_token and self.token_expires > now + 60:
|
||
return self.access_token
|
||
|
||
print("🔄 Getting new access token...")
|
||
|
||
# Подготовка запроса на получение токена
|
||
auth_header = base64.b64encode(f"{self.client_id}:{self.client_secret}".encode()).decode()
|
||
|
||
print(f"🔑 Using client_id: {self.client_id}")
|
||
# Не печатаем секрет полностью, только первые несколько символов для отладки
|
||
print(f"🔑 Using client_secret: {self.client_secret[:5]}...")
|
||
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.post(
|
||
"https://oauth.fatsecret.com/connect/token",
|
||
headers={
|
||
"Authorization": f"Basic {auth_header}",
|
||
"Content-Type": "application/x-www-form-urlencoded"
|
||
},
|
||
data={
|
||
"grant_type": "client_credentials",
|
||
"scope": "basic premier"
|
||
}
|
||
)
|
||
|
||
# Проверяем успешность запроса
|
||
if response.status_code != 200:
|
||
print(f"❌ Error getting token: {response.status_code}")
|
||
print(response.text)
|
||
raise Exception(f"Failed to get token: {response.status_code}")
|
||
|
||
token_data = response.json()
|
||
self.access_token = token_data["access_token"]
|
||
self.token_expires = now + token_data["expires_in"]
|
||
|
||
print(f"✅ Got token, expires in {token_data['expires_in']} seconds")
|
||
return self.access_token
|
||
|
||
async def search_food(self, query, page=0, max_results=10):
|
||
"""Поиск продуктов по названию"""
|
||
token = await self.get_access_token()
|
||
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.post(
|
||
self.BASE_URL,
|
||
headers={
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json"
|
||
},
|
||
json={
|
||
"method": "foods.search",
|
||
"search_expression": query,
|
||
"page_number": page,
|
||
"max_results": max_results,
|
||
"format": "json"
|
||
}
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
print(f"❌ Error searching food: {response.status_code}")
|
||
print(response.text)
|
||
raise Exception(f"Failed to search food: {response.status_code}")
|
||
|
||
return response.json()
|
||
|
||
async def get_food(self, food_id):
|
||
"""Получение детальной информации о продукте по ID"""
|
||
token = await self.get_access_token()
|
||
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.post(
|
||
self.BASE_URL,
|
||
headers={
|
||
"Authorization": f"Bearer {token}",
|
||
"Content-Type": "application/json"
|
||
},
|
||
json={
|
||
"method": "food.get",
|
||
"food_id": food_id,
|
||
"format": "json"
|
||
}
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
print(f"❌ Error getting food details: {response.status_code}")
|
||
print(response.text)
|
||
raise Exception(f"Failed to get food details: {response.status_code}")
|
||
|
||
return response.json()
|
||
|
||
|
||
async def run_tests():
|
||
"""Выполнение тестовых запросов к API FatSecret"""
|
||
client = FatSecretClient(FATSECRET_CLIENT_ID, FATSECRET_CLIENT_SECRET)
|
||
|
||
# Тест 1: Поиск продуктов
|
||
print("\n🔍 Testing food search...")
|
||
search_queries = ["apple", "bread", "chicken breast", "молоко"]
|
||
|
||
for query in search_queries:
|
||
print(f"\n📋 Searching for: {query}")
|
||
try:
|
||
result = await client.search_food(query)
|
||
|
||
# Проверяем структуру ответа
|
||
if "foods" not in result:
|
||
print(f"❌ Unexpected response format: {result}")
|
||
continue
|
||
|
||
# Если нет результатов
|
||
if "food" not in result["foods"]:
|
||
print(f"⚠️ No results found for '{query}'")
|
||
continue
|
||
|
||
food_list = result["foods"]["food"]
|
||
if not isinstance(food_list, list):
|
||
food_list = [food_list] # Если только один результат, оборачиваем в список
|
||
|
||
print(f"✅ Found {len(food_list)} results")
|
||
|
||
# Выводим первые 3 результата
|
||
first_food_id = None
|
||
for i, food in enumerate(food_list[:3]):
|
||
food_name = food.get("food_name", "Unknown")
|
||
food_id = food.get("food_id", "Unknown")
|
||
food_desc = food.get("food_description", "No description")
|
||
|
||
print(f" {i+1}. [{food_id}] {food_name}")
|
||
print(f" {food_desc}")
|
||
|
||
# Сохраняем ID первого продукта для следующего теста
|
||
if i == 0:
|
||
first_food_id = food_id
|
||
except Exception as e:
|
||
print(f"❌ Error during search: {e}")
|
||
|
||
# Тест 2: Получение информации о продукте
|
||
found_food_id = None
|
||
for query in search_queries:
|
||
try:
|
||
result = await client.search_food(query)
|
||
if "foods" in result and "food" in result["foods"]:
|
||
food_list = result["foods"]["food"]
|
||
if not isinstance(food_list, list):
|
||
food_list = [food_list]
|
||
if food_list:
|
||
found_food_id = food_list[0].get("food_id")
|
||
break
|
||
except:
|
||
continue
|
||
|
||
if found_food_id:
|
||
print(f"\n🔍 Testing food details for ID: {found_food_id}")
|
||
try:
|
||
result = await client.get_food(found_food_id)
|
||
|
||
if "food" not in result:
|
||
print(f"❌ Unexpected response format: {result}")
|
||
else:
|
||
food = result["food"]
|
||
food_name = food.get("food_name", "Unknown")
|
||
brand = food.get("brand_name", "Generic")
|
||
|
||
print(f"✅ Got details for: {food_name} [{brand}]")
|
||
|
||
# Выводим информацию о пищевой ценности
|
||
if "servings" in food:
|
||
servings = food["servings"]
|
||
if "serving" in servings:
|
||
serving_data = servings["serving"]
|
||
if not isinstance(serving_data, list):
|
||
serving_data = [serving_data]
|
||
|
||
print("\n📊 Nutrition info per serving:")
|
||
for i, serving in enumerate(serving_data[:2]): # Выводим до 2 видов порций
|
||
serving_desc = serving.get("serving_description", "Standard")
|
||
calories = serving.get("calories", "N/A")
|
||
protein = serving.get("protein", "N/A")
|
||
carbs = serving.get("carbohydrate", "N/A")
|
||
fat = serving.get("fat", "N/A")
|
||
|
||
print(f" Serving {i+1}: {serving_desc}")
|
||
print(f" Calories: {calories}")
|
||
print(f" Protein: {protein}g")
|
||
print(f" Carbohydrates: {carbs}g")
|
||
print(f" Fat: {fat}g")
|
||
except Exception as e:
|
||
print(f"❌ Error getting food details: {e}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("🚀 Starting FatSecret API test...")
|
||
asyncio.run(run_tests())
|
||
print("\n✅ Test completed!") |