""" Vinted Platform Implementation Uses Vinted's internal API for product search """ import requests import logging import time from datetime import datetime from platforms.base_platform import BasePlatform from models.article import Article REQUEST_RETRY_TIME = 5 class VintedPlatform(BasePlatform): """Vinted marketplace platform implementation""" # Mapping de dominios por país COUNTRY_DOMAINS = { 'es': 'vinted.es', 'fr': 'vinted.fr', 'de': 'vinted.de', 'it': 'vinted.it', 'pl': 'vinted.pl', 'cz': 'vinted.cz', 'lt': 'vinted.lt', 'uk': 'vinted.co.uk', 'us': 'vinted.com', 'nl': 'vinted.nl', 'be': 'vinted.be', 'at': 'vinted.at', } def __init__(self, item_monitor): super().__init__(item_monitor) self.logger = logging.getLogger(__name__) # Por defecto España, se puede configurar con un campo 'country' en item_monitor self.country = getattr(item_monitor, '_country', 'es') self.domain = self.COUNTRY_DOMAINS.get(self.country, 'vinted.es') self.session = requests.Session() self._init_session() def _init_session(self): """Initialize session with proper cookies and headers""" try: # Primera petición para obtener cookies headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'es-ES,es;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', } response = self.session.get(f'https://www.{self.domain}', headers=headers, timeout=15) self.logger.info(f"Vinted session initialized for {self.domain}") except Exception as e: self.logger.warning(f"Could not initialize Vinted session: {e}") def get_platform_name(self): return "vinted" def create_url(self): """Construir URL de búsqueda de Vinted""" # API interna de Vinted base_url = f"https://www.{self.domain}/api/v2/catalog/items" params = [] # Query de búsqueda search_query = self._item_monitor.get_search_query() if search_query: params.append(f"search_text={requests.utils.quote(search_query)}") # Ordenar por más reciente params.append("order=newest_first") # Precio (Vinted usa céntimos, multiplicamos por 100) if self._item_monitor.get_min_price() != 0: price_cents = int(self._item_monitor.get_min_price() * 100) params.append(f"price_from={price_cents}") if self._item_monitor.get_max_price() != 0: price_cents = int(self._item_monitor.get_max_price() * 100) params.append(f"price_to={price_cents}") # Resultados por página (máximo suele ser 96) params.append("per_page=96") # Página (por defecto la primera) params.append("page=1") # Mapeo de condiciones Wallapop -> Vinted condition = self._item_monitor.get_condition() if condition != "all": vinted_status = self._map_condition_to_vinted(condition) if vinted_status: params.append(f"status_ids[]={vinted_status}") url = base_url if params: url += "?" + "&".join(params) return url def _map_condition_to_vinted(self, wallapop_condition): """ Mapear condiciones de Wallapop a IDs de estado de Vinted Vinted status IDs: 1=Satisfactory, 2=Good, 3=Very Good, 6=Brand new with tag, 7=Brand new without tag """ mapping = { 'new': '6', # Brand new with tag 'as_good_as_new': '7', # Brand new without tag 'good': '3', # Very Good 'fair': '2', # Good 'has_given_it_all': '1' # Satisfactory } return mapping.get(wallapop_condition) def get_request_headers(self): """Headers específicos para Vinted""" headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'es-ES,es;q=0.9,en;q=0.8', 'Accept-Encoding': 'gzip, deflate, br', 'Referer': f'https://www.{self.domain}/', 'Origin': f'https://www.{self.domain}', 'Connection': 'keep-alive', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin', 'Sec-Ch-Ua': '"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"', 'Sec-Ch-Ua-Mobile': '?0', 'Sec-Ch-Ua-Platform': '"Windows"', } return headers def fetch_articles(self): """Obtener artículos desde Vinted""" url = self.create_url() max_retries = 3 for attempt in range(max_retries): try: headers = self.get_request_headers() response = self.session.get(url, headers=headers, timeout=30, allow_redirects=True) response.raise_for_status() json_response = response.json() # Verificar estructura de respuesta if 'items' not in json_response: self.logger.warning(f"Unexpected Vinted response structure. Keys: {list(json_response.keys())}") # Intentar ver si hay un mensaje de error if 'error' in json_response: self.logger.error(f"Vinted API error: {json_response['error']}") return [] # INSERT_YOUR_CODE json_items = json_response['items'] articles = self.parse_response(json_items) return articles except requests.exceptions.HTTPError as err: status_code = err.response.status_code self.logger.error(f"Vinted HTTP Error {status_code}: {err}") if status_code == 401 or status_code == 403: self.logger.warning("Vinted authentication issue, reinitializing session...") self._init_session() elif status_code == 429: self.logger.warning("Vinted rate limit hit, waiting longer...") time.sleep(REQUEST_RETRY_TIME * 3) elif status_code == 404: self.logger.error("Vinted API endpoint not found. Check URL.") return [] # Log response content for debugging try: self.logger.debug(f"Response content: {err.response.text[:500]}") except: pass except requests.exceptions.RequestException as err: self.logger.error(f"Vinted Request Exception: {err}") except ValueError as e: self.logger.error(f"Error parsing JSON response from Vinted: {e}") try: self.logger.debug(f"Response text: {response.text[:500]}") except: pass except Exception as e: self.logger.error(f"Unexpected error fetching from Vinted: {e}") if attempt < max_retries - 1: wait_time = REQUEST_RETRY_TIME * (attempt + 1) self.logger.info(f"Retrying in {wait_time} seconds... (attempt {attempt + 2}/{max_retries})") time.sleep(wait_time) self.logger.warning(f"Failed to fetch articles from Vinted after {max_retries} attempts") return [] def parse_response(self, json_items): """Parsear respuesta de Vinted""" articles = [] for json_article in json_items: article = self._parse_single_article(json_article) if article: articles.append(article) return articles def _parse_single_article(self, json_data): """Parsear un artículo individual de Vinted""" try: # ID del artículo article_id = str(json_data['id']) # Título title = json_data.get('title', '') # Descripción description = json_data.get('description', '') # Precio (Vinted devuelve en céntimos, convertimos a euros) price_amount = json_data.get('price', {}).get('amount', 0) if price_amount: price = float(price_amount) else: price = 0.0 # Moneda currency = json_data.get('price', {}).get('currency_code', 'EUR') # Ubicación user_data = json_data.get('user', {}) location = user_data.get('city', 'Unknown') # URL del artículo article_url = json_data.get('url', '') if article_url and not article_url.startswith('http'): article_url = f"https://www.{self.domain}{article_url}" # Imágenes images = [] photo = json_data.get('photo') if photo and 'url' in photo: images.append(photo['url']) # Imágenes adicionales photos = json_data.get('photos', []) for photo in photos[:3]: # Máximo 3 imágenes if 'url' in photo: url = photo['url'] if url not in images: # Evitar duplicados images.append(url) # Limitar a 3 imágenes images = images[:3] # Fecha de modificación updated_at_str = json_data.get('photo', {}).get('high_resolution', {}).get('timestamp') if not updated_at_str: # Alternativa: usar created_at_ts created_ts = json_data.get('created_at_ts') if created_ts: try: dt = datetime.fromtimestamp(int(created_ts)) modified_at = dt.strftime("%Y-%m-%d %H:%M:%S") except: modified_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") else: modified_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") else: try: dt = datetime.fromtimestamp(int(updated_at_str)) modified_at = dt.strftime("%Y-%m-%d %H:%M:%S") except: modified_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Envíos - Vinted generalmente permite envíos allows_shipping = True return Article( id=article_id, title=title, description=description, price=price, currency=currency, location=location, allows_shipping=allows_shipping, url=article_url, images=images, modified_at=modified_at, platform=self.get_platform_name() ) except (KeyError, ValueError, TypeError) as e: self.logger.info(f"Error parsing Vinted article: {e} {json_data}") return None