2018-02-03

8 分钟读完 (大约 1183 个字) 0次访问

记录实用的python库

　　本文专门用来记录一下python中一些好用方法/库，可以在日常使用中提高效率。

进度条

　　在爬虫和机器学习等工作中，可能需要有一个进度条能够反馈当前程序运行速度或者进度，可以考虑用以下方法实现：

tqdm

　　老朋友tqdm，就是丑了点。

安装

1	pip install tqdm

使用

# 最基本的用法
import time
from tqdm import tqdm
for i in tqdm(range(9)):
	time.sleep(0.1)
# 效果如下
>>> 100%|██████████| 10/10 [00:01<00:00,  9.79it/s]

# trange类似于tqdm
import time
from tqdm import trange
for i in trange(10):
	time.sleep(0.1)
# 效果如下
>>> 100%|██████████| 10/10 [00:01<00:00,  9.79it/s]

# 传入list
import time
from tqdm import tqdm
pbar = tqdm([1,2,3,4,5,6,7,8,9,10])  
for char in pbar:  
    pbar.set_description("Processing %s" % char) 
    time.sleep(0.1)
# 效果如下
>>> Processing 10: 100%|██████████| 10/10 [00:01<00:00,  9.49it/s]
 
# 手动控制更新
import time
from tqdm import tqdm
with tqdm(total=10) as pbar:  
    for i in range(10):  
        pbar.update(1)
        time.sleep(0.1)
# 效果如下
>>> 100%|██████████| 10/10 [00:00<00:00, 10.10it/s]

# 也可以这样
import time
from tqdm import tqdm
pbar = tqdm(total=10)  
for i in range(10):  
    pbar.update(1)
    time.sleep(0.1)
pbar.close()  
# 效果如下
>>> 100%|██████████| 10/10 [00:00<00:00, 10.10it/s]

　　在Spyder下正常了，然而在命令窗口有问题。

Rich

　　非常炫酷的包，当然这里仅仅用它的进度条。（由@kotori-y投稿）

安装

1	pip install rich

使用

　　基本用法：

from rich.progress import track

for step in track(range(100)):
    do_step(step)

　　按步更新：

from rich.progress import Progress

with Progress() as progress:
    task = progress.add_task("[red]Downloading...", total=total)
    #do sth
    progress.update(task, advance=1)

　　存在问题在于，不能在cmder或者spyder下正常显示，有请Windows Terminal（逃）。

ShowProcess类

　　在网上找到别人写的一个方法如下：

# 建立一个ShowProcess类
class ShowProcess():
    i = 0 
    max_steps = 0 
    max_arrow = 50 

    def __init__(self, max_steps):
        self.max_steps = max_steps
        self.i = 0

    def show_process(self, i=None):
        if i is not None:
            self.i = i
        else:
            self.i += 1
        num_arrow = int(self.i * self.max_arrow / self.max_steps) 
        num_line = self.max_arrow - num_arrow 
        percent = self.i * 100.0 / self.max_steps 
        process_bar = '[' + '>' * num_arrow + '-' * num_line + ']'\
                      + '%.2f' % percent + '%'
        print('\r',process_bar,end='',flush=True) 

    def close(self, words='done'):
        print('')
        print(words)
        self.i = 0

　　使用示例：

if __name__=='__main__':
    max_steps = 100

    process_bar = ShowProcess(max_steps)

    for i in range(max_steps):
        process_bar.show_process()
        time.sleep(0.05)
    process_bar.close()

# 效果如下 
>>> [>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>]100.00%
>>> done

　　在命令窗口下使用正常，但是在IDLE和Spyder中显示存在问题，考虑使用其他方法。

重试

　　进行爬虫的时候，很容易因为网络问题导致失败，这里有2个库可以很轻松地实现这个功能。

retry

安装

1	pip install retry

使用

　　只需要在函数定义前加上@retry就行了。

from retry import retry

@retry()
def make_trouble():
    '''重试直到成功'''
@retry(ZeroDivisionError, tries=3, delay=2)
def make_trouble():
    '''出现ZeroDivisionError时重试, 重试3次，每次间隔2秒'''
@retry((ValueError, TypeError), delay=1, backoff=2)
def make_trouble():
    '''出现ValueError或TypeError时重试, 每次间隔1, 2, 4, 8, ...秒'''
@retry((ValueError, TypeError), delay=1, backoff=2, max_delay=4)
def make_trouble():
    '''出现ValueError或TypeError时重试, 每次间隔1, 2, 4, 4, ...秒，最高间隔为4秒'''
@retry(ValueError, delay=1, jitter=1)
def make_trouble():
    '''出现ValueError时重试,每次间隔1, 2, 3, 4, ... 秒'''

Tenacity

　　使用类似于retry。同样只需要在函数定义前加上@retry就行了。

安装

1	pip install tenacity

使用

from tenacity import retry, retry_if_exception_type, wait_fixed, stop_after_attempt, stop_after_delay,

@retry()
def make_trouble():
    '''重试直到成功'''
@retry(retry=retry_if_exception_type(ZeroDivisionError), wait=wait_fixed(2), 	stop=stop_after_attempt(3))
def make_trouble():
    '''出现ZeroDivisionError时重试, 重试3次，每次间隔2秒'''
@retry(stop=(stop_after_delay(10) | stop_after_attempt(5)))
def make_trouble():
    '''重试10秒或者5次'''    
@retry(wait=wait_random(min=1, max=2))
def make_trouble():
    '''重试间隔在随机1-2秒'''
@retry(wait=wait_chain(*[wait_fixed(3) for i in range(3)] +
                       [wait_fixed(7) for i in range(2)] +
                       [wait_fixed(9)]))
def make_trouble():
    '''前三次重试每次间隔3秒，接下来2次间隔7秒，之后重试间隔9秒'''

超时

　　很多任务特别是多线程时，为了防止程序卡死，需要设定一个超时。

func_timeout

　　由于windows下signal的支持问题，选择使用第三方包，func_timeout就是一个给函数添加超时的包。

安装

1	pip install func_timeout

使用

import time
from func_timeout import func_set_timeout,FunctionTimedOut

# 基本用法
try:
	doitReturnValue = func_timeout(5, doit, args=('arg1', 'arg2'))
except FunctionTimedOut:
	print ( "doit('arg1', 'arg2') could not complete within 5 seconds and was terminated.\n")
except Exception as e:
	# 其他Exception

# 装饰器用法
@func_set_timeout(2)
def task():
	time.sleep(5)
# 效果如下
FunctionTimedOut: Function task (args=()) (kwargs={}) timed out after 2.000000 seconds.

# 捕获异常
from func_timeout.exceptions import FunctionTimedOut
try:
	task()
except FunctionTimedOut:
	print('task func_timeout')

　　此外它还有一个重试的函数FunctionTimedOut，就不赘述了。

异步

　　待填坑……

# Efficiency, Python