154 lines
4.9 KiB
Python
154 lines
4.9 KiB
Python
"""PiCopy – Blueprint: /api/copy/*, /api/devices, /api/storage-info, /api/status, /api/config*."""
|
||
|
||
import shutil
|
||
import subprocess
|
||
import threading
|
||
|
||
from flask import Blueprint, jsonify, request
|
||
|
||
from picopy.config import load_cfg, save_cfg, _fmt_bytes
|
||
from picopy.state import copy_state, copy_lock
|
||
from picopy.usb import usb_devices, ensure_mount, internal_dest_device
|
||
from picopy.wifi import wifi_state, wifi_lock
|
||
from picopy.wireguard import wg_state, wg_lock
|
||
from picopy.samba import internal_share_update_state
|
||
import picopy.copy_engine as _ce
|
||
|
||
copy_bp = Blueprint('copy', __name__)
|
||
|
||
|
||
def _resolve_source_ports(cfg) -> list:
|
||
ports = cfg.get('source_ports') or []
|
||
if not ports and cfg.get('source_port'):
|
||
ports = [{'port': cfg['source_port'], 'label': cfg.get('source_label', '')}]
|
||
return ports
|
||
|
||
|
||
def _configured_destination(cfg, devs):
|
||
if cfg.get('dest_type') == 'internal':
|
||
return internal_dest_device(cfg)
|
||
return next((d for d in devs if d['usb_port'] == cfg.get('dest_port')), None)
|
||
|
||
|
||
@copy_bp.route('/api/devices')
|
||
def r_devices():
|
||
return jsonify(usb_devices())
|
||
|
||
|
||
@copy_bp.route('/api/storage-info')
|
||
def r_storage_info():
|
||
cfg = load_cfg()
|
||
devs = usb_devices()
|
||
result = []
|
||
|
||
src_ports = {sp['port'] for sp in _resolve_source_ports(cfg)}
|
||
dst_port = cfg.get('dest_port')
|
||
|
||
def _du_for_dev(dev):
|
||
mp, owned = ensure_mount(dev)
|
||
if not mp:
|
||
return dict(total=None, used=None, free=None, pct=None)
|
||
try:
|
||
du = shutil.disk_usage(mp)
|
||
return dict(total=du.total, used=du.used, free=du.free,
|
||
pct=round(du.used / du.total * 100) if du.total else 0)
|
||
except Exception:
|
||
return dict(total=None, used=None, free=None, pct=None)
|
||
finally:
|
||
if owned:
|
||
subprocess.run(['umount', mp], capture_output=True)
|
||
|
||
for dev in devs:
|
||
port = dev['usb_port']
|
||
if port in src_ports:
|
||
role = 'source'
|
||
elif port == dst_port:
|
||
role = 'dest'
|
||
else:
|
||
role = 'other'
|
||
entry = dict(
|
||
role=role,
|
||
label=dev.get('label') or dev.get('device') or f'Port {port}',
|
||
port=port,
|
||
device=dev.get('device', ''),
|
||
size_str=dev.get('size', ''),
|
||
)
|
||
entry.update(_du_for_dev(dev))
|
||
result.append(entry)
|
||
|
||
if cfg.get('dest_type') == 'internal':
|
||
entry = dict(role='dest',
|
||
label=cfg.get('internal_dest_label') or 'Interner Speicher',
|
||
port='__internal__', device='internal', size_str='')
|
||
entry.update(_du_for_dev({'internal': True}))
|
||
result.append(entry)
|
||
|
||
return jsonify(result)
|
||
|
||
|
||
@copy_bp.route('/api/config', methods=['GET', 'POST'])
|
||
def r_config():
|
||
if request.method == 'POST':
|
||
cfg = load_cfg()
|
||
cfg.update(request.get_json(force=True))
|
||
save_cfg(cfg)
|
||
return jsonify(ok=True)
|
||
return jsonify(load_cfg())
|
||
|
||
|
||
@copy_bp.route('/api/config/ports/reset', methods=['POST'])
|
||
def r_ports_reset():
|
||
cfg = load_cfg()
|
||
cfg['source_ports'] = []
|
||
cfg['source_port'] = None
|
||
cfg['source_label'] = ''
|
||
cfg['dest_port'] = None
|
||
cfg['dest_label'] = ''
|
||
cfg['dest_type'] = 'usb'
|
||
save_cfg(cfg)
|
||
return jsonify(ok=True)
|
||
|
||
|
||
@copy_bp.route('/api/status')
|
||
def r_status():
|
||
with copy_lock:
|
||
cs = dict(copy_state)
|
||
with wifi_lock:
|
||
ws = dict(wifi_state)
|
||
with wg_lock:
|
||
wgs = dict(wg_state)
|
||
share = internal_share_update_state()
|
||
return jsonify(copy=cs, wifi=ws, vpn=wgs, internal_share=share)
|
||
|
||
|
||
@copy_bp.route('/api/copy/start', methods=['POST'])
|
||
def r_start():
|
||
with copy_lock:
|
||
if copy_state['running']:
|
||
return jsonify(error='Bereits aktiv'), 400
|
||
if _ce._copy_thread is not None and _ce._copy_thread.is_alive():
|
||
return jsonify(error='Abbruch wird noch abgeschlossen - bitte kurz warten und erneut versuchen.'), 400
|
||
cfg = load_cfg()
|
||
devs = usb_devices()
|
||
body = request.get_json(force=True) or {}
|
||
wanted_ports = body.get('ports') # None = alle konfigurierten Quellen
|
||
src_ports = _resolve_source_ports(cfg)
|
||
srcs = [next((d for d in devs if d['usb_port'] == sp['port']), None) for sp in src_ports]
|
||
srcs = [s for s in srcs if s is not None]
|
||
if wanted_ports is not None:
|
||
srcs = [s for s in srcs if s['usb_port'] in wanted_ports]
|
||
if not srcs: return jsonify(error='Keine Quellgeräte gefunden (Ports nicht verbunden)'), 400
|
||
dst = _configured_destination(cfg, devs)
|
||
if not dst: return jsonify(error='Zielgerät nicht gefunden'), 400
|
||
t = threading.Thread(target=_ce.do_copy, args=(srcs, dst, cfg), daemon=True)
|
||
_ce._copy_thread = t
|
||
t.start()
|
||
return jsonify(ok=True)
|
||
|
||
|
||
@copy_bp.route('/api/copy/cancel', methods=['POST'])
|
||
def r_cancel():
|
||
with copy_lock:
|
||
copy_state['running'] = False
|
||
return jsonify(ok=True)
|