Zhofang's picture
Update app.py
e9eb928 verified
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