src/web_automation.py

import time
import json
import streamlit as st
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.common.action_chains import ActionChains
from selenium.common.exceptions import TimeoutException, NoSuchElementException
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.chrome.service import Service
import base64
from PIL import Image
import io

class WebAutomation:
    def __init__(self):
        """
        کلاس اتوماسیون وب با استفاده از Selenium
        """
        self.driver = None
        self.wait = None
        self.is_initialized = False
        
    def setup_driver(self, headless=True):
        """
        راه‌اندازی درایور Chrome
        """
        try:
            chrome_options = Options()
            
            # تنظیمات برای Hugging Face Spaces
            chrome_options.add_argument('--headless') if headless else None
            chrome_options.add_argument('--no-sandbox')
            chrome_options.add_argument('--disable-dev-shm-usage')
            chrome_options.add_argument('--disable-gpu')
            chrome_options.add_argument('--window-size=1920,1080')
            chrome_options.add_argument('--disable-extensions')
            chrome_options.add_argument('--disable-plugins')
            chrome_options.add_argument('--disable-images')  # برای سرعت بیشتر
            chrome_options.add_argument('--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36')
            
            # راه‌اندازی درایور
            service = Service(ChromeDriverManager().install())
            self.driver = webdriver.Chrome(service=service, options=chrome_options)
            self.wait = WebDriverWait(self.driver, 10)
            self.is_initialized = True
            
            st.success("مرورگر با موفقیت راه‌اندازی شد!")
            return True
            
        except Exception as e:
            st.error(f"خطا در راه‌اندازی مرورگر: {str(e)}")
            return False
    
    def navigate_to_url(self, url):
        """
        رفتن به آدرس مشخص شده
        """
        if not self.is_initialized:
            if not self.setup_driver():
                return False, "مرورگر راه‌اندازی نشده است"
        
        try:
            # اضافه کردن پروتکل در صورت عدم وجود
            if not url.startswith(('http://', 'https://')):
                url = 'https://' + url
            
            self.driver.get(url)
            time.sleep(2)  # انتظار برای بارگذاری صفحه
            
            current_url = self.driver.current_url
            title = self.driver.title
            
            return True, f"صفحه بارگذاری شد: {title} ({current_url})"
            
        except Exception as e:
            return False, f"خطا در بارگذاری صفحه: {str(e)}"
    
    def take_screenshot(self):
        """
        گرفتن عکس از صفحه
        """
        if not self.is_initialized:
            return None, "مرورگر راه‌اندازی نشده است"
        
        try:
            # گرفتن اسکرین‌شات
            screenshot = self.driver.get_screenshot_as_png()
            
            # تبدیل به تصویر PIL
            image = Image.open(io.BytesIO(screenshot))
            
            return image, "عکس با موفقیت گرفته شد"
            
        except Exception as e:
            return None, f"خطا در گرفتن عکس: {str(e)}"
    
    def find_element_by_text(self, text, tag="*"):
        """
        پیدا کردن المنت بر اساس متن
        """
        try:
            xpath = f"//{tag}[contains(text(), '{text}')]"
            element = self.wait.until(EC.presence_of_element_located((By.XPATH, xpath)))
            return element, True
        except:
            try:
                xpath = f"//{tag}[contains(@value, '{text}')]"
                element = self.wait.until(EC.presence_of_element_located((By.XPATH, xpath)))
                return element, True
            except:
                return None, False
    
    def click_element(self, selector, by_type="css"):
        """
        کلیک روی المنت
        """
        if not self.is_initialized:
            return False, "مرورگر راه‌اندازی نشده است"
        
        try:
            if by_type == "css":
                element = self.wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, selector)))
            elif by_type == "xpath":
                element = self.wait.until(EC.element_to_be_clickable((By.XPATH, selector)))
            elif by_type == "id":
                element = self.wait.until(EC.element_to_be_clickable((By.ID, selector)))
            elif by_type == "text":
                element, found = self.find_element_by_text(selector)
                if not found:
                    return False, f"المنت با متن '{selector}' پیدا نشد"
            else:
                return False, "نوع selector نامعتبر است"
            
            # اسکرول به المنت
            self.driver.execute_script("arguments[0].scrollIntoView(true);", element)
            time.sleep(1)
            
            # کلیک
            element.click()
            time.sleep(1)
            
            return True, f"کلیک روی المنت انجام شد"
            
        except TimeoutException:
            return False, f"المنت '{selector}' پیدا نشد یا قابل کلیک نیست"
        except Exception as e:
            return False, f"خطا در کلیک: {str(e)}"
    
    def fill_input(self, selector, text, by_type="css", clear_first=True):
        """
        پر کردن فیلد ورودی
        """
        if not self.is_initialized:
            return False, "مرورگر راه‌اندازی نشده است"
        
        try:
            if by_type == "css":
                element = self.wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, selector)))
            elif by_type == "xpath":
                element = self.wait.until(EC.presence_of_element_located((By.XPATH, selector)))
            elif by_type == "id":
                element = self.wait.until(EC.presence_of_element_located((By.ID, selector)))
            else:
                return False, "نوع selector نامعتبر است"
            
            # اسکرول به المنت
            self.driver.execute_script("arguments[0].scrollIntoView(true);", element)
            time.sleep(1)
            
            # پاک کردن محتوای قبلی
            if clear_first:
                element.clear()
            
            # تایپ کردن متن
            element.send_keys(text)
            time.sleep(1)
            
            return True, f"متن '{text}' در فیلد وارد شد"
            
        except TimeoutException:
            return False, f"فیلد '{selector}' پیدا نشد"
        except Exception as e:
            return False, f"خطا در پر کردن فیلد: {str(e)}"
    
    def scroll_page(self, direction="down", pixels=500):
        """
        اسکرول صفحه
        """
        if not self.is_initialized:
            return False, "مرورگر راه‌اندازی نشده است"
        
        try:
            if direction == "down":
                self.driver.execute_script(f"window.scrollBy(0, {pixels});")
            elif direction == "up":
                self.driver.execute_script(f"window.scrollBy(0, -{pixels});")
            elif direction == "top":
                self.driver.execute_script("window.scrollTo(0, 0);")
            elif direction == "bottom":
                self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
            
            time.sleep(1)
            return True, f"اسکرول {direction} انجام شد"
            
        except Exception as e:
            return False, f"خطا در اسکرول: {str(e)}"
    
    def wait_for_element(self, selector, by_type="css", timeout=10):
        """
        انتظار برای ظاهر شدن المنت
        """
        if not self.is_initialized:
            return False, "مرورگر راه‌اندازی نشده است"
        
        try:
            wait = WebDriverWait(self.driver, timeout)
            
            if by_type == "css":
                element = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, selector)))
            elif by_type == "xpath":
                element = wait.until(EC.presence_of_element_located((By.XPATH, selector)))
            elif by_type == "id":
                element = wait.until(EC.presence_of_element_located((By.ID, selector)))
            
            return True, "المنت پیدا شد"
            
        except TimeoutException:
            return False, f"المنت '{selector}' در زمان مشخص شده پیدا نشد"
        except Exception as e:
            return False, f"خطا در انتظار: {str(e)}"
    
    def get_page_info(self):
        """
        دریافت اطلاعات صفحه
        """
        if not self.is_initialized:
            return None
        
        try:
            info = {
                "title": self.driver.title,
                "url": self.driver.current_url,
                "page_source_length": len(self.driver.page_source),
                "window_size": self.driver.get_window_size()
            }
            return info
        except Exception as e:
            return {"error": str(e)}
    
    def execute_javascript(self, script):
        """
        اجرای کد JavaScript
        """
        if not self.is_initialized:
            return None, "مرورگر راه‌اندازی نشده است"
        
        try:
            result = self.driver.execute_script(script)
            return result, "کد JavaScript اجرا شد"
        except Exception as e:
            return None, f"خطا در اجرای JavaScript: {str(e)}"
    
    def close_browser(self):
        """
        بستن مرورگر
        """
        if self.driver:
            try:
                self.driver.quit()
                self.is_initialized = False
                return True, "مرورگر بسته شد"
            except Exception as e:
                return False, f"خطا در بستن مرورگر: {str(e)}"
        return True, "مرورگر قبلاً بسته شده بود"

class WebAutomationCommands:
    """
    کلاس تفسیر دستورات فارسی برای اتوماسیون وب
    """
    
    def __init__(self, web_automation):
        self.web_automation = web_automation
        
    def parse_command(self, command):
        """
        تفسیر دستورات فارسی
        """
        command = command.strip().lower()
        
        # دستور باز کردن وب‌سایت
        if any(phrase in command for phrase in ['باز کن', 'برو به', 'وب‌سایت']):
            return self._extract_url_command(command)
        
        # دستور کلیک
        elif any(phrase in command for phrase in ['کلیک', 'روی', 'دکمه']):
            return self._extract_click_command(command)
        
        # دستور پر کردن فرم
        elif any(phrase in command for phrase in ['پر کن', 'وارد کن', 'بنویس']):
            return self._extract_fill_command(command)
        
        # دستور اسکرول
        elif any(phrase in command for phrase in ['اسکرول', 'پایین', 'بالا']):
            return self._extract_scroll_command(command)
        
        # دستور عکس گرفتن
        elif any(phrase in command for phrase in ['عکس', 'اسکرین‌شات']):
            return {"action": "screenshot"}
        
        # دستور انتظار
        elif any(phrase in command for phrase in ['صبر کن', 'انتظار']):
            return self._extract_wait_command(command)
        
        else:
            return {"action": "unknown", "command": command}
    
    def _extract_url_command(self, command):
        """
        استخراج URL از دستور
        """
        # جستجوی الگوهای مختلف URL
        import re
        
        # الگوی URL کامل
        url_pattern = r'https?://[^\s]+'
        url_match = re.search(url_pattern, command)
        
        if url_match:
            return {"action": "navigate", "url": url_match.group()}
        
        # جستجوی نام دامنه
        domain_patterns = [
            r'(\w+\.com)',
            r'(\w+\.ir)',
            r'(\w+\.org)',
            r'(\w+\.net)'
        ]
        
        for pattern in domain_patterns:
            match = re.search(pattern, command)
            if match:
                return {"action": "navigate", "url": match.group()}
        
        return {"action": "navigate", "url": None}
    
    def _extract_click_command(self, command):
        """
        استخراج اطلاعات کلیک از دستور
        """
        # حذف کلمات اضافی
        clean_command = command.replace('کلیک کن', '').replace('روی', '').replace('دکمه', '').strip()
        
        if clean_command:
            return {"action": "click", "target": clean_command, "by_type": "text"}
        
        return {"action": "click", "target": None}
    
    def _extract_fill_command(self, command):
        """
        استخراج اطلاعات پر کردن فرم از دستور
        """
        # الگوهای مختلف برای استخراج
        patterns = [
            r'در (.+?) بنویس (.+)',
            r'فیلد (.+?) را با (.+?) پر کن',
            r'وارد کن (.+?) در (.+)'
        ]
        
        import re
        for pattern in patterns:
            match = re.search(pattern, command)
            if match:
                field = match.group(1).strip()
                text = match.group(2).strip()
                return {"action": "fill", "field": field, "text": text}
        
        return {"action": "fill", "field": None, "text": None}
    
    def _extract_scroll_command(self, command):
        """
        استخراج اطلاعات اسکرول از دستور
        """
        if 'پایین' in command:
            return {"action": "scroll", "direction": "down"}
        elif 'بالا' in command:
            return {"action": "scroll", "direction": "up"}
        elif 'ابتدا' in command or 'بالای صفحه' in command:
            return {"action": "scroll", "direction": "top"}
        elif 'انتها' in command or 'پایین صفحه' in command:
            return {"action": "scroll", "direction": "bottom"}
        
        return {"action": "scroll", "direction": "down"}
    
    def _extract_wait_command(self, command):
        """
        استخراج زمان انتظار از دستور
        """
        import re
        
        # جستجوی عدد در دستور
        number_match = re.search(r'(\d+)', command)
        
        if number_match:
            seconds = int(number_match.group())
            return {"action": "wait", "seconds": seconds}
        
        return {"action": "wait", "seconds": 3}  # پیش‌فرض 3 ثانیه
    
    def execute_command(self, command):
        """
        اجرای دستور تفسیر شده
        """
        parsed = self.parse_command(command)
        
        if parsed["action"] == "navigate":
            if parsed["url"]:
                return self.web_automation.navigate_to_url(parsed["url"])
            else:
                return False, "آدرس وب‌سایت مشخص نشده است"
        
        elif parsed["action"] == "click":
            if parsed["target"]:
                return self.web_automation.click_element(parsed["target"], parsed["by_type"])
            else:
                return False, "هدف کلیک مشخص نشده است"
        
        elif parsed["action"] == "fill":
            if parsed["field"] and parsed["text"]:
                # تلاش برای پیدا کردن فیلد با روش‌های مختلف
                success, message = self.web_automation.fill_input(f"input[placeholder*='{parsed['field']}']", parsed["text"])
                if not success:
                    success, message = self.web_automation.fill_input(f"input[name*='{parsed['field']}']", parsed["text"])
                if not success:
                    success, message = self.web_automation.fill_input(f"textarea[placeholder*='{parsed['field']}']", parsed["text"])
                return success, message
            else:
                return False, "فیلد یا متن مشخص نشده است"
        
        elif parsed["action"] == "scroll":
            return self.web_automation.scroll_page(parsed["direction"])
        
        elif parsed["action"] == "screenshot":
            image, message = self.web_automation.take_screenshot()
            if image:
                return True, message, image
            else:
                return False, message
        
        elif parsed["action"] == "wait":
            time.sleep(parsed["seconds"])
            return True, f"{parsed['seconds']} ثانیه انتظار کشیده شد"
        
        else:
            return False, f"دستور '{command}' شناخته نشده است"

# تابع کمکی برای راه‌اندازی اتوماسیون وب در Streamlit
@st.cache_resource
def get_web_automation():
    """
    دریافت نمونه اتوماسیون وب (با کش برای بهینه‌سازی)
    """
    return WebAutomation()

def get_web_commands(web_automation):
    """
    دریافت مفسر دستورات وب
    """
    return WebAutomationCommands(web_automation)