Python编写一个优美的下载器

 更新时间:2018年04月15日 14:09:58   作者:ccpw_cn  
这篇文章主要教大家如何使用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() 

这是测试结果:

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

相关文章

  • python中的项目目录结构

    python中的项目目录结构

    这篇文章主要介绍了python中的项目目录结构,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
    2024-02-02
  • python常见的占位符总结及用法

    python常见的占位符总结及用法

    在本篇文章里小编给大家整理的是一篇关于python常见的占位符总结及用法,有兴趣的朋友们可以跟着学习参考下。
    2021-07-07
  • 基于python+pandoc实现html批量转word

    基于python+pandoc实现html批量转word

    pandoc是一个强大的文档格式转换工具,支持丰富的格式转换,并尽可能的保留原来的排版,号称文档格式转换的瑞士军刀,本文将给大家介绍一下使用python搭配pandoc实现html批量转word,感兴趣的朋友可以参考阅读下
    2023-09-09
  • TensorFlow——Checkpoint为模型添加检查点的实例

    TensorFlow——Checkpoint为模型添加检查点的实例

    今天小编就为大家分享一篇TensorFlow——Checkpoint为模型添加检查点的实例,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2020-01-01
  • Python 迭代器Iterator详情

    Python 迭代器Iterator详情

    这篇文章主要介绍了Python 迭代器Iterator详情,迭代器可以帮助我们解决面对复杂的数据场景时,快速简便的获取数据,下文关于其详细介绍,需要的小伙伴可以参考一下
    2022-05-05
  • pandas通过loc生成新的列方法

    pandas通过loc生成新的列方法

    今天小编就为大家分享一篇pandas通过loc生成新的列方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2018-11-11
  • Python用csv写入文件_消除空余行的方法

    Python用csv写入文件_消除空余行的方法

    今天小编就为大家分享一篇Python用csv写入文件_消除空余行的方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
    2018-07-07
  • Pytorch中的广播机制详解(Broadcast)

    Pytorch中的广播机制详解(Broadcast)

    这篇文章主要介绍了Pytorch中的广播机制详解(Broadcast),具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2023-01-01
  • 用Python实现BP神经网络(附代码)

    用Python实现BP神经网络(附代码)

    这篇文章主要介绍了用Python实现BP神经网络(附代码),文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
    2019-07-07
  • 使用Python Fast API发布API服务的过程详解

    使用Python Fast API发布API服务的过程详解

    这篇文章主要介绍了使用Python Fast API发布API服务,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2023-04-04

最新评论