entropy.fetchers

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 Transceivers Fetchers submodule}. 10 11 """ 12 import os 13 import sys 14 import time 15 16 if sys.hexversion >= 0x3000000: 17 import urllib.request as urlmod 18 import urllib.error as urlmod_error 19 else: 20 import urllib2 as urlmod 21 import urllib2 as urlmod_error 22 23 from entropy.tools import print_traceback, get_file_size, \ 24 convert_seconds_to_fancy_output, bytes_into_human, spliturl 25 from entropy.const import etpConst, const_isfileobj 26 from entropy.output import TextInterface, darkblue, darkred, purple, blue, \ 27 brown, darkgreen, red 28 29 from entropy.i18n import _ 30 from entropy.misc import ParallelTask 31 from entropy.core.settings.base import SystemSettings 32 33

34 -class UrlFetcher:

35

36 - def __init__(self, url, path_to_save, checksum = True, 37 show_speed = True, resume = True, 38 abort_check_func = None, disallow_redirect = False, 39 thread_stop_func = None, speed_limit = None, 40 OutputInterface = None):

41 42 self.__system_settings = SystemSettings() 43 if speed_limit == None: 44 speed_limit = \ 45 self.__system_settings['repositories']['transfer_limit'] 46 47 self.progress = None 48 import entropy.tools as entropyTools 49 import socket 50 self.entropyTools, self.socket = entropyTools, socket 51 self.__timeout = \ 52 self.__system_settings['repositories']['timeout'] 53 self.__th_id = 0 54 self.__resume = resume 55 self.__url = self.__encode_url(url) 56 self.__path_to_save = path_to_save 57 self.__checksum = checksum 58 self.__show_speed = show_speed 59 self.__abort_check_func = abort_check_func 60 self.__thread_stop_func = thread_stop_func 61 self.__disallow_redirect = disallow_redirect 62 self.__speedlimit = speed_limit # kbytes/sec 63 self.__existed_before = False 64 self.localfile = None 65 66 # important to have this here too 67 self.__datatransfer = 0 68 self.__resumed = False 69 70 uname = os.uname() 71 self.user_agent = "Entropy/%s (compatible; %s; %s: %s %s %s)" % ( 72 etpConst['entropyversion'], 73 "Entropy", 74 os.path.basename(self.__url), 75 uname[0], 76 uname[4], 77 uname[2], 78 ) 79 self.__Output = OutputInterface 80 if self.__Output == None: 81 self.__Output = TextInterface() 82 elif not hasattr(self.__Output, 'updateProgress'): 83 mytxt = _("Output interface passed doesn't have the updateProgress method") 84 raise AttributeError(mytxt) 85 elif not hasattr(self.__Output.updateProgress, '__call__'): 86 mytxt = _("Output interface passed doesn't have the updateProgress method") 87 raise AttributeError(mytxt)

88

89 - def _init_vars(self):

90 self.__resumed = False 91 self.__buffersize = 8192 92 self.__status = None 93 self.__remotefile = None 94 self.__downloadedsize = 0 95 self.__average = 0 96 self.__remotesize = 0 97 self.__oldaverage = 0.0 98 self.__last_output_time = time.time() 99 # transfer status data 100 self.__startingposition = 0 101 self.__datatransfer = 0 102 self.__time_remaining = "(infinite)" 103 self.__time_remaining_secs = 0 104 self.__elapsed = 0.0 105 self.__updatestep = 0.2 106 self.__starttime = time.time() 107 self.__existed_before = False 108 if os.path.lexists(self.__path_to_save): 109 self.__existed_before = True 110 self.__setup_resume_support() 111 self._setup_proxy()

112

113 - def __setup_resume_support(self):

114 # if client uses this instance more than 115 # once, make sure we close previously opened 116 # files. 117 if const_isfileobj(self.localfile): 118 try: 119 self.localfile.flush() 120 self.localfile.close() 121 except (IOError, OSError,): 122 pass 123 124 # resume support 125 if os.path.isfile(self.__path_to_save) and \ 126 os.access(self.__path_to_save, os.W_OK) and self.__resume: 127 128 self.localfile = open(self.__path_to_save, "ab") 129 self.localfile.seek(0, os.SEEK_END) 130 self.__startingposition = int(self.localfile.tell()) 131 self.__resumed = True 132 return 133 134 self.localfile = open(self.__path_to_save, "wb")

135

136 - def _setup_proxy(self):

137 # setup proxy, doing here because config is dynamic 138 mydict = {} 139 proxy_data = self.__system_settings['system']['proxy'] 140 if proxy_data['ftp']: 141 mydict['ftp'] = proxy_data['ftp'] 142 if proxy_data['http']: 143 mydict['http'] = proxy_data['http'] 144 if mydict: 145 mydict['username'] = proxy_data['username'] 146 mydict['password'] = proxy_data['password'] 147 self.entropyTools.add_proxy_opener(urlmod, mydict) 148 else: 149 # unset 150 urlmod._opener = None

151

152 - def __encode_url(self, url):

153 if sys.hexversion >= 0x3000000: 154 import urllib.parse as encurl 155 else: 156 import urllib as encurl 157 url = os.path.join(os.path.dirname(url), 158 encurl.quote(os.path.basename(url))) 159 return url

160

161 - def set_id(self, th_id):

162 self.__th_id = th_id

163

164 - def download(self):

165 166 self._init_vars() 167 168 if self.__url.startswith("http://"): 169 headers = { 'User-Agent' : self.user_agent } 170 req = urlmod.Request(self.__url, headers = headers) 171 else: 172 req = self.__url 173 174 u_agent_error = False 175 while True: 176 # get file size if available 177 try: 178 self.__remotefile = urlmod.urlopen(req, None, self.__timeout) 179 except KeyboardInterrupt: 180 self.__close(False) 181 raise 182 except urlmod_error.HTTPError as e: 183 if (e.code == 405) and not u_agent_error: 184 # server doesn't like our user agent 185 req = self.__url 186 u_agent_error = True 187 continue 188 self.__close(True) 189 self.__status = "-3" 190 return self.__status 191 192 except urlmod_error.URLError as err: # timeout error 193 self.__close(True) 194 self.__status = "-3" 195 return self.__status 196 197 except: 198 print_traceback() 199 raise 200 break 201 202 try: 203 self.__remotesize = int(self.__remotefile.headers.get( 204 "content-length")) 205 self.__remotefile.close() 206 except KeyboardInterrupt: 207 self.__close(False) 208 raise 209 except: 210 pass 211 212 # handle user stupidity 213 try: 214 request = self.__url 215 if ((self.__startingposition > 0) and (self.__remotesize > 0)) \ 216 and (self.__startingposition < self.__remotesize): 217 218 try: 219 request = urlmod.Request( 220 self.__url, 221 headers = { 222 "Range" : "bytes=" + \ 223 str(self.__startingposition) + "-" + \ 224 str(self.__remotesize) 225 } 226 ) 227 except KeyboardInterrupt: 228 self.__close(False) 229 raise 230 except: 231 pass 232 elif (self.__startingposition == self.__remotesize): 233 # all fine then! 234 self.__close(False) 235 return self.__prepare_return() 236 237 self.__remotefile = urlmod.urlopen(request, None, self.__timeout) 238 except KeyboardInterrupt: 239 self.__close(False) 240 raise 241 except: 242 self.__close(True) 243 self.__status = "-3" 244 return self.__status 245 246 if self.__remotesize > 0: 247 self.__remotesize = float(int(self.__remotesize))/1024 248 249 if self.__disallow_redirect and \ 250 (self.__url != self.__remotefile.geturl()): 251 252 self.__close(True) 253 self.__status = "-3" 254 return self.__status 255 256 while True: 257 try: 258 rsx = self.__remotefile.read(self.__buffersize) 259 if not rsx: 260 break 261 if self.__abort_check_func != None: 262 self.__abort_check_func() 263 if self.__thread_stop_func != None: 264 self.__thread_stop_func() 265 except KeyboardInterrupt: 266 self.__close(False) 267 raise 268 except self.socket.timeout: 269 self.__close(False) 270 self.__status = "-4" 271 return self.__status 272 except: 273 # python 2.4 timeouts go here 274 self.__close(True) 275 self.__status = "-3" 276 return self.__status 277 278 self._commit(rsx) 279 if self.__show_speed: 280 self.handle_statistics(self.__th_id, self.__downloadedsize, 281 self.__remotesize, self.__average, self.__oldaverage, 282 self.__updatestep, self.__show_speed, self.__datatransfer, 283 self.__time_remaining, self.__time_remaining_secs 284 ) 285 self.updateProgress() 286 self.__oldaverage = self.__average 287 if self.__speedlimit: 288 while self.__datatransfer > self.__speedlimit*1024: 289 time.sleep(0.1) 290 self._update_speed() 291 if self.__show_speed: 292 self.updateProgress() 293 self.__oldaverage = self.__average 294 295 self._push_progress_to_output() 296 # kill thread 297 self.__close(False) 298 return self.__prepare_return()

299 300

301 - def __prepare_return(self):

302 if self.__checksum: 303 self.__status = self.entropyTools.md5sum(self.__path_to_save) 304 return self.__status 305 self.__status = "-2" 306 return self.__status

307

308 - def _commit(self, mybuffer):

309 # writing file buffer 310 self.localfile.write(mybuffer) 311 # update progress info 312 self.__downloadedsize = self.localfile.tell() 313 kbytecount = float(self.__downloadedsize)/1024 314 average = int((kbytecount/self.__remotesize)*100) 315 if average > 100: 316 average = 100 317 self.__average = average 318 self._update_speed()

319

320 - def __close(self, errored):

321 self._update_speed() 322 try: 323 if const_isfileobj(self.localfile): 324 self.localfile.flush() 325 self.localfile.close() 326 except IOError: 327 pass 328 if (not self.__existed_before) and errored: 329 try: 330 os.remove(self.__path_to_save) 331 except OSError: 332 pass 333 try: 334 self.__remotefile.close() 335 except: 336 pass

337

338 - def _update_speed(self):

339 cur_time = time.time() 340 self.__elapsed = cur_time - self.__starttime 341 # we have the diff size 342 x_delta = self.__downloadedsize - self.__startingposition 343 self.__datatransfer = x_delta / self.__elapsed 344 if self.__datatransfer < 0: 345 self.__datatransfer = 0 346 try: 347 rounded_remote = int(round(self.__remotesize*1024, 0)) 348 rounded_downloaded = int(round(self.__downloadedsize, 0)) 349 x_delta = rounded_remote - rounded_downloaded 350 tx_round = 0 351 if self.__datatransfer > 0: 352 tx_round = int(round(x_delta/self.__datatransfer, 0)) 353 self.__time_remaining_secs = tx_round 354 355 if tx_round < 0: 356 self.__time_remaining = "(%s)" % (_("infinite"),) 357 else: 358 self.__time_remaining = \ 359 convert_seconds_to_fancy_output(self.__time_remaining_secs) 360 361 except (ValueError, TypeError,): 362 self.__time_remaining = "(%s)" % (_("infinite"),)

363

364 - def get_transfer_rate(self):

365 return self.__datatransfer

366

367 - def is_resumed(self):

368 return self.__resumed

369

370 - def handle_statistics(self, th_id, downloaded_size, total_size, 371 average, old_average, update_step, show_speed, data_transfer, 372 time_remaining, time_remaining_secs):

373 return

374

375 - def _push_progress_to_output(self):

376 377 mytxt = _("[F]") 378 eta_txt = _("ETA") 379 sec_txt = _("sec") # as in XX kb/sec 380 381 current_txt = darkred(" %s: " % (mytxt,)) + \ 382 darkgreen(str(round(float(self.__downloadedsize)/1024, 1))) + "/" \ 383 + red(str(round(self.__remotesize, 1))) + " kB" 384 # create progress bar 385 barsize = 10 386 bartext = "[" 387 curbarsize = 1 388 389 averagesize = (self.__average*barsize)/100 390 while averagesize > 0: 391 curbarsize += 1 392 bartext += "=" 393 averagesize -= 1 394 bartext += ">" 395 diffbarsize = barsize - curbarsize 396 while diffbarsize > 0: 397 bartext += " " 398 diffbarsize -= 1 399 if self.__show_speed: 400 bartext += "] => %s" % (bytes_into_human(self.__datatransfer),) 401 bartext += "/%s : %s: %s" % (sec_txt, eta_txt, 402 self.__time_remaining,) 403 else: 404 bartext += "]" 405 average = str(self.__average) 406 if len(average) < 2: 407 average = " "+average 408 current_txt += " <-> "+average+"% "+bartext 409 self.__Output.updateProgress(current_txt, back = True)

410

411 - def updateProgress(self):

412 """ 413 Main fetch progress callback. You can reimplement this to refresh 414 your output devices. 415 """ 416 update_time_delta = 0.5 417 cur_t = time.time() 418 if cur_t > (self.__last_output_time + update_time_delta): 419 self.__last_output_time = cur_t 420 self._push_progress_to_output()

421 422

423 -class MultipleUrlFetcher:

424 425 import entropy.tools as entropyTools

426 - def __init__(self, url_path_list, checksum = True, 427 show_speed = True, resume = True, 428 abort_check_func = None, disallow_redirect = False, 429 OutputInterface = None, UrlFetcherClass = None):

430 """ 431 @param url_path_list list [(url,path_to_save,),...] 432 @param checksum bool return checksum data 433 @param show_speed bool show transfer speed on the output 434 @param resume bool enable resume support 435 @param abort_check_func callable function that could 436 raise exception and stop transfer 437 @param disallow_redirect bool disable automatic HTTP redirect 438 @param OutputInterface TextInterface instance used to 439 print instance output through a common interface 440 @param UrlFetcherClass, urlFetcher instance/interface used 441 """ 442 self.__system_settings = SystemSettings() 443 self.__url_path_list = url_path_list 444 self.__resume = resume 445 self.__checksum = checksum 446 self.__show_speed = show_speed 447 self.__abort_check_func = abort_check_func 448 self.__disallow_redirect = disallow_redirect 449 450 # important to have a declaration here 451 self.__data_transfer = 0 452 self.__average = 0 453 self.__old_average = 0 454 self.__time_remaining_sec = 0 455 456 self.__Output = OutputInterface 457 if self.__Output == None: 458 self.__Output = TextInterface() 459 elif not hasattr(self.__Output, 'updateProgress'): 460 mytxt = _("Output interface passed doesn't have the updateProgress method") 461 raise AttributeError("IncorrectParameter: %s" % (mytxt,)) 462 elif not hasattr(self.__Output.updateProgress, '__call__'): 463 mytxt = _("Output interface passed doesn't have the updateProgress method") 464 raise AttributeError("IncorrectParameter: %s" % (mytxt,)) 465 466 self.__url_fetcher = UrlFetcherClass 467 if self.__url_fetcher == None: 468 self.__url_fetcher = UrlFetcher

469 470

471 - def __handle_threads_stop(self):

472 if self.__stop_threads: 473 raise InterruptError

474

475 - def _init_vars(self):

476 self.__progress_data = {} 477 self.__thread_pool = {} 478 self.__download_statuses = {} 479 self.__show_progress = False 480 self.__stop_threads = False 481 self.__first_refreshes = 50 482 self.__data_transfer = 0 483 self.__average = 0 484 self.__old_average = 0 485 self.__time_remaining_sec = 0 486 self.__progress_update_t = time.time() 487 self.__startup_time = time.time()

488

489 - def download(self):

490 self._init_vars() 491 492 th_id = 0 493 speed_limit = 0 494 dsl = self.__system_settings['repositories']['transfer_limit'] 495 if isinstance(dsl, int) and self.__url_path_list: 496 speed_limit = dsl/len(self.__url_path_list) 497 498 class MyFetcher(self.__url_fetcher): 499 500 def __init__(self, klass, multiple, *args, **kwargs): 501 klass.__init__(self, *args, **kwargs) 502 self.__multiple_fetcher = multiple

503 504 def updateProgress(self): 505 return self.__multiple_fetcher.updateProgress()

506 507 def _push_progress_to_output(self): 508 return 509 510 def handle_statistics(self, *args, **kwargs): 511 return self.__multiple_fetcher.handle_statistics(*args, 512 **kwargs) 513 514 for url, path_to_save in self.__url_path_list: 515 th_id += 1 516 downloader = MyFetcher(self.__url_fetcher, self, url, path_to_save, 517 checksum = self.__checksum, show_speed = self.__show_speed, 518 resume = self.__resume, 519 abort_check_func = self.__abort_check_func, 520 disallow_redirect = self.__disallow_redirect, 521 thread_stop_func = self.__handle_threads_stop, 522 speed_limit = speed_limit, 523 OutputInterface = self.__Output 524 ) 525 downloader.set_id(th_id) 526 527 def do_download(ds, th_id, downloader): 528 ds[th_id] = downloader.download() 529 530 t = ParallelTask(do_download, self.__download_statuses, th_id, 531 downloader) 532 self.__thread_pool[th_id] = t 533 t.start() 534 535 self._push_progress_to_output(force = True) 536 self.show_download_files_info() 537 self.__show_progress = True 538 539 while len(self.__url_path_list) != len(self.__download_statuses): 540 try: 541 time.sleep(0.5) 542 except (SystemExit, KeyboardInterrupt,): 543 self.__stop_threads = True 544 raise 545 546 return self.__download_statuses 547

548 - def get_data_transfer(self):

549 return self.__data_transfer

550

551 - def get_average(self):

552 return self.__average

553

554 - def get_seconds_remaining(self):

555 return self.__time_remaining_sec

556

557 - def show_download_files_info(self):

558 count = 0 559 pl = self.__url_path_list[:] 560 self.__Output.updateProgress( 561 "%s: %s %s" % ( 562 darkblue(_("Aggregated download")), 563 darkred(str(len(pl))), 564 darkblue(_("items")), 565 ), 566 importance = 0, 567 type = "info", 568 header = purple(" ## ") 569 ) 570 for url, save_path in pl: 571 count += 1 572 fname = os.path.basename(url) 573 uri = spliturl(url)[1] 574 self.__Output.updateProgress( 575 "[%s] %s => %s" % ( 576 darkblue(str(count)), 577 darkgreen(uri), 578 blue(fname), 579 ), 580 importance = 0, 581 type = "info", 582 header = brown(" # ") 583 )

584

585 - def handle_statistics(self, th_id, downloaded_size, total_size, 586 average, old_average, update_step, show_speed, data_transfer, 587 time_remaining, time_remaining_secs):

588 data = { 589 'th_id': th_id, 590 'downloaded_size': downloaded_size, 591 'total_size': total_size, 592 'average': average, 593 'old_average': old_average, 594 'update_step': update_step, 595 'show_speed': show_speed, 596 'data_transfer': data_transfer, 597 'time_remaining': time_remaining, 598 'time_remaining_secs': time_remaining_secs, 599 } 600 self.__progress_data[th_id] = data

601

602 - def _push_progress_to_output(self, force = False):

603 604 downloaded_size = 0 605 total_size = 0 606 time_remaining = 0 607 update_step = 0 608 pd = self.__progress_data.copy() 609 pdlen = len(pd) 610 611 # calculation 612 for th_id in sorted(pd): 613 data = pd.get(th_id) 614 downloaded_size += data.get('downloaded_size', 0) 615 total_size += data.get('total_size', 0) 616 # data_transfer from Python threading bullshit is not reliable 617 # with multiple threads and causes inaccurate informations to be 618 # printed 619 # data_transfer += data.get('data_transfer', 0) 620 tr = data.get('time_remaining_secs', 0) 621 if tr > 0: 622 time_remaining += tr 623 update_step += data.get('update_step', 0) 624 625 elapsed_t = time.time() - self.__startup_time 626 if elapsed_t < 0.1: 627 elapsed_t = 0.1 628 data_transfer = downloaded_size / elapsed_t 629 self.__data_transfer = data_transfer 630 631 average = 100 632 # total_size is in kbytes 633 # downloaded_size is in bytes 634 if total_size > 0: 635 average = int(float(downloaded_size/1024)/total_size * 100) 636 637 self.__average = average 638 if pdlen > 0: 639 update_step = update_step/pdlen 640 else: 641 update_step = 0 642 time_remaining = convert_seconds_to_fancy_output(time_remaining) 643 self.__time_remaining_sec = time_remaining 644 645 update_time_delta = 0.5 646 cur_t = time.time() 647 if ((cur_t > (self.__progress_update_t + update_time_delta)) \ 648 or force or (self.__first_refreshes > 0)) and self.__show_progress: 649 650 self.__first_refreshes -= 1 651 self.__progress_update_t = cur_t 652 653 eta_txt = _("ETA") 654 sec_txt = _("sec") # as in XX kb/sec 655 down_size_txt = str(round(float(downloaded_size)/1024, 1)) 656 total_size_txt = str(round(total_size, 1)) 657 current_txt = darkgreen(down_size_txt) + "/" + red(total_size_txt) 658 current_txt += " kB" 659 # create progress bar 660 barsize = 10 661 bartext = "[" 662 curbarsize = 1 663 averagesize = (average*barsize)/100 664 while averagesize > 0: 665 curbarsize += 1 666 bartext += "=" 667 averagesize -= 1 668 bartext += ">" 669 diffbarsize = barsize-curbarsize 670 while diffbarsize > 0: 671 bartext += " " 672 diffbarsize -= 1 673 if self.__show_speed: 674 bartext += "] => %s" % (bytes_into_human(data_transfer),) 675 bartext += "/%s : %s: %s" % (sec_txt, eta_txt, time_remaining,) 676 else: 677 bartext += "]" 678 myavg = str(average) 679 if len(myavg) < 2: 680 myavg = " "+myavg 681 current_txt += " <-> "+myavg+"% "+bartext+" " 682 self.__Output.updateProgress(current_txt, back = True) 683 684 self.__old_average = average

685

686 - def updateProgress(self):

687 return self._push_progress_to_output()

688

Source Code for Module entropy.fetchers