"""
Cliente para APIs da Shopee (Affiliate e Seller Center).
"""
import time
import hashlib
import json
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from decimal import Decimal
import requests
from src.config.settings import settings
@dataclass
class ShopeeProduct:
"""Produto da Shopee formatado."""
item_id: int
name: str
description: str
price_min: Decimal
price_max: Decimal
image_url: Optional[str]
offer_link: Optional[str]
product_link: Optional[str]
shop_id: int
shop_name: str
category_ids: List[int]
sales: int
rating: float
commission_rate: float
@dataclass
class ShopeeCategory:
"""Categoria da Shopee formatada."""
category_id: int
name: str
parent_id: Optional[int]
level: int
class ShopeeClient:
"""Cliente para APIs da Shopee."""
# URLs base
SELLER_API_URL = "https://seller.shopee.com.br/help/api/v3/global_category/list/"
def __init__(self, app_id: Optional[str] = None, secret: Optional[str] = None):
"""
Inicializa o cliente.
Args:
app_id: AppId (Credential) da plataforma de afiliados
secret: Secret da plataforma de afiliados
"""
# Prioriza o que for passado no __init__ mas faz fallback paras configs carregadas do `.env`
self.app_id = app_id or settings.SHOPEE_APP_ID
self.secret = secret or settings.SHOPEE_SECRET
self.affiliate_api_url = settings.SHOPEE_API_URL
self.session = requests.Session()
def _generate_signature(self, timestamp: int, payload: str) -> str:
"""Gera assinatura para autenticação da API de afiliados."""
if not self.secret or not self.app_id:
return ""
signature_factor = f"{self.app_id}{timestamp}{payload}{self.secret}"
return hashlib.sha256(signature_factor.encode("utf-8")).hexdigest()
def _make_graphql_request(
self, query: str, variables: Dict[str, Any] = None
) -> Dict[str, Any]:
"""Faz requisição GraphQL para API de afiliados."""
timestamp = int(time.time())
payload_dict = {"query": query, "variables": variables or {}}
payload = json.dumps(payload_dict, separators=(",", ":"))
signature = self._generate_signature(timestamp, payload)
authorization_header = (
f"SHA256 Credential={self.app_id}, "
f"Timestamp={timestamp}, "
f"Signature={signature}"
)
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
"Authorization": authorization_header,
}
try:
response = self.session.post(
self.affiliate_api_url, data=payload, headers=headers, timeout=30
)
return response.json()
except Exception as e:
return {"error": str(e)}
async def search_products(
self,
keyword: Optional[str] = None,
category_id: Optional[int] = None,
sort_type: int = 2, # ITEM_SOLD_DESC
page: int = 1,
limit: int = 20,
) -> List[ShopeeProduct]:
"""
Busca produtos na Shopee.
Args:
keyword: Palavra-chave para busca
category_id: ID da categoria para filtrar
sort_type: Tipo de ordenação
1 = RELEVANCE_DESC
2 = ITEM_SOLD_DESC (mais vendidos)
3 = PRICE_DESC
4 = PRICE_ASC
5 = COMMISSION_DESC
page: Número da página
limit: Quantidade por página
Returns:
Lista de produtos formatados
"""
query = """
query ProductOfferV2($productCatId: Int, $keyword: String, $sortType: Int, $page: Int, $limit: Int) {
productOfferV2(productCatId: $productCatId, keyword: $keyword, sortType: $sortType, page: $page, limit: $limit) {
nodes {
itemId
commissionRate
sellerCommissionRate
shopeeCommissionRate
commission
sales
priceMax
priceMin
productCatIds
ratingStar
priceDiscountRate
imageUrl
productName
shopId
shopName
shopType
productLink
offerLink
periodStartTime
periodEndTime
}
pageInfo {
page
limit
hasNextPage
}
}
}
"""
variables: Dict[str, Any] = {
"sortType": sort_type,
"page": page,
"limit": limit,
}
if keyword:
variables["keyword"] = keyword
if category_id:
variables["productCatId"] = category_id
result = self._make_graphql_request(query, variables)
if "error" in result or "errors" in result:
return []
products = []
nodes = result.get("data", {}).get("productOfferV2", {}).get("nodes", [])
for node in nodes:
try:
product = ShopeeProduct(
item_id=int(node.get("itemId", 0)),
name=node.get("productName", ""),
description=f"Loja: {node.get('shopName', '')} | Vendas: {node.get('sales', 0)}",
price_min=Decimal(str(node.get("priceMin", 0))),
price_max=Decimal(str(node.get("priceMax", 0))),
image_url=node.get("imageUrl"),
offer_link=node.get("offerLink"),
product_link=node.get("productLink"),
shop_id=int(node.get("shopId", 0)),
shop_name=node.get("shopName", ""),
category_ids=node.get("productCatIds", []),
sales=int(node.get("sales", 0)),
rating=float(node.get("ratingStar", 0) or 0),
commission_rate=float(node.get("commissionRate", 0) or 0),
)
products.append(product)
except Exception:
continue
return products
async def search_categories(
self, keyword: str, page: int = 1, size: int = 50
) -> tuple[List[ShopeeCategory], int]:
"""
Busca categorias por keyword diretamente na API da Shopee.
Args:
keyword: Palavra-chave para busca (ex: "relogio", "celular")
page: Número da página
size: Quantidade por página
Returns:
Tupla com lista de categorias encontradas e total
"""
headers = {
"accept": "application/json, text/plain, */*",
"accept-language": "pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36",
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin",
}
categories = []
total = 0
try:
url = f"{self.SELLER_API_URL}?page={page}&size={size}&keyword={keyword}"
response = self.session.get(url, headers=headers, timeout=30)
data = response.json()
if data.get("code") != 0:
return [], 0
global_cats = data.get("data", {}).get("global_cats", [])
total = data.get("data", {}).get("total", 0)
for cat in global_cats:
path = cat.get("path", [])
# Processa cada nível do path
for i, level_cat in enumerate(path):
cat_id = level_cat.get("category_id")
cat_name = level_cat.get("category_name")
parent_id = path[i - 1].get("category_id") if i > 0 else None
level = i + 1
category = ShopeeCategory(
category_id=cat_id,
name=cat_name,
parent_id=parent_id,
level=level,
)
categories.append(category)
except Exception as e:
print(f"Erro ao buscar categorias: {e}")
return [], 0
# Remove duplicatas mantendo a primeira ocorrência
seen = set()
unique_categories = []
for cat in categories:
if cat.category_id not in seen:
seen.add(cat.category_id)
unique_categories.append(cat)
return unique_categories, total
async def get_categories(
self, page: int = 1, size: int = 100
) -> tuple[List[ShopeeCategory], int]:
"""
Busca uma página de categorias do Seller Center da Shopee.
Args:
page: Número da página
size: Quantidade por página
Returns:
Tupla com lista de categorias e total de categorias disponíveis
"""
headers = {
"accept": "application/json, text/plain, */*",
"accept-language": "pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7",
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36",
"sec-fetch-dest": "empty",
"sec-fetch-mode": "cors",
"sec-fetch-site": "same-origin",
}
categories = []
total = 0
try:
url = f"{self.SELLER_API_URL}?page={page}&size={size}"
response = self.session.get(url, headers=headers, timeout=30)
data = response.json()
if data.get("code") != 0:
return [], 0
global_cats = data.get("data", {}).get("global_cats", [])
total = data.get("data", {}).get("total", 0)
for cat in global_cats:
path = cat.get("path", [])
# Processa cada nível do path
for i, level_cat in enumerate(path):
cat_id = level_cat.get("category_id")
cat_name = level_cat.get("category_name")
parent_id = path[i - 1].get("category_id") if i > 0 else None
level = i + 1
category = ShopeeCategory(
category_id=cat_id,
name=cat_name,
parent_id=parent_id,
level=level,
)
categories.append(category)
except Exception as e:
print(f"Erro ao buscar categorias: {e}")
return [], 0
return categories, total
async def get_all_categories(self) -> List[ShopeeCategory]:
"""
Busca todas as categorias da Shopee.
Faz paginação automática.
Returns:
Lista completa de categorias (sem duplicatas)
"""
all_categories = []
page = 1
size = 100
total_fetched = 0
while True:
categories, total = await self.get_categories(page=page, size=size)
if not categories:
break
all_categories.extend(categories)
total_fetched += size
if total_fetched >= total:
break
page += 1
time.sleep(0.1) # Rate limiting
# Remove duplicatas mantendo a primeira ocorrência
seen = set()
unique_categories = []
for cat in all_categories:
if cat.category_id not in seen:
seen.add(cat.category_id)
unique_categories.append(cat)
return unique_categories
