aiohttp + asyncio 异步网络请求基本操作

作者:matrix 发布时间:2019年11月26日星期二 分类:Python

asyncio异步操作需要关键字async,await
async用来声明方法函数,await用来声明耗时操作。
但是await关键字后面要求为awaitable对象 且只能在async方法内部使用,不能在外部入口中使用。asyncio的语法其实是系统内部实现了yield from协程。

aiohttp用来代替requests的请求库,且支持异步操作。
主要优点体现在并发请求多个耗时任务时,自动安排耗时时的操作,避免cpu等待一个一个请求。

单个请求操作

import aiohttp
import asyncio

#get 请求
async def get():
  async with aiohttp.request('GET','https://api.github.com/users/Ho',params={'arg1':123}) as response:
    # response.request_info # 请求信息
    return await response.json()

rel = asyncio.run(get())

# 或者使用下面方式 手动关闭异步事件循环
# loop = asyncio.get_event_loop()
# rel = loop.run_until_complete(get())
# loop.close()

print(rel)

多个并发请求操作

主要区别在于异步任务的添加操作,运行。

请求测试url:

http://link/await/1 # delay 1sec
http://link/await/2 # delay 2sec
...

请求测试:

import aiohttp
import asyncio

#get 请求
async def get():
  async with aiohttp.request('GET','http://link/await/1') as response:
    return await response.text()

# 所有请求任务
async def all_req():
#async with asyncio.Semaphore(5): 设置并发的连接数
# https://docs.python.org/zh-cn/3/library/asyncio-sync.html#asyncio.Semaphore

  task = []
  #添加请求任务
  for i in range(5):
    task.append(asyncio.create_task(get()))
  #create_task 方法等同于  ensure_future()方法
  #手册建议首选 create_task方法 
  # https://docs.python.org/zh-cn/3/library/asyncio-future.html?highlight=ensure_future#asyncio.ensure_future

  return await  asyncio.gather(*task)#传入参数 tuple类型 作为位置参数
  # 等同于 asyncio.gather(get(),get())
  # gather()方法用于收集所有任务完成的返回值,如果换成wait()方法会返回任务tuple对象,(done,pending)

rel = asyncio.run(all_req())
print(rel)

# 总共5个请求任务返回:
# 总耗时1秒多,相比同步的5秒+好N多。
"""
['sleep 1 second is done', 'sleep 1 second is done', 'sleep 1 second is done', 'sleep 1 second is done', 'sleep 1 second is done']

[Done] exited with code=0 in 1.955 seconds
"""

tell why??

测试发现Semaphore方法设置的请求并发数量跟本不起作用,nginx的access.log以及Proxifier看到的一次性请求量都不是代码中设置的数量。

使用uvloop优化异步操作

uvloop用于提升协程的速度。
uvloop使用很简单,直接设置异步策略就好了。

import asyncio
import uvloop

#声明使用 uvloop 事件循环
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

测试遇到很多报错,基本上都是await和async使用的问题。

异步请求的分块chunk并发控制

自行chunk操作
自己按照所有任务的list列表进行chunk切割,然后分块进行请求,每块中固定chunk数量的任务。基本可以实现想要的并发限制操作

async def _bulk_task(num,start_page = 1):
  """批量创建异步任务
  """
  task = []
  for i in range(num):# 每次10个连接并发进行请求
    task.append(asyncio.create_task(get(start_page)))
    start_page += 1
  return await asyncio.gather(*task) 

# 主要进行chunk操作的函数
def run_task(total,chunk):
  """运行分块处理的批量任务

  Arguments:
      total int 总请求数
      chunk int 每次并发请求数

  Yields:
      返回收集的异步任务运行结果
  """

  for i in range(math.ceil(total/chunk)):
    start_page = i * chunk + 1
    if start_page + chunk > total:
      rel = asyncio.run(_bulk_task(total if 0 == total % chunk else total % chunk,start_page))
    else:
      rel = asyncio.run(_bulk_task(chunk,start_page))
    yield rel


rel  = run_task(123,10)# 123总任务 每10条并发请求
for i in rel:
  print(i)

TODO

参考:

https://hubertroy.gitbooks.io/aiohttp-chinese-documentation/content/aiohttp%E6%96%87%E6%A1%A3/ClientUsage.html#%E6%84%89%E5%BF%AB%E5%9C%B0%E7%BB%93%E6%9D%9F

https://docs.Python.org/zh-cn/3/library/asyncio-eventloop.html#asyncio.get_running_loop

https://segmentfault.com/q/1010000008663962

http://www.ruanyifeng.com/blog/2019/11/python-asyncio.html

https://blog.csdn.net/qq_37144341/article/details/89471603

https://www.jianshu.com/p/8f65e50f39b4

python 快速读取压缩包内文件

作者:matrix 发布时间:2019年10月14日星期一 分类:Python

搜索结果一大堆但都没有找到支持url和local path两种读取方式的操作。
留着便于以后直接使用。

gits: https://gist.github.com/Hootrix/cf3e75b1fa6d3d404bc99787f89687f1


import requests,tempfile, zipfile,os def read_file_for_zip(zip_url, callback=None): """ 读取zip包内的文件 :param zip_url:zip路径/url :param callback:读取操作的回调函数 若函数返回false 则不会读取下一个文件 :return: """ with tempfile.TemporaryFile('w+b') as tmpfile: # 生成临时文件 # 判断是否为本地文件 if os.path.isfile(zip_url): #进行本地复制。没必要 # with open(zip_url,'rb') as f: # while True: # chunk = f.read(1024) # if not chunk: # break # tmpfile.write(chunk) tmpfile = zip_url else:#进行http请求 r = requests.get(zip_url, stream=True) for chunk in r.iter_content(chunk_size=1024): if chunk: tmpfile.write(chunk) assert zipfile.is_zipfile(tmpfile), '不是zip文件' zf = zipfile.ZipFile(tmpfile) for name in zf.namelist(): # list e.g. ['Brave Browser.url', 'Express VPN.url', 'ssl.txt', 'What is my IP.url'] if callable(callback): # zf.read(name) #读取 if callback(name, zf) is False:# 函数返回false 会终止下一个文件的读取 break ### 例子 def cb(filename,context): if filename.endswith('.txt'): print(context.read(filename).decode('utf-8')) # print( context.read(filename)) return False #终止下一个文件的读取 read_file_for_zip('https://cdn-01.openload.cc/S9Y7m488n8/22c3c58b-1571037628/ssl_proxies.zip',cb)

具体使用见上面例子
兼容大文件url的下载处理

p.s.
在线压缩包读取:
https://extract.me/cn/

参考:

http://www.liujiangblog.com/course/Python/62

https://docs.Python.org/2/library/tempfile.html

macOS使用TeamViewer ID避免连接失败

作者:matrix 发布时间:2019年2月27日星期三 分类:零零星星

TeamViewer使用频繁或者被检测到某些ip段中会被标记为商业行为的使用,也就会要求付费。对应的TeamViewer ID也就会被要求进行商业授权许可。
图片4393-macOS切换TeamViewer ID避免连接失败的商业许可
本来之前使用都是正常,今天给朋友远程执行脚本命令,我连接控制了半分钟就提示断开连接「超时后连接被阻断」,然后等待时间过后就提示了监测到「商业行为」。网上说TeamViewer修改了商业行为的判定导致客户大量流失,TeamViewer目前报价对于单用户单连接/年需要¥2500,有点贵啊。期间尝试过mac端的向日葵远程控制端和系统自带的屏幕共享,前者只能查看不能操作,后者完全连接不上亦或使用内网VNC地址成功,使用apple id就连接失败,mac端qq就根本没这个功能。

方法0

使用其他工具:
https://anydesk.com/zhs
https://sunlogin.oray.com/zh_CN/download
http://www.xt800.cn/download

参考:https://www.appinn.com/alternative-teamviewer/

有人建议使用俄罗斯版本:https://www.teamviewer.com/ru/

方法1

申述TeamViewerID为个人用户使用,最快7天内解决申请。
https://www.teamviewer.com/en/support/commercial-use-suspected/

2019年3月11日 22:36收到邮件“Your TeamViewer ID has been reset to free”

图片4452-macOS使用TeamViewer ID避免连接失败

方法2

TeamViewer会把ID进行标记,所以换一个新的ID就可以使用了。
使用虚拟机VirtualBox来使用TeamViewer,如果被检测有商业行为就给虚拟机重新初始化MAC地址。这样就可以切换新ID

方法3

使用脚本修改,切换TeamViewer新ID

感谢@zhovner的一键切换脚本TeamViewer ID Changer for MAC OS解决帮了大忙:

测试版本:TeamViewer for macOs V14.0.13880版本
python 2.7.10测试运行

#!/usr/bin/env python 

#coding:utf-8
import sys
import os
import glob
import platform
import re
import random
import string

print('''
--------------------------------
TeamViewer ID Changer for MAC OS
--------------------------------
''')

if platform.system() != 'Darwin':
    print('This script can be run only on MAC OS.')
    sys.exit();

if os.geteuid() != 0:
    print('This script must be run form root.')
    sys.exit();

if os.environ.has_key('SUDO_USER'):
    USERNAME = os.environ['SUDO_USER']
    if USERNAME == 'root':
       print('Can not find user name. Run this script via sudo from regular user')
       sys.exit();
else:
    print('Can not find user name. Run this script via sudo from regular user')
    sys.exit();

HOMEDIRLIB = '/Users/' + USERNAME  + '/library/preferences/'
GLOBALLIB  =  '/library/preferences/'

CONFIGS = []

# Find config files

def listdir_fullpath(d):
    return [os.path.join(d, f) for f in os.listdir(d)]

for file in listdir_fullpath(HOMEDIRLIB):
    if 'teamviewer'.lower() in file.lower():
        CONFIGS.append(file)

if not CONFIGS:
    print ('''
There is no TemViewer configs found.
Maybe you have deleted it manualy or never run TeamViewer after installation.
Nothing to delete.
''')
# Delete config files
else:
    print("Configs found:\n")
    for file in CONFIGS:
        print file

    print('''
This files will be DELETED permanently.
All TeamViewer settings will be lost
''')
    raw_input("Press Enter to continue or CTR+C to abort...")

    for file in CONFIGS:
        try:
            os.remove(file)
        except:
            print("Cannot delete config files. Permission denied?")
            sys.exit();
    print("Done.")

# Find binaryes

TMBINARYES = [
'/Applications/TeamViewer.app/Contents/MacOS/TeamViewer',
'/Applications/TeamViewer.app/Contents/MacOS/TeamViewer_Service',
'/Applications/TeamViewer.app/Contents/Helpers/TeamViewer_Desktop',
]

for file in TMBINARYES:
    if os.path.exists(file):
        pass
    else:
        print("File not found: " + file)
        print ("Install TeamViewer correctly")
        sys.exit();

# Patch files

def idpatch(fpath,platf,serial):
    file = open(fpath, 'r+b')
    binary = file.read()
    PlatformPattern = "IOPlatformExpert.{6}"
    SerialPattern =  "IOPlatformSerialNumber%s%s%sUUID"

    binary = re.sub(PlatformPattern, platf, binary)
    binary = re.sub(SerialPattern % (chr(0), "[0-9a-zA-Z]{8,8}", chr(0)), SerialPattern%(chr(0), serial, chr(0)), binary)

    file = open(fpath,'wb').write(binary)
    return True

def random_generator(size=8, chars=string.ascii_uppercase + string.digits):
    return ''.join(random.choice(chars) for _ in range(size))

RANDOMSERIAL = random_generator()
RANDOMPLATFORM = "IOPlatformExpert" + random_generator(6)


for file in TMBINARYES:
        try:
            idpatch(file,RANDOMPLATFORM,RANDOMSERIAL)
        except:
            print "Error: can not patch file " + file
            print "Wrong version?"
            sys.exit();

print "PlatformDevice: " + RANDOMPLATFORM
print "PlatformSerial: " + RANDOMSERIAL

print('''
ID changed sucessfully.
!!! Logout User Or Restart computer before using TeamViewer !!!!
''')

脚本执行成功会显示ID changed sucessfully.之后重启电脑或者注销用户启用就好了。之后打开teamViewer会发现是新的ID。

主要兼容11,12,和14.0版本

14.0.13880 mac版本:https://www.malavida.com/en/soft/teamviewer/mac/
14.0.13488 windows:https://www.filepuma.com/download/teamviewer_14.0.13488-20699/
12.0.72647 mac版本:https://teamviewer.en.uptodown.com/mac/download/1510547

P.S. 4月份发现脚本无法切换ID,软件会提示商业使用,但是使用上依然没问题

PEACE~

参考:
https://gist.github.com/zhovner/b1d72f3465c46e7b58a4ea42d625c3e8
https://community.spiceworks.com/topic/1151930-teamviewer-5-minute-timelimit

Python版PHP内置的MD5()函数

作者:matrix 发布时间:2014年9月1日星期一 分类:Python

初玩Python很不习惯那个md5函数。还好有人分享了相关代码,非常感谢。

import hashlib
def md5 (s, raw_output = False):
    res = hashlib.md5 (s)
    if raw_output:
        return res.digest ()
    return res.hexdigest ()

如果是Python2.5 :

# Python 2.5+
import hashlib
hashlib.md5("welcome").hexdigest()
# pre-2.5, removed in Python 3
import md5
md5.md5("welcome").hexdigest()

参考:Python 实现PHP内置MD5函数方法