entropy.client.interfaces.fetch

1 # -*- coding: utf-8 -*- 2 """ 3 4 @author: Fabio Erculiani <lxnay@sabayonlinux.org> 5 @contact: lxnay@sabayonlinux.org 6 @copyright: Fabio Erculiani 7 @license: GPL-2 8 9 B{Entropy Package Manager Client Packages retrieval Interface}. 10 11 """ 12 import os 13 from entropy.i18n import _ 14 from entropy.const import * 15 from entropy.exceptions import * 16 from entropy.output import purple, bold, red, blue, darkgreen, darkred, brown, darkblue 17

18 -class FetchersMixin:

19

20 - def check_needed_package_download(self, filepath, checksum = None):

21 # is the file available 22 if os.path.isfile(etpConst['entropyworkdir']+"/"+filepath): 23 if checksum is None: 24 return 0 25 else: 26 # check digest 27 md5res = self.entropyTools.compare_md5(etpConst['entropyworkdir']+"/"+filepath, checksum) 28 if (md5res): 29 return 0 30 else: 31 return -2 32 else: 33 return -1

34

35 - def fetch_files(self, url_data_list, checksum = True, resume = True, fetch_file_abort_function = None):

36 """ 37 Fetch multiple files simultaneously on URLs. 38 39 @param url_data_list list 40 [(url,dest_path [or None],checksum ['ab86fff46f6ec0f4b1e0a2a4a82bf323' or None],branch,),..] 41 @param digest bool, digest check (checksum) 42 @param resume bool enable resume support 43 @param fetch_file_abort_function callable method that could raise exceptions 44 @return general_status_code, {'url': (status_code,checksum,resumed,)}, data_transfer 45 """ 46 pkgs_bindir = etpConst['packagesbindir'] 47 url_path_list = [] 48 checksum_map = {} 49 count = 0 50 for url, dest_path, cksum, branch in url_data_list: 51 count += 1 52 filename = os.path.basename(url) 53 if dest_path == None: 54 dest_path = os.path.join(pkgs_bindir, branch, filename,) 55 56 dest_dir = os.path.dirname(dest_path) 57 if not os.path.isdir(dest_dir): 58 os.makedirs(dest_dir, 0o755) 59 60 url_path_list.append((url, dest_path,)) 61 if cksum != None: checksum_map[count] = cksum 62 63 # load class 64 fetchConn = self.MultipleUrlFetcher(url_path_list, resume = resume, 65 abort_check_func = fetch_file_abort_function, OutputInterface = self, 66 UrlFetcherClass = self.urlFetcher, checksum = checksum) 67 try: 68 data = fetchConn.download() 69 except KeyboardInterrupt: 70 return -100, {}, 0 71 72 diff_map = {} 73 if checksum_map and checksum: # verify checksums 74 diff_map = dict((url_path_list[x-1][0], checksum_map.get(x)) for x in checksum_map \ 75 if checksum_map.get(x) != data.get(x)) 76 77 data_transfer = fetchConn.get_data_transfer() 78 if diff_map: 79 defval = -1 80 for key, val in list(diff_map.items()): 81 if val == "-1": # general error 82 diff_map[key] = -1 83 elif val == "-2": 84 diff_map[key] = -2 85 elif val == "-4": # timeout 86 diff_map[key] = -4 87 elif val == "-3": # not found 88 diff_map[key] = -3 89 elif val == -100: 90 defval = -100 91 return defval, diff_map, data_transfer 92 93 return 0, diff_map, data_transfer

94

95 - def fetch_files_on_mirrors(self, download_list, checksum = False, fetch_abort_function = None):

96 """ 97 @param download_map list [(repository,branch,filename,checksum (digest),signatures,),..] 98 @param checksum bool verify checksum? 99 @param fetch_abort_function callable method that could raise exceptions 100 """ 101 repo_uris = dict(((x[0], self.SystemSettings['repositories']['available'][x[0]]['packages'][::-1],) for x in download_list)) 102 remaining = repo_uris.copy() 103 my_download_list = download_list[:] 104 105 def get_best_mirror(repository): 106 try: 107 return remaining[repository][0] 108 except IndexError: 109 return None

110 111 def update_download_list(down_list, failed_down): 112 newlist = [] 113 for repo, branch, fname, cksum, signatures in down_list: 114 myuri = get_best_mirror(repo) 115 myuri = os.path.join(myuri, fname) 116 if myuri not in failed_down: 117 continue 118 newlist.append((repo, branch, fname, cksum, signatures,)) 119 return newlist

120 121 # return True: for failing, return False: for fine 122 def mirror_fail_check(repository, best_mirror): 123 # check if uri is sane 124 if not self.MirrorStatus.get_failing_mirror_status(best_mirror) >= 30: 125 return False 126 # set to 30 for convenience 127 self.MirrorStatus.set_failing_mirror_status(best_mirror, 30) 128 mirrorcount = repo_uris[repo].index(best_mirror)+1 129 mytxt = "( mirror #%s ) " % (mirrorcount,) 130 mytxt += blue(" %s: ") % (_("Mirror"),) 131 mytxt += red(self.entropyTools.spliturl(best_mirror)[1]) 132 mytxt += " - %s." % (_("maximum failure threshold reached"),) 133 self.updateProgress( 134 mytxt, 135 importance = 1, 136 type = "warning", 137 header = red(" ## ") 138 ) 139 140 if self.MirrorStatus.get_failing_mirror_status(best_mirror) == 30: 141 self.MirrorStatus.add_failing_mirror(best_mirror, 45) 142 elif self.MirrorStatus.get_failing_mirror_status(best_mirror) > 31: 143 self.MirrorStatus.add_failing_mirror(best_mirror, -4) 144 else: 145 self.MirrorStatus.set_failing_mirror_status(best_mirror, 0) 146 147 remaining[repository].discard(best_mirror) 148 return True 149 150 def show_download_summary(down_list): 151 # fetch_files_list.append((myuri,None,cksum,branch,)) 152 for repo, branch, fname, cksum, signatures in down_list: 153 best_mirror = get_best_mirror(repo) 154 mirrorcount = repo_uris[repo].index(best_mirror)+1 155 mytxt = "( mirror #%s ) " % (mirrorcount,) 156 basef = os.path.basename(fname) 157 mytxt += "[%s] %s " % (brown(basef), blue("@"),) 158 mytxt += red(self.entropyTools.spliturl(best_mirror)[1]) 159 # now fetch the new one 160 self.updateProgress( 161 mytxt, 162 importance = 1, 163 type = "info", 164 header = red(" ## ") 165 ) 166 167 def show_successful_download(down_list, data_transfer): 168 for repo, branch, fname, cksum, signatures in down_list: 169 best_mirror = get_best_mirror(repo) 170 mirrorcount = repo_uris[repo].index(best_mirror)+1 171 mytxt = "( mirror #%s ) " % (mirrorcount,) 172 basef = os.path.basename(fname) 173 mytxt += "[%s] %s %s " % (brown(basef), darkred(_("success")), blue("@"),) 174 mytxt += red(self.entropyTools.spliturl(best_mirror)[1]) 175 self.updateProgress( 176 mytxt, 177 importance = 1, 178 type = "info", 179 header = red(" ## ") 180 ) 181 mytxt = " %s: %s%s%s" % ( 182 blue(_("Aggregated transfer rate")), 183 bold(self.entropyTools.bytes_into_human(data_transfer)), 184 darkred("/"), 185 darkblue(_("second")), 186 ) 187 self.updateProgress( 188 mytxt, 189 importance = 1, 190 type = "info", 191 header = red(" ## ") 192 ) 193 194 def show_download_error(down_list, rc): 195 for repo, branch, fname, cksum, signatures in down_list: 196 best_mirror = get_best_mirror(repo) 197 mirrorcount = repo_uris[repo].index(best_mirror)+1 198 mytxt = "( mirror #%s ) " % (mirrorcount,) 199 mytxt += blue("%s: %s") % ( 200 _("Error downloading from"), 201 red(self.entropyTools.spliturl(best_mirror)[1]), 202 ) 203 if rc == -1: 204 mytxt += " - %s." % (_("data not available on this mirror"),) 205 elif rc == -2: 206 self.MirrorStatus.add_failing_mirror(best_mirror, 1) 207 mytxt += " - %s." % (_("wrong checksum"),) 208 elif rc == -3: 209 mytxt += " - %s." % (_("not found"),) 210 elif rc == -4: # timeout! 211 mytxt += " - %s." % (_("timeout error"),) 212 elif rc == -100: 213 mytxt += " - %s." % (_("discarded download"),) 214 else: 215 self.MirrorStatus.add_failing_mirror(best_mirror, 5) 216 mytxt += " - %s." % (_("unknown reason"),) 217 self.updateProgress( 218 mytxt, 219 importance = 1, 220 type = "warning", 221 header = red(" ## ") 222 ) 223 224 def remove_failing_mirrors(repos): 225 for repo in repos: 226 best_mirror = get_best_mirror(repo) 227 if remaining[repo]: 228 remaining[repo].pop(0) 229 230 def check_remaining_mirror_failure(repos): 231 return [x for x in repos if not remaining.get(x)] 232 233 while True: 234 235 do_resume = True 236 timeout_try_count = 50 237 while True: 238 239 fetch_files_list = [] 240 for repo, branch, fname, cksum, signatures in my_download_list: 241 best_mirror = get_best_mirror(repo) 242 # set working mirror, dont care if its None 243 self.MirrorStatus.set_working_mirror(best_mirror) 244 if best_mirror != None: 245 mirror_fail_check(repo, best_mirror) 246 best_mirror = get_best_mirror(repo) 247 if best_mirror == None: 248 # at least one package failed to download 249 # properly, give up with everything 250 return 3, my_download_list 251 myuri = os.path.join(best_mirror, fname) 252 fetch_files_list.append((myuri, None, cksum, branch,)) 253 254 try: 255 256 show_download_summary(my_download_list) 257 rc, failed_downloads, data_transfer = self.fetch_files( 258 fetch_files_list, checksum = checksum, 259 fetch_file_abort_function = fetch_abort_function, 260 resume = do_resume 261 ) 262 if rc == 0: 263 show_successful_download(my_download_list, data_transfer) 264 return 0, [] 265 266 # update my_download_list 267 my_download_list = update_download_list(my_download_list, failed_downloads) 268 if rc not in (-3, -4, -100,) and failed_downloads and do_resume: 269 # disable resume 270 do_resume = False 271 continue 272 else: 273 show_download_error(my_download_list, rc) 274 if rc == -4: # timeout 275 timeout_try_count -= 1 276 if timeout_try_count > 0: 277 continue 278 elif rc == -100: # user discarded fetch 279 return 1, [] 280 myrepos = set([x[0] for x in my_download_list]) 281 remove_failing_mirrors(myrepos) 282 # make sure we don't have nasty issues 283 remaining_failure = check_remaining_mirror_failure(myrepos) 284 if remaining_failure: 285 return 3, my_download_list 286 break 287 except KeyboardInterrupt: 288 return 1, [] 289 return 0, [] 290 291

292 - def fetch_file(self, url, branch, digest = None, resume = True, 293 fetch_file_abort_function = None, filepath = None):

294 295 def do_stfu_rm(xpath): 296 try: 297 os.remove(xpath) 298 except OSError: 299 pass

300 301 filename = os.path.basename(url) 302 if not filepath: 303 filepath = os.path.join(etpConst['packagesbindir'], branch, filename) 304 filepath_dir = os.path.dirname(filepath) 305 # symlink support 306 if not os.path.isdir(os.path.realpath(filepath_dir)): 307 try: 308 os.remove(filepath_dir) 309 except OSError: 310 pass 311 os.makedirs(filepath_dir, 0o755) 312 313 existed_before = False 314 if os.path.isfile(filepath) and os.path.exists(filepath): 315 existed_before = True 316 317 # load class 318 fetchConn = self.urlFetcher(url, filepath, resume = resume, 319 abort_check_func = fetch_file_abort_function, OutputInterface = self) 320 fetchConn.progress = self.progress 321 322 # start to download 323 data_transfer = 0 324 resumed = False 325 try: 326 fetchChecksum = fetchConn.download() 327 data_transfer = fetchConn.get_transfer_rate() 328 resumed = fetchConn.is_resumed() 329 except KeyboardInterrupt: 330 return -100, data_transfer, resumed 331 except NameError: 332 raise 333 except: 334 if etpUi['debug']: 335 self.updateProgress( 336 "fetch_file:", 337 importance = 1, 338 type = "warning", 339 header = red(" ## ") 340 ) 341 self.entropyTools.print_traceback() 342 if (not existed_before) or (not resume): 343 do_stfu_rm(filepath) 344 return -1, data_transfer, resumed 345 if fetchChecksum == "-3": 346 # not found 347 return -3, data_transfer, resumed 348 elif fetchChecksum == "-4": 349 # timeout 350 return -4, data_transfer, resumed 351 352 del fetchConn 353 354 if digest and (fetchChecksum != digest): 355 # not properly downloaded 356 if (not existed_before) or (not resume): 357 do_stfu_rm(filepath) 358 return -2, data_transfer, resumed 359 360 return 0, data_transfer, resumed 361 362

363 - def fetch_file_on_mirrors(self, repository, branch, filename, 364 digest = False, fetch_abort_function = None):

365 366 uris = self.SystemSettings['repositories']['available'][repository]['packages'][::-1] 367 remaining = set(uris) 368 369 mirrorcount = 0 370 for uri in uris: 371 372 if not remaining: 373 # tried all the mirrors, quitting for error 374 self.MirrorStatus.set_working_mirror(None) 375 return 3 376 377 self.MirrorStatus.set_working_mirror(uri) 378 mirrorcount += 1 379 mirrorCountText = "( mirror #%s ) " % (mirrorcount,) 380 url = uri + "/" + filename 381 382 # check if uri is sane 383 if self.MirrorStatus.get_failing_mirror_status(uri) >= 30: 384 # ohohoh! 385 # set to 30 for convenience 386 self.MirrorStatus.set_failing_mirror_status(uri, 30) 387 mytxt = mirrorCountText 388 mytxt += blue(" %s: ") % (_("Mirror"),) 389 mytxt += red(self.entropyTools.spliturl(uri)[1]) 390 mytxt += " - %s." % (_("maximum failure threshold reached"),) 391 self.updateProgress( 392 mytxt, 393 importance = 1, 394 type = "warning", 395 header = red(" ## ") 396 ) 397 398 if self.MirrorStatus.get_failing_mirror_status(uri) == 30: 399 # put to 75 then decrement by 4 so we 400 # won't reach 30 anytime soon ahahaha 401 self.MirrorStatus.add_failing_mirror(uri, 45) 402 elif self.MirrorStatus.get_failing_mirror_status(uri) > 31: 403 # now decrement each time this point is reached, 404 # if will be back < 30, then equo will try to use it again 405 self.MirrorStatus.add_failing_mirror(uri, -4) 406 else: 407 # put to 0 - reenable mirror, welcome back uri! 408 self.MirrorStatus.set_failing_mirror_status(uri, 0) 409 410 remaining.discard(uri) 411 continue 412 413 do_resume = True 414 timeout_try_count = 50 415 while True: 416 try: 417 mytxt = mirrorCountText 418 mytxt += blue("%s: ") % (_("Downloading from"),) 419 mytxt += red(self.entropyTools.spliturl(uri)[1]) 420 # now fetch the new one 421 self.updateProgress( 422 mytxt, 423 importance = 1, 424 type = "warning", 425 header = red(" ## ") 426 ) 427 rc, data_transfer, resumed = self.fetch_file( 428 url, 429 branch, 430 digest, 431 do_resume, 432 fetch_file_abort_function = fetch_abort_function 433 ) 434 if rc == 0: 435 mytxt = mirrorCountText 436 mytxt += blue("%s: ") % (_("Successfully downloaded from"),) 437 mytxt += red(self.entropyTools.spliturl(uri)[1]) 438 human_bytes = self.entropyTools.bytes_into_human(data_transfer) 439 mytxt += " %s %s/%s" % (_("at"), human_bytes, _("second"),) 440 self.updateProgress( 441 mytxt, 442 importance = 1, 443 type = "info", 444 header = red(" ## ") 445 ) 446 447 self.MirrorStatus.set_working_mirror(None) 448 return 0 449 elif resumed and (rc not in (-3, -4, -100,)): 450 do_resume = False 451 continue 452 else: 453 error_message = mirrorCountText 454 error_message += blue("%s: %s") % ( 455 _("Error downloading from"), 456 red(self.entropyTools.spliturl(uri)[1]), 457 ) 458 # something bad happened 459 if rc == -1: 460 error_message += " - %s." % (_("file not available on this mirror"),) 461 elif rc == -2: 462 self.MirrorStatus.add_failing_mirror(uri, 1) 463 error_message += " - %s." % (_("wrong checksum"),) 464 # If file is fetched (with no resume) and its complete 465 # better to enforce resume to False. 466 if (data_transfer < 1) and do_resume: 467 error_message += " %s." % (_("Disabling resume"),) 468 do_resume = False 469 continue 470 elif rc == -3: 471 self.MirrorStatus.add_failing_mirror(uri, 3) 472 error_message += " - %s." % (_("not found"),) 473 elif rc == -4: # timeout! 474 timeout_try_count -= 1 475 if timeout_try_count > 0: 476 error_message += " - %s." % (_("timeout, retrying on this mirror"),) 477 else: 478 error_message += " - %s." % (_("timeout, giving up"),) 479 elif rc == -100: 480 error_message += " - %s." % (_("discarded download"),) 481 else: 482 self.MirrorStatus.add_failing_mirror(uri, 5) 483 error_message += " - %s." % (_("unknown reason"),) 484 self.updateProgress( 485 error_message, 486 importance = 1, 487 type = "warning", 488 header = red(" ## ") 489 ) 490 if rc == -4: # timeout 491 if timeout_try_count > 0: 492 continue 493 elif rc == -100: # user discarded fetch 494 self.MirrorStatus.set_working_mirror(None) 495 return 1 496 remaining.discard(uri) 497 # make sure we don't have nasty issues 498 if not remaining: 499 self.MirrorStatus.set_working_mirror(None) 500 return 3 501 break 502 except KeyboardInterrupt: 503 self.MirrorStatus.set_working_mirror(None) 504 return 1 505 506 self.MirrorStatus.set_working_mirror(None) 507 return 0

508

Source Code for Module entropy.client.interfaces.fetch