entropy.transceivers

1 # -*- coding: utf-8 -*- 2 ''' 3 # DESCRIPTION: 4 # Entropy Object Oriented Interface 5 6 Copyright (C) 2007-2009 Fabio Erculiani 7 8 This program is free software; you can redistribute it and/or modify 9 it under the terms of the GNU General Public License as published by 10 the Free Software Foundation; either version 2 of the License, or 11 (at your option) any later version. 12 13 This program is distributed in the hope that it will be useful, 14 but WITHOUT ANY WARRANTY; without even the implied warranty of 15 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 16 GNU General Public License for more details. 17 18 You should have received a copy of the GNU General Public License 19 along with this program; if not, write to the Free Software 20 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 21 ''' 22 from __future__ import with_statement 23 import os 24 import urllib2 25 import time 26 from entropy.const import etpConst 27 from entropy.output import TextInterface, darkblue, darkred, purple, blue, \ 28 brown, darkgreen, red, bold 29 from entropy.exceptions import * 30 from entropy.i18n import _ 31 from entropy.misc import TimeScheduled, ParallelTask 32 from entropy.core import SystemSettings 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 = self.__system_settings['repositories']['transfer_limit'] 45 46 self.progress = None 47 import entropy.tools as entropyTools 48 import socket 49 self.entropyTools, self.socket = entropyTools, socket 50 self.__th_id = 0 51 self.__resume = resume 52 self.__url = self.__encode_url(url) 53 self.__path_to_save = path_to_save 54 self.__checksum = checksum 55 self.__show_speed = show_speed 56 self.__abort_check_func = abort_check_func 57 self.__thread_stop_func = thread_stop_func 58 self.__disallow_redirect = disallow_redirect 59 self.__speedlimit = speed_limit # kbytes/sec 60 self.__existed_before = False 61 self.localfile = None 62 63 # important to have this here too 64 self.__datatransfer = 0 65 self.__resumed = False 66 67 uname = os.uname() 68 self.user_agent = "Entropy/%s (compatible; %s; %s: %s %s %s)" % ( 69 etpConst['entropyversion'], 70 "Entropy", 71 os.path.basename(self.__url), 72 uname[0], 73 uname[4], 74 uname[2], 75 ) 76 self.__extra_header_data = {} 77 self.__Output = OutputInterface 78 if self.__Output == None: 79 self.__Output = TextInterface() 80 elif not hasattr(self.__Output,'updateProgress'): 81 mytxt = _("Output interface passed doesn't have the updateProgress method") 82 raise IncorrectParameter("IncorrectParameter: %s" % (mytxt,)) 83 elif not callable(self.__Output.updateProgress): 84 mytxt = _("Output interface passed doesn't have the updateProgress method") 85 raise IncorrectParameter("IncorrectParameter: %s" % (mytxt,))

86

87 - def _init_vars(self):

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

110

111 - def __setup_resume_support(self):

112 113 # if client uses this instance more than 114 # once, make sure we close previously opened 115 # files. 116 if isinstance(self.localfile, file): 117 try: 118 self.localfile.flush() 119 self.localfile.close() 120 except (IOError, OSError,): 121 pass 122 123 # resume support 124 if os.path.isfile(self.__path_to_save) and \ 125 os.access(self.__path_to_save,os.W_OK) and self.__resume: 126 127 self.localfile = open(self.__path_to_save,"awb") 128 self.localfile.seek(0,2) 129 self.__startingposition = int(self.localfile.tell()) 130 self.__resumed = True 131 132 elif os.path.lexists(self.__path_to_save) and not \ 133 self.entropyTools.is_valid_path(self.__path_to_save): 134 try: 135 os.remove(self.__path_to_save) 136 except OSError: # I won't stop you here 137 pass 138 self.localfile = open(self.__path_to_save,"wb")

139

140 - def _setup_proxy(self):

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

155

156 - def __encode_url(self, url):

157 import urllib 158 url = os.path.join(os.path.dirname(url), 159 urllib.quote(os.path.basename(url))) 160 return url

161

162 - def set_id(self, th_id):

163 self.__th_id = th_id

164

165 - def download(self):

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

315

316 - def __close(self, errored):

317 try: 318 if isinstance(self.localfile, file): 319 self.localfile.flush() 320 self.localfile.close() 321 except IOError: 322 pass 323 if (not self.__existed_before) and errored: 324 try: 325 os.remove(self.__path_to_save) 326 except OSError: 327 pass 328 try: 329 self.__remotefile.close() 330 except: 331 pass 332 self.speedUpdater.kill() 333 self.socket.setdefaulttimeout(2)

334

335 - def __update_speed(self):

336 self.__elapsed += self.__transferpollingtime 337 # we have the diff size 338 x_delta = self.__downloadedsize - self.__startingposition 339 self.__datatransfer = x_delta / self.__elapsed 340 if self.__datatransfer < 0: 341 self.__datatransfer = 0 342 try: 343 rounded_remote = int(round(self.__remotesize*1024,0)) 344 rounded_downloaded = int(round(self.__downloadedsize,0)) 345 x_delta = rounded_remote - rounded_downloaded 346 tx_round = int(round(x_delta/self.__datatransfer,0)) 347 self.__time_remaining_secs = tx_round 348 self.__time_remaining = self.entropyTools.convert_seconds_to_fancy_output(self.__time_remaining_secs) 349 except: 350 self.__time_remaining = "(%s)" % (_("infinite"),)

351

352 - def get_transfer_rate(self):

353 return self.__datatransfer

354

355 - def is_resumed(self):

356 return self.__resumed

357

358 - def handle_statistics(self, th_id, downloaded_size, total_size, 359 average, old_average, update_step, show_speed, data_transfer, 360 time_remaining, time_remaining_secs):

361 return

362

363 - def updateProgress(self):

364 365 mytxt = _("[F]") 366 eta_txt = _("ETA") 367 sec_txt = _("sec") # as in XX kb/sec 368 369 currentText = darkred(" %s: " % (mytxt,)) + \ 370 darkgreen(str(round(float(self.__downloadedsize)/1024,1))) + "/" + \ 371 red(str(round(self.__remotesize,1))) + " kB" 372 # create progress bar 373 barsize = 10 374 bartext = "[" 375 curbarsize = 1 376 if self.__average > self.__oldaverage+self.__updatestep: 377 averagesize = (self.__average*barsize)/100 378 while averagesize > 0: 379 curbarsize += 1 380 bartext += "=" 381 averagesize -= 1 382 bartext += ">" 383 diffbarsize = barsize-curbarsize 384 while diffbarsize > 0: 385 bartext += " " 386 diffbarsize -= 1 387 if self.__show_speed: 388 bartext += "] => %s" % (self.entropyTools.bytes_into_human(self.__datatransfer),) 389 bartext += "/%s : %s: %s" % (sec_txt,eta_txt,self.__time_remaining,) 390 else: 391 bartext += "]" 392 average = str(self.__average) 393 if len(average) < 2: 394 average = " "+average 395 currentText += " <-> "+average+"% "+bartext 396 self.__Output.updateProgress(currentText, back = True)

397

398 -class MultipleUrlFetcher:

399 400 import entropy.tools as entropyTools

401 - def __init__(self, url_path_list, checksum = True, 402 show_speed = True, resume = True, 403 abort_check_func = None, disallow_redirect = False, 404 OutputInterface = None, UrlFetcherClass = None):

405 """ 406 @param url_path_list list [(url,path_to_save,),...] 407 @param checksum bool return checksum data 408 @param show_speed bool show transfer speed on the output 409 @param resume bool enable resume support 410 @param abort_check_func callable function that could 411 raise exception and stop transfer 412 @param disallow_redirect bool disable automatic HTTP redirect 413 @param OutputInterface TextInterface instance used to 414 print instance output through a common interface 415 @param UrlFetcherClass, urlFetcher instance/interface used 416 """ 417 self.__system_settings = SystemSettings() 418 self.__url_path_list = url_path_list 419 self.__resume = resume 420 self.__checksum = checksum 421 self.__show_speed = show_speed 422 self.__abort_check_func = abort_check_func 423 self.__disallow_redirect = disallow_redirect 424 425 # important to have a declaration here 426 self.__data_transfer = 0 427 self.__average = 0 428 self.__old_average = 0 429 self.__time_remaining_sec = 0 430 431 self.__Output = OutputInterface 432 if self.__Output == None: 433 self.__Output = TextInterface() 434 elif not hasattr(self.__Output,'updateProgress'): 435 mytxt = _("Output interface passed doesn't have the updateProgress method") 436 raise IncorrectParameter("IncorrectParameter: %s" % (mytxt,)) 437 elif not callable(self.__Output.updateProgress): 438 mytxt = _("Output interface passed doesn't have the updateProgress method") 439 raise IncorrectParameter("IncorrectParameter: %s" % (mytxt,)) 440 441 self.__url_fetcher = UrlFetcherClass 442 if self.__url_fetcher == None: 443 self.__url_fetcher = UrlFetcher

444 445

446 - def __handle_threads_stop(self):

447 if self.__stop_threads: 448 raise InterruptError

449

450 - def _init_vars(self):

451 self.__progress_data = {} 452 self.__thread_pool = {} 453 self.__download_statuses = {} 454 self.__show_progress = False 455 self.__stop_threads = False 456 self.__first_refreshes = 50 457 self.__data_transfer = 0 458 self.__average = 0 459 self.__old_average = 0 460 self.__time_remaining_sec = 0

461

462 - def download(self):

463 self._init_vars() 464 465 th_id = 0 466 speed_limit = 0 467 dsl = self.__system_settings['repositories']['transfer_limit'] 468 if isinstance(dsl,int) and self.__url_path_list: 469 speed_limit = dsl/len(self.__url_path_list) 470 471 for url, path_to_save in self.__url_path_list: 472 th_id += 1 473 downloader = self.__url_fetcher(url, path_to_save, 474 checksum = self.__checksum, show_speed = self.__show_speed, 475 resume = self.__resume, abort_check_func = self.__abort_check_func, 476 disallow_redirect = self.__disallow_redirect, 477 thread_stop_func = self.__handle_threads_stop, 478 speed_limit = speed_limit, 479 OutputInterface = self.__Output 480 ) 481 downloader.set_id(th_id) 482 downloader.updateProgress = self.updateProgress 483 downloader.handle_statistics = self.handle_statistics 484 485 def do_download(ds, th_id, downloader): 486 ds[th_id] = downloader.download()

487 488 t = ParallelTask(do_download, self.__download_statuses, th_id, downloader) 489 self.__thread_pool[th_id] = t 490 t.start() 491 self.show_download_files_info() 492 self.__show_progress = True 493 while len(self.__url_path_list) != len(self.__download_statuses): 494 try: 495 time.sleep(0.5) 496 except (SystemExit, KeyboardInterrupt,): 497 self.__stop_threads = True 498 raise 499 500 return self.__download_statuses

501

502 - def get_data_transfer(self):

503 return self.__data_transfer

504

505 - def get_average(self):

506 return self.__average

507

508 - def get_seconds_remaining(self):

509 return self.__time_remaining_sec

510

511 - def show_download_files_info(self):

512 count = 0 513 pl = self.__url_path_list[:] 514 self.__Output.updateProgress( 515 "%s: %s %s" % ( 516 darkblue(_("Aggregated download")), 517 darkred(str(len(pl))), 518 darkblue(_("items")), 519 ), 520 importance = 0, 521 type = "info", 522 header = purple(" ## ") 523 ) 524 for url, save_path in pl: 525 count += 1 526 fname = os.path.basename(url) 527 uri = self.entropyTools.spliturl(url)[1] 528 self.__Output.updateProgress( 529 "[%s] %s => %s" % ( 530 darkblue(str(count)), 531 darkgreen(uri), 532 blue(fname), 533 ), 534 importance = 0, 535 type = "info", 536 header = brown(" # ") 537 )

538

539 - def handle_statistics(self, th_id, downloaded_size, total_size, 540 average, old_average, update_step, show_speed, data_transfer, 541 time_remaining, time_remaining_secs):

542 data = { 543 'th_id': th_id, 544 'downloaded_size': downloaded_size, 545 'total_size': total_size, 546 'average': average, 547 'old_average': old_average, 548 'update_step': update_step, 549 'show_speed': show_speed, 550 'data_transfer': data_transfer, 551 'time_remaining': time_remaining, 552 'time_remaining_secs': time_remaining_secs, 553 } 554 self.__progress_data[th_id] = data

555

556 - def updateProgress(self):

557 558 eta_txt = _("ETA") 559 sec_txt = _("sec") # as in XX kb/sec 560 downloaded_size = 0 561 total_size = 0 562 time_remaining = 0 563 data_transfer = 0 564 update_step = 0 565 average = 100 566 pd = self.__progress_data.copy() 567 pdlen = len(pd) 568 569 # calculation 570 for th_id in sorted(pd): 571 data = pd.get(th_id) 572 downloaded_size += data.get('downloaded_size',0) 573 total_size += data.get('total_size',0) 574 data_transfer += data.get('data_transfer',0) 575 tr = data.get('time_remaining_secs',0) 576 if tr > 0: time_remaining += tr 577 update_step += data.get('update_step',0) 578 579 # total_size is in kbytes 580 # downloaded_size is in bytes 581 if total_size > 0: 582 average = int(float(downloaded_size/1024)/total_size * 100) 583 self.__data_transfer = data_transfer 584 self.__average = average 585 update_step = update_step/pdlen 586 self.__time_remaining_sec = time_remaining 587 time_remaining = self.entropyTools.convert_seconds_to_fancy_output(time_remaining) 588 589 if ((average > self.__old_average+update_step) or \ 590 (self.__first_refreshes > 0)) and self.__show_progress: 591 592 self.__first_refreshes -= 1 593 currentText = darkgreen(str(round(float(downloaded_size)/1024,1))) + "/" + \ 594 red(str(round(total_size,1))) + " kB" 595 # create progress bar 596 barsize = 10 597 bartext = "[" 598 curbarsize = 1 599 averagesize = (average*barsize)/100 600 while averagesize > 0: 601 curbarsize += 1 602 bartext += "=" 603 averagesize -= 1 604 bartext += ">" 605 diffbarsize = barsize-curbarsize 606 while diffbarsize > 0: 607 bartext += " " 608 diffbarsize -= 1 609 if self.__show_speed: 610 bartext += "] => %s" % (self.entropyTools.bytes_into_human(data_transfer),) 611 bartext += "/%s : %s: %s" % (sec_txt,eta_txt,time_remaining,) 612 else: 613 bartext += "]" 614 myavg = str(average) 615 if len(myavg) < 2: 616 myavg = " "+myavg 617 currentText += " <-> "+myavg+"% "+bartext+" " 618 self.__Output.updateProgress(currentText, back = True) 619 620 self.__old_average = average

621 622

623 -class FtpInterface:

624 625 # this must be run before calling the other functions

626 - def __init__(self, ftpuri, OutputInterface, verbose = True):

627 628 if not hasattr(OutputInterface,'updateProgress'): 629 mytxt = _("OutputInterface does not have an updateProgress method") 630 raise IncorrectParameter("IncorrectParameter: %s, (! %s !)" % (OutputInterface,mytxt,)) 631 elif not callable(OutputInterface.updateProgress): 632 mytxt = _("OutputInterface does not have an updateProgress method") 633 raise IncorrectParameter("IncorrectParameter: %s, (! %s !)" % (OutputInterface,mytxt,)) 634 635 import socket, ftplib 636 import entropy.tools as entropyTools 637 self.socket, self.ftplib, self.entropyTools = socket, ftplib, entropyTools 638 self.Entropy = OutputInterface 639 self.__verbose = verbose 640 self.__init_vars() 641 self.socket.setdefaulttimeout(60) 642 self.__ftpuri = ftpuri 643 self.__speed_updater = None 644 self.__currentdir = '.' 645 self.__ftphost = self.entropyTools.extract_ftp_host_from_uri(self.__ftpuri) 646 self.__ftpuser, self.__ftppassword, self.__ftpport, self.__ftpdir = self.entropyTools.extract_ftp_data(ftpuri) 647 648 count = 10 649 while 1: 650 count -= 1 651 try: 652 self.__ftpconn = self.ftplib.FTP(self.__ftphost) 653 break 654 except (self.socket.gaierror,), e: 655 raise ConnectionError('ConnectionError: %s' % (e,)) 656 except (self.socket.error,), e: 657 if not count: 658 raise ConnectionError('ConnectionError: %s' % (e,)) 659 continue 660 except: 661 if not count: raise 662 continue 663 664 if self.__verbose: 665 mytxt = _("connecting with user") 666 self.Entropy.updateProgress( 667 "[ftp:%s] %s: %s" % (darkgreen(self.__ftphost),mytxt,blue(self.__ftpuser),), 668 importance = 1, 669 type = "info", 670 header = darkgreen(" * ") 671 ) 672 try: 673 self.__ftpconn.login(self.__ftpuser,self.__ftppassword) 674 except self.ftplib.error_perm, e: 675 raise FtpError('FtpError: %s' % (e,)) 676 if self.__verbose: 677 mytxt = _("switching to") 678 self.Entropy.updateProgress( 679 "[ftp:%s] %s: %s" % (darkgreen(self.__ftphost),mytxt,blue(self.__ftpdir),), 680 importance = 1, 681 type = "info", 682 header = darkgreen(" * ") 683 ) 684 self.set_cwd(self.__ftpdir, dodir = True)

685

686 - def __init_vars(self):

687 self.__oldprogress = 0.0 688 self.__filesize = 0 689 self.__filekbcount = 0 690 self.__transfersize = 0 691 self.__startingposition = 0 692 self.__elapsed = 0.0 693 self.__time_remaining_secs = 0 694 self.__time_remaining = "(%s)" % (_("infinite"),) 695 self.__transferpollingtime = float(1)/4

696

697 - def set_basedir(self):

698 return self.set_cwd(self.__ftpdir)

699 700 # this can be used in case of exceptions

701 - def reconnect_host(self):

702 # import FTP modules 703 self.socket.setdefaulttimeout(60) 704 counter = 10 705 while 1: 706 counter -= 1 707 try: 708 self.__ftpconn = self.ftplib.FTP(self.__ftphost) 709 break 710 except: 711 if not counter: 712 raise 713 continue 714 if self.__verbose: 715 mytxt = _("reconnecting with user") 716 self.Entropy.updateProgress( 717 "[ftp:%s] %s: %s" % (darkgreen(self.__ftphost),mytxt,blue(self.__ftpuser),), 718 importance = 1, 719 type = "info", 720 header = darkgreen(" * ") 721 ) 722 self.__ftpconn.login(self.__ftpuser,self.__ftppassword) 723 if self.__verbose: 724 mytxt = _("switching to") 725 self.Entropy.updateProgress( 726 "[ftp:%s] %s: %s" % (darkgreen(self.__ftphost),mytxt,blue(self.__ftpdir),), 727 importance = 1, 728 type = "info", 729 header = darkgreen(" * ") 730 ) 731 self.set_cwd(self.__currentdir)

732

733 - def get_host(self):

734 return self.__ftphost

735

736 - def get_port(self):

737 return self.__ftpport

738

739 - def get_dir(self):

740 return self.__ftpdir

741

742 - def get_cwd(self):

743 pwd = self.__ftpconn.pwd() 744 return pwd

745

746 - def set_cwd(self, mydir, dodir = False):

747 try: 748 return self._set_cwd(mydir, dodir) 749 except self.ftplib.error_perm, e: 750 raise FtpError('FtpError: %s' % (e,))

751

752 - def _set_cwd(self, mydir, dodir = False):

753 if self.__verbose: 754 mytxt = _("switching to") 755 self.Entropy.updateProgress( 756 "[ftp:%s] %s: %s" % (darkgreen(self.__ftphost),mytxt,blue(mydir),), 757 importance = 1, 758 type = "info", 759 header = darkgreen(" * ") 760 ) 761 try: 762 self.__ftpconn.cwd(mydir) 763 except self.ftplib.error_perm, e: 764 if e[0][:3] == '550' and dodir: 765 self.recursive_mkdir(mydir) 766 self.__ftpconn.cwd(mydir) 767 else: 768 raise 769 self.__currentdir = self.get_cwd()

770

771 - def set_pasv(self,bool):

772 self.__ftpconn.set_pasv(bool)

773

774 - def set_chmod(self,chmodvalue,file):

775 return self.__ftpconn.voidcmd("SITE CHMOD "+str(chmodvalue)+" "+str(file))

776

777 - def get_file_mtime(self,path):

778 rc = self.__ftpconn.sendcmd("mdtm "+path) 779 return rc.split()[-1]

780

781 - def send_cmd(self,cmd):

782 return self.__ftpconn.sendcmd(cmd)

783

784 - def list_dir(self):

785 return [x.split("/")[-1] for x in self.__ftpconn.nlst()]

786

787 - def is_file_available(self, filename):

788 xx = [] 789 def cb(x): 790 if x == filename: xx.append(x)

791 self.__ftpconn.retrlines('NLST',cb) 792 if xx: return True 793 return False

794

795 - def delete_file(self,file):

796 try: 797 rc = self.__ftpconn.delete(file) 798 except self.ftplib.error_perm, e: 799 if e[0][:3] == '550': 800 return True 801 return False # not found 802 if rc.startswith("250"): 803 return True 804 return False

805

806 - def recursive_mkdir(self, mypath):

807 mydirs = [x for x in mypath.split("/") if x] 808 mycurpath = "" 809 for mydir in mydirs: 810 mycurpath = os.path.join(mycurpath,mydir) 811 if not self.is_file_available(mycurpath): 812 try: 813 self.mkdir(mycurpath) 814 except self.ftplib.error_perm, e: 815 if e[0].lower().find("permission denied") != -1: 816 raise 817 elif e[0][:3] != '550': 818 raise

819

820 - def mkdir(self,directory):

821 return self.__ftpconn.mkd(directory)

822

823 - def upload_file(self, file, ascii = False):

824 825 # this function also supports callback, because storbinary doesn't 826 def advanced_stor(cmd, fp, callback=None): 827 ''' Store a file in binary mode. Our version supports a callback function''' 828 self.__ftpconn.voidcmd('TYPE I') 829 conn = self.__ftpconn.transfercmd(cmd) 830 while 1: 831 buf = fp.readline() 832 if not buf: break 833 conn.sendall(buf) 834 if callback: callback(buf) 835 conn.close() 836 837 # that's another workaround 838 #return "226" 839 try: 840 rc = self.__ftpconn.voidresp() 841 except: 842 self.reconnect_host() 843 return "226" 844 return rc

845 846 def up_file_up_progress(buf): 847 self.updateProgress(len(buf)) 848 849 tries = 0 850 while tries < 10: 851 852 tries += 1 853 filename = os.path.basename(file) 854 self.__init_vars() 855 self.__start_speed_counter() 856 try: 857 858 with open(file,"r") as f: 859 860 self.__filesize = round(float(self.entropyTools.get_file_size(file))/1024,1) 861 self.__filekbcount = 0 862 863 if self.is_file_available(filename+".tmp"): 864 self.delete_file(filename+".tmp") 865 866 if ascii: 867 rc = self.__ftpconn.storlines("STOR "+filename+".tmp",f) 868 else: 869 rc = advanced_stor("STOR "+filename+".tmp", f, callback = up_file_up_progress) 870 871 # now we can rename the file with its original name 872 self.rename_file(filename+".tmp",filename) 873 874 if rc.find("226") != -1: # upload complete 875 return True 876 return False 877 878 except Exception, e: # connection reset by peer 879 880 self.entropyTools.print_traceback() 881 mytxt = red("%s: %s, %s... #%s") % ( 882 _("Upload issue"), 883 e, 884 _("retrying"), 885 tries+1, 886 ) 887 self.Entropy.updateProgress( 888 mytxt, 889 importance = 1, 890 type = "warning", 891 header = " " 892 ) 893 self.reconnect_host() # reconnect 894 if self.is_file_available(filename): 895 self.delete_file(filename) 896 if self.is_file_available(filename+".tmp"): 897 self.delete_file(filename+".tmp") 898 899 finally: 900 self.__stop_speed_counter() 901

902 - def download_file(self, filename, downloaddir, ascii = False):

903 904 def df_up(buf): 905 # writing file buffer 906 f.write(buf) 907 # update progress 908 self.__filekbcount += float(len(buf))/1024 909 # create text 910 cnt = round(self.__filekbcount,1) 911 mytxt = _("Download status") 912 currentText = brown(" <-> %s: " % (mytxt,)) + darkgreen(str(cnt)) + "/" + \ 913 red(str(self.__filesize)) + " kB" 914 self.Entropy.updateProgress( 915 currentText, 916 importance = 0, 917 type = "info", 918 back = True, 919 count = (cnt, self.__filesize), 920 percent = True 921 )

922 923 tries = 10 924 while tries: 925 tries -= 1 926 927 self.__init_vars() 928 self.__start_speed_counter() 929 try: 930 931 # look if the file exist 932 if self.is_file_available(filename): 933 self.__filekbcount = 0 934 # get the file size 935 self.__filesize = self.get_file_size_compat(filename) 936 if (self.__filesize): 937 self.__filesize = round(float(int(self.__filesize))/1024,1) 938 if (self.__filesize == 0): 939 self.__filesize = 1 940 else: 941 self.__filesize = 0 942 if not ascii: 943 f = open(downloaddir+"/"+filename,"wb") 944 rc = self.__ftpconn.retrbinary('RETR '+filename, df_up, 1024) 945 else: 946 f = open(downloaddir+"/"+filename,"w") 947 rc = self.__ftpconn.retrlines('RETR '+filename, f.write) 948 f.flush() 949 f.close() 950 if rc.find("226") != -1: # upload complete 951 return True 952 return False 953 954 except Exception, e: # connection reset by peer 955 956 self.entropyTools.print_traceback() 957 mytxt = red("%s: %s, %s... #%s") % ( 958 _("Download issue"), 959 e, 960 _("retrying"), 961 tries+1, 962 ) 963 self.Entropy.updateProgress( 964 mytxt, 965 importance = 1, 966 type = "warning", 967 header = " " 968 ) 969 self.reconnect_host() # reconnect 970 971 finally: 972 self.__stop_speed_counter() 973 974 # also used to move files

975 - def rename_file(self, fromfile, tofile):

976 rc = self.__ftpconn.rename(fromfile,tofile) 977 return rc

978

979 - def get_file_md5(self, filename):

980 # PROFTPD with mod_md5 supports it! 981 try: 982 rc_data = self.__ftpconn.sendcmd("SITE MD5 %s" % (filename,)) 983 except self.ftplib.error_perm: 984 return None # not supported 985 try: 986 return rc_data.split("\n")[0].split("\t")[0].split("-")[1] 987 except (IndexError, TypeError,): # wrong output 988 return None

989

990 - def get_file_size(self, filename):

991 return self.__ftpconn.size(filename)

992

993 - def get_file_size_compat(self, filename):

994 try: 995 data = [x.split() for x in self.__ftpconn.sendcmd("stat %s" % (filename,)).split("\n")] 996 except self.ftplib.error_temp: 997 return "" 998 for item in data: 999 if item[-1] == filename: 1000 return item[4] 1001 return ""

1002

1003 - def get_raw_list(self):

1004 mybuffer = [] 1005 def bufferizer(buf): 1006 mybuffer.append(buf)

1007 self.__ftpconn.dir(bufferizer) 1008 return mybuffer 1009

1010 - def close(self):

1011 try: 1012 self.__ftpconn.quit() 1013 except (EOFError,AttributeError,self.socket.timeout,self.ftplib.error_reply,): 1014 # AttributeError is raised when socket gets trashed 1015 # EOFError is raised when the connection breaks 1016 # timeout, who cares! 1017 pass

1018

1019 - def __start_speed_counter(self):

1020 self.__speed_updater = TimeScheduled( 1021 self.__transferpollingtime, 1022 self.__update_speed, 1023 ) 1024 self.__speed_updater.start()

1025

1026 - def __stop_speed_counter(self):

1027 if self.__speed_updater != None: 1028 self.__speed_updater.kill()

1029

1030 - def __update_speed(self):

1031 self.__elapsed += self.__transferpollingtime 1032 # we have the diff size 1033 self.__datatransfer = (self.__transfersize-self.__startingposition) / self.__elapsed 1034 if self.__datatransfer < 0: 1035 self.__datatransfer = 0 1036 try: 1037 self.__time_remaining_secs = int(round((int(round(self.__filesize*1024,0))-int(round(self.__transfersize,0)))/self.__datatransfer,0)) 1038 self.__time_remaining = self.entropyTools.convert_seconds_to_fancy_output(self.__time_remaining_secs) 1039 except: 1040 self.__time_remaining = "(%s)" % (_("infinite"),)

1041

1042 - def updateProgress(self, buf_len):

1043 # get the buffer size 1044 self.__filekbcount += float(buf_len)/1024 1045 self.__transfersize += buf_len 1046 # create percentage 1047 myUploadPercentage = 100.0 1048 if self.__filesize >= 1: 1049 myUploadPercentage = round((round(self.__filekbcount,1)/self.__filesize)*100,1) 1050 currentprogress = myUploadPercentage 1051 myUploadSize = round(self.__filekbcount,1) 1052 if (currentprogress > self.__oldprogress+1.0) and \ 1053 (myUploadPercentage < 100.1) and \ 1054 (myUploadSize <= self.__filesize): 1055 1056 myUploadPercentage = str(myUploadPercentage)+"%" 1057 # create text 1058 mytxt = _("Transfer status") 1059 currentText = brown(" <-> %s: " % (mytxt,)) + \ 1060 darkgreen(str(myUploadSize)) + "/" + red(str(self.__filesize)) + " kB " + \ 1061 brown("[") + str(myUploadPercentage) + brown("]") + " " + self.__time_remaining + \ 1062 " " + self.entropyTools.bytes_into_human(self.__datatransfer) + "/"+_("sec") 1063 # WARN: re-enabled updateProgress, this may cause slowdowns 1064 # print_info(currentText, back = True) 1065 self.Entropy.updateProgress(currentText, back = True) 1066 self.__oldprogress = currentprogress

1067 1068

1069 -class FtpServerHandler:

1070 1071 import entropy.tools as entropyTools

1072 - def __init__(self, ftp_interface, entropy_interface, uris, files_to_upload, 1073 download = False, remove = False, ftp_basedir = None, local_basedir = None, 1074 critical_files = [], use_handlers = False, handlers_data = {}, repo = None):

1075 1076 self.FtpInterface = ftp_interface 1077 self.Entropy = entropy_interface 1078 if not isinstance(uris,list): 1079 raise InvalidDataType("InvalidDataType: %s" % (_("uris must be a list instance"),)) 1080 if not isinstance(files_to_upload,(list,dict)): 1081 raise InvalidDataType("InvalidDataType: %s" % ( 1082 _("files_to_upload must be a list or dict instance"), 1083 ) 1084 ) 1085 self.uris = uris 1086 if isinstance(files_to_upload,list): 1087 self.myfiles = files_to_upload[:] 1088 else: 1089 self.myfiles = sorted([x for x in files_to_upload]) 1090 self.download = download 1091 self.remove = remove 1092 self.repo = repo 1093 if self.repo == None: 1094 self.repo = self.Entropy.default_repository 1095 self.use_handlers = use_handlers 1096 if self.remove: 1097 self.download = False 1098 self.use_handlers = False 1099 if not ftp_basedir: 1100 # default to database directory 1101 branch = self.Entropy.SystemSettings['repositories']['branch'] 1102 my_path = os.path.join(self.Entropy.get_remote_database_relative_path(repo), branch) 1103 self.ftp_basedir = unicode(my_path) 1104 else: 1105 self.ftp_basedir = unicode(ftp_basedir) 1106 if not local_basedir: 1107 # default to database directory 1108 self.local_basedir = os.path.dirname(self.Entropy.get_local_database_file(self.repo)) 1109 else: 1110 self.local_basedir = unicode(local_basedir) 1111 self.critical_files = critical_files 1112 self.handlers_data = handlers_data.copy()

1113

1114 - def handler_verify_upload(self, local_filepath, uri, counter, maxcount, 1115 tries, remote_md5 = None):

1116 1117 crippled_uri = self.entropyTools.extract_ftp_host_from_uri(uri) 1118 1119 self.Entropy.updateProgress( 1120 "[%s|#%s|(%s/%s)] %s: %s" % ( 1121 blue(crippled_uri), 1122 darkgreen(str(tries)), 1123 blue(str(counter)), 1124 bold(str(maxcount)), 1125 darkgreen(_("verifying upload (if supported)")), 1126 blue(os.path.basename(local_filepath)), 1127 ), 1128 importance = 0, 1129 type = "info", 1130 header = red(" @@ "), 1131 back = True 1132 ) 1133 1134 valid_remote_md5 = True 1135 # if remote server supports MD5 commands, remote_md5 is filled 1136 if isinstance(remote_md5, basestring): 1137 valid_md5 = self.entropyTools.is_valid_md5(remote_md5) 1138 ckres = False 1139 if valid_md5: # seems valid 1140 ckres = self.entropyTools.compare_md5(local_filepath, 1141 remote_md5) 1142 if ckres: 1143 self.Entropy.updateProgress( 1144 "[%s|#%s|(%s/%s)] %s: %s: %s" % ( 1145 blue(crippled_uri), 1146 darkgreen(str(tries)), 1147 blue(str(counter)), 1148 bold(str(maxcount)), 1149 blue(_("digest verification")), 1150 os.path.basename(local_filepath), 1151 darkgreen(_("so far, so good!")), 1152 ), 1153 importance = 0, 1154 type = "info", 1155 header = red(" @@ ") 1156 ) 1157 return True 1158 # ouch! 1159 elif not valid_md5: 1160 # mmmh... malformed md5, try with handlers 1161 self.Entropy.updateProgress( 1162 "[%s|#%s|(%s/%s)] %s: %s: %s" % ( 1163 blue(crippled_uri), 1164 darkgreen(str(tries)), 1165 blue(str(counter)), 1166 bold(str(maxcount)), 1167 blue(_("digest verification")), 1168 os.path.basename(local_filepath), 1169 bold(_("malformed md5 provided to function")), 1170 ), 1171 importance = 0, 1172 type = "warning", 1173 header = brown(" @@ ") 1174 ) 1175 else: # it's really bad! 1176 self.Entropy.updateProgress( 1177 "[%s|#%s|(%s/%s)] %s: %s: %s" % ( 1178 blue(crippled_uri), 1179 darkgreen(str(tries)), 1180 blue(str(counter)), 1181 bold(str(maxcount)), 1182 blue(_("digest verification")), 1183 os.path.basename(local_filepath), 1184 bold(_("remote md5 is invalid")), 1185 ), 1186 importance = 0, 1187 type = "warning", 1188 header = brown(" @@ ") 1189 ) 1190 valid_remote_md5 = False 1191 1192 if not self.use_handlers: 1193 # handlers usage is disabled 1194 return valid_remote_md5 # always valid 1195 1196 checksum = self.Entropy.get_remote_package_checksum( 1197 self.repo, 1198 os.path.basename(local_filepath), 1199 self.handlers_data['branch'] 1200 ) 1201 if checksum == None: 1202 self.Entropy.updateProgress( 1203 "[%s|#%s|(%s/%s)] %s: %s: %s" % ( 1204 blue(crippled_uri), 1205 darkgreen(str(tries)), 1206 blue(str(counter)), 1207 bold(str(maxcount)), 1208 blue(_("digest verification")), 1209 os.path.basename(local_filepath), 1210 darkred(_("not supported")), 1211 ), 1212 importance = 0, 1213 type = "info", 1214 header = red(" @@ ") 1215 ) 1216 return valid_remote_md5 1217 elif isinstance(checksum, bool) and not checksum: 1218 self.Entropy.updateProgress( 1219 "[%s|#%s|(%s/%s)] %s: %s: %s" % ( 1220 blue(crippled_uri), 1221 darkgreen(str(tries)), 1222 blue(str(counter)), 1223 bold(str(maxcount)), 1224 blue(_("digest verification")), 1225 os.path.basename(local_filepath), 1226 bold(_("file not found")), 1227 ), 1228 importance = 0, 1229 type = "warning", 1230 header = brown(" @@ ") 1231 ) 1232 return False 1233 elif self.entropyTools.is_valid_md5(checksum): 1234 # valid? checking 1235 ckres = self.entropyTools.compare_md5(local_filepath,checksum) 1236 if ckres: 1237 self.Entropy.updateProgress( 1238 "[%s|#%s|(%s/%s)] %s: %s: %s" % ( 1239 blue(crippled_uri), 1240 darkgreen(str(tries)), 1241 blue(str(counter)), 1242 bold(str(maxcount)), 1243 blue(_("digest verification")), 1244 os.path.basename(local_filepath), 1245 darkgreen(_("so far, so good!")), 1246 ), 1247 importance = 0, 1248 type = "info", 1249 header = red(" @@ ") 1250 ) 1251 return True 1252 else: 1253 self.Entropy.updateProgress( 1254 "[%s|#%s|(%s/%s)] %s: %s: %s" % ( 1255 blue(crippled_uri), 1256 darkgreen(str(tries)), 1257 blue(str(counter)), 1258 bold(str(maxcount)), 1259 blue(_("digest verification")), 1260 os.path.basename(local_filepath), 1261 darkred(_("invalid checksum")), 1262 ), 1263 importance = 0, 1264 type = "warning", 1265 header = brown(" @@ ") 1266 ) 1267 return False 1268 else: 1269 self.Entropy.updateProgress( 1270 "[%s|#%s|(%s/%s)] %s: %s: %s" % ( 1271 blue(crippled_uri), 1272 darkgreen(str(tries)), 1273 blue(str(counter)), 1274 bold(str(maxcount)), 1275 blue(_("digest verification")), 1276 os.path.basename(local_filepath), 1277 darkred(_("unknown data returned")), 1278 ), 1279 importance = 0, 1280 type = "warning", 1281 header = brown(" @@ ") 1282 ) 1283 return valid_remote_md5

1284

1285 - def go(self):

1286 1287 broken_uris = set() 1288 fine_uris = set() 1289 errors = False 1290 action = 'upload' 1291 if self.download: 1292 action = 'download' 1293 elif self.remove: 1294 action = 'remove' 1295 1296 for uri in self.uris: 1297 1298 crippled_uri = self.entropyTools.extract_ftp_host_from_uri(uri) 1299 self.Entropy.updateProgress( 1300 "[%s|%s] %s..." % ( 1301 blue(crippled_uri), 1302 brown(action), 1303 blue(_("connecting to mirror")), 1304 ), 1305 importance = 0, 1306 type = "info", 1307 header = blue(" @@ ") 1308 ) 1309 try: 1310 ftp = self.FtpInterface(uri, self.Entropy) 1311 except ConnectionError: 1312 self.entropyTools.print_traceback() 1313 return True,fine_uris,broken_uris # issues 1314 branch = self.Entropy.SystemSettings['repositories']['branch'] 1315 my_path = os.path.join(self.Entropy.get_remote_database_relative_path(self.repo), branch) 1316 self.Entropy.updateProgress( 1317 "[%s|%s] %s %s..." % ( 1318 blue(crippled_uri), 1319 brown(action), 1320 blue(_("changing directory to")), 1321 darkgreen(my_path), 1322 ), 1323 importance = 0, 1324 type = "info", 1325 header = blue(" @@ ") 1326 ) 1327 1328 ftp.set_cwd(self.ftp_basedir, dodir = True) 1329 maxcount = len(self.myfiles) 1330 counter = 0 1331 1332 for mypath in self.myfiles: 1333 1334 ftp.set_basedir() 1335 ftp.set_cwd(self.ftp_basedir, dodir = True) 1336 1337 mycwd = None 1338 if isinstance(mypath,tuple): 1339 if len(mypath) < 2: continue 1340 mycwd = mypath[0] 1341 mypath = mypath[1] 1342 ftp.set_cwd(mycwd, dodir = True) 1343 1344 syncer = ftp.upload_file 1345 myargs = [mypath] 1346 if self.download: 1347 syncer = ftp.download_file 1348 myargs = [os.path.basename(mypath),self.local_basedir] 1349 elif self.remove: 1350 syncer = ftp.delete_file 1351 1352 counter += 1 1353 tries = 0 1354 done = False 1355 lastrc = None 1356 while tries < 5: 1357 tries += 1 1358 self.Entropy.updateProgress( 1359 "[%s|#%s|(%s/%s)] %s: %s" % ( 1360 blue(crippled_uri), 1361 darkgreen(str(tries)), 1362 blue(str(counter)), 1363 bold(str(maxcount)), 1364 blue(action+"ing"), 1365 red(os.path.basename(mypath)), 1366 ), 1367 importance = 0, 1368 type = "info", 1369 header = red(" @@ ") 1370 ) 1371 rc = syncer(*myargs) 1372 if rc and not self.download: 1373 # try with "SITE MD5 command first" 1374 # proftpd's mod_md5 supports it 1375 remote_md5 = ftp.get_file_md5(os.path.basename(mypath)) 1376 rc = self.handler_verify_upload(mypath, uri, 1377 counter, maxcount, tries, remote_md5 = remote_md5) 1378 if rc: 1379 self.Entropy.updateProgress( 1380 "[%s|#%s|(%s/%s)] %s %s: %s" % ( 1381 blue(crippled_uri), 1382 darkgreen(str(tries)), 1383 blue(str(counter)), 1384 bold(str(maxcount)), 1385 blue(action), 1386 _("successful"), 1387 red(os.path.basename(mypath)), 1388 ), 1389 importance = 0, 1390 type = "info", 1391 header = darkgreen(" @@ ") 1392 ) 1393 done = True 1394 break 1395 else: 1396 self.Entropy.updateProgress( 1397 "[%s|#%s|(%s/%s)] %s %s: %s" % ( 1398 blue(crippled_uri), 1399 darkgreen(str(tries)), 1400 blue(str(counter)), 1401 bold(str(maxcount)), 1402 blue(action), 1403 brown(_("failed, retrying")), 1404 red(os.path.basename(mypath)), 1405 ), 1406 importance = 0, 1407 type = "warning", 1408 header = brown(" @@ ") 1409 ) 1410 lastrc = rc 1411 continue 1412 1413 if not done: 1414 1415 self.Entropy.updateProgress( 1416 "[%s|(%s/%s)] %s %s: %s - %s: %s" % ( 1417 blue(crippled_uri), 1418 blue(str(counter)), 1419 bold(str(maxcount)), 1420 blue(action), 1421 darkred("failed, giving up"), 1422 red(os.path.basename(mypath)), 1423 _("error"), 1424 lastrc, 1425 ), 1426 importance = 1, 1427 type = "error", 1428 header = darkred(" !!! ") 1429 ) 1430 1431 if mypath not in self.critical_files: 1432 self.Entropy.updateProgress( 1433 "[%s|(%s/%s)] %s: %s, %s..." % ( 1434 blue(crippled_uri), 1435 blue(str(counter)), 1436 bold(str(maxcount)), 1437 blue(_("not critical")), 1438 os.path.basename(mypath), 1439 blue(_("continuing")), 1440 ), 1441 importance = 1, 1442 type = "warning", 1443 header = brown(" @@ ") 1444 ) 1445 continue 1446 1447 ftp.close() 1448 errors = True 1449 broken_uris.add((uri,lastrc)) 1450 # next mirror 1451 break 1452 1453 # close connection 1454 ftp.close() 1455 fine_uris.add(uri) 1456 1457 return errors,fine_uris,broken_uris

1458

Source Code for Module entropy.transceivers