@ -33,6 +33,16 @@ def _stored_name(original_name: str) -> str:
ts = datetime . now ( timezone . utc ) . strftime ( " % Y % m %d T % H % M % S %f Z " )
return f " { ts } __ { safe } "
def _recovered_filename ( original_name : str ) - > tuple [ str , str , str ] :
base_name = original_name . rsplit ( " / " , 1 ) [ - 1 ] . rsplit ( " \\ " , 1 ) [ - 1 ]
if " . " in base_name :
basename , extension = base_name . rsplit ( " . " , 1 )
recovered_name = f " { basename } -recovered. { extension } "
return recovered_name , f " { basename } -recovered " , extension
recovered_name = f " { base_name } -recovered "
return recovered_name , f " { base_name } -recovered " , " "
def _safe_path_from_relative ( relative_path : str ) - > Path :
return _tenant_root ( ) / relative_path
@ -695,6 +705,102 @@ def deleted_files():
files = files ,
)
@bp . route ( " /deleted/<int:file_id>/recover " , methods = [ " POST " ] )
@portal_session_required
def recover_deleted_file ( file_id : int ) :
db = get_db ( )
with db . cursor ( ) as cur :
cur . execute (
"""
SELECT
f . id ,
f . device_id ,
f . original_filename ,
f . relative_path ,
f . is_deleted ,
d . device_name ,
d . device_type ,
d . relative_path AS device_relative_path
FROM files f
LEFT JOIN devices d ON f . device_id = d . id
WHERE f . id = % s
AND f . tenant_id = % s
""" ,
( file_id , session [ " otb_tenant_id " ] ) ,
)
file_row = cur . fetchone ( )
if not file_row or not file_row [ " is_deleted " ] :
flash ( " Deleted file not found. " , " warning " )
return redirect ( url_for ( " main.deleted_files " ) )
if not file_row [ " device_relative_path " ] :
flash ( " Cannot recover this file because its device record is missing. " , " warning " )
return redirect ( url_for ( " main.deleted_files " ) )
source_path = _safe_path_from_relative ( file_row [ " relative_path " ] )
if not source_path . exists ( ) :
flash ( " Deleted file is missing from storage. " , " warning " )
return redirect ( url_for ( " main.deleted_files " ) )
recovered_name , recovered_basename , recovered_extension = _recovered_filename ( file_row [ " original_filename " ] )
stored_name = _stored_name ( recovered_name )
target_rel = f " { file_row [ ' device_relative_path ' ] } /originals/ { stored_name } "
target_dir = f " { file_row [ ' device_relative_path ' ] } /originals "
target_path = _safe_path_from_relative ( target_rel )
target_path . parent . mkdir ( parents = True , exist_ok = True )
shutil . move ( str ( source_path ) , str ( target_path ) )
cur . execute (
"""
UPDATE files
SET original_filename = % s ,
basename = % s ,
extension = % s ,
relative_path = % s ,
directory_path = % s ,
is_deleted = 0 ,
deleted_at = NULL
WHERE id = % s AND tenant_id = % s
""" ,
(
recovered_name ,
recovered_basename ,
recovered_extension ,
target_rel ,
target_dir ,
file_id ,
session [ " otb_tenant_id " ] ,
) ,
)
cur . execute (
"""
INSERT INTO audit_logs (
tenant_id , user_id , actor_type , event_type , file_id , ip_address , user_agent , event_detail
) VALUES ( % s , % s , ' user ' , ' file_recovered ' , % s , % s , % s , % s )
""" ,
(
session [ " otb_tenant_id " ] ,
session [ " otb_user_id " ] ,
file_id ,
_client_ip ( ) ,
request . headers . get ( " User-Agent " , " " ) ,
f " Recovered deleted file as ' { recovered_name } ' back into originals " ,
) ,
)
db . commit ( )
flash ( f " Recovered file as ' { recovered_name } ' . You can find it back in the device file browser. " , " success " )
return redirect ( url_for ( " main.deleted_files " ) )
@bp . route ( " /deleted/<int:file_id>/hard-delete " , methods = [ " POST " ] )
@portal_session_required
def hard_delete_file ( file_id : int ) :
