import requests import json from datetime import datetime, timedelta CLIENT_ID = "i1kgxjWVvHRqACp8GrtZyueXMc0bDlPz" CLIENT_SECRET = "gVPsMNU8uwqpladEz9joT1GYO3QH0i2v" class EPNAPIClient: def __init__(self, client_id, client_secret): self.client_id = client_id self.client_secret = client_secret self.session = requests.Session() self.ssid_token = None self.access_token = None self.refresh_token = None self.token_expires_at = None def step1_get_ssid_token(self): """Этап 1: Получение SSID токена""" print("🔑 Этап 1: Получение SSID токена...") url = f"https://oauth2.epn.bz/ssid?v=2&client_id={self.client_id}" try: response = self.session.get(url) print(f"Статус ответа: {response.status_code}") print(f"Ответ: {response.text}") if response.status_code == 200: data = response.json() if 'data' in data and 'attributes' in data['data']: self.ssid_token = data['data']['attributes']['ssid_token'] print(f"✅ SSID токен получен: {self.ssid_token}") return True else: print(f"❌ Неожиданная структура ответа: {data}") else: # Возможно потребуется капча data = response.json() if 'errors' in data and data['errors'][0]['error'] == 429001: print("❌ Требуется решение капчи") print("Откройте в браузере:", url) self.ssid_token = input("Введите SSID токен из браузера: ").strip() if self.ssid_token: print(f"✅ SSID токен введен: {self.ssid_token}") return True return False except Exception as e: print(f"❌ Ошибка при получении SSID токена: {e}") return False def step2_get_access_token(self): """Этап 2: Получение access_token и refresh_token""" if not self.ssid_token: print("❌ Сначала нужно получить SSID токен") return False print("🎫 Этап 2: Получение access_token...") url = "https://oauth2.epn.bz/token?v=2" headers = { "Content-Type": "application/json" } body = { "ssid_token": self.ssid_token, "client_id": self.client_id, "client_secret": self.client_secret, "grant_type": "client_credentials", "check_ip": False } try: response = self.session.post(url, headers=headers, json=body) print(f"Статус ответа: {response.status_code}") print(f"Ответ: {response.text}") if response.status_code == 200: data = response.json() if 'access_token' in data: self.access_token = data['access_token'] self.refresh_token = data.get('refresh_token') # Вычисляем время истечения токена (обычно 3600 секунд) expires_in = data.get('expires_in', 3600) self.token_expires_at = datetime.now() + timedelta(seconds=expires_in) print(f"✅ Access токен получен: {self.access_token[:20]}...") print(f"✅ Refresh токен получен: {self.refresh_token[:20] if self.refresh_token else 'Нет'}") print(f"⏰ Токен истекает: {self.token_expires_at}") return True else: print(f"❌ Access токен не найден в ответе: {data}") return False except Exception as e: print(f"❌ Ошибка при получении access токена: {e}") return False def refresh_access_token(self): """Обновление access токена с помощью refresh токена""" if not self.refresh_token: print("❌ Refresh токен отсутствует") return False print("🔄 Обновление access токена...") url = "https://oauth2.epn.bz/token/refresh" headers = { "Content-Type": "application/json" } body = { "refresh_token": self.refresh_token, "client_id": self.client_id, "client_secret": self.client_secret } try: response = self.session.post(url, headers=headers, json=body) if response.status_code == 200: data = response.json() if 'access_token' in data: self.access_token = data['access_token'] if 'refresh_token' in data: self.refresh_token = data['refresh_token'] expires_in = data.get('expires_in', 3600) self.token_expires_at = datetime.now() + timedelta(seconds=expires_in) print("✅ Токен успешно обновлен") return True print(f"❌ Ошибка обновления токена: {response.text}") return False except Exception as e: print(f"❌ Ошибка при обновлении токена: {e}") return False def is_token_expired(self): """Проверка, истек ли токен""" if not self.token_expires_at: return True return datetime.now() >= self.token_expires_at def ensure_valid_token(self): """Убеждаемся, что у нас есть действительный токен""" if not self.access_token or self.is_token_expired(): if self.refresh_token: return self.refresh_access_token() else: print("❌ Нужно заново получить токены") return False return True def create_affiliate_link(self, product_url, offer_id=1, description="API Generated Link"): """Создание партнерской ссылки""" if not self.ensure_valid_token(): print("❌ Нет действительного токена") return None print(f"🔗 Создание партнерской ссылки для: {product_url}") url = "https://app.epn.bz/creative/create" headers = { "X-ACCESS-TOKEN": self.access_token, "Content-Type": "application/json" } body = { "link": product_url, "offerId": offer_id, "description": description, "type": "deeplink" } try: response = self.session.post(url, headers=headers, json=body) print(f"Статус ответа: {response.status_code}") print(f"Ответ: {response.text}") if response.status_code == 200: data = response.json() print("✅ Партнерская ссылка создана успешно!") return data elif response.status_code == 401: print("🔄 Токен истек, пытаемся обновить...") if self.refresh_access_token(): # Повторяем запрос с новым токеном headers["X-ACCESS-TOKEN"] = self.access_token response = self.session.post(url, headers=headers, json=body) if response.status_code == 200: return response.json() print(f"❌ Ошибка создания ссылки: {response.text}") return None except Exception as e: print(f"❌ Ошибка при создании ссылки: {e}") return None def get_user_info(self): """Тестовый запрос для проверки API""" if not self.ensure_valid_token(): return None print("📋 Получение информации о пользователе...") url = "https://app.epn.bz/test/user-info" headers = { "X-ACCESS-TOKEN": self.access_token } try: response = self.session.get(url, headers=headers) if response.status_code == 200: data = response.json() print("✅ Информация о пользователе получена") return data else: print(f"❌ Ошибка получения информации: {response.text}") return None except Exception as e: print(f"❌ Ошибка при запросе информации: {e}") return None def save_tokens(self, filename="epn_tokens.json"): """Сохранение токенов в файл""" data = { "client_id": self.client_id, "client_secret": self.client_secret, "ssid_token": self.ssid_token, "access_token": self.access_token, "refresh_token": self.refresh_token, "expires_at": self.token_expires_at.isoformat() if self.token_expires_at else None, "saved_at": datetime.now().isoformat() } with open(filename, 'w') as f: json.dump(data, f, indent=2) print(f"💾 Токены сохранены в {filename}") def load_tokens(self, filename="epn_tokens.json"): """Загрузка токенов из файла""" try: with open(filename, 'r') as f: data = json.load(f) self.ssid_token = data.get('ssid_token') self.access_token = data.get('access_token') self.refresh_token = data.get('refresh_token') if data.get('expires_at'): self.token_expires_at = datetime.fromisoformat(data['expires_at']) print(f"📂 Токены загружены из {filename}") return True except FileNotFoundError: print(f"📂 Файл {filename} не найден") return False except Exception as e: print(f"❌ Ошибка загрузки токенов: {e}") return False def main(): print("=== ePN API Client (Официальная инструкция) ===\n") # Создаем клиент client = EPNAPIClient(CLIENT_ID, CLIENT_SECRET) # Пытаемся загрузить сохраненные токены if not client.load_tokens(): print("🔑 Получаем новые токены...\n") # Этап 1: Получение SSID токена if not client.step1_get_ssid_token(): print("❌ Не удалось получить SSID токен") return print() # Этап 2: Получение access токена if not client.step2_get_access_token(): print("❌ Не удалось получить access токен") return # Сохраняем токены client.save_tokens() print() # Тестируем API user_info = client.get_user_info() if user_info: print(f"Пользователь: {json.dumps(user_info, indent=2, ensure_ascii=False)}") print() # Создаем партнерскую ссылку test_url = "https://aliexpress.ru/item/1005004321449827.html" affiliate_link = client.create_affiliate_link( product_url=test_url, offer_id=1, description="Тестовая партнерская ссылка" ) if affiliate_link: print(f"🎉 Партнерская ссылка создана!") print(f"Данные: {json.dumps(affiliate_link, indent=2, ensure_ascii=False)}") # Обновляем сохраненные токены client.save_tokens() if __name__ == "__main__": main()