# -*- coding: utf-8 -*-
"""
Created on Tue May 15 13:59:43 2018
@author: hanye
"""
import json
import datetime
import random
import time
import re
from typing import Dict, List
import pymysql
import requests
from elasticsearch.exceptions import TransportError
from crawler_sys.framework.redis_interact import feed_url_into_redis
from crawler_sys.framework.redis_interact import redis_path
from crawler_sys.framework.es_ccr_index_defination import es_framework as es_site_crawler
from crawler_sys.framework.es_ccr_index_defination import index_url_register
from crawler_sys.framework.es_ccr_index_defination import doc_type_url_register
from crawler_sys.framework.es_ccr_index_defination import fields_url_register
from write_data_into_es.func_cal_doc_id import cal_doc_id
from crawler_sys.utils.write_into_file import write_str_into_file
from crawler.crawler_sys.proxy_pool.func_get_proxy_form_kuaidaili import get_proxy
from lxml import html
from lxml.html.clean import Cleaner
from crawler.gm_upload.gm_upload import upload, upload_file
index_site_crawler = 'crawler-data-raw'
doc_type_site_crawler = 'doc'
# 实例化mysql连接对象
class mysql_conn():
def __init__(self, mysql_name):
if mysql_name == "test":
self.conn = pymysql.connect(host='bj-cdb-6slgqwlc.sql.tencentcdb.com', port=62120, user='work',
passwd='Gengmei1',
db='mimas_test', charset='utf8')
elif mysql_name == "mimas":
self.conn = pymysql.connect(host='172.16.30.138', port=3306, user='mimas',
passwd='GJL3UJe1Ck9ggL6aKnZCq4cRvM',
db='mimas_prod', charset='utf8mb4')
self.cur = self.conn.cursor()
def form_data_Lst_for_url_register(data_Lst_ori):
ts = int(datetime.datetime.now().timestamp() * 1e3)
data_Lst_reg = []
for line in data_Lst_ori:
try:
fields_ori = set(line.keys())
fields_rem = set(fields_url_register)
fields_to_remove = fields_ori.difference(fields_rem)
for field in fields_to_remove:
line.pop(field)
line['timestamp'] = ts
data_Lst_reg.append(line)
except:
print('attributeerror at %s' % data_Lst_ori.index(line))
return data_Lst_reg
def hot_words_output_result(result_Lst, output_index="short-video-hotwords"):
bulk_all_body = ""
for count, result in enumerate(result_Lst):
doc_id = result["platform"] + "_" + result["title"]
bulk_head = '{"index": {"_id":"%s"}}' % doc_id
data_str = json.dumps(result, ensure_ascii=False)
bulk_one_body = bulk_head + '\n' + data_str + '\n'
bulk_all_body += bulk_one_body
if count % 500 == 0 and count > 0:
eror_dic = es_site_crawler.bulk(index=output_index,
body=bulk_all_body, request_timeout=200)
bulk_all_body = ''
if eror_dic['errors'] is True:
print(eror_dic['items'])
print(bulk_all_body)
print(count)
if bulk_all_body != '':
eror_dic = es_site_crawler.bulk(body=bulk_all_body,
index=output_index,
request_timeout=200)
if eror_dic['errors'] is True:
print(eror_dic)
WHITE_TAGS = {
"basic": ["div", "p", "span", "img", "br", "video", 'a'], # 暂定小程序及爬取数据使用
"all": [
"div", "p", "span", "img", "br", "video", "audio", "a", "b", "strong", "i", "ul", "ol", "li", "em", "h1",
"h2", "h3", "h4", "h5", "h6", "iframe",
] # 可以展示的所有白标签
}
img_type = {
"OTHER": 1,
# '其他图片'
"GIF": 2,
# "GIF动图")
"JPG": 3,
# "JPG图片")
"JPEG": 4,
# "JPEG图片")
"PNG": 5,
# "PNG图片")
"BMP": 6,
# "BMP位图")
"WEBP": 7,
# "WEBP图片类型")
"TIFF": 8,
# "TIFF图片类型")
}
def gm_convert_html_tags(rich_text, all_tags=False, remove_tags=None):
"""
富文本内容重新清洗，剔除不需要的样式
:param rich_text: 富文本
:param all_tags: 是否需要匹配所有白名单中的标签
:param remove_tags: 需要剔除的，白名单标签 []
:return:
"""
if not rich_text:
return ""
# rich_text = _get_rich_text(rich_text)
# 标签清洗 + 补齐 参数
tags = WHITE_TAGS["all"] if all_tags else WHITE_TAGS["basic"]
if remove_tags:
tags = [tag for tag in tags if tag not in remove_tags]
kw = {
"remove_unknown_tags": False,
"allow_tags": tags,
"safe_attrs": ["src", ],
}
if "a" in tags:
kw["safe_attrs"].append("href")
elif all_tags:
kw["safe_attrs"].extend(["class", "style"])
if "iframe" in kw["allow_tags"]:
kw["embedded"] = False
clear = Cleaner(**kw)
rich_text = clear.clean_html(rich_text)
# 增加样式
element_obj = html.fromstring(rich_text)
for element in element_obj.xpath(u"//img|//video"):
if not all_tags: # 小程序，普通用户，爬取数据
element.attrib["width"] = "100%" # 图片、视频增加宽度 100%
if element.tag == "video" and all_tags:
element.attrib["class"] = "js_richtext_video"
# 移除a标签中跳转链不是gengmei开头的链接
for item in element_obj.xpath("//a[not(starts-with(@href, 'gengmei://'))]"):
item.getparent().remove(item)
# a 标签追加样式
for item in element_obj.xpath("//a"):
item.attrib["style"] = 'color:#3FB5AF' # a标签颜色
rich_text = html.tostring(element_obj, encoding="unicode")
return rich_text
def push_data_to_user(res_data: Dict) -> Dict:
"""
处理数据为可以入库的格式
:param res_data:
:return:
"""
qiniu_img_list = []
if res_data["img_list"]:
for img_url in res_data["img_list"]:
try:
img_wb = retry_get_url(img_url).content
res = upload(img_wb)
print(res)
img_info = retry_get_url(res + "-imageinfo")
img_info_json = img_info.json()
qiniu_img_list.append((res + "-w", img_info_json))
except Exception as e:
print("down load img error %s" % e)
return {}
# 替换图片
if res_data["platform"] == "weibo":
res_data["qiniu_img_list"] = qiniu_img_list
if "http://t.cn/" in res_data["title"]:
res_data["title"] = res_data["title"].split("http://t.cn/")[0]
res_data["content"] = res_data["title"]
elif res_data["platform"] == "douban":
content = res_data.get("content")
if content:
for count, img_url in enumerate(res_data["img_list"]):
# print(qiniu_img_list[count][0])
content = content.replace(img_url, qiniu_img_list[count][0])
res_data["qiniu_img_list"] = qiniu_img_list
res_data["content"] = content
if res_data["platform"] == "weibo":
res_data["content"] = gm_convert_html_tags(res_data["title"], all_tags=True)
res_data["title"] = ""
elif res_data["platform"] == "douban":
res_data["content"] = gm_convert_html_tags(res_data["content"], all_tags=True)
return res_data
def write_data_into_mysql(res_data,mysql):
now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
# 清洗数据为可以入库的格式
data = push_data_to_user(res_data)
if not data.get("content"):
return None
if not data.get("qiniu_img_list"):
return None
tractate_id = 0
try:
sql_query = """insert into api_tractate
(user_id,content,is_online,status,platform,content_level,is_excellent,create_time,last_modified,user_del,low_quality,low_quality_deal,platform_id,pgc_type,title)
values ({user_id},'{content}',{is_online},{status},{platform},{content_level},{is_excellent},'{create_time}','{last_modified}',{user_del},{low_quality},{low_quality_deal},'{platform_id}',{pgc_type},'{title}');""".format(
user_id=random.choice(user_id_list), content=data["content"], is_online=0, status=2, platform=15,
content_level=data["level"],
is_excellent=0, create_time=now_str,
last_modified=now_str, user_del=0,
low_quality=0, low_quality_deal=0, platform_id=data["doc_id"], pgc_type=0, title=data["title"])
res = mysql.cur.execute(sql_query)
tractate_id = int(mysql.conn.insert_id())
if res:
mysql.conn.commit()
except Exception as e:
print("commit error %s" % e)
print(data)
mysql.conn.rollback()
if data.get("qiniu_img_list"):
for img_info in data.get("qiniu_img_list"):
if img_info[0] in data.get("content"):
image_url_source = 2
else:
image_url_source = 3
try:
image_type = img_type.get(img_info[1]["format"].upper())
except:
image_type = 1
try:
width = img_info[1]["width"]
height = img_info[1]["height"]
except:
width = 0
height = 0
try:
if img_type == 7:
sql_query = """
insert into api_tractate_images (tractate_id,image_url,width,image_webp,height,image_url_source,image_type,image_webp,create_time,update_time)
values ({tractate_id},'{image_url}',{width},{height},{image_webp},{image_url_source},{image_type},{image_webp},'{create_time}','{update_time}')
""".format(tractate_id=tractate_id, image_url=img_info[0], width=width,
height=height, image_url_source=image_url_source,
image_type=image_type, image_webp=img_info[0],
create_time=now_str, update_time=now_str)
else:
sql_query = """
insert into api_tractate_images (tractate_id,image_url,width,height,image_url_source,image_type,create_time,update_time)
values ({tractate_id},'{image_url}',{width},{height},{image_url_source},{image_type},'{create_time}','{update_time}')
""".format(tractate_id=tractate_id, image_url=img_info[0], width=width,
height=height, image_url_source=image_url_source, image_type=image_type,
create_time=now_str, update_time=now_str)
res = mysql.cur.execute(sql_query)
if res:
mysql.conn.commit()
except Exception as e:
print("commit error %s" % e)
mysql.conn.rollback()
if tractate_id:
return tractate_id
else:
return None
def output_result(result_Lst, platform,
output_to_file=False, filepath=None,
output_to_es_raw=False,
output_to_es_register=False,
push_to_redis=False,
output_to_test_mysql=False,
output_to_mimas_mysql=False,
es_index=index_site_crawler,
rds_path="on_line",
**kwargs):
# write data into es crawler-raw index
if output_to_es_raw:
bulk_write_into_es(result_Lst, es_index)
# write data into es crawler-url-register index
if output_to_es_register:
# data_Lst_reg = form_data_Lst_for_url_register(result_Lst)
bulk_write_into_es(result_Lst,
index=es_index,
construct_id=True,
platform=platform,
)
if output_to_test_mysql:
pass
# feed url into redis
if push_to_redis:
rds = redis_path(rds_path)
feed_url_into_redis(
result_Lst, expire=kwargs.get("expire"),rds=rds)
# output into file according to passed in parameters
if output_to_file is True and filepath is not None:
output_fn = ('crawler_%s_on_%s_json'
% (platform, datetime.datetime.now().isoformat()[:10]))
output_f = open(filepath + '/' + output_fn, 'a', encoding='utf-8')
write_into_file(result_Lst, output_f)
output_f.close()
else:
return result_Lst
def retry_get_url(url, retrys=3, proxies=None, timeout=10, **kwargs):
retry_c = 0
while retry_c < retrys:
try:
if proxies:
proxies_dic = get_proxy(proxies)
if not proxies_dic:
get_resp = requests.get(url, timeout=timeout, **kwargs)
else:
get_resp = requests.get(url, proxies=proxies_dic, timeout=timeout, **kwargs)
else:
get_resp = requests.get(url, timeout=timeout, **kwargs)
return get_resp
except Exception as e:
retry_c += 1
time.sleep(1)
print(e)
print('Failed to get page %s after %d retries, %s'
% (url, retrys, datetime.datetime.now()))
return None
def get_ill_encoded_str_posi(UnicodeEncodeError_msg):
posi_nums = []
get_err_posi = re.findall('\s+[0-9]+-[0-9]+:', UnicodeEncodeError_msg)
if get_err_posi != []:
posi = get_err_posi[0].replace(' ', '').replace(':', '').split('-')
for pp in posi:
try:
ppn = int(pp)
posi_nums.append(ppn)
except:
pass
else:
pass
return posi_nums
def bulk_write_into_es(dict_Lst,
index,
construct_id=False,
platform=None):
bulk_write_body = ''
write_counter = 0
def bulk_write_with_retry_UnicodeEncodeError(index, bulk_write_body,
retry_counter_for_UnicodeEncodeError):
if bulk_write_body != '':
try:
bulk_write_resp = es_site_crawler.bulk(index=index,
body=bulk_write_body,
request_timeout=100)
if bulk_write_resp['errors'] is True:
print(bulk_write_resp)
bulk_write_body = ''
# print(bulk_write_resp)
print('Writing into es done')
except UnicodeEncodeError as ue:
print('Got UnicodeEncodeError, will remove ill formed string and retry.')
retry_counter_for_UnicodeEncodeError += 1
UnicodeEncodeError_msg = ue.__str__()
ill_str_idxs = get_ill_encoded_str_posi(UnicodeEncodeError_msg)
if len(ill_str_idxs) == 2:
ill_str = bulk_write_body[ill_str_idxs[0]: ill_str_idxs[1] + 1]
bulk_write_body = bulk_write_body.replace(ill_str, '')
bulk_write_with_retry_UnicodeEncodeError(index, bulk_write_body,
retry_counter_for_UnicodeEncodeError
)
except TransportError:
print("output to es register error")
write_str_into_file(file_path='/home/',
file_name='debug',
var=bulk_write_body)
return retry_counter_for_UnicodeEncodeError
for line in dict_Lst:
write_counter += 1
if construct_id and platform is not None:
doc_id = cal_doc_id(platform, url=line["url"], doc_id_type='all-time-url', data_dict=line)
action_str = ('{ "index" : { "_index" : "%s", "_id" : "%s" } }'
% (index, doc_id))
else:
action_str = ('{ "index" : { "_index" : "%s" }'
% (index))
data_str = json.dumps(line, ensure_ascii=False)
line_body = action_str + '\n' + data_str + '\n'
bulk_write_body += line_body
if write_counter % 1000 == 0 or write_counter == len(dict_Lst):
print('Writing into es %s %d/%d' % (index,
write_counter,
len(dict_Lst)))
if bulk_write_body != '':
retry_counter_for_UnicodeEncodeError = 0
retry_counter_for_UnicodeEncodeError = bulk_write_with_retry_UnicodeEncodeError(index,
bulk_write_body,
retry_counter_for_UnicodeEncodeError)
bulk_write_body = ''
def write_into_file(dict_Lst, file_obj):
for data_dict in dict_Lst:
json_str = json.dumps(data_dict)
file_obj.write(json_str)
file_obj.write('\n')
file_obj.flush()
def load_json_file_into_dict_Lst(filename, path):
if path[-1] != '/':
path += '/'
data_Lst = []
with open(path + filename, 'r', encoding='utf-8') as f:
for line in f:
line_d = json.loads(line)
if 'data_provider' not in line_d:
line_d['data_provider'] = 'CCR'
if 'releaser_id' in line_d:
try:
line_d['releaser_id'] = int(line_d['releaser_id'])
except:
line_d.pop('releaser_id')
data_Lst.append(line_d)
return data_Lst
def crawl_a_url_and_update_redis(url, platform, urlhash, rds,processID=-1,):
# find crawler
# perform crawling, get the data
# write es or output to files
# update redis
ts = int(datetime.datetime.now().timestamp() * 1e3)
redis_hmset_dict = {'push_time': ts, 'is_fetched': 1,
'url': url, 'platform': platform}
rds.hmset(urlhash, redis_hmset_dict)
def crawl_batch_task(url_Lst):
for url_info in url_Lst:
crawl_a_url_and_update_redis(url_info['url'],
url_info['platform'],
url_info['urlhash'])
def scan_redis_to_crawl(rds):
batch_size = 1000
cur = 0
task_batchs = []
scan_counter = 0
while True:
scan_counter += 1
if scan_counter % 5 == 0:
print(scan_counter, 'cur:', cur, datetime.datetime.now())
cur, hash_keys = rds.scan(cur)
for urlhash in hash_keys:
if len(urlhash) == 40:
url_d = rds.hgetall(urlhash)
url = url_d[b'url'].decode()
platform = url_d[b'platform'].decode()
is_fetched = int(url_d[b'is_fetched'].decode())
if is_fetched == 0:
task_batchs.append({'url': url,
'platform': platform,
'urlhash': urlhash})
if len(task_batchs) == batch_size:
# multi-processing here
crawl_batch_task(rds,task_batchs)
task_batchs.clear()
if cur == 0:
break
def remove_fetched_url_from_redis(rds,remove_interval=10):
time.sleep(remove_interval)
cur = 0
delete_counter = 0
while True:
cur, hash_keys = rds.scan(cur)
for urlhash in hash_keys:
if len(urlhash) == 40:
url_d = rds.hgetall(urlhash)
try:
is_fetched = int(url_d[b'is_fetched'].decode())
if is_fetched == 1:
rds.delete(urlhash)
delete_counter += 1
except:
pass
if cur == 0:
break
print('delete_counter', delete_counter)
return delete_counter