import base64 import hashlib import hmac import json import logging import os import random import time import urllib.parse from datetime import datetime, timedelta from typing import Dict, List, Optional, Any, Union import httpx from shared.config import settings logger = logging.getLogger(__name__) class FatSecretClient: """Клиент для работы с API FatSecret""" BASE_URL = "https://platform.fatsecret.com/rest/server.api" def __init__(self): """Инициализация клиента для работы с API FatSecret""" # Используем CUSTOMER_KEY для OAuth 1.0, если он доступен, иначе CLIENT_ID self.api_key = settings.FATSECRET_CUSTOMER_KEY or settings.FATSECRET_CLIENT_ID self.api_secret = settings.FATSECRET_CLIENT_SECRET # Логируем информацию о ключах (без полного раскрытия) logger.info(f"FatSecretClient initialized with key: {self.api_key[:8]}...") def _generate_oauth_params(self, http_method: str, url: str, params: Dict[str, Any]) -> Dict[str, Any]: """Создание и подписание OAuth 1.0 параметров""" # Текущее время в секундах timestamp = str(int(time.time())) # Случайная строка для nonce nonce = ''.join([str(random.randint(0, 9)) for _ in range(8)]) # Базовый набор параметров OAuth oauth_params = { 'oauth_consumer_key': self.api_key, 'oauth_nonce': nonce, 'oauth_signature_method': 'HMAC-SHA1', 'oauth_timestamp': timestamp, 'oauth_version': '1.0' } # Объединяем с параметрами запроса all_params = {**params, **oauth_params} # Сортируем параметры по ключу sorted_params = sorted(all_params.items()) # Создаем строку параметров для подписи param_string = "&".join([ f"{urllib.parse.quote(str(k), safe='')}={urllib.parse.quote(str(v), safe='')}" for k, v in sorted_params ]) # Создаем строку для подписи signature_base = f"{http_method}&{urllib.parse.quote(url, safe='')}&{urllib.parse.quote(param_string, safe='')}" # Создаем ключ для подписи signing_key = f"{urllib.parse.quote(str(self.api_secret), safe='')}&" # Создаем HMAC-SHA1 подпись signature = base64.b64encode( hmac.new( signing_key.encode(), signature_base.encode(), hashlib.sha1 ).digest() ).decode() # Добавляем подпись к параметрам OAuth all_params['oauth_signature'] = signature return all_params async def search_foods(self, query: str, page_number: int = 0, max_results: int = 10) -> Dict[str, Any]: """Поиск продуктов по запросу""" params = { 'method': 'foods.search', 'search_expression': query, 'page_number': str(page_number), 'max_results': str(max_results), 'format': 'json' } # Получаем подписанные OAuth параметры oauth_params = self._generate_oauth_params("GET", self.BASE_URL, params) try: async with httpx.AsyncClient() as client: response = await client.get( self.BASE_URL, params=oauth_params ) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Error searching foods: {e}") raise async def get_food_details(self, food_id: Union[str, int]) -> Dict[str, Any]: """Получить детальную информацию о продукте по ID""" params = { 'method': 'food.get.v2', 'food_id': str(food_id), 'format': 'json' } # Получаем подписанные OAuth параметры oauth_params = self._generate_oauth_params("GET", self.BASE_URL, params) try: async with httpx.AsyncClient() as client: response = await client.get( self.BASE_URL, params=oauth_params ) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Error getting food details: {e}") raise async def parse_food_data(self, food_json: Dict[str, Any]) -> Dict[str, Any]: """Разбирает данные о продукте из API в более удобный формат""" try: food = food_json.get('food', {}) # Извлечение основной информации о продукте food_id = food.get('food_id') food_name = food.get('food_name', '') food_type = food.get('food_type', '') brand_name = food.get('brand_name', '') # Обработка информации о питании servings = food.get('servings', {}).get('serving', []) # Если есть только одна порция, преобразуем ее в список if isinstance(servings, dict): servings = [servings] # Берем первую порцию по умолчанию (обычно это 100г или стандартная порция) serving_data = {} for serving in servings: if serving.get('is_default_serving', 0) == "1" or serving.get('serving_description', '').lower() == '100g': serving_data = serving break # Если не нашли стандартную порцию, берем первую if not serving_data and servings: serving_data = servings[0] # Извлечение данных о пищевой ценности serving_description = serving_data.get('serving_description', '') serving_amount = serving_data.get('metric_serving_amount', serving_data.get('serving_amount', '')) serving_unit = serving_data.get('metric_serving_unit', serving_data.get('serving_unit', '')) # Формирование читаемого текста размера порции serving_size = f"{serving_amount} {serving_unit}" if serving_amount and serving_unit else serving_description # Извлечение данных о пищевой ценности calories = float(serving_data.get('calories', 0) or 0) protein = float(serving_data.get('protein', 0) or 0) fat = float(serving_data.get('fat', 0) or 0) carbs = float(serving_data.get('carbohydrate', 0) or 0) fiber = float(serving_data.get('fiber', 0) or 0) sugar = float(serving_data.get('sugar', 0) or 0) sodium = float(serving_data.get('sodium', 0) or 0) cholesterol = float(serving_data.get('cholesterol', 0) or 0) # Формирование результата result = { "fatsecret_id": food_id, "name": food_name, "brand": brand_name, "food_type": food_type, "serving_size": serving_size, "serving_weight_grams": float(serving_amount) if serving_unit == 'g' else None, "calories": calories, "protein_grams": protein, "fat_grams": fat, "carbs_grams": carbs, "fiber_grams": fiber, "sugar_grams": sugar, "sodium_mg": sodium, "cholesterol_mg": cholesterol, "is_verified": True } return result except Exception as e: logger.error(f"Error parsing food data: {e}") raise