#!/usr/bin/env python3 """ Micro-serviço Python para scraping e busca Pipeline: 1. Scraping: curl_cffi / Playwright Stealth 2. Busca: duckduckgo-search """ import sys import json import logging import traceback import random import time import os import asyncio from itertools import cycle from typing import Dict, Optional, Union, List from urllib.parse import urljoin # Bibliotecas de Scraping e Parsing from bs4 import BeautifulSoup from curl_cffi import requests import html2text # DuckDuckGo Search try: from duckduckgo_search import DDGS HAS_DDG = True except ImportError: HAS_DDG = False # Fallback libs try: import requests as pyrequests HAS_PYREQUESTS = True except ImportError: HAS_PYREQUESTS = False try: import httpx HAS_HTTPX = True except ImportError: HAS_HTTPX = False # Playwright & Anti-Detect try: from playwright.async_api import async_playwright, TimeoutError as PlaywrightTimeoutError HAS_PLAYWRIGHT = True except ImportError: HAS_PLAYWRIGHT = False try: from fake_useragent import UserAgent ua_generator = UserAgent() HAS_FAKE_UA = True except ImportError: HAS_FAKE_UA = False ua_generator = None # --- CONFIGURAÇÃO DO AMBIENTE --- try: sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') except (AttributeError, Exception): pass logging.basicConfig( level=logging.INFO, format='[PYTHON SCRAPER] %(levelname)s: %(message)s', stream=sys.stderr ) # --- CONSTANTES E POOLS --- UA_POOL_FALLBACK = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36', 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15', 'Mozilla/5.0 (X11; Linux x86_64; rv:132.0) Gecko/20100101 Firefox/132.0', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0' ] ua_cycle = cycle(UA_POOL_FALLBACK) REQUEST_TIMEOUT = 20 MAX_RETRIES = 2 PROXY_POOL = None env_proxies = os.environ.get('SCRAPER_PROXIES') or os.environ.get('SCRAPER_HTTP_PROXIES') if env_proxies: parts = [p.strip() for p in env_proxies.split(',') if p.strip()] if parts: PROXY_POOL = cycle(parts) # --- FUNÇÕES AUXILIARES --- def get_random_ua() -> str: if HAS_FAKE_UA: try: return ua_generator.random except Exception: pass return next(ua_cycle) def generate_consistent_headers(user_agent: str) -> Dict[str, str]: base_headers = { 'User-Agent': user_agent, 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8', 'Accept-Language': 'pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7', 'Accept-Encoding': 'gzip, deflate, br', 'DNT': '1', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'none', 'Sec-Fetch-User': '?1', 'Cache-Control': 'max-age=0' } if 'Chrome' in user_agent: try: version = user_agent.split('Chrome/')[1].split('.')[0] except IndexError: version = "120" base_headers.update({ 'sec-ch-ua': f'"Chromium";v="{version}", "Google Chrome";v="{version}", "Not-A.Brand";v="99"', 'sec-ch-ua-mobile': '?0', 'sec-ch-ua-platform': '"Windows"' if 'Windows' in user_agent else ('"macOS"' if 'Macintosh' in user_agent else '"Linux"') }) return base_headers def get_proxy_dict() -> Optional[Dict[str, str]]: if PROXY_POOL: proxy = next(PROXY_POOL) return {"http": proxy, "https": proxy} return None def random_jitter(min_sec=0.5, max_sec=2.0): time.sleep(random.uniform(min_sec, max_sec)) # --- LIMPEZA E EXTRAÇÃO --- def clean_html(soup: BeautifulSoup) -> BeautifulSoup: for tag in soup(["script", "style", "meta", "link", "noscript", "iframe", "object", "embed", "applet", "canvas", "svg"]): tag.decompose() for tag in soup.find_all(["nav", "header", "footer", "aside", "menu", "dialog"]): tag.decompose() ad_keywords = ["ads", "advertisement", "banner", "ad-", "popup", "modal", "cookie", "subscription", "newsletter", "share", "social", "sidebar", "promo", "sponsor"] selectors_to_remove = ['.ad', '.advertisement', '[class*="ad-"]', '[id*="ad-"]', '[id*="cookie"]', '[class*="cookie"]', '.share-buttons', '.social-media', '.comments-section', '#comments'] for selector in selectors_to_remove: for tag in soup.select(selector): tag.decompose() def is_ad(tag): if not tag.name: return False attrs = str(tag.attrs).lower() return any(k in attrs for k in ad_keywords) for tag in soup.find_all(is_ad): tag.decompose() return soup def extract_main_content(soup: BeautifulSoup) -> BeautifulSoup: selectors = ['article', 'main', '[role="main"]', '.post-content', '.article-body', '#content', '.content', '.entry-content', '.page-content'] for selector in selectors: found = soup.select_one(selector) if found and len(found.get_text(strip=True)) > 200: return found body = soup.find('body') return body if body else soup def html_to_markdown(html_content: str) -> str: try: h = html2text.HTML2Text() h.ignore_links = False h.ignore_images = False h.images_to_alt = True h.body_width = 0 h.protect_links = True return h.handle(html_content) except Exception as e: logging.warning(f"Erro na conversão Markdown: {e}") return html_content # --- ESTRATÉGIAS DE SCRAPING --- async def scrape_with_playwright(url: str) -> Dict: if not HAS_PLAYWRIGHT: return {"error": "Playwright not installed"} logging.info(f"[PLAYWRIGHT] Iniciando scraping stealth para: {url}") async with async_playwright() as p: args = ['--disable-blink-features=AutomationControlled', '--disable-dev-shm-usage', '--no-sandbox', '--disable-setuid-sandbox', '--disable-infobars', '--window-size=1920,1080', '--start-maximized'] try: browser = await p.chromium.launch(headless=True, args=args) ua = get_random_ua() context = await browser.new_context(user_agent=ua, viewport={'width': 1920, 'height': 1080}, locale='pt-BR', timezone_id='America/Sao_Paulo', java_script_enabled=True, ignore_https_errors=True) await context.add_init_script(""" Object.defineProperty(navigator, 'webdriver', {get: () => undefined}); window.chrome = { runtime: {} }; Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); Object.defineProperty(navigator, 'languages', { get: () => ['pt-BR', 'pt', 'en-US', 'en'] }); """) page = await context.new_page() try: await page.goto(url, wait_until='domcontentloaded', timeout=25000) await page.wait_for_timeout(2000) except PlaywrightTimeoutError: logging.warning(f"[PLAYWRIGHT] Timeout carregando {url}") content = await page.content() title = await page.title() await browser.close() return {"content": content, "title": title, "method": "playwright"} except Exception as e: logging.error(f"[PLAYWRIGHT] Erro fatal: {e}") return {"error": str(e)} def scrape_with_curl_cffi(url: str) -> Dict: ua = get_random_ua() headers = generate_consistent_headers(ua) proxies = get_proxy_dict() try: response = requests.get(url, headers=headers, proxies=proxies, timeout=15, impersonate="chrome120", allow_redirects=True) response.raise_for_status() if response.status_code in [403, 429, 503]: raise requests.RequestsError(f"Blocked with status {response.status_code}") return {"content": response.text, "title": "", "method": "curl_cffi", "headers": response.headers} except Exception as e: return {"error": str(e)} async def scrape_url_smart(url: str) -> Dict: random_jitter() logging.info(f"Tentativa 1: curl_cffi -> {url}") result = scrape_with_curl_cffi(url) html_content = None title = "" if "error" not in result: html_content = result["content"] if "challenge-platform" in html_content or "Cloudflare" in html_content and len(html_content) < 5000: logging.warning("Detectado desafio Cloudflare/WAF no curl_cffi. Acionando fallback.") html_content = None if not html_content: if HAS_PLAYWRIGHT: logging.info(f"Tentativa 2: Playwright Stealth -> {url}") pw_result = await scrape_with_playwright(url) if "error" not in pw_result: html_content = pw_result["content"] title = pw_result.get("title", "") else: logging.error(f"Playwright falhou: {pw_result['error']}") else: logging.warning("Playwright não disponível para fallback.") if html_content: try: soup = BeautifulSoup(html_content, 'html.parser') if not title: title = soup.title.string.strip() if soup.title else url soup = clean_html(soup) main_content = extract_main_content(soup) cleaned_html = str(main_content) markdown_text = html_to_markdown(cleaned_html) final_markdown = '\n'.join([line.strip() for line in markdown_text.split('\n') if line.strip()]) if len(final_markdown) < 50: return {"response_type": "error", "message": "Conteúdo insuficiente após processamento."} return {"response_type": "success", "url": url, "title": title, "markdown": final_markdown, "content": final_markdown} except Exception as e: logging.error(f"Erro no processamento HTML: {traceback.format_exc()}") return {"response_type": "error", "message": f"Processing error: {str(e)}"} return {"response_type": "error", "message": f"Falha em todos os métodos de scraping. Erro inicial: {result.get('error', 'Unknown')}"} # --- DUCKDUCKGO SEARCH LOGIC --- async def search_duckduckgo(keywords: str, region: str = "wt-wt", safesearch: str = "moderate", max_results: int = 10) -> Dict: if not HAS_DDG: return {"response_type": "error", "message": "duckduckgo-search module not installed"} logging.info(f"Searching DuckDuckGo: {keywords} (region={region}, safe={safesearch})") try: results_list = [] # DDGS is synchronous, run in thread executor to avoid blocking async loop if needed # But since this script main loop is simple, direct call is fine, or wrap in to_thread with DDGS() as ddgs: # ddgs.text() is a generator ddg_gen = ddgs.text( keywords=keywords, region=region, safesearch=safesearch, max_results=max_results ) for r in ddg_gen: results_list.append({ "title": r.get("title", ""), "url": r.get("href", ""), "content": r.get("body", ""), "engine": "duckduckgo", "score": 1.0 }) return { "response_type": "success", "query": keywords, "results": results_list, "number_of_results": len(results_list) } except Exception as e: logging.error(f"DuckDuckGo Search Error: {e}") return {"response_type": "error", "message": str(e)} # --- MAIN --- def write_json_response(data: Dict): try: json_line = json.dumps(data, separators=(',', ':'), ensure_ascii=False) print(json_line) sys.stdout.flush() except Exception as e: logging.error(f"Erro enviando resposta JSON: {e}") def main(): logging.info("Scraper/Search Service v4.0 (DuckDuckGo Edition) Initialized") while True: try: line = sys.stdin.readline() if not line: break if not line.strip(): continue try: request = json.loads(line) except json.JSONDecodeError: write_json_response({"response_type": "error", "message": "Invalid JSON"}) continue command = request.get("command") if command == "scrape": url = request.get("url") if url: result = asyncio.run(scrape_url_smart(url)) write_json_response(result) else: write_json_response({"response_type": "error", "message": "No URL provided"}) elif command == "search": # Generic search command, implemented via DuckDuckGo keywords = request.get("keywords") or request.get("query") region = request.get("region", "wt-wt") safesearch = request.get("safesearch", "moderate") max_results = request.get("max_results", 10) if keywords: result = asyncio.run(search_duckduckgo(keywords, region, safesearch, max_results)) write_json_response(result) else: write_json_response({"response_type": "error", "message": "Missing keywords"}) elif command == "ping": write_json_response({"response_type": "pong"}) else: write_json_response({"response_type": "error", "message": "Unknown command"}) except KeyboardInterrupt: break except Exception as e: logging.critical(f"Fatal error in main loop: {e}") write_json_response({"response_type": "error", "message": str(e)}) if __name__ == "__main__": main()