1010
1111import dataclasses
1212import logging
13-
13+ import re
1414import pyudev
1515
1616from arm .models import SystemDrives
@@ -26,10 +26,18 @@ def _apply_masked_repr(cls):
2626 original_repr = cls .__repr__ if hasattr (cls , '__repr__' ) else object .__repr__
2727
2828 def masked_repr (self ):
29- """Mask the serial with asterisk in the repr"""
29+ """Mask the serial with asterisks in the repr"""
3030 repr_str = original_repr (self )
31- mask_last = self .serial [:- 6 ] + '*' * 6
32- return repr_str .replace (self .serial , mask_last )
31+ serial = getattr (self , "serial" , None )
32+
33+ if serial and len (serial ) > 6 :
34+ masked = serial [:- 6 ] + "*" * 6
35+ elif serial : # shorter than 6 chars
36+ masked = "*" * len (serial )
37+ else : # None or empty
38+ masked = "UNKNOWN"
39+
40+ return repr_str .replace (str (serial ), masked , 1 )
3341
3442 cls .__repr__ = masked_repr
3543
@@ -114,9 +122,14 @@ class DriveInformationExtended(DriveInformation):
114122
115123 @staticmethod
116124 def _convert_bool (value ):
117- if isinstance (value , (str , int , float , bool )):
125+ # Allow some of the data to be None
126+ if value in (None , "" , "unknown" ):
127+ return False
128+ try :
129+ # Test if we have filled values
118130 return bool (int (value ))
119- return False
131+ except (ValueError , TypeError ):
132+ return False
120133
121134 def __post_init__ (self ):
122135 super ().__post_init__ ()
@@ -160,17 +173,53 @@ def __post_init__(self):
160173
161174
162175def drives_search ():
163- """Search the system for optical drives.
176+ """
177+ Search the system for optical drives and yield DriveInformationMedium objects.
178+
179+ In a container environment, falls back to treating /dev/sr* as optical drives.
164180 """
165181 context = pyudev .Context ()
166182 for device in context .list_devices (subsystem = "block" ):
167- if device .properties .get ("ID_TYPE" ) == "cd" :
168- fields = (
169- DRIVE_INFORMATION +
170- DRIVE_INFORMATION_EXTENDED +
171- DRIVE_INFORMATION_MEDIUM
183+ try :
184+ devnode = device .device_node
185+ if not devnode :
186+ continue
187+
188+ # Ignore loop&nvme devices
189+ if devnode and (devnode .startswith ("/dev/loop" ) or devnode .startswith ("/dev/nvme" )):
190+ # Logging here might not be helpful, unsure yet
191+ # logging.debug("Ignoring loop/nvme device: %s", devnode)
192+ continue
193+
194+ # Log all properties - Helps with debugging if a drive has loaded all properties, or just the base
195+ logging .debug ("Device: %s" , devnode )
196+ for key , value in device .properties .items ():
197+ app .logger .debug (" %s = %s" , key , value )
198+
199+ # Optical drive detection - Try to use ID_TYPE then ID_CDROM but fall back to all drives matching /dev/sr*
200+ # NOTE: this may be better to check if MAJOR = 11
201+ # + devname `/dev/sr*` and possibly DEVTYPE as this always means its an optical drive on linux
202+ # But just the first two MAJOR + devname should be more than enough to verify its a CD/DVD drive
203+ is_optical = (
204+ device .properties .get ("ID_TYPE" ) == "cd" or
205+ device .properties .get ("ID_CDROM" ) == "1" or
206+ re .match (r"^/dev/sr\d+$" , devnode )
172207 )
173- yield DriveInformationMedium (* map (device .properties .get , fields ))
208+
209+ if is_optical :
210+ logging .info ("Optical drive detected: %s" , devnode )
211+
212+ # Try to populate fields, but allow missing values incase the drive hasn't been mounted/activated yet
213+ fields = (
214+ DRIVE_INFORMATION +
215+ DRIVE_INFORMATION_EXTENDED +
216+ DRIVE_INFORMATION_MEDIUM
217+ )
218+ values = [device .properties .get (field ) or "" for field in fields ]
219+ yield DriveInformationMedium (* values )
220+
221+ except Exception as e :
222+ app .logger .error ("Error processing device %s: %s" , device , e , exc_info = True )
174223
175224
176225def drives_update (startup = False ):
@@ -192,8 +241,11 @@ def drives_update(startup=False):
192241 db .session .commit ()
193242
194243 # Update drive information:
195- for drive in sorted (drives_search ()): # sorted by mount point
196- app .logger .debug (drive )
244+ system_drives = sorted (drives_search ())
245+ if len (system_drives ) < 1 :
246+ logging .error (f"We Cant find any system drives!. { system_drives } " )
247+ for drive in system_drives : # sorted by mount point
248+ logging .debug (f"Drive info: { drive } " )
197249 # Retrieve the drive matching `drive.serial_id` from the database or
198250 # create a new entry if it doesn't exist. Since `drive.serial_id` *may*
199251 # not be unique, we update only the first drive that misses the mdisc
0 commit comments