724 lines
26 KiB
Python
724 lines
26 KiB
Python
import os
|
||
import re
|
||
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"
|
||
|
||
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 = 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()
|
||
|
||
# Запоминаем количество бэкапов до создания
|
||
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()
|
||
|
||
# Считаем только ГОТОВЫЕ файлы (размер вида "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 _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 _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("\n[OK] Готово")
|
||
|
||
|
||
# === ИСПОЛЬЗОВАНИЕ ===
|
||
|
||
if __name__ == "__main__":
|
||
mgr = BackupManager(
|
||
username="rav",
|
||
password="KB3e8rbtQv",
|
||
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()
|