dinlo
f624fee5e0
- SSH/SFTP управление без агентов - Пакеты (apt), Docker, файлы, systemd, ufw, cron - Плейбуки, групповые операции, планировщик - Бэкапы, отчёты, Portainer, Proxmox VE, Gitea Deploy - Терминал, Telegram-уведомления, тёмная/светлая тема
127 lines
4.2 KiB
Python
127 lines
4.2 KiB
Python
"""Cron: чтение/запись crontab, версии, описание расписаний."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import List, Optional
|
|
|
|
from croniter import croniter
|
|
|
|
from app.core.ssh import SSHClient
|
|
from app.db.database import get_session
|
|
from app.db.models import CronBackup
|
|
|
|
|
|
@dataclass
|
|
class CronLine:
|
|
raw: str
|
|
enabled: bool
|
|
schedule: str
|
|
command: str
|
|
comment: str = ""
|
|
|
|
@property
|
|
def is_entry(self) -> bool:
|
|
return bool(self.schedule)
|
|
|
|
|
|
def parse_crontab(content: str) -> List[CronLine]:
|
|
lines: List[CronLine] = []
|
|
for raw in content.splitlines():
|
|
stripped = raw.strip()
|
|
enabled = True
|
|
candidate = stripped
|
|
if not candidate:
|
|
lines.append(CronLine(raw, True, "", ""))
|
|
continue
|
|
if candidate.startswith("#"):
|
|
enabled = False
|
|
candidate = candidate.lstrip("#").strip()
|
|
parts = candidate.split(None, 5)
|
|
if candidate.startswith("@") and len(candidate.split(None, 1)) == 2:
|
|
sched, cmd = candidate.split(None, 1)
|
|
lines.append(CronLine(raw, enabled, sched, cmd))
|
|
elif len(parts) == 6 and _looks_like_schedule(parts[:5]):
|
|
lines.append(CronLine(raw, enabled, " ".join(parts[:5]), parts[5]))
|
|
else:
|
|
lines.append(CronLine(raw, enabled, "", "", comment=stripped))
|
|
return lines
|
|
|
|
|
|
def _looks_like_schedule(fields: List[str]) -> bool:
|
|
for f in fields:
|
|
for ch in f:
|
|
if ch not in "0123456789*,-/":
|
|
return False
|
|
return True
|
|
|
|
|
|
def humanize(schedule: str) -> str:
|
|
try:
|
|
if schedule.startswith("@"):
|
|
return {"@reboot": "при загрузке", "@daily": "ежедневно",
|
|
"@hourly": "ежечасно", "@weekly": "еженедельно",
|
|
"@monthly": "ежемесячно", "@yearly": "ежегодно"}.get(schedule, schedule)
|
|
it = croniter(schedule)
|
|
nxt = it.get_next(__import__("datetime").datetime)
|
|
return f"следующий запуск: {nxt:%Y-%m-%d %H:%M}"
|
|
except Exception:
|
|
return "неверное выражение"
|
|
|
|
|
|
def render(lines: List[CronLine]) -> str:
|
|
out = []
|
|
for l in lines:
|
|
if not l.is_entry:
|
|
out.append(l.raw)
|
|
continue
|
|
entry = f"{l.schedule} {l.command}"
|
|
out.append(entry if l.enabled else f"# {entry}")
|
|
return "\n".join(out) + "\n"
|
|
|
|
|
|
async def get_crontab(c: SSHClient, user: Optional[str] = None) -> str:
|
|
cmd = f"crontab -u {user} -l" if user else "crontab -l"
|
|
r = await c.sudo(cmd, timeout=30)
|
|
if not r.ok and "no crontab" in (r.stderr + r.stdout):
|
|
return ""
|
|
return r.stdout
|
|
|
|
|
|
async def set_crontab(c: SSHClient, content: str, server_id: int,
|
|
user: Optional[str] = None) -> None:
|
|
# бэкап текущего
|
|
current = await get_crontab(c, user)
|
|
with get_session() as s:
|
|
s.add(CronBackup(server_id=server_id, user=user or "current", content=current))
|
|
s.commit()
|
|
import base64
|
|
b64 = base64.b64encode(content.encode()).decode()
|
|
target = f"crontab -u {user} -" if user else "crontab -"
|
|
r = await c.sudo(f"echo {b64} | base64 -d | {target}", timeout=30)
|
|
if not r.ok:
|
|
raise RuntimeError(r.stderr)
|
|
|
|
|
|
async def list_cron_files(c: SSHClient) -> dict:
|
|
"""Системные cron-файлы: /etc/crontab + /etc/cron.d/*"""
|
|
out = {}
|
|
r = await c.sudo("cat /etc/crontab 2>/dev/null", timeout=30)
|
|
if r.ok:
|
|
out["/etc/crontab"] = r.stdout
|
|
r = await c.sudo("ls /etc/cron.d/ 2>/dev/null", timeout=30)
|
|
for name in r.stdout.split():
|
|
rr = await c.sudo(f"cat '/etc/cron.d/{name}'", timeout=30)
|
|
if rr.ok:
|
|
out[f"/etc/cron.d/{name}"] = rr.stdout
|
|
return out
|
|
|
|
|
|
def backups(server_id: int, user: str, limit: int = 10) -> List[CronBackup]:
|
|
from sqlmodel import select
|
|
with get_session() as s:
|
|
return list(s.exec(
|
|
select(CronBackup)
|
|
.where(CronBackup.server_id == server_id, CronBackup.user == user)
|
|
.order_by(CronBackup.id.desc()).limit(limit)
|
|
).all())
|