117 lines
4.2 KiB
Python
117 lines
4.2 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from app.models.models import SystemSetting
|
|
from app.core.config import settings
|
|
from pathlib import Path
|
|
import os
|
|
|
|
DEFAULT_SETTINGS = {
|
|
"storage_root": settings.STORAGE_ROOT,
|
|
"global_quota_gb": "1000",
|
|
"default_client_quota_gb": "100",
|
|
"default_keep_daily": str(settings.DEFAULT_RETENTION_DAILY),
|
|
"default_keep_weekly": str(settings.DEFAULT_RETENTION_WEEKLY),
|
|
"default_keep_monthly": str(settings.DEFAULT_RETENTION_MONTHLY),
|
|
"auth_mode": "local",
|
|
"radius_host": "",
|
|
"radius_port": "1812",
|
|
"radius_secret": "",
|
|
"storage_network_enabled": "false",
|
|
"storage_network_user": "",
|
|
"storage_network_pass": "",
|
|
}
|
|
|
|
def connect_network_share(path: str, username: str, password: str) -> bool:
|
|
"""
|
|
Mounts a UNC network share on Windows using 'net use'.
|
|
"""
|
|
if not (path.startswith(r"\\") or path.startswith("//")):
|
|
return True
|
|
|
|
parts = [p for p in path.replace("/", "\\").split("\\") if p]
|
|
if len(parts) < 2:
|
|
return False
|
|
unc_base = f"\\\\{parts[0]}\\{parts[1]}"
|
|
|
|
import subprocess
|
|
print(f"Connecting to network share {unc_base} with user '{username}'...")
|
|
|
|
# 1. Remove existing connection if any (ignore errors)
|
|
subprocess.run(["net", "use", unc_base, "/delete", "/y"], capture_output=True)
|
|
|
|
# 2. Add connection
|
|
cmd = ["net", "use", unc_base]
|
|
if password:
|
|
cmd.append(password)
|
|
if username:
|
|
cmd.append(f"/user:{username}")
|
|
cmd.append("/persistent:yes")
|
|
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
return result.returncode == 0
|
|
|
|
async def get_setting(db: AsyncSession, key: str) -> str:
|
|
result = await db.execute(select(SystemSetting).where(SystemSetting.key == key))
|
|
setting = result.scalar_one_or_none()
|
|
if setting:
|
|
return setting.value
|
|
return DEFAULT_SETTINGS.get(key, "")
|
|
|
|
async def get_all_settings(db: AsyncSession) -> dict:
|
|
result = await db.execute(select(SystemSetting))
|
|
db_settings = {s.key: s.value for s in result.scalars().all()}
|
|
|
|
# Merge defaults and add them if not present in db
|
|
updated = False
|
|
for k, v in DEFAULT_SETTINGS.items():
|
|
if k not in db_settings:
|
|
db_settings[k] = v
|
|
db.add(SystemSetting(key=k, value=v))
|
|
updated = True
|
|
if updated:
|
|
await db.commit()
|
|
return db_settings
|
|
|
|
async def update_settings_service(db: AsyncSession, new_settings: dict) -> dict:
|
|
from app.storage.local import storage_provider
|
|
|
|
current = await get_all_settings(db)
|
|
merged = {**current, **new_settings}
|
|
|
|
# Connect network share if enabled and it's a UNC path
|
|
net_enabled = str(merged.get("storage_network_enabled", "false")).lower() == "true"
|
|
net_user = merged.get("storage_network_user", "")
|
|
net_pass = merged.get("storage_network_pass", "")
|
|
root_path = merged.get("storage_root", "")
|
|
|
|
if net_enabled and root_path and (root_path.startswith(r"\\") or root_path.startswith("//")):
|
|
ok = connect_network_share(root_path, net_user, net_pass)
|
|
if not ok:
|
|
raise ValueError("Failed to authenticate or connect to the specified network share (UNC). Verify path and credentials.")
|
|
|
|
# Perform standard storage root validation
|
|
if "storage_root" in new_settings:
|
|
v = new_settings["storage_root"]
|
|
path = Path(v).resolve()
|
|
try:
|
|
os.makedirs(path, exist_ok=True)
|
|
test_file = path / ".write_test"
|
|
test_file.touch()
|
|
test_file.unlink()
|
|
except Exception as e:
|
|
raise ValueError(f"Invalid or unwritable storage root path: {str(e)}")
|
|
|
|
storage_provider.root_dir = path
|
|
|
|
for k, v in new_settings.items():
|
|
if k in DEFAULT_SETTINGS and v is not None:
|
|
result = await db.execute(select(SystemSetting).where(SystemSetting.key == k))
|
|
setting = result.scalar_one_or_none()
|
|
if setting:
|
|
setting.value = str(v)
|
|
else:
|
|
db.add(SystemSetting(key=k, value=str(v)))
|
|
|
|
await db.commit()
|
|
return await get_all_settings(db)
|