# -*- coding: utf-8 -*-

"""

Created on 2018-09-06 14:21

---------

@summary: 工具

---------

@author: Boris

@email: boris_liu@foxmail.com

"""

import asyncio

import base64

import calendar

import codecs

import configparser # 读配置文件的

import datetime

import functools

import hashlib

import hmac

import html

import importlib

import json

import os

import pickle

import random

import re

import signal

import socket

import ssl

import string

import sys

import time

import traceback

import urllib

import urllib.parse

import uuid

import weakref

from functools import partial, wraps

from hashlib import md5

from pprint import pformat

from pprint import pprint

from urllib import request

from urllib.parse import urljoin

import redis

import requests

import six

from requests.cookies import RequestsCookieJar

from w3lib.url import canonicalize_url as _canonicalize_url

import feapder.setting as setting

from feapder.db.redisdb import RedisDB

from feapder.utils.email_sender import EmailSender

from feapder.utils.log import log

try:

import execjs # pip install PyExecJS

except Exception as e:

pass

os.environ["EXECJS_RUNTIME"] = "Node" # 设置使用node执行js

# 全局取消ssl证书验证

ssl._create_default_https_context = ssl._create_unverified_context

TIME_OUT = 30

TIMER_TIME = 5

redisdb = None

def get_redisdb():

global redisdb

if not redisdb:

redisdb = RedisDB()

return redisdb

# 装饰器

class Singleton(object):

def __init__(self, cls):

self._cls = cls

self._instance = {}

def __call__(self, *args, **kwargs):

if self._cls not in self._instance:

self._instance[self._cls] = self._cls(*args, **kwargs)

return self._instance[self._cls]

class LazyProperty:

"""

属性延时初始化，且只初始化一次

"""

def __init__(self, func):

self.func = func

def __get__(self, instance, owner):

if instance is None:

return self

else:

value = self.func(instance)

setattr(instance, self.func.__name__, value)

return value

def log_function_time(func):

try:

@functools.wraps(func) # 将函数的原来属性付给新函数

def calculate_time(*args, **kw):

began_time = time.time()

callfunc = func(*args, **kw)

end_time = time.time()

log.debug(func.__name__ + " run time = " + str(end_time - began_time))

return callfunc

return calculate_time

except:

log.debug("求取时间无效 因为函数参数不符")

return func

def run_safe_model(module_name):

def inner_run_safe_model(func):

try:

@functools.wraps(func) # 将函数的原来属性付给新函数

def run_func(*args, **kw):

callfunc = None

try:

callfunc = func(*args, **kw)

except Exception as e:

log.error(module_name + ": " + func.__name__ + " - " + str(e))

traceback.print_exc()

return callfunc

return run_func

except Exception as e:

log.error(module_name + ": " + func.__name__ + " - " + str(e))

traceback.print_exc()

return func

return inner_run_safe_model

def memoizemethod_noargs(method):

"""Decorator to cache the result of a method (without arguments) using a

weak reference to its object

"""

cache = weakref.WeakKeyDictionary()

@functools.wraps(method)

def new_method(self, *args, **kwargs):

if self not in cache:

cache[self] = method(self, *args, **kwargs)

return cache[self]

return new_method

def retry(retry_times=3, interval=0):

"""

普通函数的重试装饰器

Args:

retry_times: 重试次数

interval: 每次重试之间的间隔

Returns:

"""

def _retry(func):

@functools.wraps(func) # 将函数的原来属性付给新函数

def wapper(*args, **kwargs):

for i in range(retry_times):

try:

return func(*args, **kwargs)

except Exception as e:

log.error(

"函数 {} 执行失败 重试 {} 次. error {}".format(func.__name__, i + 1, e)

)

time.sleep(interval)

if i + 1 >= retry_times:

raise e

return wapper

return _retry

def retry_asyncio(retry_times=3, interval=0):

"""

协程的重试装饰器

Args:

retry_times: 重试次数

interval: 每次重试之间的间隔

Returns:

"""

def _retry(func):

@functools.wraps(func) # 将函数的原来属性付给新函数

async def wapper(*args, **kwargs):

for i in range(retry_times):

try:

return await func(*args, **kwargs)

except Exception as e:

log.error(

"函数 {} 执行失败 重试 {} 次. error {}".format(func.__name__, i + 1, e)

)

await asyncio.sleep(interval)

if i + 1 >= retry_times:

raise e

return wapper

return _retry

def func_timeout(timeout):

"""

函数运行时间限制装饰器

注: 不支持window

Args:

timeout: 超时的时间

Eg:

@set_timeout(3)

def test():

...

Returns:

"""

def wapper(func):

def handle(

signum, frame

): # 收到信号 SIGALRM 后的回调函数，第一个参数是信号的数字，第二个参数是the interrupted stack frame.

raise TimeoutError

def new_method(*args, **kwargs):

signal.signal(signal.SIGALRM, handle) # 设置信号和回调函数

signal.alarm(timeout) # 设置 timeout 秒的闹钟

r = func(*args, **kwargs)

signal.alarm(0) # 关闭闹钟

return r

return new_method

return wapper

########################【网页解析相关】###############################

# @log_function_time

def get_html_by_requests(

url, headers=None, code="utf-8", data=None, proxies={}, with_response=False

):

html = ""

r = None

try:

if data:

r = requests.post(

url, headers=headers, timeout=TIME_OUT, data=data, proxies=proxies

)

else:

r = requests.get(url, headers=headers, timeout=TIME_OUT, proxies=proxies)

if code:

r.encoding = code

html = r.text

except Exception as e:

log.error(e)

finally:

r and r.close()

if with_response:

return html, r

else:

return html

def get_json_by_requests(

url,

params=None,

headers=None,

data=None,

proxies={},

with_response=False,

cookies=None,

):

json = {}

response = None

try:

# response = requests.get(url, params = params)

if data:

response = requests.post(

url,

headers=headers,

data=data,

params=params,

timeout=TIME_OUT,

proxies=proxies,

cookies=cookies,

)

else:

response = requests.get(

url,

headers=headers,

params=params,

timeout=TIME_OUT,

proxies=proxies,

cookies=cookies,

)

response.encoding = "utf-8"

json = response.json()

except Exception as e:

log.error(e)

finally:

response and response.close()

if with_response:

return json, response

else:

return json

def get_cookies(response):

cookies = requests.utils.dict_from_cookiejar(response.cookies)

return cookies

def get_cookies_from_str(cookie_str):

"""

>>> get_cookies_from_str("key=value; key2=value2; key3=; key4=; ")

{'key': 'value', 'key2': 'value2', 'key3': '', 'key4': ''}

Args:

cookie_str: key=value; key2=value2; key3=; key4=

Returns:

"""

cookies = {}

for cookie in cookie_str.split(";"):

cookie = cookie.strip()

if not cookie:

continue

key, value = cookie.split("=", 1)

key = key.strip()

value = value.strip()

cookies[key] = value

return cookies

def get_cookies_jar(cookies):

"""

@summary: 适用于selenium生成的cookies转requests的cookies

requests.get(xxx, cookies=jar)

参考：https://www.cnblogs.com/small-bud/p/9064674.html

---------

@param cookies: [{},{}]

---------

@result: cookie jar

"""

cookie_jar = RequestsCookieJar()

for cookie in cookies:

cookie_jar.set(cookie["name"], cookie["value"])

return cookie_jar

def get_cookies_from_selenium_cookie(cookies):

"""

@summary: 适用于selenium生成的cookies转requests的cookies

requests.get(xxx, cookies=jar)

参考：https://www.cnblogs.com/small-bud/p/9064674.html

---------

@param cookies: [{},{}]

---------

@result: cookie jar

"""

cookie_dict = {}

for cookie in cookies:

if cookie.get("name"):

cookie_dict[cookie["name"]] = cookie["value"]

return cookie_dict

def cookiesjar2str(cookies):

str_cookie = ""

for k, v in requests.utils.dict_from_cookiejar(cookies).items():

str_cookie += k

str_cookie += "="

str_cookie += v

str_cookie += "; "

return str_cookie

def cookies2str(cookies):

str_cookie = ""

for k, v in cookies.items():

str_cookie += k

str_cookie += "="

str_cookie += v

str_cookie += "; "

return str_cookie

def get_urls(

html,

stop_urls=(

"javascript",

"+",

".css",

".js",

".rar",

".xls",

".exe",

".apk",

".doc",

".jpg",

".png",

".flv",

".mp4",

),

):

# 不匹配javascript、 +、 # 这样的url

regex = r'<a.*?href.*?=.*?["|\'](.*?)["|\']'

urls = get_info(html, regex)

urls = sorted(set(urls), key=urls.index)

if stop_urls:

stop_urls = isinstance(stop_urls, str) and [stop_urls] or stop_urls

use_urls = []

for url in urls:

for stop_url in stop_urls:

if stop_url in url:

break

else:

use_urls.append(url)

urls = use_urls

return urls

def get_full_url(root_url, sub_url):

"""

@summary: 得到完整的ur

---------

@param root_url: 根url （网页的url）

@param sub_url: 子url （带有相对路径的 可以拼接成完整的）

---------

@result: 返回完整的url

"""

return urljoin(root_url, sub_url)

def joint_url(url, params):

# param_str = "?"

# for key, value in params.items():

# value = isinstance(value, str) and value or str(value)

# param_str += key + "=" + value + "&"

#

# return url + param_str[:-1]

if not params:

return url

params = urlencode(params)

separator = "?" if "?" not in url else "&"

return url + separator + params

def canonicalize_url(url):

"""

url 归一化 会参数排序 及去掉锚点

"""

return _canonicalize_url(url)

def get_url_md5(url):

url = canonicalize_url(url)

url = re.sub("^http://", "https://", url)

return get_md5(url)

def fit_url(urls, identis):

identis = isinstance(identis, str) and [identis] or identis

fit_urls = []

for link in urls:

for identi in identis:

if identi in link:

fit_urls.append(link)

return list(set(fit_urls))

def get_param(url, key):

pattern = r"(?:[?&])" + re.escape(key) + r"=([^&]+)"

match = re.search(pattern, url)

if match:

return match.group(1)

return None

def get_all_params(url):

"""

>>> get_all_params("https://www.baidu.com/s?wd=feapder")

{'wd': 'feapder'}

"""

params_json = {}

params = url.split("?", 1)[-1].split("&")

for param in params:

key_value = param.split("=", 1)

if len(key_value) == 2:

params_json[key_value[0]] = unquote_url(key_value[1])

else:

params_json[key_value[0]] = ""

return params_json

def parse_url_params(url):

"""

解析url参数

:param url:

:return:

>>> parse_url_params("https://www.baidu.com/s?wd=%E4%BD%A0%E5%A5%BD")

('https://www.baidu.com/s', {'wd': '你好'})

>>> parse_url_params("wd=%E4%BD%A0%E5%A5%BD")

('', {'wd': '你好'})

>>> parse_url_params("https://www.baidu.com/s?wd=%E4%BD%A0%E5%A5%BD&pn=10")

('https://www.baidu.com/s', {'wd': '你好', 'pn': '10'})

>>> parse_url_params("wd=%E4%BD%A0%E5%A5%BD&pn=10")

('', {'wd': '你好', 'pn': '10'})

>>> parse_url_params("https://www.baidu.com")

('https://www.baidu.com', {})

>>> parse_url_params("https://www.spidertools.cn/#/")

('https://www.spidertools.cn/#/', {})

"""

root_url = ""

params = {}

if "?" not in url:

if re.search("[&=]", url) and not re.search("/", url):

# 只有参数

params = get_all_params(url)

else:

root_url = url

else:

root_url = url.split("?", 1)[0]

params = get_all_params(url)

return root_url, params

def urlencode(params):

"""

字典类型的参数转为字符串

@param params:

{

'a': 1,

'b': 2

}

@return: a=1&b=2

"""

return urllib.parse.urlencode(params)

def urldecode(url):

"""

将字符串类型的参数转为json

@param url: xxx?a=1&b=2

@return:

{

'a': 1,

'b': 2

}

"""

params_json = {}

params = url.split("?")[-1].split("&")

for param in params:

key, value = param.split("=", 1)

params_json[key] = unquote_url(value)

return params_json

def unquote_url(url, encoding="utf-8"):

"""

@summary: 将url解码

---------

@param url:

---------

@result:

"""

return urllib.parse.unquote(url, encoding=encoding)

def quote_url(url, encoding="utf-8"):

"""

@summary: 将url编码 编码意思http://www.w3school.com.cn/tags/html_ref_urlencode.html

---------

@param url:

---------

@result:

"""

return urllib.parse.quote(url, safe="%;/?:@&=+$,", encoding=encoding)

def quote_chinese_word(text, encoding="utf-8"):

def quote_chinese_word_func(text):

chinese_word = text.group(0)

return urllib.parse.quote(chinese_word, encoding=encoding)

return re.sub("([\u4e00-\u9fa5]+)", quote_chinese_word_func, text, flags=re.S)

def unescape(str):

"""

反转译

"""

return html.unescape(str)

def excape(str):

"""

转译

"""

return html.escape(str)

_regexs = {}

# @log_function_time

def get_info(html, regexs, allow_repeat=True, fetch_one=False, split=None):

regexs = isinstance(regexs, str) and [regexs] or regexs

infos = []

for regex in regexs:

if regex == "":

continue

if regex not in _regexs.keys():

_regexs[regex] = re.compile(regex, re.S)

if fetch_one:

infos = _regexs[regex].search(html)

if infos:

infos = infos.groups()

else:

continue

else:

infos = _regexs[regex].findall(str(html))

if len(infos) > 0:

# print(regex)

break

if fetch_one:

infos = infos if infos else ("",)

return infos if len(infos) > 1 else infos[0]

else:

infos = allow_repeat and infos or sorted(set(infos), key=infos.index)

infos = split.join(infos) if split else infos

return infos

def table_json(table, save_one_blank=True):

"""

将表格转为json 适应于 key：value 在一行类的表格

@param table: 使用selector封装后的具有xpath的selector

@param save_one_blank: 保留一个空白符

@return:

"""

data = {}

trs = table.xpath(".//tr")

for tr in trs:

tds = tr.xpath("./td|./th")

for i in range(0, len(tds), 2):

if i + 1 > len(tds) - 1:

break

key = tds[i].xpath("string(.)").extract_first(default="").strip()

value = tds[i + 1].xpath("string(.)").extract_first(default="").strip()

value = replace_str(value, "[\f\n\r\t\v]", "")

value = replace_str(value, " +", " " if save_one_blank else "")

if key:

data[key] = value

return data

def get_table_row_data(table):

"""

获取表格里每一行数据

@param table: 使用selector封装后的具有xpath的selector

@return: [[],[]..]

"""

datas = []

rows = table.xpath(".//tr")

for row in rows:

cols = row.xpath("./td|./th")

row_datas = []

for col in cols:

data = col.xpath("string(.)").extract_first(default="").strip()

row_datas.append(data)

datas.append(row_datas)

return datas

def rows2json(rows, keys=None):

"""

将行数据转为json

@param rows: 每一行的数据

@param keys: json的key，空时将rows的第一行作为key

@return:

"""

data_start_pos = 0 if keys else 1

datas = []

keys = keys or rows[0]

for values in rows[data_start_pos:]:

datas.append(dict(zip(keys, values)))

return datas

def get_form_data(form):

"""

提取form中提交的数据

:param form: 使用selector封装后的具有xpath的selector

:return:

"""

data = {}

inputs = form.xpath(".//input")

for input in inputs:

name = input.xpath("./@name").extract_first()

value = input.xpath("./@value").extract_first()

if name:

data[name] = value

return data

def get_domain(url):

return urllib.parse.urlparse(url).netloc

def get_index_url(url):

return "/".join(url.split("/")[:3])

def get_ip(domain):

ip = socket.getaddrinfo(domain, "http")[0][4][0]

return ip

def get_localhost_ip():

"""

利用 UDP 协议来实现的，生成一个UDP包，把自己的 IP 放如到 UDP 协议头中，然后从UDP包中获取本机的IP。

这个方法并不会真实的向外部发包，所以用抓包工具是看不到的

:return:

"""

s = None

try:

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

s.connect(("8.8.8.8", 80))

ip = s.getsockname()[0]

except:

ip = ""

finally:

if s:

s.close()

return ip

def ip_to_num(ip):

import struct

ip_num = socket.ntohl(struct.unpack("I", socket.inet_aton(str(ip)))[0])

return ip_num

def is_valid_proxy(proxy, check_url=None):

"""

检验代理是否有效

@param proxy: xxx.xxx.xxx:xxx

@param check_url: 利用目标网站检查，目标网站url。默认为None， 使用代理服务器的socket检查, 但不能排除Connection closed by foreign host

@return: True / False

"""

is_valid = False

if check_url:

proxies = {"http": f"http://{proxy}", "https": f"https://{proxy}"}

headers = {

"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.100 Safari/537.36"

}

response = None

try:

response = requests.get(

check_url, headers=headers, proxies=proxies, stream=True, timeout=20

)

is_valid = True

except Exception as e:

log.error("check proxy failed: {} {}".format(e, proxy))

finally:

if response:

response.close()

else:

ip, port = proxy.split(":")

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sk:

sk.settimeout(7)

try:

sk.connect((ip, int(port))) # 检查代理服务器是否开着

is_valid = True

except Exception as e:

log.error("check proxy failed: {} {}:{}".format(e, ip, port))

return is_valid

def is_valid_url(url):

"""

验证url是否合法

:param url:

:return:

"""

if re.match(r"(^https?:/{2}\w.+$)|(ftp://)", url):

return True

else:

return False

def get_text(soup, *args):

try:

return soup.get_text()

except Exception as e:

log.error(e)

return ""

def del_html_tag(content, save_line_break=True, save_p=False, save_img=False):

"""

删除html标签

@param content: html内容

@param save_p: 保留p标签

@param save_img: 保留图片标签

@param save_line_break: 保留\n换行

@return:

"""

if not content:

return content

# js

content = re.sub("(?i)<script(.|\n)*?</script>", "", content) # (?)忽略大小写

# css

content = re.sub("(?i)<style(.|\n)*?</style>", "", content) # (?)忽略大小写

# 注释

content = re.sub("<!--(.|\n)*?-->", "", content)

# 干掉&nbsp;等无用的字符 但&xxx= 这种表示参数的除外

content = re.sub("(?!&[a-z]+=)&[a-z]+;?", "", content)

if save_p and save_img:

content = re.sub("<(?!(p[ >]|/p>|img ))(.|\n)+?>", "", content)

elif save_p:

content = re.sub("<(?!(p[ >]|/p>))(.|\n)+?>", "", content)

elif save_img:

content = re.sub("<(?!img )(.|\n)+?>", "", content)

elif save_line_break:

content = re.sub("<(?!/p>)(.|\n)+?>", "", content)

content = re.sub("</p>", "\n", content)

else:

content = re.sub("<(.|\n)*?>", "", content)

if save_line_break:

# 把非换行符的空白符替换为一个空格

content = re.sub("[^\S\n]+", " ", content)

# 把多个换行符替换为一个换行符 如\n\n\n 或 \n \n \n 替换为\n

content = re.sub("(\n ?)+", "\n", content)

else:

content = re.sub("\s+", " ", content)

content = content.strip()

return content

def del_html_js_css(content):

content = replace_str(content, "(?i)<script(.|\n)*?</script>") # (?)忽略大小写

content = replace_str(content, "(?i)<style(.|\n)*?</style>")

content = replace_str(content, "<!--(.|\n)*?-->")

return content

def is_have_chinese(content):

regex = "[\u4e00-\u9fa5]+"

chinese_word = get_info(content, regex)

return chinese_word and True or False

def is_have_english(content):

regex = "[a-zA-Z]+"

english_words = get_info(content, regex)

return english_words and True or False

def get_chinese_word(content):

regex = "[\u4e00-\u9fa5]+"

chinese_word = get_info(content, regex)

return chinese_word

def get_english_words(content):

regex = "[a-zA-Z]+"

english_words = get_info(content, regex)

return english_words or ""

##################################################

def get_json(json_str):

"""

@summary: 取json对象

---------

@param json_str: json格式的字符串

---------

@result: 返回json对象

"""

try:

return json.loads(json_str) if json_str else {}

except Exception as e1:

try:

json_str = json_str.strip()

json_str = json_str.replace("'", '"')

keys = get_info(json_str, "(\w+):")

for key in keys:

json_str = json_str.replace(key, '"%s"' % key)

return json.loads(json_str) if json_str else {}

except Exception as e2:

log.error(

"""

e1: %s

format json_str: %s

e2: %s

"""

% (e1, json_str, e2)

)

return {}

def jsonp2json(jsonp):

"""

将jsonp转为json

@param jsonp: jQuery172013600082560040794_1553230569815({})

@return:

"""

try:

return json.loads(re.match(".*?({.*}).*", jsonp, re.S).group(1))

except:

raise ValueError("Invalid Input")

def dumps_json(data, indent=4, sort_keys=False):

"""

@summary: 格式化json 用于打印

---------

@param data: json格式的字符串或json对象

---------

@result: 格式化后的字符串

"""

try:

if isinstance(data, str):