194 lines
7.6 KiB
Python
194 lines
7.6 KiB
Python
import time
|
|
import socket
|
|
import platform
|
|
import threading
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional, Callable, Dict, Any, List
|
|
import httpx
|
|
|
|
from agent.config import AgentConfig, load_config, save_config, LocalFolderJob
|
|
from agent.scanner import DirectoryScanner, is_file_stable
|
|
from agent.chunker import compute_file_sha256
|
|
from agent.uploader import ChunkUploader
|
|
from agent.state_db import state_db
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] [OnEver Agent] %(message)s"
|
|
)
|
|
logger = logging.getLogger("OnEverAgent")
|
|
|
|
class AgentDaemon:
|
|
"""Background service worker for Windows: handles heartbeats, job polling and scheduled backups."""
|
|
|
|
def __init__(
|
|
self,
|
|
config: Optional[AgentConfig] = None,
|
|
on_started: Optional[Callable[[str, int], None]] = None,
|
|
on_progress: Optional[Callable[[str, int, int, float], None]] = None,
|
|
on_completed: Optional[Callable[[str, str, int], None]] = None,
|
|
on_error: Optional[Callable[[str, str], None]] = None,
|
|
on_status: Optional[Callable[[str, str], None]] = None
|
|
):
|
|
self.config = config or load_config()
|
|
self.running = False
|
|
self.uploader = ChunkUploader(self.config)
|
|
self._heartbeat_thread: Optional[threading.Thread] = None
|
|
self._worker_thread: Optional[threading.Thread] = None
|
|
|
|
# Event callbacks
|
|
self.on_started = on_started
|
|
self.on_progress = on_progress
|
|
self.on_completed = on_completed
|
|
self.on_error = on_error
|
|
self.on_status = on_status
|
|
|
|
def start(self):
|
|
self.config = load_config()
|
|
if not self.config.device_id or not self.config.device_token:
|
|
logger.warning("Agent is not registered yet. Waiting for registration.")
|
|
if self.on_status:
|
|
self.on_status("UNREGISTERED", "El agente no está registrado en el servidor.")
|
|
return
|
|
|
|
self.running = True
|
|
logger.info(f"Starting OnEver Drive Windows Agent ({self.config.client_code} - {self.config.client_name})")
|
|
logger.info(f"Target Server: {self.config.server_url}")
|
|
|
|
if self.on_status:
|
|
self.on_status("ONLINE", f"Conectado a {self.config.server_url} ({self.config.client_code})")
|
|
|
|
self._heartbeat_thread = threading.Thread(target=self._heartbeat_loop, daemon=True)
|
|
self._worker_thread = threading.Thread(target=self._backup_worker_loop, daemon=True)
|
|
|
|
self._heartbeat_thread.start()
|
|
self._worker_thread.start()
|
|
|
|
def stop(self):
|
|
logger.info("Stopping agent daemon...")
|
|
self.running = False
|
|
if self.on_status:
|
|
self.on_status("PAUSED", "Servicio en pausa.")
|
|
|
|
def _get_headers(self):
|
|
return {
|
|
"X-Device-Id": self.config.device_id,
|
|
"X-Device-Token": self.config.device_token
|
|
}
|
|
|
|
def _heartbeat_loop(self):
|
|
while self.running:
|
|
try:
|
|
base_url = self.config.server_url.rstrip("/")
|
|
with httpx.Client(base_url=base_url, headers=self._get_headers(), timeout=10.0) as client:
|
|
resp = client.get("/api/jobs/agent/assigned")
|
|
if resp.status_code == 200:
|
|
logger.debug("Heartbeat acknowledged by server.")
|
|
except Exception as ex:
|
|
logger.warning(f"Heartbeat failed: {str(ex)}")
|
|
|
|
time.sleep(self.config.heartbeat_interval_seconds)
|
|
|
|
def _backup_worker_loop(self):
|
|
while self.running:
|
|
try:
|
|
self._run_backup_cycle()
|
|
except Exception as ex:
|
|
logger.error(f"Error during backup cycle: {str(ex)}")
|
|
|
|
time.sleep(30)
|
|
|
|
def _run_backup_cycle(self):
|
|
self.config = load_config()
|
|
self.uploader.config = self.config
|
|
|
|
# 1. Fetch server-assigned jobs
|
|
server_jobs = []
|
|
try:
|
|
base_url = self.config.server_url.rstrip("/")
|
|
with httpx.Client(base_url=base_url, headers=self._get_headers(), timeout=15.0) as client:
|
|
resp = client.get("/api/jobs/agent/assigned")
|
|
if resp.status_code == 200:
|
|
server_jobs = resp.json()
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. Combine server jobs + user local folders
|
|
all_jobs = []
|
|
for sj in server_jobs:
|
|
all_jobs.append({
|
|
"job_id": sj.get("id"),
|
|
"name": sj.get("name"),
|
|
"source_path": sj.get("source_path"),
|
|
"file_patterns": sj.get("file_patterns", "*.*"),
|
|
"min_stable_seconds": sj.get("min_stable_time_seconds", 60)
|
|
})
|
|
|
|
for lj in self.config.local_folders:
|
|
if lj.is_active:
|
|
all_jobs.append({
|
|
"job_id": None,
|
|
"local_job_id": lj.id,
|
|
"name": lj.name,
|
|
"source_path": lj.source_path,
|
|
"file_patterns": lj.file_patterns,
|
|
"min_stable_seconds": lj.min_stable_seconds
|
|
})
|
|
|
|
# 3. Process jobs
|
|
for job in all_jobs:
|
|
source_path = job["source_path"]
|
|
file_patterns = job["file_patterns"]
|
|
min_stable = job["min_stable_seconds"]
|
|
job_name = job["name"]
|
|
|
|
scanner = DirectoryScanner(source_path, file_patterns, min_stable_seconds=min_stable)
|
|
files = scanner.scan()
|
|
|
|
for filepath in files:
|
|
if not is_file_stable(filepath, min_stable_seconds=min_stable):
|
|
logger.warning(f"File {filepath.name} is currently locked or growing. Skipping.")
|
|
continue
|
|
|
|
current_sha = compute_file_sha256(filepath)
|
|
if state_db.is_file_already_backed_up(str(filepath), current_sha):
|
|
continue
|
|
|
|
file_size_bytes = filepath.stat().st_size
|
|
logger.info(f"Starting backup for file: {filepath.name} ({file_size_bytes / (1024*1024):.2f} MB)")
|
|
|
|
# Single notification when the process starts
|
|
if self.on_started:
|
|
self.on_started(filepath.name, file_size_bytes)
|
|
|
|
def on_chunk_progress(done, total, pct):
|
|
if self.on_progress:
|
|
self.on_progress(filepath.name, done, total, pct)
|
|
|
|
try:
|
|
res = self.uploader.upload_file(
|
|
filepath,
|
|
job_id=job.get("job_id"),
|
|
progress_callback=on_chunk_progress
|
|
)
|
|
logger.info(f"Successfully backed up {filepath.name}!")
|
|
|
|
# Update local folder last backup status
|
|
if "local_job_id" in job:
|
|
for folder in self.config.local_folders:
|
|
if folder.id == job["local_job_id"]:
|
|
folder.last_backup_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
|
|
folder.last_status = "Backup Exitoso"
|
|
save_config(self.config)
|
|
|
|
# Single notification when the process finishes
|
|
if self.on_completed:
|
|
self.on_completed(filepath.name, res.get("sha256", ""), file_size_bytes)
|
|
|
|
except Exception as ex:
|
|
logger.error(f"Failed to backup {filepath.name}: {str(ex)}")
|
|
if self.on_error:
|
|
self.on_error(filepath.name, str(ex))
|