This commit is contained in:
root
2026-01-22 18:32:18 +05:00
parent 98f6188e12
commit 5fd5de3baf
14 changed files with 1045 additions and 173 deletions

1
services/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Services package

380
services/browser_service.py Normal file
View File

@@ -0,0 +1,380 @@
"""
Модуль для автоматизации работы с браузером через Selenium
"""
import time
import logging
import os
import sys
import platform
import subprocess
from pathlib import Path
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.firefox.options import Options as FirefoxOptions
from selenium.webdriver.firefox.service import Service as FirefoxService
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
try:
from webdriver_manager.firefox import GeckoDriverManager
HAS_WEBDRIVER_MANAGER = True
except ImportError:
HAS_WEBDRIVER_MANAGER = False
class BrowserService:
"""Сервис для автоматизации работы с браузером."""
def __init__(self, download_dir, firefox_binary_path=None):
"""Инициализирует браузер с указанными параметрами."""
self.download_dir = download_dir
self.firefox_binary_path = firefox_binary_path or self._find_firefox_path()
self.driver = None
self.wait = None
self._setup_driver()
@staticmethod
def _find_firefox_path():
"""Автоматически находит путь к Firefox для текущей ОС."""
system = platform.system()
found_path = None
if system == "Windows":
# Windows пути - разные места установки Firefox
possible_paths = [
r"C:\Program Files\Mozilla Firefox\firefox.exe",
r"C:\Program Files (x86)\Mozilla Firefox\firefox.exe",
os.path.expandvars(r"%PROGRAMFILES%\Mozilla Firefox\firefox.exe"),
os.path.expandvars(r"%PROGRAMFILES(X86)%\Mozilla Firefox\firefox.exe"),
os.path.expandvars(r"%LOCALAPPDATA%\Mozilla Firefox\firefox.exe"),
]
# Пытаемся найти через реестр (Windows)
try:
import winreg
try:
reg_path = r"SOFTWARE\Mozilla\Mozilla Firefox"
with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, reg_path) as key:
install_dir = winreg.QueryValueEx(key, "InstallDir")[0]
possible_paths.insert(0, os.path.join(install_dir, "firefox.exe"))
except:
pass
except:
pass
elif system == "Darwin":
# macOS пути
possible_paths = [
"/Applications/Firefox.app/Contents/MacOS/firefox",
"/usr/local/bin/firefox",
]
else: # Linux
# Linux пути
possible_paths = [
"/snap/bin/firefox",
"/usr/bin/firefox",
"/usr/local/bin/firefox",
"/opt/firefox/firefox",
]
# Ищем Firefox в возможных местах
for path in possible_paths:
if path and os.path.exists(path):
logging.info(f"✅ Firefox найден: {path}")
return path
# Пытаемся найти через which/where команды
try:
if system == "Windows":
result = subprocess.run(["where", "firefox.exe"], capture_output=True, text=True, timeout=5)
else:
result = subprocess.run(["which", "firefox"], capture_output=True, text=True, timeout=5)
if result.returncode == 0 and result.stdout.strip():
firefox_path = result.stdout.strip().split('\n')[0]
logging.info(f"✅ Firefox найден через {system} search: {firefox_path}")
return firefox_path
except:
pass
logging.warning(f"⚠️ Firefox не найден в стандартных местах для {system}")
return None
def _setup_driver(self):
"""Настраивает и запускает веб-драйвер."""
try:
firefox_options = FirefoxOptions()
# ===== ОПЦИИ ЗАГРУЗКИ =====
firefox_options.set_preference("browser.download.folderList", 2)
firefox_options.set_preference("browser.download.dir", self.download_dir)
firefox_options.set_preference("browser.download.useDownloadDir", True)
firefox_options.set_preference("browser.helperApps.neverAsk.saveToDisk", "application/octet-stream")
# ===== ОТКЛЮЧАЕМ HEADLESS (окно ДОЛЖНО быть видимым) =====
# Убедимся, что headless режим НЕ установлен
# (это заставит браузер открыться в нормальном режиме)
# Устанавливаем путь к Firefox только если он найден
if self.firefox_binary_path:
logging.info(f" ✅ Используется Firefox: {self.firefox_binary_path}")
firefox_options.binary_location = self.firefox_binary_path
else:
logging.info(" ⚠️ Firefox путь не определен, буду использовать системный поиск...")
logging.info("🚀 Запуск Firefox драйвера с ВИДИМЫМ окном...")
# Попытка 1: Использовать webdriver-manager (самый надежный способ)
if HAS_WEBDRIVER_MANAGER:
try:
logging.info(" Попытка 1: webdriver-manager...")
service = FirefoxService(GeckoDriverManager().install())
self.driver = webdriver.Firefox(service=service, options=firefox_options)
except Exception as e:
logging.warning(f" webdriver-manager не сработал: {e}")
self.driver = None
# Попытка 2: Прямой запуск (Selenium сам найдет geckodriver в PATH)
if not self.driver:
try:
logging.info(" Попытка 2: Прямой запуск (Selenium ищет geckodriver в PATH)...")
self.driver = webdriver.Firefox(options=firefox_options)
except Exception as e:
logging.warning(f" Прямой запуск не сработал: {e}")
self.driver = None
if not self.driver:
raise Exception("Не удалось инициализировать Firefox драйвер ни одним способом")
# Даем браузеру время открыться
time.sleep(2)
self.driver.set_page_load_timeout(300)
self.wait = WebDriverWait(self.driver, 10)
logging.info("✅ Браузер открыт и готов к работе.")
except Exception as e:
logging.error(f"❌ Ошибка при настройке драйвера: {e}")
logging.error(f"\n Решения:")
logging.error(f" 1. Убедитесь, что Firefox установлен")
logging.error(f" 2. На Windows: установите Firefox с https://mozilla.org/firefox")
raise
def open_page(self, url):
"""Открывает страницу по URL."""
try:
logging.info(f"🌐 Переход на страницу: {url}")
self.driver.get(url)
logging.info("✅ Страница загружена.")
time.sleep(2)
except Exception as e:
logging.error(f"❌ Ошибка при открытии страницы: {e}")
raise
def wait_page_load(self, seconds=5):
"""Ждет загрузки страницы."""
logging.info(f"⏳ Ожидание {seconds} сек...")
time.sleep(seconds)
def click_button_by_text(self, button_text, exact_match=False, timeout=10):
"""
Ищет и нажимает кнопку по тексту в различных формах.
Args:
button_text (str): Текст кнопки для поиска
exact_match (bool): Точное совпадение или частичное
timeout (int): Время ожидания в секундах
Returns:
bool: True если кнопка найдена и нажата, False иначе
"""
try:
logging.info(f"🔎 Ищу кнопку: '{button_text}'")
button = self._find_button_by_text(button_text, exact_match, timeout)
if button:
logging.info(f"✅ Кнопка найдена, нажимаю...")
# Скролл к кнопке если нужно
self.driver.execute_script("arguments[0].scrollIntoView(true);", button)
time.sleep(0.5)
button.click()
logging.info(f"✅ Кнопка '{button_text}' нажата успешно.")
time.sleep(1)
return True
else:
logging.warning(f"⚠️ Кнопка '{button_text}' не найдена.")
return False
except Exception as e:
logging.error(f"❌ Ошибка при нажатии кнопки '{button_text}': {e}")
return False
def _find_button_by_text(self, button_text, exact_match=False, timeout=10):
"""
Находит кнопку по тексту различными способами.
Args:
button_text (str): Текст для поиска
exact_match (bool): Точное совпадение
timeout (int): Время ожидания
Returns:
WebElement или None
"""
button_text_lower = button_text.lower().strip()
end_time = time.time() + timeout
while time.time() < end_time:
# Вариант 1: Поиск по тегу <button>
buttons = self.driver.find_elements(By.TAG_NAME, 'button')
for button in buttons:
if self._text_matches(button.text, button_text_lower, exact_match):
if self._is_clickable(button):
return button
# Вариант 2: Поиск по роли button
buttons = self.driver.find_elements(By.CSS_SELECTOR, '[role="button"]')
for button in buttons:
if self._text_matches(button.text, button_text_lower, exact_match):
if self._is_clickable(button):
return button
# Вариант 3: Поиск по содержимому текста (для div/span с текстом)
try:
elements = self.driver.find_elements(By.XPATH,
f"//*[contains(translate(text(), 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), '{button_text_lower}')][not(self::script or self::style)]")
for elem in elements:
if self._is_clickable(elem):
return elem
except:
pass
# Вариант 4: Поиск по атрибутам (aria-label, title, etc)
for attr in ['aria-label', 'title', 'data-text', 'placeholder', 'value']:
try:
elements = self.driver.find_elements(By.XPATH, f"//*[@{attr}]")
for elem in elements:
attr_value = elem.get_attribute(attr)
if attr_value and self._text_matches(attr_value, button_text_lower, exact_match):
if self._is_clickable(elem):
return elem
except:
pass
# Вариант 5: Поиск label и возврат его родителя (для div+label случаев)
try:
labels = self.driver.find_elements(By.TAG_NAME, 'label')
for label in labels:
if self._text_matches(label.text, button_text_lower, exact_match):
if self._is_clickable(label):
return label
# Если сам label не кликабелен, ищем кликабельного родителя
try:
parent = label.find_element(By.XPATH, "ancestor::div[@onclick] | ancestor::div[@role='button'] | ancestor::button")
if self._is_clickable(parent):
return parent
except:
pass
except:
pass
# Вариант 6: <input> элементы (button, submit, reset)
try:
inputs = self.driver.find_elements(By.CSS_SELECTOR, 'input[type="button"], input[type="submit"], input[type="reset"]')
for inp in inputs:
value = inp.get_attribute('value')
if value and self._text_matches(value, button_text_lower, exact_match):
if self._is_clickable(inp):
return inp
except:
pass
time.sleep(0.5)
return None
def _text_matches(self, element_text, search_text, exact_match):
"""Проверяет совпадение текста."""
element_text_clean = element_text.lower().strip()
if exact_match:
return element_text_clean == search_text
else:
return search_text in element_text_clean
def _is_clickable(self, element):
"""Проверяет, можно ли кликнуть на элемент."""
try:
return element.is_displayed() and element.is_enabled()
except:
return False
def upload_file(self, file_path, selector='input[type="file"]', by=By.CSS_SELECTOR):
"""Загружает файл через input."""
try:
logging.info(f"📤 Загружаю файл: {Path(file_path).name}")
# Попытка кликнуть на область загрузки
try:
drop_area = self.driver.find_element(By.CSS_SELECTOR, '.upload-area, .drop-zone, [data-upload]')
drop_area.click()
time.sleep(1)
logging.info(" Найдена область загрузки")
except:
pass
file_input = self.wait.until(EC.presence_of_element_located((by, selector)))
file_input.send_keys(file_path)
logging.info(f"✅ Файл загружен: {Path(file_path).name}")
time.sleep(1)
return True
except Exception as e:
logging.error(f"❌ Ошибка при загрузке файла {Path(file_path).name}: {e}")
return False
def close(self):
"""Закрывает браузер."""
if self.driver:
self.driver.quit()
logging.info("🛑 Браузер закрыт.")
def get_page_title(self):
"""Возвращает заголовок страницы."""
return self.driver.title
def open_new_tab(self, url):
"""Открывает новую вкладку с заданным URL."""
try:
logging.info(f"📑 Открываю новую вкладку: {url}")
# Открываем новую вкладку
self.driver.execute_script("window.open(arguments[0], '_blank');", url)
# Переключаемся на последнюю открытую вкладку
self.driver.switch_to.window(self.driver.window_handles[-1])
time.sleep(2)
logging.info("✅ Новая вкладка открыта и активна.")
return True
except Exception as e:
logging.error(f"❌ Ошибка при открытии новой вкладки: {e}")
return False
def close_current_tab(self):
"""Закрывает текущую вкладку и переходит на предыдущую."""
try:
if len(self.driver.window_handles) > 1:
logging.info("📑 Закрываю текущую вкладку...")
self.driver.close()
# Переходим на последнюю оставшуюся вкладку
self.driver.switch_to.window(self.driver.window_handles[-1])
time.sleep(1)
logging.info("✅ Перешел на предыдущую вкладку.")
return True
else:
logging.warning("⚠️ Это последняя вкладка, не закрываю.")
return False
except Exception as e:
logging.error(f"❌ Ошибка при закрытии вкладки: {e}")
return False
def get_tabs_count(self):
"""Возвращает количество открытых вкладок."""
return len(self.driver.window_handles)

97
services/file_finder.py Normal file
View File

@@ -0,0 +1,97 @@
"""
Модуль для поиска файлов формата .xlsx в директории
"""
import logging
from pathlib import Path
class FileFinderService:
"""Сервис для поиска и управления файлами .xlsx"""
def __init__(self):
"""Инициализирует сервис поиска файлов."""
self.xlsx_files = []
logging.info("FileFinderService инициализирован.")
def find_xlsx_files(self, directory_path, recursive=True):
"""
Ищет все файлы формата .xlsx в директории и сортирует их.
Args:
directory_path (str или Path): Путь к директории для поиска
recursive (bool): Искать ли рекурсивно по подпапкам
Returns:
list: Список путей до файлов .xlsx в порядке возрастания
"""
try:
dir_path = Path(directory_path)
if not dir_path.exists():
logging.error(f"❌ Директория не найдена: {directory_path}")
return []
if not dir_path.is_dir():
logging.error(f"❌ Это не директория: {directory_path}")
return []
logging.info(f"🔍 Начинаю поиск .xlsx файлов в: {dir_path}")
# Ищем все файлы .xlsx
if recursive:
xlsx_files = list(dir_path.glob('**/*.xlsx'))
logging.info(" (поиск во всех подпапках)")
else:
xlsx_files = list(dir_path.glob('*.xlsx'))
logging.info(" (только текущая папка)")
# Сортируем в порядке возрастания по имени файла
xlsx_files.sort(key=lambda x: x.name)
# Сохраняем результаты
self.xlsx_files = xlsx_files
logging.info(f"✅ Найдено файлов .xlsx: {len(xlsx_files)}")
# Логируем каждый найденный файл
if xlsx_files:
for idx, file_path in enumerate(xlsx_files, 1):
logging.info(f" [{idx}] {file_path.name}")
else:
logging.warning("⚠️ Файлы .xlsx не найдены в указанной директории")
return xlsx_files
except Exception as e:
logging.error(f"❌ Ошибка при поиске .xlsx файлов: {e}")
return []
def get_file_paths(self):
"""Возвращает список сохраненных путей файлов."""
return self.xlsx_files
def get_file_paths_as_strings(self):
"""Возвращает список путей в виде строк."""
return [str(file_path) for file_path in self.xlsx_files]
def get_file_count(self):
"""Возвращает количество найденных файлов."""
return len(self.xlsx_files)
def get_file_by_index(self, index):
"""Возвращает файл по индексу (0-based)."""
try:
if 0 <= index < len(self.xlsx_files):
file_path = self.xlsx_files[index]
logging.info(f"📄 Получен файл #{index + 1}: {file_path.name}")
return file_path
else:
logging.warning(f"⚠️ Индекс {index} выходит за границы (всего: {len(self.xlsx_files)})")
return None
except Exception as e:
logging.error(f"❌ Ошибка при получении файла по индексу: {e}")
return None
def clear(self):
"""Очищает список найденных файлов."""
self.xlsx_files = []
logging.info("🗑️ Список файлов очищен.")

186
services/processor.py Normal file
View File

@@ -0,0 +1,186 @@
"""
Модуль для обработки файлов и управления циклом загрузок
"""
import logging
import time
from pathlib import Path
from services.file_finder import FileFinderService
from services.browser_service import BrowserService
class FileProcessorService:
"""Сервис для обработки файлов и управления процессом загрузок"""
def __init__(self, directory_path, url, download_dir, firefox_path=None):
"""
Инициализирует процессор файлов.
Args:
directory_path (str): Путь к директории с .xlsx файлами
url (str): URL страницы для загрузки файлов
download_dir (str): Путь для загрузок
firefox_path (str): Путь к Firefox (опционально)
"""
self.directory_path = directory_path
self.url = url
self.download_dir = download_dir
self.firefox_path = firefox_path
self.file_finder = FileFinderService()
self.browser = None
self.processed_files = 0
self.failed_files = []
logging.info("=" * 70)
logging.info("🚀 ИНИЦИАЛИЗАЦИЯ ПРОЦЕССОРА ФАЙЛОВ")
logging.info("=" * 70)
def start_processing(self, upload_button_texts=None, process_button_texts=None):
"""
Запускает процесс поиска и загрузки файлов.
Args:
upload_button_texts (list): Варианты текста кнопки загрузки файла
process_button_texts (list): Варианты текста кнопки обработки
Returns:
dict: Результаты обработки
"""
if upload_button_texts is None:
upload_button_texts = ["Загрузить файл", "Выбрать файл", "Добавить файл"]
if process_button_texts is None:
process_button_texts = ["Загрузить", "Обработать", "Преобразовать", "Изменить", "Конвертировать"]
results = {
"total_files": 0,
"processed_files": 0,
"failed_files": [],
"errors": []
}
try:
# Шаг 1: Поиск файлов
logging.info("\n📂 ШАГ 1: ПОИСК ФАЙЛОВ")
logging.info("-" * 70)
xlsx_files = self.file_finder.find_xlsx_files(self.directory_path, recursive=True)
results["total_files"] = len(xlsx_files)
if not xlsx_files:
logging.error("❌ Файлы не найдены. Процесс прерван.")
results["errors"].append("Файлы .xlsx не найдены в указанной директории")
return results
# Шаг 2: Инициализация браузера
logging.info("\n🌐 ШАГ 2: ИНИЦИАЛИЗАЦИЯ БРАУЗЕРА")
logging.info("-" * 70)
self.browser = BrowserService(self.download_dir, self.firefox_path)
self.browser.open_page(self.url)
self.browser.wait_page_load(5)
# Шаг 3: Обработка каждого файла
logging.info("\n📋 ШАГ 3: ОБРАБОТКА ФАЙЛОВ")
logging.info("-" * 70)
for idx, file_path in enumerate(xlsx_files, 1):
# Задержка 5 сек перед следующим файлом (кроме первого)
if idx > 1:
logging.info(f"\n⏳ Задержка 5 сек перед следующим файлом...")
time.sleep(5)
logging.info(f"\n📄 Файл {idx}/{len(xlsx_files)}: {file_path.name}")
logging.info("=" * 70)
try:
# Открываем новую вкладку для каждого файла (кроме первого)
if idx > 1:
logging.info(f" Открываю новую вкладку для файла...")
if not self.browser.open_new_tab(self.url):
logging.warning(f"⚠️ Не удалось открыть новую вкладку")
self.browser.wait_page_load(3)
# Поиск и клик по кнопке загрузки
upload_button_found = False
for btn_text in upload_button_texts:
if self.browser.click_button_by_text(btn_text):
upload_button_found = True
break
if not upload_button_found:
logging.warning(f"⚠️ Не найдена кнопка загрузки файла")
self.failed_files.append({
"file": file_path.name,
"reason": "Кнопка загрузки не найдена"
})
results["failed_files"].append(file_path.name)
continue
# Загрузка файла
self.browser.wait_page_load(1)
upload_success = self.browser.upload_file(str(file_path))
if not upload_success:
logging.warning(f"⚠️ Ошибка при загрузке файла")
self.failed_files.append({
"file": file_path.name,
"reason": "Ошибка при загрузке файла"
})
results["failed_files"].append(file_path.name)
continue
# Поиск и клик по кнопке обработки
self.browser.wait_page_load(2)
process_button_found = False
for btn_text in process_button_texts:
if self.browser.click_button_by_text(btn_text):
process_button_found = True
break
if not process_button_found:
logging.warning(f"⚠️ Не найдена кнопка обработки")
# Это не критичная ошибка, файл уже загружен
# Ожидание обработки
self.browser.wait_page_load(3)
self.processed_files += 1
results["processed_files"] += 1
logging.info(f"✅ Файл обработан успешно")
# Закрываем вкладку после обработки (если это не первый файл)
if idx > 1:
logging.info(f" Закрываю вкладку после обработки...")
self.browser.close_current_tab()
except Exception as e:
logging.error(f"❌ Ошибка при обработке файла: {e}")
self.failed_files.append({
"file": file_path.name,
"reason": str(e)
})
results["failed_files"].append(file_path.name)
# Итоговый отчет
logging.info("\n" + "=" * 70)
logging.info("📊 ИТОГОВЫЙ ОТЧЕТ")
logging.info("=" * 70)
logging.info(f"Всего файлов: {results['total_files']}")
logging.info(f"Обработано успешно: {results['processed_files']}")
logging.info(f"Ошибок: {len(results['failed_files'])}")
if results["failed_files"]:
logging.warning("\n❌ Файлы с ошибками:")
for file_info in self.failed_files:
logging.warning(f" - {file_info['file']}: {file_info['reason']}")
logging.info("\n✅ ОБРАБОТКА ЗАВЕРШЕНА")
logging.info("=" * 70)
return results
except Exception as e:
logging.error(f"❌ Критическая ошибка: {e}")
results["errors"].append(str(e))
return results
finally:
if self.browser:
self.browser.close()