Python编写一个优美的下载器

本文实例为大家分享了python编写下载器的具体代码，供大家参考，具体内容如下

#!/bin/python3 
# author: lidawei 
# create: 2016-07-11 
# version: 1.0 
# 功能说明： 
#  从指定的url将文件取回本地 
##################################################### 
 
import http.client 
import os 
import threading 
import time 
import logging 
import unittest 
from queue import queue 
from urllib.parse import urlparse 
 
logging.basicconfig(level = logging.debug, 
     format = '%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s', 
     datefmt = '%a, %d %b %y %h:%m:%s', 
     filename = 'downloader_%s.log' % (time.strftime('%y-%m-%d')), 
     filemode = 'a') 
 
class downloader(object): 
 '''''文件下载器''' 
 url = '' 
 filename = '' 
 
 def __init__(self, full_url_str, filename): 
  '''''初始化''' 
  self.url = urlparse(full_url_str) 
  self.filename = filename 
 
 def download(self): 
  '''''执行下载，返回true或false''' 
  if self.url == '' or self.url == none or self.filename == '' or self.filename == none: 
   logging.error('invalid parameter for downloader') 
   return false 
 
  successed = false 
  conn = none 
  if self.url.scheme == 'https': 
   conn = http.client.httpsconnection(self.url.netloc) 
  else: 
   conn = http.client.httpconnection(self.url.netloc) 
  conn.request('get', self.url.path) 
  response = conn.getresponse() 
  if response.status == 200: 
   total_size = response.getheader('content-length') 
   total_size = (int)(total_size) 
   if total_size > 0: 
    finished_size = 0 
    file = open(self.filename, 'wb') 
    if file: 
     progress = progress() 
     progress.start() 
     while not response.closed: 
      buffers = response.read(1024) 
      file.write(buffers) 
 
      finished_size += len(buffers) 
      progress.update(finished_size, total_size) 
      if finished_size >= total_size: 
       break 
     # ... end while statment 
     file.close() 
     progress.stop() 
     progress.join() 
    else: 
     logging.error('create local file %s failed' % (self.filename)) 
    # ... end if statment 
   else: 
    logging.error('request file %s size failed' % (self.filename)) 
   # ... end if statment 
  else: 
   logging.error('http/https request failed, status code:%d' % (response.status)) 
  # ... end if statment 
  conn.close() 
 
  return successed 
 # ... end download() method 
# ... end downloader class 
 
class datawriter(threading.thread): 
 filename = '' 
 data_dict = {'offset' : 0, 'buffers_byte' : b''} 
 queue = queue(128) 
 __stop = false 
 
 def __init__(self, filename): 
  self.filename = filename 
  threading.thread.__init__(self) 
 
 #override 
 def run(self): 
  while not self.__stop: 
   self.queue.get(true, 1) 
 
 def put_data(data_dict): 
  '''''将data_dict的数据放入队列，data_dict是一个字典，有两个元素：offset是偏移量，buffers_byte是二进制字节串''' 
  self.queue.put(data_dict) 
 
 def stop(self): 
  self.__stop = true 
 
class progress(threading.thread): 
 interval = 1 
 total_size = 0 
 finished_size = 0 
 old_size = 0 
 __stop = false 
 
 def __init__(self, interval = 0.5): 
  self.interval = interval 
  threading.thread.__init__(self) 
 
 #override 
 def run(self): 
  # logging.info('  total  finished  percent  speed') 
  print('  total  finished  percent  speed') 
  while not self.__stop: 
   time.sleep(self.interval) 
   if self.total_size > 0: 
    percent = self.finished_size / self.total_size * 100 
    speed = (self.finished_size - self.old_size) / self.interval 
    msg = '%12d %12d %10.2f%% %12d' % (self.total_size, self.finished_size, percent, speed) 
    # logging.info(msg) 
    print(msg) 
 
    self.old_size = self.finished_size 
   else: 
    logging.error('total size is zero') 
  # ... end while statment 
 # ... end run() method 
 
 def stop(self): 
  self.__stop = true 
 
 def update(self, finished_size, total_size): 
  self.finished_size = finished_size 
  self.total_size = total_size 
 
class testdownloaderfunctions(unittest.testcase): 
 
 def setup(self): 
  print('setup') 
 
 def test_download(self): 
  url = 'http://dldir1.qq.com/qqfile/qq/qq8.4/18376/qq8.4.exe' 
  filename = 'qq8.4.exe' 
  dl = downloader(url, filename) 
  dl.download() 
 
 def teardown(self): 
  print('teardown') 
 
if __name__ == '__main__': 
 unittest.main() 

这是测试结果：

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。