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

"""

Created on 2018-09-06 14:21

---------

@summary: 工具

---------

@author: Boris

@email: boris@bzkj.tech

"""

import calendar

import codecs

import configparser # 读配置文件的

import datetime

import functools

import hashlib

import html

import json

import os

import pickle

import random

import re

import socket

import ssl

import string

import sys

import time

import traceback

import urllib

import urllib.parse

import uuid

import weakref

from hashlib import md5

from pprint import pformat

from pprint import pprint

from urllib import request

from urllib.parse import urljoin

import execjs # pip install PyExecJS

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.utils.email_sender import EmailSender

from feapder.utils.log import log

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:

ip, port = setting.REDISDB_IP_PORTS.split(":")

redisdb = redis.Redis(

host=ip,

port=port,

db=setting.REDISDB_DB,

password=setting.REDISDB_USER_PASS,

decode_responses=True,

) # redis默认端口是6379

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]

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

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

# @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_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):

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

for param in params:

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

if key == key_value[0]:

return key_value[1]

return None

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("=")

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

# mac上不好使

# def get_domain(url):

# domain = ''

# try:

# domain = get_tld(url)

# except Exception as e:

# log.debug(e)

# return domain

def get_domain(url):

proto, rest = urllib.parse.splittype(url)

domain, rest = urllib.parse.splithost(rest)

return domain

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]

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, except_line_break=False, save_img=False, white_replaced=""):

"""

删除html标签

@param content: html内容

@param except_line_break: 保留p标签

@param save_img: 保留图片

@param white_replaced: 空白符替换

@return:

"""

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

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

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

content = replace_str(

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

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

if except_line_break:

content = content.replace("</p>", "/p")

content = replace_str(content, "<[^p].*?>")

content = content.replace("/p", "</p>")

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

elif save_img:

content = replace_str(content, "(?!<img.+?>)<.+?>") # 替换掉除图片外的其他标签

content = replace_str(content, "(?! +)\s+", "\n") # 保留空格

content = content.strip()

else:

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

content = replace_str(content, "\s", white_replaced)

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(json_, indent=4, sort_keys=False):

"""

@summary: 格式化json 用于打印

---------

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

---------

@result: 格式化后的字符串

"""

try:

if isinstance(json_, str):

json_ = get_json(json_)

json_ = json.dumps(

json_, ensure_ascii=False, indent=indent, skipkeys=True, sort_keys=sort_keys

)

except Exception as e:

log.error(e)

json_ = pformat(json_)

return json_

def get_json_value(json_object, key):

"""

@summary:

---------

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

@param key: 建值 如果在多个层级目录下 可写 key1.key2 如{'key1':{'key2':3}}

---------

@result: 返回对应的值，如果没有，返回''

"""

current_key = ""

value = ""

try:

json_object = (

isinstance(json_object, str) and get_json(json_object) or json_object

)

current_key = key.split(".")[0]

value = json_object[current_key]

key = key[key.find(".") + 1 :]

except Exception as e:

return value

if key == current_key:

return value

else:

return get_json_value(value, key)

def get_all_keys(datas, depth=None, current_depth=0):

"""

@summary: 获取json李所有的key

---------

@param datas: dict / list

@param depth: 字典key的层级 默认不限制层级 层级从1开始

@param current_depth: 字典key的当前层级 不用传参

---------

@result: 返回json所有的key

"""

keys = []

if depth and current_depth >= depth:

return keys

if isinstance(datas, list):

for data in datas:

keys.extend(get_all_keys(data, depth, current_depth=current_depth + 1))

elif isinstance(datas, dict):

for key, value in datas.items():

keys.append(key)

if isinstance(value, dict):

keys.extend(get_all_keys(value, depth, current_depth=current_depth + 1))

return keys

def to_chinese(unicode_str):

format_str = json.loads('{"chinese":"%s"}' % unicode_str)

return format_str["chinese"]

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

def replace_str(source_str, regex, replace_str=""):

"""

@summary: 替换字符串

---------

@param source_str: 原字符串

@param regex: 正则

@param replace_str: 用什么来替换 默认为''

---------

@result: 返回替换后的字符串

"""

str_info = re.compile(regex)

return str_info.sub(replace_str, source_str)

def del_redundant_blank_character(text):

"""

删除冗余的空白符， 只保留一个

:param text:

:return:

"""

return re.sub("\s+", " ", text)

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

def get_conf_value(config_file, section, key):

cp = configparser.ConfigParser(allow_no_value=True)

with codecs.open(config_file, "r", encoding="utf-8") as f:

cp.read_file(f)

return cp.get(section, key)

def mkdir(path):

try:

os.makedirs(path)

except OSError as exc: # Python >2.5

pass

def write_file(filename, content, mode="w", encoding="utf-8"):

"""

@summary: 写文件

---------

@param filename: 文件名（有路径）

@param content: 内容

@param mode: 模式 w/w+ (覆盖/追加)

---------

@result:

"""

directory = os.path.dirname(filename)

mkdir(directory)

with open(filename, mode, encoding=encoding) as file:

file.writelines(content)

def read_file(filename, readlines=False, encoding="utf-8"):

"""

@summary: 读文件

---------

@param filename: 文件名（有路径）

@param readlines: 按行读取 （默认False）

---------

@result: 按行读取返回List，否则返回字符串

"""

content = None

try:

with open(filename, "r", encoding=encoding) as file:

content = file.readlines() if readlines else file.read()

except Exception as e:

log.error(e)

return content

def get_oss_file_list(oss_handler, prefix, date_range_min, date_range_max=None):

"""

获取文件列表

@param prefix: 路径前缀 如 data/car_service_line/yiche/yiche_serial_zongshu_info

@param date_range_min: 时间范围 最小值 日期分隔符为/ 如 2019/03/01 或 2019/03/01/00/00/00

@param date_range_max: 时间范围 最大值 日期分隔符为/ 如 2019/03/01 或 2019/03/01/00/00/00

@return: 每个文件路径 如 html/e_commerce_service_line/alibaba/alibaba_shop_info/2019/03/22/15/53/15/8ca8b9e4-4c77-11e9-9dee-acde48001122.json.snappy

"""

# 计算时间范围

date_range_max = date_range_max or date_range_min

date_format = "/".join(

["%Y", "%m", "%d", "%H", "%M", "%S"][: date_range_min.count("/") + 1]

)

time_interval = [

{"days": 365},

{"days": 31},

{"days": 1},

{"hours": 1},

{"minutes": 1},

{"seconds": 1},

][date_range_min.count("/")]

date_range = get_between_date(

date_range_min, date_range_max, date_format=date_format, **time_interval

)

for date in date_range:

file_folder_path = os.path.join(prefix, date)

objs = oss_handler.list(prefix=file_folder_path)

for obj in objs:

filename = obj.key

yield filename

def is_html(url):

if not url:

return False

try:

content_type = request.urlopen(url).info().get("Content-Type", "")