152 lines
4.5 KiB
Python
152 lines
4.5 KiB
Python
"""PiCopy – Interner Speicher/Samba: internal_share_state, alle Samba-Funktionen."""
|
||
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import threading
|
||
from pathlib import Path
|
||
|
||
from picopy.config import INTERNAL_DEST_DIR, load_cfg, save_cfg, log
|
||
|
||
SAMBA_CONF = Path('/etc/samba/smb.conf')
|
||
SAMBA_BEGIN = '# BEGIN PICOPY INTERNAL SHARE'
|
||
SAMBA_END = '# END PICOPY INTERNAL SHARE'
|
||
|
||
internal_share_state = {
|
||
'installed': False,
|
||
'enabled': False,
|
||
'active': False,
|
||
'path': str(INTERNAL_DEST_DIR),
|
||
'share': 'PiCopy',
|
||
'pkg_running': False,
|
||
'pkg_error': None,
|
||
'error': None,
|
||
}
|
||
internal_share_lock = threading.Lock()
|
||
|
||
|
||
def _internal_usage():
|
||
INTERNAL_DEST_DIR.mkdir(parents=True, exist_ok=True)
|
||
usage = shutil.disk_usage(INTERNAL_DEST_DIR)
|
||
return {
|
||
'path': str(INTERNAL_DEST_DIR),
|
||
'total': usage.total,
|
||
'used': usage.used,
|
||
'free': usage.free,
|
||
}
|
||
|
||
|
||
def smbd_installed():
|
||
return shutil.which('smbd') is not None
|
||
|
||
|
||
def _systemctl(*args, timeout=20):
|
||
try:
|
||
return subprocess.run(['systemctl'] + list(args), capture_output=True,
|
||
text=True, timeout=timeout)
|
||
except Exception as e:
|
||
return subprocess.CompletedProcess(['systemctl'] + list(args), 1,
|
||
stdout='', stderr=str(e))
|
||
|
||
|
||
def _smbd_active():
|
||
if not smbd_installed():
|
||
return False
|
||
r = _systemctl('is-active', 'smbd', timeout=5)
|
||
return r.returncode == 0 and r.stdout.strip() == 'active'
|
||
|
||
|
||
def internal_share_update_state():
|
||
cfg = load_cfg()
|
||
usage = _internal_usage()
|
||
with internal_share_lock:
|
||
internal_share_state.update(
|
||
installed=smbd_installed(),
|
||
enabled=bool(cfg.get('internal_share_enabled')),
|
||
active=_smbd_active(),
|
||
path=usage['path'],
|
||
total=usage['total'],
|
||
used=usage['used'],
|
||
free=usage['free'],
|
||
)
|
||
return dict(internal_share_state)
|
||
|
||
|
||
def _write_samba_share(enabled: bool):
|
||
old = SAMBA_CONF.read_text(encoding='utf-8') if SAMBA_CONF.exists() else ''
|
||
pattern = re.compile(rf'\n?{re.escape(SAMBA_BEGIN)}.*?{re.escape(SAMBA_END)}\n?', re.S)
|
||
cleaned = pattern.sub('\n', old).rstrip() + '\n'
|
||
if enabled:
|
||
INTERNAL_DEST_DIR.mkdir(parents=True, exist_ok=True)
|
||
INTERNAL_DEST_DIR.chmod(0o755)
|
||
block = f"""
|
||
{SAMBA_BEGIN}
|
||
[PiCopy]
|
||
path = {INTERNAL_DEST_DIR}
|
||
browseable = yes
|
||
read only = yes
|
||
guest ok = yes
|
||
force user = root
|
||
{SAMBA_END}
|
||
"""
|
||
cleaned += block
|
||
tmp = SAMBA_CONF.with_suffix('.conf.picopy_tmp')
|
||
tmp.write_text(cleaned, encoding='utf-8')
|
||
os.replace(str(tmp), str(SAMBA_CONF))
|
||
|
||
|
||
def _install_samba_if_needed():
|
||
if smbd_installed():
|
||
return True, ''
|
||
with internal_share_lock:
|
||
internal_share_state.update(pkg_running=True, pkg_error=None)
|
||
try:
|
||
r = subprocess.run(['apt-get', 'install', '-y', 'samba'],
|
||
capture_output=True, text=True, timeout=300,
|
||
env={**os.environ, 'DEBIAN_FRONTEND': 'noninteractive'})
|
||
if r.returncode != 0:
|
||
err = (r.stderr.strip().splitlines()[-1]
|
||
if r.stderr.strip() else 'samba-Installation fehlgeschlagen')
|
||
with internal_share_lock:
|
||
internal_share_state['pkg_error'] = err
|
||
return False, err
|
||
return True, ''
|
||
except Exception as e:
|
||
with internal_share_lock:
|
||
internal_share_state['pkg_error'] = str(e)
|
||
return False, str(e)
|
||
finally:
|
||
with internal_share_lock:
|
||
internal_share_state['pkg_running'] = False
|
||
|
||
|
||
def set_internal_share_enabled(enabled: bool):
|
||
ok, err = (True, '')
|
||
if enabled:
|
||
ok, err = _install_samba_if_needed()
|
||
if not ok:
|
||
return False, err
|
||
elif not smbd_installed():
|
||
cfg = load_cfg()
|
||
cfg['internal_share_enabled'] = False
|
||
save_cfg(cfg)
|
||
internal_share_update_state()
|
||
return True, ''
|
||
try:
|
||
_write_samba_share(enabled)
|
||
if enabled:
|
||
_systemctl('enable', '--now', 'smbd', timeout=60)
|
||
_systemctl('restart', 'smbd', timeout=60)
|
||
else:
|
||
_systemctl('restart', 'smbd', timeout=60)
|
||
cfg = load_cfg()
|
||
cfg['internal_share_enabled'] = enabled
|
||
save_cfg(cfg)
|
||
internal_share_update_state()
|
||
return True, ''
|
||
except Exception as e:
|
||
with internal_share_lock:
|
||
internal_share_state['error'] = str(e)
|
||
return False, str(e)
|