Первая рабочая версия скрипта по созданию бекапов портала hivmo через
взаимодействие с порталом с использованием библиотеки playwright
This commit is contained in:
VolandSZ
35
.gitignore
vendored
Normal file
35
.gitignore
vendored
Normal file
@ -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
|
||||
|
||||
|
||||
703
main.py
Normal file
703
main.py
Normal file
@ -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()
|
||||
Reference in New Issue
Block a user