import argparse import os import re import sys import time # from datetime import datetime from dataclasses import dataclass # 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" self._stdout_original = None self._log_file = None def enable_output_log(self, log_path: str = "backup.log"): """Duplicate all stdout to a log file.""" if self._stdout_original is None: self._stdout_original = sys.stdout if self._log_file is None: self._log_file = open(log_path, "a", encoding="utf-8", buffering=1) class _Tee: def __init__(self, *streams): self._streams = streams def write(self, data): for s in self._streams: s.write(data) return len(data) def flush(self): for s in self._streams: s.flush() sys.stdout = _Tee(self._stdout_original, self._log_file) 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._stdout_original is not None: sys.stdout = self._stdout_original self._stdout_original = None if self._log_file is not None: self._log_file.close() self._log_file = None 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 _refresh_backup_list(self): """Обновить список бэкапов кликом по ссылке 'обновить список / статусы'""" self.page.click("a:has-text('обновить список / статусы')") 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 _is_file_ready(self, backup_file: BackupFile) -> bool: """Проверяет что файл готов (не в процессе создания)""" size = backup_file.size.lower() # Готовый файл: "1033513 KB" # В процессе: "файл формируется, 50048 KB" if "формируется" in size or "создаётся" in size or "progress" in size: return False return True 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 _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 create_backup( self, backup_data: bool = False, backup_mysql: bool = True, wait_for_completion: bool = True, poll_interval: int = 60, 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() # Запоминаем количество бэкапов до создания ready_files_before = [ f for f in self.parse_backup_table() if self._is_file_ready(f) ] count_before = len(ready_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 _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() # Считаем только ГОТОВЫЕ файлы (размер вида "12345 KB" без "формируется") ready_files = [f for f in files_now if self._is_file_ready(f)] ready_count = len(ready_files) new_ready = ready_count - count_before # Считаем файлы в процессе создания in_progress = [f for f in files_now if not self._is_file_ready(f)] print( f"готовых: {ready_count} (новых: {new_ready}), в процессе: {len(in_progress)}" ) if in_progress: for f in in_progress: print(f" ⏳ {f.name} — {f.size}") if new_ready >= expected_new: print(f"\n[OK] Бэкап создан! Новых готовых файлов: {new_ready}") 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 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("\n[OK] Готово") def parse_args(): parser = argparse.ArgumentParser( description="Менеджер резервных копий 1gb.ru", formatter_class=argparse.RawTextHelpFormatter, ) parser.add_argument( "--action", choices=["create", "list", "download", "delete", "create-download"], default="list", help="""Действие: list — показать список бэкапов (по умолчанию) create — создать бэкап download — скачать последний бэкап create-download — создать и скачать delete — удалить старые бэкапы""", ) parser.add_argument( "--type", choices=["all", "data", "db"], default="all", help="""Тип бэкапа (для create/create-download/download): db — только база данных MySQL (по умолчанию) data — только данные сайта FTP all — данные + база""", ) parser.add_argument( "--keep", type=int, default=3, help="Сколько бэкапов ОСТАВИТЬ при удалении (по умолчанию: 3)", ) parser.add_argument( "--year", type=int, default=2026, help="За какой год удалять бэкапы (по умолчанию: 2026)", ) parser.add_argument( "--save-dir", type=str, default="F:/test_folder/MyReports", help="Папка для скачивания (по умолчанию: F:/test_folder/MyReports)", ) parser.add_argument( "--no-wait", action="store_true", help="Не ждать завершения создания бэкапа" ) parser.add_argument("--username", type=str, default="minzdravmo", help="Логин") parser.add_argument( "--password", type=str, default="KB3ZS7*sv6j4zYr@bPdeFWQN38rbtQv", help="Пароль" ) parser.add_argument("--account", type=str, default="564622", help="ID аккаунта") return parser.parse_args() # === ИСПОЛЬЗОВАНИЕ === if __name__ == "__main__": args = parse_args() backup_data = args.type in ("all", "data") backup_mysql = args.type in ("all", "db") mgr = BackupManager( username=args.username, password=args.password, account_id=args.account ) mgr.enable_output_log("F:/test_folder/MyReports/backup.log") try: mgr.start() if args.action == "list": mgr.print_backups() elif args.action == "create": mgr.print_backups() print(f"\n{'=' * 70}") mgr.create_backup( backup_data=backup_data, backup_mysql=backup_mysql, wait_for_completion=not args.no_wait, ) mgr.print_backups() elif args.action == "download": mgr.download_latest( save_dir=args.save_dir, include_data=backup_data, include_sql_raw=backup_mysql, include_sql_zip=backup_mysql, ) elif args.action == "create-download": mgr.create_and_download( save_dir=args.save_dir, backup_data=backup_data, backup_mysql=backup_mysql, ) elif args.action == "delete": groups = mgr.get_grouped_backups() year_groups = [g for g in groups if g.timestamp.startswith(str(args.year))] total = len(year_groups) if total <= args.keep: print( f"[*] Бэкапов за {args.year}: {total}, оставляем: {args.keep}. Удалять нечего." ) else: delete_count = total - args.keep mgr.delete_old_backups(delete_count=delete_count, year=args.year) # 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() # # Показать список бэкапов # python main.py --action list # # Создать бэкап только БД # python main.py --action create --type db # # Создать бэкап всего (данные + БД) # python main.py --action create --type all # # Создать и сразу скачать БД # python main.py --action create-download --type db --save-dir ./backups # # Скачать последний бэкап # python main.py --action download --type db # # Удалить старые, оставить 3 за 2026 год # python main.py --action delete --keep 3 --year 2026 # # Удалить старые, оставить 1 # python main.py --action delete --keep 1 --year 2026 # # Полный цикл: создать БД + скачать + удалить старые (оставить 3) # python main.py --action create-download --type db && python main.py --action delete --keep 3