From 60476e2c1fc94c97f9a018f2ac7df4ad315b0e57 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Mon, 29 Jul 2019 09:59:10 +0800 Subject: [PATCH] update --- appPublic/httpclient.py | 83 +++++++++++++++++++ appPublic/myTE.py | 65 +++++++++++++++ appPublic/tworkers.py | 62 ++++++++++++++ appPublic/worker.py | 42 ++++++++++ appPublic/zmqapi.py | 176 ++++++++++++++++++++++++++++++++++++++++ setup.py | 18 ++-- 6 files changed, 439 insertions(+), 7 deletions(-) create mode 100644 appPublic/httpclient.py create mode 100644 appPublic/myTE.py create mode 100644 appPublic/tworkers.py create mode 100644 appPublic/worker.py create mode 100644 appPublic/zmqapi.py diff --git a/appPublic/httpclient.py b/appPublic/httpclient.py new file mode 100644 index 0000000..69734da --- /dev/null +++ b/appPublic/httpclient.py @@ -0,0 +1,83 @@ +import asyncio +import aiohttp +import re + +RESPONSE_BIN = 0 +RESPONSE_TEXT = 1 +RESPONSE_JSON = 2 +RESPONSE_FILE = 3 + +class HttpClient: + def __init__(self,coding='utf-8'): + self.coding = coding + self.session = None + self.cookies = {} + + async def close(self): + await self.session.close() + + def url2domain(self,url): + parts = url.split('/')[:3] + pre = '/'.join(parts) + return pre + + def setCookie(self,url,cookies): + name = self.url2domain(url) + self.cookies[name] = cookies + + def getCookies(self,url): + name = url2domain(url) + return self.cookies.get(name,None) + + def getsession(self,url): + if self.session is None: + jar = aiohttp.CookieJar(unsafe=True) + self.session = aiohttp.ClientSession(cookie_jar=jar) + return self.session + + async def handleResp(self,url,resp,resp_type): + if resp.cookies is not None: + self.setCookie(url,resp.cookies) + + if resp_type == RESPONSE_BIN: + return await resp.read() + if resp_type == RESPONSE_JSON: + return await resp.json() + # default resp_type == RESPONSE_TEXT: + return await resp.text(self.coding) + + def grapCookie(self,url): + session = self.getsession(url) + domain = self.url2domain(url) + filtered = session.cookie_jar.filter_cookies(domain) + return filtered + print(f'=====domain={domain},cookies={fltered},type={type(fltered)}===') + + async def get(self,url, + params={},headers={},resp_type=RESPONSE_TEXT): + session = self.getsession(url) + resp = await session.get(url,params=params,headers=headers) + if resp.status==200: + return await self.handleResp(url,resp,resp_type) + return None + + async def post(self,url, + data=b'',headers={},cookies=None, + resp_type=RESPONSE_TEXT): + session = self.getsession(url) + resp = await session.post(url,data=data, + headers=headers,cookies=cookies) + if resp.status==200: + return await self.handleResp(url,resp,resp_type) + print(f'**ERROR**{url} response code={resp.status}') + return None + +if __name__ == '__main__': + async def gbaidu(hc): + r = await hc.get('https://www.baidu.com') + print(r) + await hc.close() + loop = asyncio.get_event_loop() + hc = HttpClient() + loop.run_until_complete(gbaidu(hc)) + diff --git a/appPublic/myTE.py b/appPublic/myTE.py new file mode 100644 index 0000000..4f58e3c --- /dev/null +++ b/appPublic/myTE.py @@ -0,0 +1,65 @@ +import os +import sys +import appPublic.myjson as json +from jinja2 import Environment,FileSystemLoader +import codecs +from appPublic.argsConvert import ArgsConvert +from appPublic.dictObject import DictObject +def isNone(obj): + return obj is None + + +class MyTemplateEngine: + def __init__(self,pathList,file_coding='utf-8',out_coding='utf-8'): + self.file_coding = file_coding + self.out_coding = out_coding + loader = FileSystemLoader(pathList, encoding=self.file_coding) + self.env = Environment(loader=loader) + denv={ + 'json':json, + 'hasattr':hasattr, + 'int':int, + 'float':float, + 'str':str, + 'type':type, + 'isNone':isNone, + 'len':len, + 'recordFind':recordFind, + 'render':self.render, + 'renders':self.renders, + 'ArgsConvert':ArgsConvert, + 'renderJsonFile':self.renderJsonFile, + 'ospath':lambda x:os.path.sep.join(x.split(os.altsep)), + 'basename':lambda x:os.path.basename(x), + 'basenameWithoutExt':lambda x:os.path.splitext(os.path.basename(x))[0], + 'extname':lambda x:os.path.splitext(x)[-1], + } + self.env.globals.update(denv) + + def setGlobal(self,dic): + self.env.globals.update(dic) + + def _render(self,template,data): + self._setEnv() + uRet = template.render(**data) + return uRet.encode(self.out_coding) + + def renders(self,tmplstring,data): + def getGlobal(): + return data + self.set('global',getGlobal) + template = self.env.from_string(tmplstring) + return self._render(template,data) + + def render(self,tmplfile,data): + def getGlobal(): + return data + self.set('global',getGlobal) + template = self.env.get_template(tmplfile) + return self._render(template,data) + + def renderJsonFile(self,tmplfile,jsonfile): + f = codecs.open(jsonfile,"r",self.file_coding) + data = json.load(f) + f.close() + return self.render(tmplfile,data) diff --git a/appPublic/tworkers.py b/appPublic/tworkers.py new file mode 100644 index 0000000..4a393c3 --- /dev/null +++ b/appPublic/tworkers.py @@ -0,0 +1,62 @@ +import sys +import threading +from threading import Thread +from queue import Queue, Empty + +class Worker(Thread): + def __init__(self, rqueue, timeout=1): + Thread.__init__(self) + self.timeout = timeout + self.setDaemon(False) + self.r_queue = rqueue + self.start() + + def run(self): + emptyQueue = False + while True: + try: + callable,args,kw = self.r_queue.get(timeout=self.timeout) + if callable is None: + break + callable(*args,**kw) + + except Empty: + time.sleep(1) + + def resulthandler(self,rez): + pass + +class ThreadWorkers: + def __init__(self,num_workers=20): + self.workQueue = Queue() + self.worker_cnt = num_workers + self.workers = [] + self.__createThreadPool(num_workers) + + def __createThreadPool(self,num): + for i in range(num): + thread = Worker(self.workQueue) + self.workers.append(thread) + + def wait_for_complete(self): + for i in range(self.worker_cnt): + self.add_job(None,None,None) + + while len(self.workers): + thread = self.workers.pop() + if thread.isAlive(): + thread.join() + + def add_job(self,callable,args=[],kw={}): + self.workQueue.put([callable,args,kw]) +if __name__ == '__main__': + import requests + def get(url): + x = requests.get(url) + print(x.status_code) + + tw = ThreadWorkers() + for i in range(10000): + tw.add_job(get,['http://www.baidu.com']) + tw.wait_for_complete() + print('finished') diff --git a/appPublic/worker.py b/appPublic/worker.py new file mode 100644 index 0000000..f6a955c --- /dev/null +++ b/appPublic/worker.py @@ -0,0 +1,42 @@ + +import asyncio +from functools import wraps + +def asyncCall(func): + @wraps(func) + def wraped_func(*args,**kw): + task = asyncio.ensure_future(func(*args,**kw)) + asyncio.gather(task) + return wraped_func + +class Worker: + def __init__(self,max=50): + self.semaphore = asyncio.Semaphore(max) + + async def __call__(self,callee,*args,**kw): + async with self.semaphore: + return await callee(*args,**kw) + + async def run(self,cmd): + async with self.semaphore: + proc = await asyncio.create_subprocess_shell(cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE) + + stdout, stderr = await proc.comunicate() + return stdout, stderr + +if __name__ == '__main__': + async def hello(cnt,greeting): + await asyncio.sleep(1) + print(cnt,greeting) + + + async def run(): + w = Worker() + tasks = [ w(hello,i,'hello world') for i in range(1000) ] + await asyncio.wait(tasks) + + loop = asyncio.get_event_loop() + loop.run_until_complete(run()) + diff --git a/appPublic/zmqapi.py b/appPublic/zmqapi.py new file mode 100644 index 0000000..bc643a9 --- /dev/null +++ b/appPublic/zmqapi.py @@ -0,0 +1,176 @@ + +import asyncio +from collections.abc import Coroutine +# from asyncio.coroutines import iscoroutine + +import zmq +import zmq.asyncio +import json + +class Publisher: + def __init__(self,port,coding='utf-8',msgid=1000): + self.port = port + self.socket = None + self.coding = coding + self.msgid = msgid + context = zmq.asyncio.Context() + self.socket = context.socket(zmq.PUB) + self.socket.bind('tcp://*:%d' % self.port) + + async def publish(self,msg,msgtype='text',msgid=-1): + print(msg,msgtype) + if msgid == -1: + msgid = self.msgid + if msgtype != 'text': + msg = json.dumps(msg) + msgtype = 'json' + s = '%d %s %s' % (msgid,msgtype,msg) + print(s,msgtype,msgid) + b = s.encode(self.coding) + await self.socket.send(b) + + def __del__(self): + self.socket.close() + +class Subscriber: + def __init__(self,host,ports,msgid,coding='utf-8'): + self.host = host + self.ports = ports + self.msgid = msgid + self.coding = coding + context = zmq.asyncio.Context() + self.socket = context.socket(zmq.SUB) + f = b'%d' % self.msgid + self.socket.setsockopt(zmq.SUBSCRIBE, f) + for p in self.ports: + self.socket.connect("tcp://%s:%d" % (self.host,p)) + + def addPort(self,port): + self.socket.connect("tcp://%s:%d" % (self.host,port)) + #f = b'%d' % self.msgid + #self.socket.setsockopt(zmq.SUBSCRIBE, f) + + async def subscribe(self): + ret = await self.socket.recv() + ret = ret.decode(self.coding) + msgid, msgtype, body = ret.split(' ',2) + print('msgid=',msgid,'msgtype=',msgtype,'body=',body) + if msgtype == 'json': + return json.loads(body) + return body + + def __del__(self): + self.socket.close() + +class RRServer: + """ + a request / response mode server + """ + def __init__(self,port,handler=None): + self.port = port + self.handler = handler + print(type(self.handler)) + + async def run(self): + running = True + context = zmq.asyncio.Context() + socket = context.socket(zmq.REP) + socket.bind('tcp://*:%s' % self.port) + while running: + rmsg = await socket.recv() + wmsg = rmsg + if self.handler is not None: + wmsg = self.handler(rmsg) + if isinstance(wmsg,Coroutine): + wmsg = await wmsg + await socket.send(wmsg) + socket.close() + +class RRClient: + """ + a request / response mode client + """ + def __init__(self,host,port): + self.host = host + self.port = port + context = zmq.asyncio.Context() + self.socket = context.socket(zmq.REQ) + self.socket.connect('tcp://%s:%d' % (self.host,self.port)) + + async def request(self,msg): + await self.socket.send(msg) + return await self.socket.recv() + + +class PPPusher: + """ + pusher of Push / Pull mode + """ + def __init__(self,host,port): + self.host = host + self.port = port + context = zmq.asyncio.Context() + self.socket = context.socket(zmq.PUSH) + self.socket.bind('tcp://%s:%d' % (self.host,self.port)) + + async def push(self,msg): + await self.socket.send(msg) + +class PPPuller: + """ + puller of Push / Pull mode + """ + def __init__(self,host,port,handler=None): + self.host = host + self.port = port + self.handler = handler + + async def run(self): + self.running = True + context = zmq.asyncio.Context() + socket = context.socket(zmq.PULL) + socket.bind('tcp://%s:%d' % (self.host,self.port)) + while self.running: + msg = await self.socket.recv() + if self.handler is not None: + x = self.handler(msg) + if isinstance(x,Coroutine): + await x + +class PairClient: + """ + client of Pair mode + """ + def __init__(self,host,port): + self.host = host + self.port = port + context = zmq.asyncio.Context() + self.socket = context.socket(zmq.PAIR) + self.socket.bind('tcp://%s:%d' % (self.host,self.port)) + + async def request(self,msg): + await self.socket.send(msg) + return await self.socket.recv() + +class PairServer: + """ + server of Pair mode + """ + def __init__(self,port,handler=None): + self.port = port + self.handler = handler + self.running = True + + async def run(self): + self.running = True + context = zmq.asyncio.Context() + socket = context.socket(zmq.PAIR) + socket.bind('tcp://*:%d' % self.port) + while self.running: + msg = await socket.recv() + ret = msg + if self.handler is not None: + ret = self.handler() + if isinstance(ret,Coroutine): + ret = await ret + await socket.send(ret) diff --git a/setup.py b/setup.py index 691a519..1018bcf 100755 --- a/setup.py +++ b/setup.py @@ -9,14 +9,18 @@ from setuptools import setup, find_packages # python setup.py bdist_egg generate a egg file # Release information about eway -version = "5.0.16" +version = "5.0.18" +name = "appPublic" description = "appPublic" author = "yumoqing" -email = "yumoqing@icloud.com" +email = "yumoqing@gmail.com" packages=find_packages() package_data = {} +with open("README.md", "r") as fh: + long_description = fh.read() + setup( name="appPublic", version=version, @@ -32,12 +36,12 @@ setup( package_data=package_data, keywords = [ ], + url="https://github.com/yumoqing/appPublic", + long_description=long_description, + long_description_content_type="text/markdown", classifiers = [ - 'Development Status :: 3 - Alpha', 'Operating System :: OS Independent', - 'Programming Language :: Python', - 'Topic :: Software Development :: Libraries :: Python Modules', - 'Framework :: utils', + 'Programming Language :: Python :: 3', + 'License :: OSI Approved :: MIT License', ], - platforms= 'any' )