Spaces:
Running
Running
import os | |
import time | |
import gradio as gr | |
from selenium import webdriver | |
from selenium.webdriver.chrome.options import Options | |
from selenium.webdriver.support.ui import WebDriverWait | |
from selenium.webdriver.support import expected_conditions as EC | |
from selenium.webdriver.common.by import By | |
from selenium.common.exceptions import WebDriverException # Import specific exception | |
import tempfile | |
import logging | |
import atexit # Untuk cleanup saat skrip exit | |
from threading import Lock # Untuk thread safety jika Gradio menangani request secara konkuren | |
# --- Konfigurasi --- | |
CHROME_INSTANCE_TIMEOUT = 600 # Detik (10 menit) | |
# Konfigurasi logging dasar | |
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
class ChromeDriverManager: | |
"""Mengelola satu instance WebDriver Chrome yang persisten dengan timeout.""" | |
def __init__(self, timeout_seconds): | |
self.driver = None | |
self.last_used_time = None | |
self.timeout_seconds = timeout_seconds | |
self.lock = Lock() # Lock untuk mencegah race condition saat mengakses/membuat driver | |
self.options = self._configure_options() | |
logging.info(f"ChromeDriverManager diinisialisasi dengan timeout {timeout_seconds} detik.") | |
# Daftarkan fungsi cleanup untuk dijalankan saat skrip Python keluar | |
atexit.register(self.quit_driver) | |
def _configure_options(self): | |
"""Konfigurasi opsi Chrome (tanpa ukuran jendela awal).""" | |
chrome_options = Options() | |
chrome_options.add_argument('--headless') | |
chrome_options.add_argument('--disable-gpu') | |
chrome_options.add_argument('--no-sandbox') | |
chrome_options.add_argument('--disable-dev-shm-usage') | |
chrome_options.add_argument('--remote-debugging-port=9222') # Opsional | |
# Jangan set window size di sini, akan diatur per permintaan | |
# chrome_options.add_argument('--window-size=1920,1080') | |
return chrome_options | |
def _initialize_driver(self): | |
"""Membuat instance WebDriver baru.""" | |
logging.info("Menginisialisasi instance WebDriver Chrome baru...") | |
try: | |
# Pertimbangkan menggunakan webdriver-manager jika perlu | |
# from selenium.webdriver.chrome.service import Service as ChromeService | |
# from webdriver_manager.chrome import ChromeDriverManager | |
# service = ChromeService(executable_path=ChromeDriverManager().install()) | |
# driver = webdriver.Chrome(service=service, options=self.options) | |
driver = webdriver.Chrome(options=self.options) | |
self.last_used_time = time.time() | |
logging.info("Instance WebDriver baru berhasil dibuat.") | |
return driver | |
except Exception as e: | |
logging.error(f"Gagal menginisialisasi WebDriver: {e}", exc_info=True) | |
# Set driver ke None jika gagal agar dicoba lagi nanti | |
self.driver = None | |
self.last_used_time = None | |
return None | |
def _is_driver_alive(self): | |
"""Memeriksa apakah driver masih responsif.""" | |
if self.driver is None: | |
return False | |
try: | |
# Perintah ringan untuk memeriksa koneksi, misal mendapatkan URL saat ini | |
_ = self.driver.current_url | |
return True | |
except WebDriverException as e: | |
# Tangkap error spesifik yang mengindikasikan driver mati/tidak terhubung | |
logging.warning(f"Driver tampaknya tidak responsif: {e}") | |
return False | |
except Exception as e: | |
# Tangkap error lain yang mungkin terjadi | |
logging.warning(f"Error tidak terduga saat memeriksa status driver: {e}") | |
return False | |
def get_driver(self, requested_width, requested_height): | |
"""Mendapatkan instance driver yang ada atau membuat yang baru jika perlu.""" | |
with self.lock: # Pastikan operasi ini thread-safe | |
current_time = time.time() | |
needs_new_driver = False | |
if self.driver is None: | |
logging.info("Tidak ada driver aktif, membuat yang baru.") | |
needs_new_driver = True | |
elif not self._is_driver_alive(): | |
logging.warning("Driver yang ada tidak responsif, membuat yang baru.") | |
self.quit_driver(acquire_lock=False) # Tutup driver lama (lock sudah dipegang) | |
needs_new_driver = True | |
elif self.last_used_time and (current_time - self.last_used_time > self.timeout_seconds): | |
logging.info(f"Driver melewati batas waktu idle ({self.timeout_seconds}s). Membuat yang baru.") | |
self.quit_driver(acquire_lock=False) # Tutup driver lama (lock sudah dipegang) | |
needs_new_driver = True | |
if needs_new_driver: | |
self.driver = self._initialize_driver() | |
# Jika inisialisasi gagal, driver akan None, tangani di luar | |
if self.driver is None: | |
logging.error("Gagal membuat instance driver baru setelah mencoba.") | |
return None # Kembalikan None untuk menandakan kegagalan | |
# Jika driver berhasil didapatkan (baru atau lama) | |
if self.driver: | |
try: | |
# Atur ukuran jendela sesuai permintaan SETIAP kali driver digunakan | |
logging.info(f"Mengatur ukuran jendela driver ke {requested_width}x{requested_height}") | |
self.driver.set_window_size(requested_width, requested_height) | |
# Update waktu penggunaan terakhir HANYA jika berhasil mendapatkan/menggunakan driver | |
self.last_used_time = current_time | |
logging.info("Menggunakan instance driver yang ada/baru.") | |
except WebDriverException as e: | |
logging.error(f"Gagal mengatur ukuran jendela atau driver bermasalah: {e}. Menutup driver.", exc_info=True) | |
self.quit_driver(acquire_lock=False) # Tutup driver bermasalah | |
self.driver = None # Pastikan driver di-reset | |
return None # Kembalikan None untuk menandakan kegagalan | |
else: | |
# Kasus di mana _initialize_driver gagal dan self.driver masih None | |
logging.warning("Tidak dapat menyediakan instance driver.") | |
return self.driver # Kembalikan driver (bisa None jika gagal total) | |
def quit_driver(self, acquire_lock=True): | |
"""Menutup instance WebDriver jika ada.""" | |
# Fungsi ini bisa dipanggil dari atexit atau saat timeout/error, | |
# jadi perlu penanganan lock yang fleksibel | |
if acquire_lock: | |
acquired = self.lock.acquire(timeout=5) # Coba kunci, timeout jika macet | |
if not acquired: | |
logging.error("Gagal mendapatkan lock untuk quit_driver. Melewati penutupan.") | |
return | |
try: | |
if self.driver: | |
logging.info("Menutup instance WebDriver...") | |
try: | |
self.driver.quit() | |
logging.info("WebDriver berhasil ditutup.") | |
except Exception as e: | |
logging.error(f"Error saat menutup WebDriver: {e}", exc_info=True) | |
finally: | |
# Selalu set ke None meskipun quit gagal | |
self.driver = None | |
self.last_used_time = None | |
else: | |
logging.debug("Tidak ada instance WebDriver aktif untuk ditutup.") # Log level debug | |
finally: | |
if acquire_lock and acquired: # Hanya release jika lock didapatkan di sini | |
self.lock.release() | |
# --- Buat instance manager global --- | |
chrome_manager = ChromeDriverManager(timeout_seconds=CHROME_INSTANCE_TIMEOUT) | |
# Function to capture a screenshot using the managed driver | |
def capture_screenshot_gradio(url, window_width, window_height, internal_delay): | |
""" | |
Mengambil screenshot halaman web menggunakan instance Chrome yang dikelola. | |
""" | |
# Validasi URL sederhana | |
if not url.startswith(('http://', 'https://')): | |
logging.info(f"Menambahkan 'http://' ke URL: {url}") | |
url = 'http://' + url | |
# Validasi input ukuran dan delay | |
try: | |
window_width = int(window_width) | |
window_height = int(window_height) | |
internal_delay = int(internal_delay) | |
if window_width <= 0 or window_height <= 0 or internal_delay < 0: | |
raise ValueError("Width, Height harus positif dan Delay tidak boleh negatif") | |
except (ValueError, TypeError) as e: | |
logging.error(f"Input tidak valid untuk ukuran/delay: {e}") | |
raise gr.Error(f"Input tidak valid: Lebar, Tinggi harus angka positif dan Delay tidak boleh negatif.") | |
logging.info(f"Permintaan screenshot untuk URL: {url}") | |
logging.info(f"Opsi: Lebar={window_width}, Tinggi={window_height}, Delay={internal_delay} detik") | |
driver = None # Inisialisasi driver di scope ini | |
screenshot_path = None | |
try: | |
# Dapatkan driver dari manager (akan membuat baru atau menggunakan yang ada) | |
# Ukuran jendela akan diatur di dalam get_driver | |
driver = chrome_manager.get_driver(window_width, window_height) | |
# Jika get_driver gagal mengembalikan driver | |
if driver is None: | |
raise gr.Error("Gagal mendapatkan atau menginisialisasi instance WebDriver Chrome.") | |
logging.info(f"Navigasi ke URL: {url}") | |
# Set page load timeout untuk mencegah hang jika halaman sangat lambat | |
driver.set_page_load_timeout(60) # Timeout 60 detik untuk memuat halaman | |
try: | |
driver.get(url) | |
except Exception as page_load_error: | |
logging.error(f"Timeout atau error saat memuat URL {url}: {page_load_error}") | |
# Pertimbangkan apakah driver harus ditutup jika hanya load gagal | |
# chrome_manager.quit_driver() # Opsional: tutup driver jika load gagal total | |
raise gr.Error(f"Gagal memuat URL {url}: {page_load_error}") | |
# 1. Tunggu hingga status dokumen 'complete' | |
logging.info("Menunggu document.readyState menjadi 'complete'...") | |
try: | |
WebDriverWait(driver, 45).until( # Timeout sedikit lebih lama | |
lambda d: d.execute_script('return document.readyState') == 'complete' | |
) | |
logging.info("document.readyState sudah 'complete'.") | |
except Exception as wait_error: | |
logging.warning(f"Timeout atau error saat menunggu readyState complete untuk {url}: {wait_error}") | |
# Lanjutkan saja, mungkin halaman masih bisa di-screenshot sebagian | |
# 2. Tambahkan delay internal setelah 'complete' | |
if internal_delay > 0: | |
logging.info(f"Menunggu delay tambahan selama {internal_delay} detik...") | |
time.sleep(internal_delay) | |
logging.info("Delay selesai.") | |
else: | |
logging.info("Tidak ada delay tambahan (0 detik).") | |
# Buat file sementara untuk menyimpan screenshot | |
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp_file: | |
screenshot_path = tmp_file.name | |
logging.info(f"Mengambil screenshot ke: {screenshot_path}") | |
driver.save_screenshot(screenshot_path) | |
logging.info("Screenshot berhasil diambil.") | |
return screenshot_path | |
except WebDriverException as wd_error: | |
# Jika terjadi error spesifik WebDriver, kemungkinan besar instance-nya bermasalah | |
logging.error(f"WebDriverException terjadi saat memproses {url}: {wd_error}", exc_info=True) | |
logging.warning("Menutup instance driver karena WebDriverException.") | |
chrome_manager.quit_driver() # Paksa tutup driver yang bermasalah | |
# Hapus file sementara jika ada | |
if screenshot_path and os.path.exists(screenshot_path): | |
try: os.remove(screenshot_path) | |
except OSError: pass | |
raise gr.Error(f"Terjadi masalah dengan browser: {wd_error}") | |
except Exception as e: | |
logging.error(f"Terjadi kesalahan umum saat memproses URL {url}: {e}", exc_info=True) | |
# Hapus file sementara jika ada | |
if screenshot_path and os.path.exists(screenshot_path): | |
try: | |
os.remove(screenshot_path) | |
logging.info(f"Membersihkan file sementara yang gagal: {screenshot_path}") | |
except OSError as remove_err: | |
logging.error(f"Gagal menghapus file sementara {screenshot_path}: {remove_err}") | |
# Tidak perlu menutup driver di sini kecuali errornya WebDriverException (sudah ditangani) | |
raise gr.Error(f"Gagal mengambil screenshot: {e}") | |
# finally: | |
# TIDAK ADA driver.quit() di sini lagi. Dikelola oleh ChromeDriverManager. | |
# Cukup pastikan instance manager dibuat di luar dan atexit terdaftar. | |
# logging.debug("Blok finally capture_screenshot_gradio selesai.") # Untuk debugging jika perlu | |
# --- Membuat Antarmuka Gradio (Sama seperti sebelumnya) --- | |
# 1. Definisikan Komponen Input | |
url_input = gr.Textbox( | |
label="Masukkan URL Website", | |
placeholder="contoh: https://www.google.com atau example.com" | |
) | |
width_input = gr.Number(label="Lebar Jendela (px)", value=1920, minimum=300, step=10, info="Lebar viewport browser.") | |
height_input = gr.Number(label="Tinggi Jendela (px)", value=1080, minimum=200, step=10, info="Tinggi viewport browser.") | |
delay_input = gr.Number(label="Delay Tambahan (detik)", value=5, minimum=0, step=1, info="Waktu tunggu ekstra setelah halaman 'lengkap'.") | |
# 2. Definisikan Komponen Output | |
output_image = gr.Image(type="filepath", label="Hasil Screenshot") | |
# 3. Membuat instance Interface Gradio | |
iface = gr.Interface( | |
fn=capture_screenshot_gradio, | |
inputs=[url_input, width_input, height_input, delay_input], | |
outputs=output_image, | |
title="Pengambil Screenshot Website Cepat (Chrome Persisten)", | |
description="Masukkan URL, atur ukuran jendela dan delay. Instance Chrome akan berjalan hingga 10 menit untuk mempercepat respons.", | |
allow_flagging='never', | |
examples=[ | |
["https://gradio.app", 1280, 720, 3], | |
["https://github.com", 1920, 1080, 5], | |
["https://www.google.com/maps", 1024, 768, 8] | |
] | |
) | |
# Menjalankan aplikasi Gradio | |
if __name__ == "__main__": | |
print("Menjalankan aplikasi Gradio dengan instance Chrome persisten...") | |
print(f"Instance Chrome akan ditutup otomatis setelah {CHROME_INSTANCE_TIMEOUT} detik tidak aktif atau saat aplikasi berhenti.") | |
print("Pastikan Google Chrome dan ChromeDriver yang sesuai sudah terinstall.") | |
print("Anda mungkin perlu menginstal: pip install selenium webdriver-manager gradio") # webdriver-manager jika digunakan | |
iface.launch() | |
print("Aplikasi Gradio ditutup.") | |
# atexit akan memanggil chrome_manager.quit_driver() secara otomatis di sini |