代码
# -*- coding: utf-8 -*-
# @Author : zbz
import asyncio
import random
import ddddocr
import httpx
class AsyncSpider:
def __init__(self, ocr=False):
self.ocr = ddddocr.DdddOcr() if ocr else None
def get_ua(self) -> str:
a = random.randint(55, 62)
c = random.randint(0, 3200)
d = random.randint(0, 140)
os_type = [
'(Windows NT 6.1; WOW64)', '(Windows NT 10.0; WOW64)', '(X11; Linux x86_64)',
'(Macintosh; Intel Mac OS X 10_12_6)'
]
chrome_version = 'Chrome/{}.0.{}.{}'.format(a, c, d)
ua = ' '.join(
['Mozilla/5.0', random.choice(os_type), 'AppleWebKit/537.36',
'(KHTML, like Gecko)', chrome_version, 'Safari/537.36']
)
return ua
def only_ua(self) -> dict:
headers = {
"user-agent": self.get_ua()
}
return headers
async def parse(self, url, method="GET", headers=None, params=None, cookies=None, data=None, json=None,
follow_redirects=True, timeout=3):
if method == "GET":
async with httpx.AsyncClient(proxies=None, verify=False) as cli:
resp = await cli.get(url, headers=headers, params=params, cookies=cookies,
follow_redirects=follow_redirects, timeout=timeout)
return resp
elif method == "POST":
async with httpx.AsyncClient(proxies=None, verify=False) as cli:
resp = await cli.post(url, headers=headers, params=params, cookies=cookies, data=data, json=json,
follow_redirects=follow_redirects, timeout=timeout)
return resp
async def dddd_ident(self, param):
if isinstance(param, str):
resp = await self.parse(param)
result = self.ocr.classification(resp.content).lower()
return result
elif isinstance(param, bytes):
result = self.ocr.classification(param).lower()
return result