conduit.Vfs

1 import os.path 2 import logging 3 import gobject 4 log = logging.getLogger("Vfs") 5 6 try: 7 import gnomevfs 8 except ImportError: 9 from gnome import gnomevfs 10 11 import conduit.utils.Singleton as Singleton 12 13 # 14 # URI Functions 15 #

16 -def uri_is_valid(uri):

17 """ 18 (weakly) checks if a uri is valid by looking for a scheme seperator 19 """ 20 assert type(uri) == str 21 return uri[0] != "/" and uri.find("://") != -1

22

23 -def uri_join(first, *rest):

24 """ 25 Joins multiple uri components. Performs safely if the first 26 argument contains a uri scheme 27 """ 28 assert type(first) == str 29 return os.path.join(first,*rest)

30 #idx = first.rfind("://") 31 #if idx == -1: 32 # start = 0 33 #else: 34 # start = idx + 3 35 #return first[0:start]+os.path.join(first[start:],*rest) 36

37 -def uri_get_relative(fromURI, toURI):

38 """ 39 Returns the relative path fromURI --> toURI 40 """ 41 assert type(fromURI) == str 42 assert type(toURI) == str 43 rel = toURI.replace(fromURI,"") 44 #strip leading / 45 if rel[0] == os.sep: 46 return rel[1:] 47 else: 48 return rel

49

50 -def uri_open(uri):

51 """ 52 Opens a gnomevfs or xdg compatible uri. 53 """ 54 assert type(uri) == str 55 APP = "xdg-open" 56 os.spawnlp(os.P_NOWAIT, APP, APP, uri)

57

58 -def uri_to_local_path(uri):

59 """ 60 @returns: The local path (/foo/bar) for the given URI. Throws a 61 RuntimeError (wtf??) if the uri is not a local one 62 """ 63 assert type(uri) == str 64 return gnomevfs.get_local_path_from_uri(uri)

65

66 -def uri_get_volume_root_uri(uri):

67 """ 68 @returns: The root path of the volume at the given uri, or None 69 """ 70 assert type(uri) == str 71 try: 72 path = uri_to_local_path(uri) 73 return VolumeMonitor().volume_get_root_uri(path) 74 except: 75 return None

76

77 -def uri_is_on_removable_volume(uri):

78 """ 79 @returns: True if the specified uri is on a removable volume, like a USB key 80 or removable/mountable disk. 81 """ 82 assert type(uri) == str 83 scheme = gnomevfs.get_uri_scheme(uri) 84 if scheme == "file": 85 #FIXME: Unfortunately this approach actually works better than gnomevfs 86 #return uri.startswith("file:///media/") 87 try: 88 path = uri_to_local_path(uri) 89 return VolumeMonitor().volume_is_removable(path) 90 except Exception, e: 91 log.warn("Could not determine if uri on removable volume: %s (%s)" % (uri, e)) 92 return False 93 return False

94 95

96 -def uri_get_filesystem_type(uri):

97 """ 98 @returns: The filesystem that uri is stored on or None if it cannot 99 be determined 100 """ 101 assert type(uri) == str 102 scheme = gnomevfs.get_uri_scheme(uri) 103 if scheme == "file": 104 try: 105 path = uri_to_local_path(uri) 106 return VolumeMonitor().volume_get_fstype(path) 107 except RuntimeError: 108 log.warn("Could not get local path from URI") 109 return None 110 except AttributeError: 111 log.warn("Could not determine volume for path") 112 return None 113 return None

114

115 -def uri_make_canonical(uri):

116 """ 117 Standardizes the format of the uri 118 @param uri:an absolute or relative stringified uri. It might have scheme. 119 """ 120 assert type(uri) == str 121 return gnomevfs.make_uri_canonical(uri)

122

123 -def uri_escape(uri):

124 """ 125 Escapes a uri, replacing only special characters that would not be found in 126 paths or host names. 127 (so '/', '&', '=', ':' and '@' will not be escaped by this function) 128 """ 129 assert type(uri) == str 130 #FIXME: This function lies, it escapes @ 131 #return gnomevfs.escape_host_and_path_string(uri) 132 import urllib 133 return urllib.quote(uri,safe='/&=:@')

134

135 -def uri_unescape(uri):

136 """ 137 Replace "%xx" escapes by their single-character equivalent. 138 """ 139 assert type(uri) == str 140 import urllib 141 return urllib.unquote(uri)

142

143 -def uri_get_protocol(uri):

144 """ 145 Returns the protocol (file, smb, etc) for a URI 146 """ 147 assert type(uri) == str 148 if uri.rfind("://")==-1: 149 return "" 150 protocol = uri[:uri.index("://")+3] 151 return protocol.lower()

152

153 -def uri_get_filename(uri):

154 """ 155 Method to return the filename of a file. Could use GnomeVFS for this 156 is it wasnt so slow 157 """ 158 assert type(uri) == str 159 return uri.split(os.sep)[-1]

160

161 -def uri_get_filename_and_extension(uri):

162 """ 163 Returns filename,file_extension 164 """ 165 assert type(uri) == str 166 return os.path.splitext(uri_get_filename(uri))

167

168 -def uri_sanitize_for_filesystem(uri, filesystem=None):

169 """ 170 Removes illegal characters in uri that cannot be stored on the 171 given filesystem - particuarly fat and ntfs types 172 """ 173 assert type(uri) == str 174 import string 175 176 ILLEGAL_CHARS = { 177 "vfat" : "\\:*?\"<>|", 178 "ntfs" : "\\:*?\"<>|" 179 } 180 181 illegal = ILLEGAL_CHARS.get(filesystem,None) 182 if illegal: 183 #dont escape the scheme part 184 idx = uri.rfind("://") 185 if idx == -1: 186 start = 0 187 else: 188 start = idx + 3 189 190 #replace illegal chars with a space, ignoring the scheme 191 return uri[0:start]+uri[start:].translate(string.maketrans( 192 illegal, 193 " "*len(illegal) 194 ) 195 ) 196 return uri

197

198 -def uri_is_folder(uri):

199 """ 200 @returns: True if the uri is a folder and not a file 201 """ 202 assert type(uri) == str 203 info = gnomevfs.get_file_info(uri) 204 return info.type == gnomevfs.FILE_TYPE_DIRECTORY

205

206 -def uri_format_for_display(uri):

207 """ 208 Formats the uri so it can be displayed to the user (strips passwords, etc) 209 """ 210 assert type(uri) == str 211 return gnomevfs.format_uri_for_display(uri)

212

213 -def uri_exists(uri):

214 """ 215 @returns: True if the uri exists 216 """ 217 assert type(uri) == str 218 try: 219 return gnomevfs.exists(gnomevfs.URI(uri)) == 1 220 except Exception, err: 221 log.warn("Error checking if location exists") 222 return False

223

224 -def uri_make_directory(uri):

225 """ 226 Makes a directory with the default permissions. Does not catch any 227 error 228 """ 229 assert type(uri) == str 230 gnomevfs.make_directory( 231 uri, 232 gnomevfs.PERM_USER_ALL | gnomevfs.PERM_GROUP_READ | gnomevfs.PERM_GROUP_EXEC | gnomevfs.PERM_OTHER_READ | gnomevfs.PERM_OTHER_EXEC 233 )

234

235 -def uri_make_directory_and_parents(uri):

236 """ 237 Because gnomevfs.make_dir does not perform as mkdir -p this function 238 is required to make a heirarchy of directories. 239 240 @param uri: A directory that does not exist 241 @type uri: str 242 """ 243 assert type(uri) == str 244 exists = False 245 dirs = [] 246 247 directory = gnomevfs.URI(uri) 248 while not exists: 249 dirs.append(directory) 250 directory = directory.parent 251 exists = gnomevfs.exists(directory) 252 253 dirs.reverse() 254 for d in dirs: 255 log.debug("Making directory %s" % d) 256 uri_make_directory(str(d))

257

258 -class FileMonitor(gobject.GObject):

259 260 __gsignals__ = { 261 "changed" : (gobject.SIGNAL_RUN_LAST, gobject.TYPE_NONE, [ 262 gobject.TYPE_PYOBJECT, 263 gobject.TYPE_PYOBJECT, 264 gobject.TYPE_PYOBJECT]) 265 } 266 267 MONITOR_EVENT_CREATED = gnomevfs.MONITOR_EVENT_CREATED 268 MONITOR_EVENT_CHANGED = gnomevfs.MONITOR_EVENT_CHANGED 269 MONITOR_EVENT_DELETED = gnomevfs.MONITOR_EVENT_DELETED 270 MONITOR_EVENT_METADATA_CHANGED = gnomevfs.MONITOR_EVENT_METADATA_CHANGED 271 MONITOR_EVENT_STARTEXECUTING = gnomevfs.MONITOR_EVENT_STARTEXECUTING 272 MONITOR_EVENT_STOPEXECUTING = gnomevfs.MONITOR_EVENT_STOPEXECUTING 273 MONITOR_FILE = gnomevfs.MONITOR_FILE 274 MONITOR_DIRECTORY = gnomevfs.MONITOR_DIRECTORY 275

276 - def __init__(self):

277 gobject.GObject.__init__(self) 278 self._monitor_folder_id = None

279

280 - def _monitor_cb(self, monitor_uri, event_uri, event):

281 self.emit("changed", monitor_uri,

Source Code for Module conduit.Vfs