oemspot/onever_drive
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

se agregaron nuevas caracteristicas

Browse Source
This commit is contained in:
Carlos Tello
2026-08-13 23:07:01 -03:00
parent 1bfb808c79
commit d736db982e
23 changed files with 1244 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
windows-agent/agent/config.py
+2
View File
@@ -17,9 +17,11 @@ CONFIG_FILE = AGENT_HOME / "config.json"
class LocalFolderJob(BaseModel):
id: str = Field(default_factory=lambda: str(uuid.uuid4())[:8])
job_id: Optional[int] = None
name: str
source_path: str
file_patterns: str = "*.bak,*.mdf"
schedule_cron: str = "daily"
schedule_interval_minutes: int = 60
min_stable_seconds: int = 60
is_active: bool = True
windows-agent/agent/service.py
+139 -42
View File
@@ -20,6 +20,36 @@ logging.basicConfig(
)
logger = logging.getLogger("OnEverAgent")
def is_job_due(schedule_str: str, last_run_str: Optional[str]) -> bool:
if not schedule_str:
return True
if not last_run_str:
return True
try:
# Try parsing ISO (from server) or standard YYYY-MM-DD HH:MM:SS (local)
if "T" in last_run_str:
last_run = datetime.fromisoformat(last_run_str.replace("Z", "+00:00"))
else:
last_run = datetime.strptime(last_run_str, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
except Exception:
return True
now = datetime.now(timezone.utc)
delta = now - last_run
sched = schedule_str.lower().strip()
if sched == "hourly":
return delta.total_seconds() >= 3600
elif sched == "daily":
return delta.total_seconds() >= 86400
elif sched == "weekly":
return delta.total_seconds() >= 86400 * 7
elif sched == "monthly":
return delta.total_seconds() >= 86400 * 30
else:
return True
class AgentDaemon:
"""Background service worker for Windows: handles heartbeats, job polling and scheduled backups."""
@@ -35,6 +65,7 @@ class AgentDaemon:
self.config = config or load_config()
self.running = False
self.uploader = ChunkUploader(self.config)
self.last_server_jobs = []
self._heartbeat_thread: Optional[threading.Thread] = None
self._worker_thread: Optional[threading.Thread] = None
@@ -100,53 +131,122 @@ class AgentDaemon:
time.sleep(30)
def _run_backup_cycle(self):
def _run_backup_cycle(self, force: bool = False):
self.config = load_config()
self.uploader.config = self.config
# 1. Fetch server-assigned jobs
# 1. Fetch server-assigned jobs and run bidirectional sync
server_jobs = []
sync_success = False
try:
base_url = self.config.server_url.rstrip("/")
with httpx.Client(base_url=base_url, headers=self._get_headers(), timeout=15.0) as client:
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:
server_jobs = resp.json()
except Exception:
pass
self.last_server_jobs = server_jobs
sync_success = True
except Exception as e:
logger.warning(f"Could not fetch server jobs: {e}")
# 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)
})
config_changed = False
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
})
if sync_success:
# A. Sync Server -> Local
server_job_ids = {sj["id"] for sj in server_jobs}
# Remove local jobs that have a job_id but are not on the server anymore (deleted on server)
local_jobs_to_keep = []
for lj in self.config.local_folders:
if lj.job_id is None:
# New local job, keep it so we register it next
local_jobs_to_keep.append(lj)
elif lj.job_id in server_job_ids:
# Keep it and update local properties from server
sj = next(x for x in server_jobs if x["id"] == lj.job_id)
lj.name = sj.get("name", lj.name)
lj.source_path = sj.get("source_path", lj.source_path)
lj.file_patterns = sj.get("file_patterns", lj.file_patterns)
lj.schedule_cron = sj.get("schedule_cron", lj.schedule_cron)
lj.min_stable_seconds = sj.get("min_stable_time_seconds", lj.min_stable_seconds)
# Also sync last_run_at from server if available and newer
if sj.get("last_run_at"):
lj.last_backup_at = sj["last_run_at"].replace("T", " ")[:19]
lj.last_status = sj.get("status", lj.last_status)
local_jobs_to_keep.append(lj)
else:
# Deleted on server, don't keep it
config_changed = True
self.config.local_folders = local_jobs_to_keep
# 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"]
# Add server jobs that are missing locally
local_job_ids = {lj.job_id for lj in self.config.local_folders if lj.job_id is not None}
for sj in server_jobs:
if sj["id"] not in local_job_ids:
new_job = LocalFolderJob(
job_id=sj["id"],
name=sj["name"],
source_path=sj["source_path"],
file_patterns=sj["file_patterns"],
schedule_cron=sj["schedule_cron"],
min_stable_seconds=sj["min_stable_time_seconds"],
last_status=sj.get("status", "En espera")
)
if sj.get("last_run_at"):
new_job.last_backup_at = sj["last_run_at"].replace("T", " ")[:19]
self.config.local_folders.append(new_job)
config_changed = True
# B. Sync Local -> Server (Register new local folders on the server)
for lj in self.config.local_folders:
if lj.job_id is None:
try:
base_url = self.config.server_url.rstrip("/")
payload = {
"name": lj.name,
"source_path": lj.source_path,
"file_patterns": lj.file_patterns,
"schedule_cron": lj.schedule_cron,
"min_stable_time_seconds": lj.min_stable_seconds
}
resp = httpx.post(f"{base_url}/api/jobs/agent/register", headers=self._get_headers(), json=payload, timeout=10.0)
if resp.status_code == 200:
data = resp.json()
lj.job_id = data["id"]
config_changed = True
logger.info(f"Registered local job '{lj.name}' on server with ID {lj.job_id}")
except Exception as e:
logger.warning(f"Could not register local job '{lj.name}' on server: {e}")
if config_changed:
save_config(self.config)
# 2. Process active jobs
for job in self.config.local_folders:
if not job.is_active:
continue
# Check if job is due or forced
if not force and not is_job_due(job.schedule_cron, job.last_backup_at):
continue
source_path = job.source_path
file_patterns = job.file_patterns
min_stable = job.min_stable_seconds
job_name = job.name
# Skip if path does not exist
if not Path(source_path).exists():
logger.warning(f"Source path {source_path} for job {job_name} does not exist. Skipping.")
continue
scanner = DirectoryScanner(source_path, file_patterns, min_stable_seconds=min_stable)
files = scanner.scan()
# Track files successfully backed up in this run
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.")
@@ -159,7 +259,6 @@ class AgentDaemon:
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)
@@ -170,24 +269,22 @@ class AgentDaemon:
try:
res = self.uploader.upload_file(
filepath,
job_id=job.get("job_id"),
job_id=job.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)}")
job.last_status = "Error"
save_config(self.config)
if self.on_error:
self.on_error(filepath.name, str(ex))
# Update job state in config after checking directory
job.last_backup_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
job.last_status = "Backup Exitoso" if job.last_status != "Error" else "Error"
save_config(self.config)