Supporta utente target per percorsi watchdog

This commit is contained in:
2026-07-02 09:22:12 +02:00
parent b0e29b4cc5
commit e5a76992d4
9 changed files with 130 additions and 11 deletions

View File

@@ -21,8 +21,22 @@ Per creare o trovare il file di configurazione:
```powershell ```powershell
python -m bakrest.config_cli init python -m bakrest.config_cli init
python -m bakrest.config_cli path python -m bakrest.config_cli path
python -m bakrest.config_cli show
``` ```
Se il servizio gira come `LocalSystem`, imposta l'utente a cui devono riferirsi i percorsi nel tab `Monitoraggio`, campo `Utente percorsi`, oppure nel TOML:
```toml
[watch]
target_user = "pettirosso"
include_dirs = [
"%USERPROFILE%\\Documents",
"%USERPROFILE%\\Desktop",
]
```
Con `target_user = "pettirosso"`, `%USERPROFILE%` viene espanso come `C:\Users\pettirosso` anche se il servizio gira come `SYSTEM`.
In sviluppo si puo' usare un percorso diverso: In sviluppo si puo' usare un percorso diverso:
```powershell ```powershell

View File

@@ -198,12 +198,15 @@ La strategia consigliata e':
- permettere all'utente di aggiungere cartelle di lavoro dalla GUI; - permettere all'utente di aggiungere cartelle di lavoro dalla GUI;
- prevedere una modalita' di audit per capire quali cartelle producono file rilevanti. - prevedere una modalita' di audit per capire quali cartelle producono file rilevanti.
Se il servizio gira come `LocalSystem`, la configurazione deve supportare un utente target per l'espansione dei percorsi utente. Esempio: con `target_user = "pettirosso"`, `%USERPROFILE%\\Documents` deve essere espanso in `C:\\Users\\pettirosso\\Documents`, non nel profilo di sistema.
### Directory candidate da includere ### Directory candidate da includere
Esempi: Esempi:
```yaml ```yaml
watch: watch:
target_user: "pettirosso"
include_dirs: include_dirs:
- "%USERPROFILE%\\Desktop" - "%USERPROFILE%\\Desktop"
- "%USERPROFILE%\\Documents" - "%USERPROFILE%\\Documents"

View File

@@ -5,6 +5,7 @@ autostart = true
autostart = true autostart = true
[watch] [watch]
target_user = ""
include_dirs = [ include_dirs = [
"%USERPROFILE%\\Desktop", "%USERPROFILE%\\Desktop",
"%USERPROFILE%\\Documents", "%USERPROFILE%\\Documents",

View File

@@ -35,6 +35,7 @@ class BackupConfig:
@dataclass(frozen=True) @dataclass(frozen=True)
class WatchConfig: class WatchConfig:
target_user: str | None
include_dirs: tuple[Path, ...] include_dirs: tuple[Path, ...]
include_extensions: frozenset[str] include_extensions: frozenset[str]
exclude_dirs: tuple[Path | str, ...] exclude_dirs: tuple[Path | str, ...]
@@ -158,11 +159,12 @@ def parse_config(data: dict[str, Any], path: Path) -> AppConfig:
storage = data.get("storage", {}) storage = data.get("storage", {})
stability = data.get("stability", {}) stability = data.get("stability", {})
include_dirs = tuple(expand_path(item) for item in watch.get("include_dirs", [])) target_user = _optional_string(watch.get("target_user"))
include_dirs = tuple(expand_path(item, target_user) for item in watch.get("include_dirs", []))
include_extensions = frozenset( include_extensions = frozenset(
_normalize_extension(item) for item in watch.get("include_extensions", []) _normalize_extension(item) for item in watch.get("include_extensions", [])
) )
exclude_dirs = tuple(_normalize_exclude_dir(item) for item in watch.get("exclude_dirs", [])) exclude_dirs = tuple(_normalize_exclude_dir(item, target_user) for item in watch.get("exclude_dirs", []))
storage_config = StorageConfig( storage_config = StorageConfig(
change_registry=expand_path( change_registry=expand_path(
@@ -180,6 +182,7 @@ def parse_config(data: dict[str, Any], path: Path) -> AppConfig:
service_autostart=bool(service.get("autostart", True)), service_autostart=bool(service.get("autostart", True)),
tray_autostart=bool(tray.get("autostart", True)), tray_autostart=bool(tray.get("autostart", True)),
watch=WatchConfig( watch=WatchConfig(
target_user=target_user,
include_dirs=include_dirs, include_dirs=include_dirs,
include_extensions=include_extensions, include_extensions=include_extensions,
exclude_dirs=exclude_dirs, exclude_dirs=exclude_dirs,
@@ -219,13 +222,20 @@ def _normalize_extension(value: str) -> str:
return value if value.startswith(".") else f".{value}" return value if value.startswith(".") else f".{value}"
def _normalize_exclude_dir(value: str) -> Path | str: def _normalize_exclude_dir(value: str, target_user: str | None = None) -> Path | str:
expanded = expand_path(value) expanded = expand_path(value, target_user)
if "%" in value or not expanded.is_absolute(): if "%" in str(expanded) or not expanded.is_absolute():
return value.lower() return value.lower()
return expanded return expanded
def _optional_string(value: object) -> str | None:
if value is None:
return None
text = str(value).strip()
return text or None
def _remote_destinations(backup: dict[str, Any]) -> tuple[str, ...]: def _remote_destinations(backup: dict[str, Any]) -> tuple[str, ...]:
destinations = backup.get("remote_destinations") destinations = backup.get("remote_destinations")
if isinstance(destinations, list): if isinstance(destinations, list):

View File

@@ -56,16 +56,22 @@ class BakRestConfigApp(ctk.CTk):
def _build_monitor_tab(self) -> None: def _build_monitor_tab(self) -> None:
self.monitor_tab.grid_columnconfigure(0, weight=1) self.monitor_tab.grid_columnconfigure(0, weight=1)
self.monitor_tab.grid_columnconfigure(1, weight=1) self.monitor_tab.grid_columnconfigure(1, weight=1)
self.monitor_tab.grid_rowconfigure(0, weight=1) self.monitor_tab.grid_rowconfigure(1, weight=1)
target_frame = ctk.CTkFrame(self.monitor_tab)
target_frame.grid(row=0, column=0, columnspan=2, sticky="ew", padx=8, pady=(8, 0))
target_frame.grid_columnconfigure(1, weight=1)
self.target_user_var = tk.StringVar()
_label_entry(target_frame, "Utente percorsi", self.target_user_var, 0)
left = ctk.CTkFrame(self.monitor_tab) left = ctk.CTkFrame(self.monitor_tab)
left.grid(row=0, column=0, sticky="nsew", padx=(0, 8), pady=8) left.grid(row=1, column=0, sticky="nsew", padx=(0, 8), pady=8)
left.grid_columnconfigure(0, weight=1) left.grid_columnconfigure(0, weight=1)
left.grid_rowconfigure(1, weight=1) left.grid_rowconfigure(1, weight=1)
left.grid_rowconfigure(3, weight=1) left.grid_rowconfigure(3, weight=1)
right = ctk.CTkFrame(self.monitor_tab) right = ctk.CTkFrame(self.monitor_tab)
right.grid(row=0, column=1, sticky="nsew", padx=(8, 0), pady=8) right.grid(row=1, column=1, sticky="nsew", padx=(8, 0), pady=8)
right.grid_columnconfigure(0, weight=1) right.grid_columnconfigure(0, weight=1)
right.grid_rowconfigure(1, weight=1) right.grid_rowconfigure(1, weight=1)
right.grid_rowconfigure(3, weight=1) right.grid_rowconfigure(3, weight=1)
@@ -83,7 +89,7 @@ class BakRestConfigApp(ctk.CTk):
self.exclude_patterns.grid(row=2, column=0, rowspan=2, sticky="nsew", padx=12, pady=12) self.exclude_patterns.grid(row=2, column=0, rowspan=2, sticky="nsew", padx=12, pady=12)
actions = ctk.CTkFrame(self.monitor_tab, fg_color="transparent") actions = ctk.CTkFrame(self.monitor_tab, fg_color="transparent")
actions.grid(row=1, column=0, columnspan=2, sticky="ew", pady=(0, 8)) actions.grid(row=2, column=0, columnspan=2, sticky="ew", pady=(0, 8))
actions.grid_columnconfigure(0, weight=1) actions.grid_columnconfigure(0, weight=1)
save_button = ctk.CTkButton(actions, text="Salva configurazione", command=self._save_config) save_button = ctk.CTkButton(actions, text="Salva configurazione", command=self._save_config)
@@ -169,6 +175,7 @@ class BakRestConfigApp(ctk.CTk):
backup = self.data.setdefault("backup", {}) backup = self.data.setdefault("backup", {})
server_check = backup.setdefault("server_check", {}) server_check = backup.setdefault("server_check", {})
self.target_user_var.set(str(watch.get("target_user", "")))
self.include_dirs.set_items(watch.get("include_dirs", [])) self.include_dirs.set_items(watch.get("include_dirs", []))
self.extensions.set_items(watch.get("include_extensions", [])) self.extensions.set_items(watch.get("include_extensions", []))
self.exclude_dirs.set_items(watch.get("exclude_dirs", [])) self.exclude_dirs.set_items(watch.get("exclude_dirs", []))
@@ -187,6 +194,7 @@ class BakRestConfigApp(ctk.CTk):
server_check = backup.setdefault("server_check", {}) server_check = backup.setdefault("server_check", {})
watch["include_dirs"] = self.include_dirs.items() watch["include_dirs"] = self.include_dirs.items()
watch["target_user"] = self.target_user_var.get().strip()
watch["include_extensions"] = self.extensions.items() watch["include_extensions"] = self.extensions.items()
watch["exclude_dirs"] = self.exclude_dirs.items() watch["exclude_dirs"] = self.exclude_dirs.items()
watch["exclude_patterns"] = self.exclude_patterns.items() watch["exclude_patterns"] = self.exclude_patterns.items()

View File

@@ -5,6 +5,7 @@ autostart = true
autostart = true autostart = true
[watch] [watch]
target_user = ""
include_dirs = [ include_dirs = [
"%USERPROFILE%\\\\Desktop", "%USERPROFILE%\\\\Desktop",
"%USERPROFILE%\\\\Documents", "%USERPROFILE%\\\\Documents",

View File

@@ -22,5 +22,45 @@ def default_config_path() -> Path:
return program_data_dir() / "config.toml" return program_data_dir() / "config.toml"
def expand_path(value: str) -> Path: def expand_path(value: str, target_user: str | None = None) -> Path:
return Path(os.path.expandvars(value)).expanduser() expanded = expand_user_vars(value, target_user)
return Path(os.path.expandvars(expanded)).expanduser()
def expand_user_vars(value: str, target_user: str | None = None) -> str:
if not target_user:
return value
user = target_user.strip().lstrip(".\\")
if not user:
return value
profile = Path("C:/Users") / user
replacements = {
"%USERPROFILE%": str(profile),
"%APPDATA%": str(profile / "AppData" / "Roaming"),
"%LOCALAPPDATA%": str(profile / "AppData" / "Local"),
"%TEMP%": str(profile / "AppData" / "Local" / "Temp"),
"%TMP%": str(profile / "AppData" / "Local" / "Temp"),
}
expanded = value
for token, replacement in replacements.items():
expanded = _replace_case_insensitive(expanded, token, replacement)
return expanded
def _replace_case_insensitive(value: str, old: str, new: str) -> str:
lower_value = value.lower()
lower_old = old.lower()
start = 0
parts: list[str] = []
while True:
index = lower_value.find(lower_old, start)
if index == -1:
parts.append(value[start:])
break
parts.append(value[start:index])
parts.append(new)
start = index + len(old)
return "".join(parts)

View File

@@ -22,6 +22,25 @@ def test_parse_config_normalizes_extensions(tmp_path: Path) -> None:
assert config.watch.include_extensions == frozenset({".jpg", ".png"}) assert config.watch.include_extensions == frozenset({".jpg", ".png"})
def test_parse_config_expands_watch_paths_for_target_user(tmp_path: Path) -> None:
config = parse_config(
{
"watch": {
"target_user": "pettirosso",
"include_dirs": ["%USERPROFILE%\\Documents"],
"include_extensions": [".docx"],
"exclude_dirs": ["%APPDATA%"],
"exclude_patterns": [],
}
},
tmp_path / "config.toml",
)
assert config.watch.target_user == "pettirosso"
assert config.watch.include_dirs == (Path("C:/Users/pettirosso/Documents"),)
assert config.watch.exclude_dirs == (Path("C:/Users/pettirosso/AppData/Roaming"),)
def test_parse_config_reads_multiple_remote_destinations(tmp_path: Path) -> None: def test_parse_config_reads_multiple_remote_destinations(tmp_path: Path) -> None:
config = parse_config( config = parse_config(
{ {

23
tests/test_paths.py Normal file
View File

@@ -0,0 +1,23 @@
from __future__ import annotations
from pathlib import Path
from bakrest.paths import expand_path
def test_expand_path_uses_target_user_for_userprofile() -> None:
assert expand_path("%USERPROFILE%\\Documents", "pettirosso") == Path(
"C:/Users/pettirosso/Documents"
)
def test_expand_path_uses_target_user_for_appdata() -> None:
assert expand_path("%APPDATA%\\Microsoft", ".\\pettirosso") == Path(
"C:/Users/pettirosso/AppData/Roaming/Microsoft"
)
def test_expand_path_uses_target_user_for_temp() -> None:
assert expand_path("%TEMP%", "pettirosso") == Path(
"C:/Users/pettirosso/AppData/Local/Temp"
)