From d06615b0b33fd53fe34240d26599a359c5306db8 Mon Sep 17 00:00:00 2001 From: VolandSZ <{E-MAIL}> Date: Sun, 15 Mar 2026 11:36:22 +0300 Subject: [PATCH] =?UTF-8?q?=D0=9F=D0=B5=D1=80=D0=B2=D0=B0=D1=8F=20=D1=80?= =?UTF-8?q?=D0=B0=D0=B1=D0=BE=D1=87=D0=B0=D1=8F=20=D0=B2=D0=B5=D1=80=D1=81?= =?UTF-8?q?=D0=B8=D1=8F=20=D1=81=D0=BA=D1=80=D0=B8=D0=BF=D1=82=D0=B0=20?= =?UTF-8?q?=D0=BF=D0=BE=20=D1=81=D0=BE=D0=B7=D0=B4=D0=B0=D0=BD=D0=B8=D1=8E?= =?UTF-8?q?=20=D0=B1=D0=B5=D0=BA=D0=B0=D0=BF=D0=BE=D0=B2=20=D0=BF=D0=BE?= =?UTF-8?q?=D1=80=D1=82=D0=B0=D0=BB=D0=B0=20hivmo=20=D1=87=D0=B5=D1=80?= =?UTF-8?q?=D0=B5=D0=B7=20=D0=B2=D0=B7=D0=B0=D0=B8=D0=BC=D0=BE=D0=B4=D0=B5?= =?UTF-8?q?=D0=B9=D1=81=D1=82=D0=B2=D0=B8=D0=B5=20=D1=81=20=D0=BF=D0=BE?= =?UTF-8?q?=D1=80=D1=82=D0=B0=D0=BB=D0=BE=D0=BC=20=D1=81=20=D0=B8=D1=81?= =?UTF-8?q?=D0=BF=D0=BE=D0=BB=D1=8C=D0=B7=D0=BE=D0=B2=D0=B0=D0=BD=D0=B8?= =?UTF-8?q?=D0=B5=D0=BC=20=D0=B1=D0=B8=D0=B1=D0=BB=D0=B8=D0=BE=D1=82=D0=B5?= =?UTF-8?q?=D0=BA=D0=B8=20playwright?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 35 +++ main.py | 703 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 738 insertions(+) create mode 100644 .gitignore create mode 100644 main.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f59ab04 --- /dev/null +++ b/.gitignore @@ -0,0 +1,35 @@ +# Python +__pycache__/ +*.py[cod] +*.pyo +*.pyd +*.so +*.egg-info/ +.eggs/ + +# Virtual environments +venv/ +.venv/ +env/ + +# Tool caches +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.pyright/ +.coverage +.coverage.* +htmlcov/ + +# Reflex generated artifacts +.web/ +.states/ +*.db + +# IDE / OS +.vscode/ +.idea/ +.DS_Store +Thumbs.db + + diff --git a/main.py b/main.py new file mode 100644 index 0000000..3581deb --- /dev/null +++ b/main.py @@ -0,0 +1,703 @@ +import os +import re +import time + +# from datetime import datetime +from dataclasses import dataclass, field + +# from pathlib import Path +from playwright.sync_api import ( + Browser, + BrowserContext, + Page, + Playwright, + sync_playwright, +) + + +@dataclass +class BackupFile: + """Одна строка таблицы — файл бэкапа""" + + name: str + size: str + type: str + download_url: str = "" + delete_js: str = "" + timestamp: str = "" + + @property + def category(self) -> str: + if self.name.startswith("ftp_"): + return "data" + elif self.type == "SQL dump ZIP": + return "sql_zip" + elif self.type == "raw SQL file": + return "sql_raw" + return "unknown" + + +@dataclass +class BackupGroup: + """Группа файлов одного бэкапа (данные + SQL + ZIP)""" + + timestamp: str + data: BackupFile | None = None + sql_raw: BackupFile | None = None + sql_zip: BackupFile | None = None + + @property + def label(self) -> str: + parts = self.timestamp.split("_") + if len(parts) == 6: + return f"{parts[0]}-{parts[1]}-{parts[2]} {parts[3]}:{parts[4]}:{parts[5]}" + return self.timestamp + + @property + def files(self) -> list[BackupFile]: + return [f for f in [self.data, self.sql_raw, self.sql_zip] if f] + + +class BackupManager: + pw: Playwright + browser: Browser + context: BrowserContext + page: Page + + def __init__(self, username: str, password: str, account_id: str = "564622"): + self.username = username + self.password = password + self.account_id = account_id + # self.backup_folder = backup_folder / datetime.now().strftime("%Y-%m-%d") + self.base_url = "https://www.1gb.ru/adm-alv5/" + self.backup_url = f"{self.base_url}?u={account_id}#/c/b1" + + def start(self): + self.pw = sync_playwright().start() + self.browser = self.pw.chromium.launch(headless=False) + self.context = self.browser.new_context() + self.page = self.context.new_page() + self._login() + + def stop(self): + if self.browser: + self.browser.close() + if self.pw: + self.pw.stop() + + def _login(self): + print("[*] Авторизация...") + self.page.goto(self.base_url, wait_until="networkidle") + self.page.fill("input[name='login_login']", self.username) + self.page.fill("input[name='login_pwd']", self.password) + with self.page.expect_navigation(wait_until="networkidle"): + self.page.click("button[type='submit']") + if "login_login" in self.page.content(): + raise Exception("Авторизация не удалась!") + print("[OK] Авторизация успешна") + + def _go_to_backups(self): + self.page.goto(self.backup_url, wait_until="networkidle") + self.page.wait_for_timeout(10000) + + def _extract_timestamp(self, name: str) -> str: + match = re.search(r"(\d{4}_\d{2}_\d{2}_\d{2}_\d{2}_\d{2})", name) + return match.group(1) if match else "" + + def parse_backup_table(self) -> list[BackupFile]: + rows_data = self.page.evaluate("""() => { + const table = document.querySelector('table.gb-ttc-2'); + if (!table) return []; + + const results = []; + const rows = table.querySelectorAll('tr'); + + for (const row of rows) { + const cells = row.querySelectorAll('td'); + if (cells.length < 4) continue; + + const name = cells[0].innerText.trim(); + const size = cells[1].innerText.trim(); + const type = cells[2].innerText.trim(); + + const actionLinks = cells[3].querySelectorAll('a'); + let downloadUrl = ''; + let deleteJs = ''; + + for (const link of actionLinks) { + const text = link.innerText.trim().toLowerCase(); + if (text === 'скачать') { + downloadUrl = link.getAttribute('href') || ''; + } else if (text === 'удалить') { + deleteJs = link.getAttribute('href') || ''; + } + } + + results.push({ + name: name, + size: size, + type: type, + download_url: downloadUrl, + delete_js: deleteJs + }); + } + + return results; + }""") + + files = [] + for row in rows_data: + f = BackupFile( + name=row["name"], + size=row["size"], + type=row["type"], + download_url=row["download_url"], + delete_js=row["delete_js"], + timestamp=self._extract_timestamp(row["name"]), + ) + files.append(f) + + return files + + def get_grouped_backups(self) -> list[BackupGroup]: + self._go_to_backups() + files = self.parse_backup_table() + + groups: dict[str, BackupGroup] = {} + + for f in files: + ts = f.timestamp + if not ts: + continue + + if ts not in groups: + groups[ts] = BackupGroup(timestamp=ts) + + group = groups[ts] + + if f.category == "data": + group.data = f + elif f.category == "sql_raw": + group.sql_raw = f + elif f.category == "sql_zip": + group.sql_zip = f + + sorted_groups = sorted(groups.values(), key=lambda g: g.timestamp) + return sorted_groups + + def print_backups(self): + groups = self.get_grouped_backups() + + print(f"\n{'=' * 70}") + print(f" РЕЗЕРВНЫЕ КОПИИ: {len(groups)} шт.") + print(f"{'=' * 70}") + + for i, group in enumerate(groups): + print(f"\n Бэкап #{i + 1} [{group.label}]") + print(f" {'─' * 50}") + + if group.data: + f = group.data + print(" 📁 Данные сайта (FTP)") + print(f" Файл: {f.name}") + print(f" Размер: {f.size}") + print(f" Тип: {f.type}") + print(f" Скачать: {f.download_url[:80]}...") + print(f" Удалить: {'есть' if f.delete_js else 'нет'}") + + if group.sql_raw: + f = group.sql_raw + print(" 📄 База данных (SQL)") + print(f" Файл: {f.name}") + print(f" Размер: {f.size}") + print(f" Тип: {f.type}") + print(f" Скачать: {f.download_url[:80]}...") + print(f" Удалить: {'есть' if f.delete_js else 'нет'}") + + if group.sql_zip: + f = group.sql_zip + print(" 📦 База данных (ZIP)") + print(f" Файл: {f.name}") + print(f" Размер: {f.size}") + print(f" Тип: {f.type}") + print(f" Скачать: {f.download_url[:80]}...") + print(f" Удалить: {'есть' if f.delete_js else 'нет'}") + + return groups + + # ===================================================== + # СОЗДАНИЕ БЭКАПОВ + # ===================================================== + + def create_backup( + self, + backup_data: bool = False, + backup_mysql: bool = True, + wait_for_completion: bool = True, + poll_interval: int = 30, + max_wait: int = 600, + ): + """ + Создать резервную копию. + + Args: + backup_data: True — бэкап данных FTP (большой, долгий) + backup_mysql: True — бэкап базы данных MySQL + wait_for_completion: ждать завершения создания + poll_interval: интервал проверки статуса (секунды) + max_wait: максимальное время ожидания (секунды) + """ + if not backup_data and not backup_mysql: + print("[!] Нужно выбрать хотя бы один тип бэкапа") + return False + + self._go_to_backups() + + # Запоминаем количество бэкапов до создания + files_before = self.parse_backup_table() + count_before = len(files_before) + print(f"[*] Бэкапов до создания: {count_before}") + + types = [] + if backup_data: + types.append("FTP-данные") + if backup_mysql: + types.append("MySQL") + print(f"[*] Создание бэкапа: {', '.join(types)}") + + # Ставим чекбоксы + if backup_data: + self.page.click("#t_iis_690148") + self.page.wait_for_timeout(300) + + if backup_mysql: + self.page.click("#t_mysql_444096") + self.page.wait_for_timeout(300) + + # Проверяем + checkbox_state = self.page.evaluate("""() => { + return { + ftp: document.querySelector('#t_iis_690148')?.checked || false, + mysql: document.querySelector('#t_mysql_444096')?.checked || false + } + }""") + print(f" Чекбокс FTP: {'✓' if checkbox_state['ftp'] else '✗'}") + print(f" Чекбокс MySQL: {'✓' if checkbox_state['mysql'] else '✗'}") + + # Нажимаем "создать архивы" + print("[*] Отправка команды на создание...") + self.page.click("input[name='do_upd']") + self.page.wait_for_timeout(5000) + + print("[OK] Команда на создание бэкапа отправлена") + self.page.screenshot(path="create_backup_sent.png") + + if not wait_for_completion: + print("[*] Ожидание отключено. Проверьте статус позже.") + return True + + # Ждём завершения + return self._wait_for_backup_completion( + count_before=count_before, + expected_new=self._expected_new_files(backup_data, backup_mysql), + poll_interval=poll_interval, + max_wait=max_wait, + ) + + def _expected_new_files(self, backup_data: bool, backup_mysql: bool) -> int: + """Сколько новых файлов ожидаем""" + count = 0 + if backup_data: + count += 1 # ftp_... + if backup_mysql: + count += 2 # sqldump_... (raw + zip) + return count + + def _refresh_backup_list(self): + """Обновить список бэкапов кликом по ссылке 'обновить список / статусы'""" + self.page.click("a:has-text('обновить список / статусы')") + self.page.wait_for_timeout(10000) + + def _wait_for_backup_completion( + self, count_before: int, expected_new: int, poll_interval: int, max_wait: int + ) -> bool: + """Ожидание завершения создания бэкапа""" + print(f"\n[*] Ожидание завершения (макс. {max_wait} сек)...") + print(f" Ожидаем {expected_new} новых файлов") + + start_time = time.time() + attempt = 0 + + while True: + elapsed = time.time() - start_time + if elapsed > max_wait: + print(f"\n[!] Таймаут {max_wait} сек. Бэкап возможно ещё создаётся.") + return False + + attempt += 1 + print(f"\n Проверка #{attempt} ({int(elapsed)} сек)...", end=" ") + + # Обновляем список через ссылку на странице + self._refresh_backup_list() + files_now = self.parse_backup_table() + count_now = len(files_now) + new_count = count_now - count_before + + print(f"файлов: {count_now} (новых: {new_count})") + + if new_count >= expected_new: + print(f"\n[OK] Бэкап создан! Новых файлов: {new_count}") + return True + + # Проверяем статус задания + status = self._check_task_status() + if status: + for s in status: + print(f" Задание: {s}") + + print(f" Следующая проверка через {poll_interval} сек...") + time.sleep(poll_interval) + + def _check_task_status(self) -> list[str]: + """Проверяет статус заданий в таблице Schedule/Operation""" + statuses = self.page.evaluate("""() => { + const tables = document.querySelectorAll('table'); + const results = []; + + for (const table of tables) { + const headers = table.querySelectorAll('th'); + const headerTexts = Array.from(headers).map(h => h.innerText.trim()); + + if (headerTexts.some(h => h.includes('Schedule') || h.includes('Operation'))) { + const rows = table.querySelectorAll('tr'); + for (const row of rows) { + const cells = row.querySelectorAll('td'); + if (cells.length >= 3) { + const schedule = cells[0]?.innerText?.trim() || ''; + const server = cells[1]?.innerText?.trim() || ''; + const operation = cells[2]?.innerText?.trim() || ''; + if (schedule || operation) { + results.push(`${schedule} | ${server} | ${operation}`); + } + } + } + } + } + return results; + }""") + return statuses + + def create_backup_mysql_only(self, wait: bool = True): + """Создать только бэкап базы данных""" + return self.create_backup( + backup_data=False, backup_mysql=True, wait_for_completion=wait + ) + + def create_backup_full(self, wait: bool = True): + """Создать полный бэкап (данные + база)""" + return self.create_backup( + backup_data=True, backup_mysql=True, wait_for_completion=wait + ) + + def download_file( + self, backup_file: BackupFile, save_dir: str = "./backups" + ) -> str | None: + """Скачать один файл бэкапа""" + os.makedirs(save_dir, exist_ok=True) + + if not backup_file.download_url: + print(f"[!] Нет ссылки для скачивания: {backup_file.name}") + return None + + print(f"[*] Скачивание: {backup_file.name} ({backup_file.size})...") + + with self.page.expect_download(timeout=600000) as download_info: + # Открываем ссылку в текущем контексте (ссылка имеет target="_blank") + self.page.evaluate(f"""() => {{ + const a = document.createElement('a'); + a.href = '{backup_file.download_url}'; + a.download = ''; + document.body.appendChild(a); + a.click(); + a.remove(); + }}""") + + download = download_info.value + filename = download.suggested_filename or backup_file.name + save_path = os.path.join(save_dir, filename) + download.save_as(save_path) + + file_size = os.path.getsize(save_path) + print(f"[OK] Сохранено: {save_path} ({file_size:,} байт)") + return save_path + + def download_group( + self, + group: BackupGroup, + save_dir: str = "./backups", + include_data: bool = True, + include_sql_raw: bool = True, + include_sql_zip: bool = True, + ) -> list[str]: + """Скачать все файлы одной группы бэкапа""" + downloaded = [] + + print(f"\n[*] Скачивание бэкапа [{group.label}]") + + if include_data and group.data: + path = self.download_file(group.data, save_dir) + if path: + downloaded.append(path) + + if include_sql_raw and group.sql_raw: + path = self.download_file(group.sql_raw, save_dir) + if path: + downloaded.append(path) + + if include_sql_zip and group.sql_zip: + path = self.download_file(group.sql_zip, save_dir) + if path: + downloaded.append(path) + + print(f"[OK] Скачано файлов: {len(downloaded)}") + return downloaded + + def download_latest( + self, + save_dir: str = "./backups", + include_data: bool = False, + include_sql_raw: bool = True, + include_sql_zip: bool = True, + ) -> list[str]: + """Скачать самый свежий бэкап""" + groups = self.get_grouped_backups() + if not groups: + print("[!] Нет бэкапов для скачивания") + return [] + + latest = groups[-1] + print(f"[*] Последний бэкап: [{latest.label}]") + return self.download_group( + latest, + save_dir, + include_data=include_data, + include_sql_raw=include_sql_raw, + include_sql_zip=include_sql_zip, + ) + + def create_and_download( + self, + save_dir: str = "./backups", + backup_data: bool = False, + backup_mysql: bool = True, + ) -> list[str]: + """Создать бэкап и сразу скачать""" + # Запоминаем текущие timestamp'ы + groups_before = self.get_grouped_backups() + timestamps_before = {g.timestamp for g in groups_before} + + # Создаём бэкап + success = self.create_backup( + backup_data=backup_data, backup_mysql=backup_mysql, wait_for_completion=True + ) + + if not success: + print("[!] Бэкап не был создан, скачивание отменено") + return [] + + # Находим новые бэкапы + groups_after = self.get_grouped_backups() + new_groups = [g for g in groups_after if g.timestamp not in timestamps_before] + + if not new_groups: + print("[!] Новые бэкапы не найдены") + return [] + + print(f"\n[*] Найдено новых бэкапов: {len(new_groups)}") + + # Скачиваем все новые + all_downloaded = [] + for group in new_groups: + downloaded = self.download_group( + group, + save_dir, + include_data=backup_data, + include_sql_raw=backup_mysql, + include_sql_zip=backup_mysql, + ) + all_downloaded.extend(downloaded) + + print(f"\n{'=' * 50}") + print(f"[OK] Итого скачано файлов: {len(all_downloaded)}") + for path in all_downloaded: + size = os.path.getsize(path) + print(f" {path} ({size:,} байт)") + print(f"{'=' * 50}") + + return all_downloaded + + def find_old_backups(self, delete_count: int = 1, year: int = 2026): + """ + Найти самые старые бэкапы за указанный год для удаления. + + Args: + delete_count: сколько самых старых удалить + year: за какой год + + Returns: + кортеж (на удаление, оставляем) + """ + groups = self.get_grouped_backups() + + # Фильтруем только бэкапы за указанный год + year_groups = [g for g in groups if g.timestamp.startswith(str(year))] + + if not year_groups: + print(f"[*] Бэкапов за {year} год не найдено") + return None + + print(f"[*] Всего бэкапов за {year} год: {len(year_groups)}") + print(f"[*] Удаляем самых старых: {delete_count}") + + # Сортировка по timestamp + year_groups.sort(key=lambda g: g.timestamp) + + if delete_count >= len(year_groups): + print(f"[!] Нельзя удалить все бэкапы. Максимум: {len(year_groups) - 1}") + delete_count = len(year_groups) - 1 + if delete_count <= 0: + return None + + to_delete = year_groups[:delete_count] + to_keep = year_groups[delete_count:] + + return to_delete, to_keep + + def print_deletion_plan(self, delete_count: int = 1, year: int = 2026): + """Показать план удаления без реального удаления""" + result = self.find_old_backups(delete_count=delete_count, year=year) + + if not result: + return [] + + to_delete, to_keep = result + + print(f"\n{'=' * 70}") + print(f" ПЛАН УДАЛЕНИЯ БЭКАПОВ ЗА {year} ГОД") + print(f"{'=' * 70}") + + print(f"\n 🗑️ НА УДАЛЕНИЕ ({len(to_delete)} шт.):") + print(f" {'─' * 50}") + for i, group in enumerate(to_delete): + file_count = len(group.files) + total_size = sum( + int(f.size.replace(" KB", "").replace(",", "").strip()) + for f in group.files + if "KB" in f.size + ) + print( + f" {i + 1}. [{group.label}] — {file_count} файлов, ~{total_size:,} KB" + ) + + for f in group.files: + print( + f" {'📁' if f.category == 'data' else '📄' if f.category == 'sql_raw' else '📦'} {f.name} ({f.size})" + ) + + print(f"\n ✅ ОСТАВЛЯЕМ ({len(to_keep)} шт.):") + print(f" {'─' * 50}") + for i, group in enumerate(to_keep): + file_count = len(group.files) + print(f" {i + 1}. [{group.label}] — {file_count} файлов") + + return to_delete + + def delete_old_backups( + self, delete_count: int = 1, year: int = 2026, confirm: bool = True + ): + """Удалить старые бэкапы за указанный год""" + to_delete = self.print_deletion_plan(delete_count=delete_count, year=year) + + if not to_delete: + return + + if confirm: + answer = input(f"\nУдалить {len(to_delete)} бэкапов? (y/n): ") + if answer.lower() != "y": + print("Отменено.") + return + + # Автоматически подтверждаем confirm() диалоги + self.page.on("dialog", lambda dialog: dialog.accept()) + + for i, group in enumerate(to_delete): + print(f"\n[*] Удаление бэкапа [{group.label}] ({i + 1}/{len(to_delete)})") + + for f in group.files: + if not f.delete_js: + print(f" [!] Нет ссылки удаления: {f.name}") + continue + + print(f" Удаление: {f.name}...", end="\n") + + js_code = f.delete_js.replace("javascript:", "", 1) + self.page.evaluate(js_code) + self.page.wait_for_timeout(3000) + self._refresh_backup_list() + + # print("ПРОПУЩЕНО (удаление закомментировано)") + + print(f"\n[OK] Готово") + + +# === ИСПОЛЬЗОВАНИЕ === + +if __name__ == "__main__": + mgr = BackupManager( + username="minzdravmo", + password="KB3ZS7*sv6j4zYr@bPdeFWQN38rbtQv", + # backup_folder=Path("F:/test_folder/MyReports"), + account_id="564622", + ) + + try: + mgr.start() + + # 1. Показать текущие бэкапы + groups = mgr.print_backups() + print("\n" + "=" * 70) + + # 2. Создать бэкап только базы данных + # mgr.create_backup_mysql_only(wait=True) + + # 2. Создать бэкап и сразу скачать + mgr.create_and_download( + save_dir="F:/test_folder/MyReports", + backup_data=True, # данные FTP (закомментировано) + backup_mysql=True, # база данных + ) + + # 3. Показать, что удаляем и удалить + mgr.print_deletion_plan(delete_count=1, year=2026) + + mgr.delete_old_backups(delete_count=1, year=2026) + + # Скачать только последний бэкап (без создания нового) + # mgr.download_latest(save_dir="./backups") + + # Скачать конкретный бэкап по номеру + # mgr.download_group(groups[0], save_dir="./backups") + + # Создать бэкап без скачивания + # mgr.create_backup_mysql_only(wait=True) + + # 3. Показать бэкапы после создания + # groups = mgr.print_backups() + + # 4. Создать полный бэкап (раскомментировать когда нужно) + # mgr.create_backup_full(wait=True) + + input("\nEnter для выхода...") + + finally: + mgr.stop()