Оптимизация веб-сканера на основе Selenium с непостоянным временем загрузки страницPython

Программы на Python
Ответить
Anonymous
 Оптимизация веб-сканера на основе Selenium с непостоянным временем загрузки страниц

Сообщение Anonymous »

Я написал скрипт Python с использованием Selenium для сканирования нескольких веб-страниц IEC и обнаружения изменений в опубликованных документах.

Страницы динамически обрабатываются и защищаются CloudFront.
Сценарий работает правильно, но время выполнения нестабильно, а иногда и значительно медленнее.
Замеченные проблемы:
  • driver.get() иногда требуется 10–30 секунд для одного и того же URL
  • WebDriverWait время от времени истекает, хотя таблица в конечном итоге появляется.
  • Общее время выполнения варьируется от ~ 40 секунд до более 3 минут для одного и того же набора URL-адресов.
Что я уже пробовал:
  • Использование пользовательского агента Chrome для настольных компьютеров
  • Отключение флагов автоматизации Selenium
  • Добавление случайных задержек между запросами
  • Использование WebDriverWait вместо фиксированного сна
  • Разбор с помощью BeautifulSoup вместо запросов Selenium DOM
Мне нужен совет по:
  • Определению вероятных узких мест в производительности
  • Структурным улучшениям для уменьшения блокировок или ненужных ожиданий
  • Следует ли в этой конструкции полностью избегать Selenium, если это возможно
Ниже представлена упрощенная версия скрипта.
import os
import re
import time
import random
import pandas as pd
from datetime import datetime
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
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.common.exceptions import TimeoutException, WebDriverException
from bs4 import BeautifulSoup
from webdriver_manager.chrome import ChromeDriverManager

BASE_DIR = r'C:\asdafasfafasfasfasf/asffafsafsaf'
os.makedirs(BASE_DIR, exist_ok=True)
TARGET_FILE = os.path.join(BASE_DIR, 'vote_log.xlsx')

URL_MAP = {
'TC22': 'https://www.iec.ch/dyn/www/f?p=103:26:2 ... ID:1293,25',
'SC22E': 'https://iec.ch/dyn/www/f?p=103:26:21387 ... ID:1414,25',
'SC22F': 'https://www.iec.ch/dyn/www/f?p=103:26:2 ... ID:1415,25',
'SC22G': 'https://www.iec.ch/dyn/www/f?p=103:26:2 ... ID:1416,25',
'SC22H': 'https://www.iec.ch/dyn/www/f?p=103:26:2 ... ID:1441,25',
'SC23H': 'https://www.iec.ch/dyn/www/f?p=103:26:2 ... ID:1426,25',
'SC23K': 'https://iec.ch/dyn/www/f?p=103:26:21387 ... D:10046,25',
'TC115': 'https://www.iec.ch/dyn/www/f?p=103:26:2 ... ID:3988,25',
'TC120': 'https://www.iec.ch/dyn/www/f?p=103:26:2 ... ID:9463,25'
}

def extract_reference(text: str) -> str:
if not text:
return ""
s = " ".join(str(text).split()).strip()
# 예: 22F/846/Q, 120/446/ACWG 5 같은 패턴 먼저 잡기
m = re.match(r'^([A-Z0-9]+(?:[\/\-][A-Z0-9]+)+)', s, re.IGNORECASE)
if m:
return m.group(1).strip().upper()
return s.split(" ")[0].strip().upper()

def is_cloudfront_403(html: str) -> bool:
return ("403 ERROR" in html) and ("cloudfront" in html.lower())

def create_driver():
options = Options()
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')

# ✅ 풀 UA로 (이게 은근히 핵심)
options.add_argument("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")

# ✅ 네 “되는 코드” 쪽 옵션과 동일하게
options.add_argument("--disable-blink-features=AutomationControlled")
options.add_experimental_option("excludeSwitches", ["enable-automation"])
options.add_experimental_option("useAutomationExtension", False)

driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)
driver.set_page_load_timeout(30)

# ✅ 네 “되는 코드”처럼 webdriver 흔적 제거
driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")
return driver

def crawl_vote_table(driver, tc_code, url):
print(f"🔍 [{tc_code}] 확인 중...")

# ✅ 너무 빠르게 연속 호출하면 막힐 수 있어서 랜덤 딜레이
time.sleep(random.uniform(1.5, 4.0))

try:
driver.get(url)
except TimeoutException:
try:
driver.execute_script("window.stop();")
except:
pass
except WebDriverException as e:
print(f"❌ [{tc_code}] 접속 실패: {e}")
return None

html = driver.page_source
if is_cloudfront_403(html):
print(f"🚫 [{tc_code}] CloudFront 403 차단 감지 (이번 실행 중단 권장)")
return None

try:
WebDriverWait(driver, 15).until(EC.presence_of_element_located((By.ID, "example")))
except TimeoutException:
print(f"❌ [{tc_code}] 테이블 로딩 타임아웃")
return []

soup = BeautifulSoup(driver.page_source, 'html.parser')
table = soup.find('table', {'id': 'example'})
if not table:
return []

rows = table.find_all('tr')[2:]
results = []

for row in rows:
cols = row.find_all('td')
if len(cols) < 6:
continue

raw_text = cols[0].get_text(" ", strip=True)
reference = extract_reference(raw_text)

title_tag = cols[0].find('span')
title = title_tag.get_text(" ", strip=True) if title_tag else raw_text.replace(reference, "", 1).strip()

link = cols[5].find('a')
download_url = link['href'] if link else ''
size = link.get_text(strip=True) if link else ''

results.append({
'Reference': reference,
'Title': title,
'Type': cols[1].get_text(strip=True),
'Circulation Date': cols[2].get_text(strip=True),
'Closing Date': cols[3].get_text(strip=True),
'CLC': cols[4].get_text(strip=True),
'Download URL': download_url,
'Size': size,
'Committee': tc_code
})

return results

def main():
if os.path.exists(TARGET_FILE):
df_existing = pd.read_excel(TARGET_FILE)
if 'Reference' in df_existing.columns:
df_existing['Reference'] = df_existing['Reference'].astype(str).apply(extract_reference)
existing_refs = set(df_existing['Reference'].astype(str)) if 'Reference' in df_existing.columns else set()
print(f"✅ 기존 데이터: {len(df_existing)}건")
else:
df_existing = pd.DataFrame()
existing_refs = set()
print("🆕 기존 파일 없음")

driver = create_driver()
crawled = []

try:
for tc, url in URL_MAP.items():
r = crawl_vote_table(driver, tc, url)

# 403 감지(None)이면 여기서 끊는 게 안전
if r is None:
print("🚫 차단 신호가 있어 크롤링을 중단합니다.")
break

crawled.extend(r)
finally:
driver.quit()

if not crawled:
print("❌ 수집 결과 없음")
return

df_new_all = pd.DataFrame(crawled)
df_new_all['Reference'] = df_new_all['Reference'].astype(str).apply(extract_reference)

df_new = df_new_all[~df_new_all['Reference'].isin(existing_refs)].copy()
if df_new.empty:
print("👌 신규 문서 없음")
return

df_new['Date Added'] = datetime.now().strftime('%Y-%m-%d')
df_final = pd.concat([df_existing, df_new], ignore_index=True)

cols = [
'Reference', 'Title', 'Type', 'Circulation Date',
'Closing Date', 'CLC', 'Download URL', 'Size',
'Committee', 'Date Added'
]
for c in cols:
if c not in df_final.columns:
df_final[c] = ''
df_final = df_final[cols]

df_final.to_excel(TARGET_FILE, index=False)

print("\n🎉 업데이트 완료")
print(f" - 신규: {len(df_new)}건 / 총: {len(df_final)}건")
print("\n[새로 추가된 문서]")
for _, r in df_new[['Reference', 'Committee']].iterrows():
print(f"{r['Reference']:

Подробнее здесь: https://stackoverflow.com/questions/798 ... load-times
Ответить

Быстрый ответ

Изменение регистра текста: 
Смайлики
:) :( :oops: :roll: :wink: :muza: :clever: :sorry: :angel: :read: *x)
Ещё смайлики…
   
К этому ответу прикреплено по крайней мере одно вложение.

Если вы не хотите добавлять вложения, оставьте поля пустыми.

Максимально разрешённый размер вложения: 15 МБ.

Вернуться в «Python»