Skip to content

Z
zzsn_spider
提交 31e25a8d 作者: LiuLiYuan
操作

Merge remote-tracking branch 'origin/master'

上级 1c9b2f85 efcc39af
正在显示 包含 1349 行增加 和 149 行删除
base/BaseCore.py
......@@ -650,7 +650,7 @@ class BaseCore:
return selects
#插入到att表 返回附件id
def tableUpdate(self,retData,com_name,year,pdf_name,num):
def tableUpdate(self,retData,com_name,year,pdf_name,num,pub_time):
item_id = retData['item_id']
type_id = retData['type_id']
group_name = retData['group_name']
......@@ -670,12 +670,12 @@ class BaseCore:
id = ''
return id
else:
Upsql = '''insert into clb_sys_attachment(year,name,type_id,item_id,group_name,path,full_path,category,file_size,order_by,status,create_by,create_time,page_size) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)'''
Upsql = '''insert into clb_sys_attachment(year,name,type_id,item_id,group_name,path,full_path,category,file_size,order_by,status,create_by,create_time,page_size,publish_time) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s.%s)'''
values = (
year, pdf_name, type_id, item_id, group_name, path, full_path, category, file_size, order_by,
status, create_by,
create_time, page_size)
create_time, page_size,pub_time)
self.cursor_.execute(Upsql, values) # 插入
self.cnx_.commit() # 提交
......@@ -759,7 +759,7 @@ class BaseCore:
try:
time_now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
retData['state'] = True
retData['path'] = result['body']['objectUrl'].split('.com')[1]
retData['path'] = unquote(result['body']['objectUrl'].split('.com')[1])
retData['full_path'] = unquote(result['body']['objectUrl'])
retData['file_size'] = self.convert_size(file_size)
retData['create_time'] = time_now
......
comData/annualReport/证监会-年报.py
import json
import json
......@@ -133,7 +133,7 @@ def SpiderByZJH(url, payload, dic_info, num, start_time):
return False
#插入数据库获取att_id
num = num + 1
att_id = baseCore.tableUpdate(retData, short_name, year, name_pdf, num)
att_id = baseCore.tableUpdate(retData, short_name, year, name_pdf, num,pub_time)
if att_id:
pass
else:
......
comData/annualReport/雪球网-年报.py
# -*- coding: utf-8 -*-
# -*- coding: utf-8 -*-
......@@ -164,7 +164,7 @@ def spider_annual_report(dict_info,num):
return False
num = num + 1
try:
att_id = baseCore.tableUpdate(retData,com_name,year,name_pdf,num)
att_id = baseCore.tableUpdate(retData,com_name,year,name_pdf,num,pub_time)
content = retData['content']
state = 1
takeTime = baseCore.getTimeCost(start_time, time.time())
......
comData/annualReport_US/annualreportUS.py
......@@ -174,7 +174,7 @@ def spider(com_name,cik,up_okCount):
'summary': '',
'title': title,
'type': 1,
'socialCreditCode': '',
'socialCreditCode': social_code,
'year': year
}
# print(dic_news)
......@@ -207,7 +207,7 @@ def spider(com_name,cik,up_okCount):
# 采集一条资讯记录一条，记录该企业采到了多少的资讯
log.info(f'{social_code}----{news_url}:新增一条')
except Exception as e:
log.error(f'传输失败:{social_code}----{news_url}-----{e}')
log.error(f'数据库传输失败:{social_code}----{news_url}-----{e}')
e = '数据库传输失败'
state = 0
takeTime = baseCore.getTimeCost(start_time, time.time())
......
comData/caiwushuju/YAHOO财务数据4.py
# -*- coding: utf-8 -*-
# -*- coding: utf-8 -*-
......@@ -64,8 +64,9 @@ class YahooCaiwu(object):
doc_items = pq(resp1_table[1]).children()
if len(doc_items)<1:
resp1_table = doc_resp('#Col1-1-Financials-Proxy section div:nth-child(4)>div>div').children()
catalogue_title = pq(resp1_table[0]).text().split('\n')
doc_items = pq(resp1_table[1]).children()
if resp1_table:
catalogue_title = pq(resp1_table[0]).text().split('\n')
doc_items = pq(resp1_table[1]).children()
catalogue_dict = {}
content_dict = {}
for doc_item in doc_items:
......@@ -376,6 +377,10 @@ class YahooCaiwu(object):
#对比指标计算
def calculateIndexReq(self):
get_url = 'http://114.115.236.206:8088/sync/calculateIndex'
# 获取当前时间
current_time = datetime.datetime.now()
# 将时间转换为字符串
currentdate = current_time.strftime("%Y-%m-%d %H:%M:%S")
try:
params={
'type':2
......@@ -399,6 +404,7 @@ if __name__ == '__main__':
# parse_excel()
#get_content1()
yahoo=YahooCaiwu()
while True:
securitiescode=''
try:
......
comData/caiwushuju/YAHOO财务数据tmp.py
# -*- coding: utf-8 -*-
# -*- coding: utf-8 -*-
......@@ -58,74 +58,80 @@ class YahooCaiwu(object):
# 雅虎财经处理表格
def deal_table(self,doc_resp):
all_dict = {}
resp1_table = doc_resp('#Col1-1-Financials-Proxy section div:nth-child(3)>div>div').children()
catalogue_title = pq(resp1_table[0]).text().split('\n')
doc_items = pq(resp1_table[1]).children()
if len(doc_items)<1:
resp1_table = doc_resp('#Col1-1-Financials-Proxy section div:nth-child(4)>div>div').children()
try:
all_dict = {}
resp1_table = doc_resp('#Col1-1-Financials-Proxy section div:nth-child(3)>div>div').children()
catalogue_title = pq(resp1_table[0]).text().split('\n')
doc_items = pq(resp1_table[1]).children()
catalogue_dict = {}
content_dict = {}
for doc_item in doc_items:
if pq(doc_item).text() == '':
continue
a = pq(pq(doc_item).children()[0]).text().split('\n')[0]
a_list = pq(pq(doc_item).children()[0]).text().split('\n')[1:]
content_dict[a] = a_list
b_dict = {}
for doc_item1 in pq(doc_item).children()[1]:
b = pq(pq(doc_item1).children()[0]).text().split('\n')[0]
if not b:
if len(doc_items)<1:
resp1_table = doc_resp('#Col1-1-Financials-Proxy section div:nth-child(4)>div>div').children()
if resp1_table:
catalogue_title = pq(resp1_table[0]).text().split('\n')
doc_items = pq(resp1_table[1]).children()
catalogue_dict = {}
content_dict = {}
for doc_item in doc_items:
if pq(doc_item).text() == '':
continue
b_list = pq(pq(doc_item1).children()[0]).text().split('\n')[1:]
content_dict[b] = b_list
c_dict = {}
for doc_item2 in pq(doc_item1).children()[1]:
c = pq(pq(doc_item2).children()[0]).text().split('\n')[0]
if not c:
a = pq(pq(doc_item).children()[0]).text().split('\n')[0]
a_list = pq(pq(doc_item).children()[0]).text().split('\n')[1:]
content_dict[a] = a_list
b_dict = {}
for doc_item1 in pq(doc_item).children()[1]:
b = pq(pq(doc_item1).children()[0]).text().split('\n')[0]
if not b:
continue
c_list = pq(pq(doc_item2).children()[0]).text().split('\n')[1:]
content_dict[c] = c_list
d_dict = {}
for doc_item3 in pq(doc_item2).children()[1]:
d = pq(pq(doc_item3).children()[0]).text().split('\n')[0]
if not d:
b_list = pq(pq(doc_item1).children()[0]).text().split('\n')[1:]
content_dict[b] = b_list
c_dict = {}
for doc_item2 in pq(doc_item1).children()[1]:
c = pq(pq(doc_item2).children()[0]).text().split('\n')[0]
if not c:
continue
d_list = pq(pq(doc_item3).children()[0]).text().split('\n')[1:]
content_dict[d] = d_list
e_dict = {}
for doc_item4 in pq(doc_item3).children()[1]:
e = pq(pq(doc_item4).children()[0]).text().split('\n')[0]
if not e:
c_list = pq(pq(doc_item2).children()[0]).text().split('\n')[1:]
content_dict[c] = c_list
d_dict = {}
for doc_item3 in pq(doc_item2).children()[1]:
d = pq(pq(doc_item3).children()[0]).text().split('\n')[0]
if not d:
continue
e_list = pq(pq(doc_item4).children()[0]).text().split('\n')[1:]
content_dict[e] = e_list
f_dict = {}
for doc_item5 in pq(doc_item4).children()[1]:
f = pq(pq(doc_item5).children()[0]).text().split('\n')[0]
if not f:
d_list = pq(pq(doc_item3).children()[0]).text().split('\n')[1:]
content_dict[d] = d_list
e_dict = {}
for doc_item4 in pq(doc_item3).children()[1]:
e = pq(pq(doc_item4).children()[0]).text().split('\n')[0]
if not e:
continue
f_list = pq(pq(doc_item5).children()[0]).text().split('\n')[1:]
content_dict[f] = f_list
g_dict = {}
for doc_item6 in pq(doc_item5).children()[1]:
g = pq(pq(doc_item6).children()[0]).text().split('\n')[0]
if not g:
e_list = pq(pq(doc_item4).children()[0]).text().split('\n')[1:]
content_dict[e] = e_list
f_dict = {}
for doc_item5 in pq(doc_item4).children()[1]:
f = pq(pq(doc_item5).children()[0]).text().split('\n')[0]
if not f:
continue
g_list = pq(pq(doc_item6).children()[0]).text().split('\n')[1:]
content_dict[g] = g_list
g_dict[g] = {}
f_dict[f] = g_dict
e_dict[e] = f_dict
d_dict[d] = e_dict
c_dict[c] = d_dict
b_dict[b] = c_dict
catalogue_dict[a] = b_dict
all_dict['表头'] = catalogue_title
all_dict['目录'] = catalogue_dict
all_dict['内容'] = content_dict
f_list = pq(pq(doc_item5).children()[0]).text().split('\n')[1:]
content_dict[f] = f_list
g_dict = {}
for doc_item6 in pq(doc_item5).children()[1]:
g = pq(pq(doc_item6).children()[0]).text().split('\n')[0]
if not g:
continue
g_list = pq(pq(doc_item6).children()[0]).text().split('\n')[1:]
content_dict[g] = g_list
g_dict[g] = {}
f_dict[f] = g_dict
e_dict[e] = f_dict
d_dict[d] = e_dict
c_dict[c] = d_dict
b_dict[b] = c_dict
catalogue_dict[a] = b_dict
all_dict['表头'] = catalogue_title
all_dict['目录'] = catalogue_dict
all_dict['内容'] = content_dict
except Exception as e:
all_dict['表头'] = {}
all_dict['目录'] = {}
all_dict['内容'] = {}
return all_dict
......@@ -157,7 +163,8 @@ class YahooCaiwu(object):
conn,cursor=self.conn11()
try:
sql1 = """select social_credit_code,securities_code,securities_short_name from sys_base_enterprise_ipo where exchange='8' """ # and stock_code = "SYNH"
# sql1 = """select social_credit_code,securities_code,securities_short_name from sys_base_enterprise_ipo where securities_code='RAIZ4.SA' """ # and stock_code = "SYNH"
sql1 = """select social_credit_code,securities_code,securities_short_name from sys_base_enterprise_ipo where exchange='8' and any_data='0' """ # and stock_code = "SYNH"
# sql1 = f"select social_credit_code,securities_code,securities_short_name from sys_base_enterprise_ipo where securities_code='{securitiescode}' " # and stock_code = "SYNH"
cursor.execute(sql1)
result_data = cursor.fetchall()
......@@ -396,6 +403,7 @@ class YahooCaiwu(object):
print('调用接口成功！！')
except:
print('调用失败！')
if __name__ == '__main__':
# parse_excel()
#get_content1()
......
comData/caiwushuju/test.py
# import redis
#
#
# r = redis.Redis(host="114.115.236.206", port=6379, password='clbzzsn',db=3)
#
# # 获取所有键
# keys = r.keys('*')
# # print(keys)
# for key in keys:
# f_key = key.decode()
# print(f_key)
# print("----------")
# res = r.exists(f_key)
# value = list(r.smembers(f_key))
# # 对列表进行排序
# value.sort()
# # 遍历排序后的列表
# list_data = []
# for member in value:
# member = member.decode()
# members = member.strip('[').strip(']').replace('\'','').strip().split(',')
# #获取每一个报告期
# for date in members:
# data = date.strip()
# # print(date.strip())
# list_data.append(data)
# # 放入redis
# for item in list_data:
# r.sadd(key, item)
#
# # 获取Set中的所有元素
# items = r.smembers(key)
# # print(items)
# print("======================================")
import re
from urllib.parse import quote, unquote
import requests
from pyquery import PyQuery as pq
from bs4 import BeautifulSoup
import json
import difflib
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
import datetime
timestamp = 1688054400 # 示例时间戳
date = datetime.datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')
headers={
'Connection':'keep-alive',
'Pragma':'no-cache',
'Cache-Control':'no-cache',
'sec-ch-ua':'"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',
'sec-ch-ua-mobile':'?0',
'sec-ch-ua-platform':'"Windows"',
'Upgrade-Insecure-Requests':'1',
'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36',
'Accept':'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'Sec-Fetch-Site':'same-origin',
'Sec-Fetch-Mode':'navigate',
'Sec-Fetch-User':'?1',
'Sec-Fetch-Dest':'document',
'Referer':'https://quotes.sina.com.cn/usstock/hq/income.php?s=brk.a&t=quarter',
'Accept-Encoding':'gzip, deflate, br',
'Accept-Language':'zh-CN,zh;q=0.9',
'Cookie':'UOR=,finance.sina.com.cn,; SINAGLOBAL=123.149.3.173_1695815968.404462; Apache=123.149.3.173_1695815968.404463; ULV=1695816017391:2:2:2:123.149.3.173_1695815968.404463:1695815967476; lxlrttp=1578733570; U_TRS1=000000ad.bc7f83f51.651419db.690100f2; U_TRS2=000000ad.bc8a83f51.651419db.138fca70; SUB=_2AkMSSJVgf8NxqwFRmP0XzG3kbIxxyA_EieKkFGS7JRMyHRl-yD9kqhY-tRB6Oci7j27VGy6gikgIaUYBZsIPzk3PbLLC; hqEtagMode=1',
}
proxy = {'https': 'http://127.0.0.1:1080', 'http': 'http://127.0.0.1:1080'}
def reqHtml(url):
res=requests.get(url,headers=headers,verify=False,timeout=10)
res.encoding='GB18030'
text=res.text
return text
headers2={
'Accept':'*/*',
'Accept-Encoding':'gzip, deflate, br',
'Accept-Language':'zh-CN,zh;q=0.9',
'Cache-Control':'no-cache',
'Connection':'keep-alive',
'Content-Length':'0',
'Cookie':'HWWAFSESID=fd8b573695b0ce804b; HWWAFSESTIME=1695799275143',
'Host':'www.qyyjt.cn',
'Origin':'https://www.qyyjt.cn',
'Pragma':'no-cache',
'Referer':'https://www.qyyjt.cn/detail/enterprise/overview?code=56CD928FAD278663E73BE7486C764DA7&type=company',
'Sec-Fetch-Dest':'empty',
'Sec-Fetch-Mode':'cors',
'Sec-Fetch-Site':'same-origin',
'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Safari/537.36',
'client':'pc-web;pro',
'dataid':'869',
'pcuss':'eyJ0eXAiOiJKV1QiLCJ0eXBlIjoiand0IiwiYWxnIjoiSFMyNTYifQ.eyJjcmVhdGVUaW1lIjoiMjAyMy0wOS0yNyAyMDoxODowMy40NDkiLCJleHAiOjE2OTU4MTc5ODMsInVzZXJJZCI6IjIwMjMwOTI3MTUyMzA0XzEzNTkyNDgxODM5IiwiZXhwaXJlZFRpbWUiOiIyMDIzLTA5LTI3IDIwOjMzOjAzLjQ0OSJ9.SouwRylKogHfJILh97JMnYRzcJuj2Hg30BmQa9gc-Nc',
'sec-ch-ua':'"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',
'sec-ch-ua-mobile':'?0',
'sec-ch-ua-platform':'"Windows"',
'system':'new',
'terminal':'pc-web;pro',
'user':'847E223529194582C37A02EEEC8AC09F0D7AD12E40778D6CA9CFB91F69F8C537',
'ver':'20230914',
'x-request-id':'x1eCRO-X8D7',
}
def reqPostMsg(url):
res=requests.post(url,headers=headers2,verify=False,timeout=10)
res.encoding='utf-8'
text=res.text
return text
def get_realurl(tmpurl):
try:
pattern='url=(.{1,}?)&aid'
match = re.search(pattern, tmpurl)
# 判断是否匹配成功
if match:
# 获取匹配的结果
result = match.group(1)
result=unquote(result)
else:
result=''
except:
result=''
return result
def getFormatedate(timestamp):
date = datetime.datetime.fromtimestamp(timestamp)
formatted_date = date.strftime('%Y-%m-%d')
return formatted_date
print(date)
url='https://quotes.sina.com.cn/usstock/hq/income.php?s=brk.a&t=quarter'
ttext=reqHtml(url)
soup=BeautifulSoup(ttext,'html.parser')
tdoc=soup.select('div[class="tbl_wrap"]>table[class="data_tbl os_tbl"]')[0]
print(str(tdoc))
comData/caiwushuju/yahoo_stock.py
......@@ -461,7 +461,7 @@ def listPage():
}
]
for operand in operands:
logger.info(f'采集地域股票信息{operand}')
rego=operand['operands'][1]
#第一次请求获取地域总共有的股票代码数量
try:
stockmsg=reqmsg(0,operand)
......@@ -469,21 +469,23 @@ def listPage():
except Exception as e:
logger.info(f'region该地域没有股票信息{operand}')
continue
logger.info(f'采集地域股票信息{rego}---对应的数量{total}')
for i in range(0,total,100):
logger.info(f"offset的值{i}")
stockmsg=reqmsg(i,operand)
if stockmsg:
try:
getStock(stockmsg)
getStock(stockmsg,rego)
except Exception as e:
logger.info(f"解析失败{e}")
time.sleep(3)
def getStock(stockmsg):
def getStock(stockmsg,rego):
quotes=stockmsg['finance']['result'][0]['quotes']
for quote in quotes:
symbol=quote['symbol']
logger.info(f"{rego}地区对应的股票代码{symbol}")
try:
longName=quote['longName']
except:
......
comData/noticeReport/证监会-公告.py
import json
import json
......@@ -37,7 +37,7 @@ def convert_size(size_bytes):
def uptoOBS(pdf_url,pdf_name,type_id,social_code):
headers = {}
retData = {'state': False, 'type_id': type_id, 'item_id': social_code, 'group_name': 'group1', 'path': '',
retData = {'state': False, 'type_id': type_id, 'item_id': social_code, 'group_name': '', 'path': '',
'full_path': '',
'category': 'pdf', 'file_size': '', 'status': 1, 'create_by': 'XueLingKun',
'create_time': '', 'page_size': '', 'content': ''}
......@@ -55,7 +55,7 @@ def uptoOBS(pdf_url,pdf_name,type_id,social_code):
try:
name = pdf_name + '.pdf'
now_time = time.strftime("%Y-%m")
result = obsClient.putContent('zzsn', f'ZJH/{now_time}/'+name, content=response.content)
result = obsClient.putContent('zzsn', 'QYNotice/'+name, content=response.content)
with fitz.open(stream=response.content, filetype='pdf') as doc:
page_size = doc.page_count
for page in doc.pages():
......@@ -113,12 +113,12 @@ def tableUpdate(retData, com_name, year, pdf_name, num):
# id = ''
# return id
# else:
Upsql = '''insert into clb_sys_attachment(year,name,type_id,item_id,group_name,path,full_path,category,file_size,order_by,status,create_by,create_time,page_size) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)'''
Upsql = '''insert into clb_sys_attachment(year,name,type_id,item_id,group_name,path,full_path,category,file_size,order_by,status,create_by,create_time,page_size,object_key,bucket_name) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)'''
values = (
year, pdf_name, type_id, item_id, group_name, path, full_path, category, file_size, order_by,
status, create_by,
create_time, page_size)
create_time, page_size,path,'zzsn')
cursor_.execute(Upsql, values) # 插入
cnx_.commit() # 提交
......@@ -277,9 +277,9 @@ def InsterInto(social_code, pdf_url,pub_time):
def GetContent(pdf_url, pdf_name, social_code, year, pub_time, start_time,com_name,num):
#判断文件是否已经存在obs服务器中
# file_path = 'XQWAnnualReport/2023-10/浙江国祥股份有限公司首次公开发行股票并在主板上市暂缓发行公告.doc'
# file_path = 'QYNotice//浙江国祥股份有限公司首次公开发行股票并在主板上市暂缓发行公告'
now_time = time.strftime("%Y-%m")
file_path = 'ZJH/'+now_time+'/'+pdf_name+'.pdf'
file_path = 'QYNotice/'+pdf_name+'.pdf'
response = obsClient.getObjectMetadata('zzsn', file_path)
if response.status >= 300:
log.info('=====文件不存在obs=====')
......@@ -372,19 +372,23 @@ def SpiderByZJH(url, payload, dic_info, start_time,num): # dic_info 数据库
except:
pass
# # 先获取页数
# page = soup.find('div', class_='pages').find('ul', class_='g-ul').text
#
# total = re.findall(r'\d+', page)[0]
#
# r_page = int(total) % 15
# if r_page == 0:
# Maxpage = int(total) // 15
# else:
# Maxpage = int(total) // 15 + 1
# log.info(f'{short_name}====={code}===========一共{total}条,{Maxpage}页')
# 先获取页数
page = soup.find('div', class_='pages').find('ul', class_='g-ul').text
total = re.findall(r'\d+', page)[0]
r_page = int(total) % 15
if r_page == 0:
Maxpage = int(total) // 15
else:
Maxpage = int(total) // 15 + 1
log.info(f'{short_name}====={code}===========一共{total}条,{Maxpage}页')
# # 首页和其他页不同，遍历 如果是首页 修改一下链接
for i in range(1,51):
if Maxpage < 50:
pass
else:
Maxpage = 50
for i in range(1,Maxpage):
log.info(f'==========正在采集第{i}页=========')
if i == 1:
href = url
......
comData/policylaw/BaseCore.py
......@@ -533,12 +533,12 @@ class BaseCore:
order_by = num
Upsql = '''insert into clb_sys_attachment(name,type_id,item_id,group_name,path,full_path,category,file_size,order_by,status,create_by,create_time) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)'''
Upsql = '''insert into clb_sys_attachment(name,type_id,item_id,group_name,path,full_path,category,file_size,order_by,status,create_by,create_time,object_key,bucket_name) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)'''
values = (
file_name, type_id, item_id, group_name, path, full_path, category, file_size, order_by,
status, create_by,
create_time)
create_time,path,'zzsn')
self.cursor_.execute(Upsql, values) # 插入
self.cnx_.commit() # 提交
......@@ -558,11 +558,11 @@ class BaseCore:
i += 1
return f"{size_bytes:.2f} {units[i]}"
def uptoOBS(self,file_href,item_id,pathType,file_name):
def uptoOBS(self,file_href,item_id,file_name):
headers = {}
category = os.path.splitext(file_href)[1]
retData = {'state': False, 'type_id': 7, 'item_id': item_id, 'group_name': 'group1', 'path': '',
retData = {'state': False, 'type_id': 7, 'item_id': item_id, 'group_name': '', 'path': '',
'full_path': '',
'category': category, 'file_size': '', 'status': 1, 'create_by': 'XueLingKun',
'create_time': '', 'page_size': '', 'content': ''}
......@@ -583,7 +583,7 @@ class BaseCore:
pass
else:
file_name = file_name + '.' + category
result = obsClient.putContent('zzsn', f'{pathType}' + file_name, content=response.content)
result = obsClient.putContent('zzsn', 'PolicyDocuments/' + file_name, content=response.content)
break
except:
time.sleep(3)
......
comData/policylaw/tingtype.py
......@@ -40,7 +40,8 @@ def save_data(dic_news):
'网址':dic_news['sourceAddress'],
'tid':dic_news['labels'][0]['relationId'],
'来源':dic_news['labels'][0]['relationName'],
'创建时间':dic_news['createDate']
'创建时间':dic_news['createDate'],
'带标签内容': dic_news['contentWithTag'][:100]
}
db_storage.insert_one(aaa_dic)
......
jrtt1news_comm/config.ini 0 → 100644
[redis]
host=114.115.236.206
port=6379
pass=clbzzsn
[mysql]
host=114.115.159.144
username=caiji
password=zzsn9988
database=caiji
url=jdbc:mysql://114.115.159.144:3306/caiji?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&useSSL=false
[kafka]
bootstrap_servers=114.115.159.144:9092
topic=keyWordsInfo
groupId=python_sougou
[selenium]
chrome_driver=C:\Users\WIN10\DataspellProjects\crawlerProjectDemo\tmpcrawler\cmd100\chromedriver.exe
binary_location=D:\crawler\baidu_crawler\tool\Google\Chrome\Application\chrome.exe
jrtt1news_comm/entity.py 0 → 100644
# -*- coding: utf-8 -*-
# 智能采集请求
# 1、考虑：请求智能采集时，不再使用实体类
# a. 仍使用：通过HTTP的 raw 请求体，直接传递HTML源文件，通过query参数传递 lang-code、link-text 参数
# b. 原因：在 postman 中，不方便进行测试，无法使用粘贴后的HTML源文件
# 2、不考虑：使用实体类，利大于弊
# a. 使用实体类，方便扩展参数字段
# b. 方便展示接口文档：调用 json_parameter_utility.get_json_parameters 函数，可显示请求实体类
class ExtractionRequest:
# 语言代码
# 1、采集“非中文”的文章时，需要用到语言代码
lang_code = ""
# 链接文本
# 1、用于采集标题，如果不提供，标题的准确度会下降
link_text = ""
# 文章页面源文件
# 1、用于采集标题、发布时间、内容等
article_html = ""
@staticmethod
def from_dict(dictionary: dict):
extraction_request = ExtractionRequest()
# 尝试方法：
# 1、将字典，更新到内部的 __dict__ 对象
# extraction_request.__dict__.update(dictionary)
# 将字典值，设置到当前对象
for key in dictionary:
setattr(extraction_request, key, dictionary[key])
return extraction_request
def to_dict(self):
# 转换为字典对象：
# 1、序列化为JSON时，需要调用此方法
# 2、转换为JSON字符串：json.dumps(extraction_result, default=ExtractionResult.to_dict)
data = {}
# 借助内部的 __dict__ 对象
# 1、将内部的 __dict__ 对象，更新到新的字典对象中
data.update(self.__dict__)
return data
# 采集结果
class ExtractionResult:
# 标题
title = ""
# 发布日期
publish_date = ""
# 正文（保留所有HTML标记，如：br、img）
text = ""
# URL
url = ""
# 摘要
meta_description = ""
# 干净正文（不带HTML）
cleaned_text = ""
# 来源（目前只支持采集中文网站中的“来源”）
# source = ""
# 顶部图片（top_image：采集不到任何内容，不再使用此属性）
# top_image = ""
def to_dict(self):
# 转换为字典对象：
# 1、序列化为JSON时，需要调用此方法
# 2、转换为JSON字符串：json.dumps(extraction_result, default=ExtractionResult.to_dict)
data = {}
# 借助内部的 __dict__ 对象
# 1、将内部的 __dict__ 对象，更新到新的字典对象中
data.update(self.__dict__)
return data
class UrlPickingRequest:
# 列表页面的响应URL
# 1、作为Base URL，用于拼接提取到的相对URL
# 2、Base URL：必须使用响应URL
# 3、示例：在 Python中，通过 requests.get(url) 请求URL后，需要使用 resp.url 作为 Base URL
list_page_resp_url = ""
# 列表页面源文件
# 1、用于提取文章网址
list_page_html = ""
@staticmethod
def from_dict(dictionary: dict):
url_picking_request = UrlPickingRequest()
# 将字典值，设置到当前对象
for key in dictionary:
setattr(url_picking_request, key, dictionary[key])
return url_picking_request
def to_dict(self):
# 转换为字典对象：
# 1、序列化为JSON时，需要调用此方法
# 2、转换为JSON字符串：json.dumps(extraction_result, default=ExtractionResult.to_dict)
data = {}
# 借助内部的 __dict__ 对象
# 1、将内部的 __dict__ 对象，更新到新的字典对象中
data.update(self.__dict__)
return data
jrtt1news_comm/jrttnewstaskJob_comm.py 0 → 100644
# -*- coding: utf-8 -*-
"""
任务集成测试
1、连接redis做取出
2、连接kafka做信息的获取，与存储
"""
import time
import redis
from kafka import KafkaProducer
from kafka import KafkaConsumer
import json
import itertools
from jrttnewspider import SouhunewsSpider
import concurrent.futures
from baseCore import BaseCore
from queue import Queue
import configparser
class SouhunewsTaskJob(object):
def __init__(self):
# 创建ConfigParser对象
self.config = configparser.ConfigParser()
# 读取配置文件
self.config.read('config.ini')
self.r = redis.Redis(host=self.config.get('redis', 'host'),
port=self.config.get('redis', 'port'),
password=self.config.get('redis', 'pass'), db=0)
def getkafka(self):
# Kafka集群的地址
bootstrap_servers = self.config.get('kafka', 'bootstrap_servers')
# 要订阅的主题
topic = self.config.get('kafka', 'topic')
groupId=self.config.get('kafka', 'groupId')
consumer = KafkaConsumer(topic, group_id=groupId,
bootstrap_servers=[bootstrap_servers],
value_deserializer=lambda m: json.loads(m.decode('utf-8')))
try:
for record in consumer:
try:
logger.info("value:",record.value)
keymsg=record.value
if keymsg:
break
else:
continue
#print("%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value))
except Exception as e:
logger.info("msg.value error:",e)
except KeyboardInterrupt as e:
keymsg={}
finally:
consumer.close()
return keymsg
def getkeyFromredis(self,codeid):
kvalue=self.r.get('KEY_WORDS_TO_REDIS::'+codeid)
kvalue=kvalue.decode('utf-8')
kvalue=json.loads(kvalue)
return kvalue
def getkeywords(self,keywords):
kwList=[]
if ')+(' in keywords:
k1List=keywords.split('+')
kk2=[]
for k2 in k1List:
k2=k2.strip("()")
k2List=k2.split('|')
kk2.append(k2List)
if len(kk2)==2:
result = list(itertools.product(kk2[0], kk2[1]))
elif len(kk2)==3:
result = list(itertools.product(kk2[0], kk2[1],kk2[2]))
elif len(kk2)==4:
result = list(itertools.product(kk2[0], kk2[1],kk2[2],kk2[3]))
for res in result:
kwstr=''
for kw in res:
kwstr+=kw+"+"
kwList.append(kwstr.strip('+'))
elif '+(' in keywords:
k1List=keywords.split('+')
kk2=[]
for k2 in k1List:
k2=k2.strip("()")
k2List=k2.split('|')
kk2.append(k2List)
if len(kk2)==2:
result = list(itertools.product(kk2[0], kk2[1]))
for res in result:
kwstr=''
for kw in res:
kwstr+=kw+"+"
kwList.append(kwstr.strip('+'))
else:
k3=keywords.split("|")
kwList=k3
return kwList
def paserKeyMsg(self,keymsg):
logger.info('----------')
wordsCode=keymsg['wordsCode']
id=keymsg['id']
try:
searchEngines=keymsg['searchEngines']
except Exception as e:
searchEngines=[]
kwList=[]
if searchEngines:
if '3' in searchEngines:
keyword=keymsg['keyWord']
keymsglist=self.getkeywords(keyword)
for kw in keymsglist:
kwmsg={
'kw':kw,
'wordsCode':wordsCode,
'sid':id
}
kwList.append(kwmsg)
else:
pass
# logger.info('+++++')
# keyword=keymsg['keyWord']
# keymsglist=self.getkeywords(keyword)
# for kw in keymsglist:
# kwmsg={
# 'kw':kw,
# 'wordsCode':wordsCode,
# 'sid':id
# }
# kwList.append(kwmsg)
return kwList
# def runSpider(self,kwmsg):
# try:
# searchkw=kwmsg['kw']
# wordsCode=kwmsg['wordsCode']
# sid=kwmsg['sid']
#
# baiduSpider=BaiduSpider(searchkw,wordsCode,sid)
# baiduSpider.get_page_html()
# baiduSpider.get_detail_html()
# except Exception as e:
# logger.info('百度搜索异常'+searchkw)
# finally:
# baiduSpider.driver.quit()
# logger.info("关键词采集结束！"+searchkw)
def runSpider(self,kwmsg):
searchkw=kwmsg['kw']
wordsCode=kwmsg['wordsCode']
sid=kwmsg['sid']
souhunewsSpider=SouhunewsSpider(searchkw,wordsCode,sid)
try:
souhunewsSpider.get_page_html()
except Exception as e:
try:
souhunewsSpider.get_page_html()
except Exception as e:
logger.info('搜狗搜索异常'+searchkw)
if souhunewsSpider.detailList.qsize() != 0:
try:
souhunewsSpider.get_detail_html()
except Exception as e:
logger.info('详情解析异常'+searchkw)
logger.info("关键词采集结束！"+searchkw)
if __name__ == '__main__':
# ss='道地西洋参+(销售市场|交易市场|直播带货|借助大会平台|网店|微商|电商|农民博主|推介宣传|高品质定位|西洋参产品经营者加盟|引进龙头企业|西洋参冷风库|建设农旅中心|农产品展销中心|精品民宿|温泉)'
# keymsglist=getkeywords(ss)
# print(keymsglist)
# 创建Redis连接
souhunewsTaskJob=SouhunewsTaskJob()
baseCore=BaseCore()
logger=baseCore.getLogger()
print('---------------')
while True:
try:
try:
keymsg=souhunewsTaskJob.getkafka()
kwList=souhunewsTaskJob.paserKeyMsg(keymsg)
except Exception as e:
logger.info("从kafka拿取信息失败！")
time.sleep(5)
continue
if kwList:
# 创建一个线程池，指定线程数量为4
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
# 提交任务给线程池，每个任务处理一个数据
results = [executor.submit(souhunewsTaskJob.runSpider, data) for data in kwList]
# 获取任务的执行结果
for future in concurrent.futures.as_completed(results):
try:
result = future.result()
# 处理任务的执行结果
logger.info(f"任务执行结束: {result}")
except Exception as e:
# 处理任务执行过程中的异常
logger.info(f"任务执行exception: {e}")
except Exception as e:
logger.info('采集异常')
jrtt1news_comm/requirements.txt 0 → 100644
pip install langid -i https://mirrors.aliyun.com/pypi/simple/
pip install redis==4.3.5 -i https://pypi.douban.com/simple
pip install kafka-python==2.0.2 -i https://pypi.douban.com/simple
pip install PyMySQL -i https://pypi.douban.com/simple
pip install gne==0.3.0 -i https://pypi.douban.com/simple
pip install selenium==4.9.1 -i https://pypi.douban.com/simple
pip install logbook -i https://pypi.douban.com/simple
pip install tqdm -i https://pypi.douban.com/simple
pip install goose3 -i https://mirrors.aliyun.com/pypi/simple
pip install Beautifulsoup4 -i https://mirrors.aliyun.com/pypi/simple
pip install langid -i https://mirrors.aliyun.com/pypi/simple/
pip install jieba -i https://mirrors.aliyun.com/pypi/simple
selenium==3.141.0
selenium-wire==5.1.0
pip install --upgrade selenium
pip install --upgrade urllib3
pip3 uninstall urllib3
ImportError: urllib3 v2.0 only supports OpenSSL 1.1.1+, currently the 'ssl' module is compiled with 'OpenSSL 1.1.0i 14 Aug 2018'. See: https://github.com/urllib3/urllib3/issues/2168
\ No newline at end of file
jrtt1news_comm/start.bat 0 → 100644
title baidu_comm
chcp 65001
cd /d %~dp0
python baidutaskJob.py
\ No newline at end of file
qqnews_comm/qqnewstaskJob_comm.py
......@@ -169,15 +169,13 @@ class QQnewsTaskJob(object):
qqnewsSpider.get_page_html()
except Exception as e:
logger.info('搜狗搜索异常'+searchkw)
finally:
qqnewsSpider.driver.quit()
if qqnewsSpider.detailList.qsize() != 0:
try:
qqnewsSpider.get_detail_html()
except Exception as e:
logger.info('详情解析异常'+searchkw)
finally:
qqnewsSpider.driver.quit()
logger.info("关键词采集结束！"+searchkw)
if __name__ == '__main__':
# ss='道地西洋参+(销售市场|交易市场|直播带货|借助大会平台|网店|微商|电商|农民博主|推介宣传|高品质定位|西洋参产品经营者加盟|引进龙头企业|西洋参冷风库|建设农旅中心|农产品展销中心|精品民宿|温泉)'
......
souhunews_comm/config.ini 0 → 100644
[redis]
host=114.115.236.206
port=6379
pass=clbzzsn
[mysql]
host=114.115.159.144
username=caiji
password=zzsn9988
database=caiji
url=jdbc:mysql://114.115.159.144:3306/caiji?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&useSSL=false
[kafka]
bootstrap_servers=114.115.159.144:9092
topic=keyWordsInfo
groupId=python_sougou
[selenium]
chrome_driver=C:\Users\WIN10\DataspellProjects\crawlerProjectDemo\tmpcrawler\cmd100\chromedriver.exe
binary_location=D:\crawler\baidu_crawler\tool\Google\Chrome\Application\chrome.exe
souhunews_comm/entity.py 0 → 100644
# -*- coding: utf-8 -*-
# 智能采集请求
# 1、考虑：请求智能采集时，不再使用实体类
# a. 仍使用：通过HTTP的 raw 请求体，直接传递HTML源文件，通过query参数传递 lang-code、link-text 参数
# b. 原因：在 postman 中，不方便进行测试，无法使用粘贴后的HTML源文件
# 2、不考虑：使用实体类，利大于弊
# a. 使用实体类，方便扩展参数字段
# b. 方便展示接口文档：调用 json_parameter_utility.get_json_parameters 函数，可显示请求实体类
class ExtractionRequest:
# 语言代码
# 1、采集“非中文”的文章时，需要用到语言代码
lang_code = ""
# 链接文本
# 1、用于采集标题，如果不提供，标题的准确度会下降
link_text = ""
# 文章页面源文件
# 1、用于采集标题、发布时间、内容等
article_html = ""
@staticmethod
def from_dict(dictionary: dict):
extraction_request = ExtractionRequest()
# 尝试方法：
# 1、将字典，更新到内部的 __dict__ 对象
# extraction_request.__dict__.update(dictionary)
# 将字典值，设置到当前对象
for key in dictionary:
setattr(extraction_request, key, dictionary[key])
return extraction_request
def to_dict(self):
# 转换为字典对象：
# 1、序列化为JSON时，需要调用此方法
# 2、转换为JSON字符串：json.dumps(extraction_result, default=ExtractionResult.to_dict)
data = {}
# 借助内部的 __dict__ 对象
# 1、将内部的 __dict__ 对象，更新到新的字典对象中
data.update(self.__dict__)
return data
# 采集结果
class ExtractionResult:
# 标题
title = ""
# 发布日期
publish_date = ""
# 正文（保留所有HTML标记，如：br、img）
text = ""
# URL
url = ""
# 摘要
meta_description = ""
# 干净正文（不带HTML）
cleaned_text = ""
# 来源（目前只支持采集中文网站中的“来源”）
# source = ""
# 顶部图片（top_image：采集不到任何内容，不再使用此属性）
# top_image = ""
def to_dict(self):
# 转换为字典对象：
# 1、序列化为JSON时，需要调用此方法
# 2、转换为JSON字符串：json.dumps(extraction_result, default=ExtractionResult.to_dict)
data = {}
# 借助内部的 __dict__ 对象
# 1、将内部的 __dict__ 对象，更新到新的字典对象中
data.update(self.__dict__)
return data
class UrlPickingRequest:
# 列表页面的响应URL
# 1、作为Base URL，用于拼接提取到的相对URL
# 2、Base URL：必须使用响应URL
# 3、示例：在 Python中，通过 requests.get(url) 请求URL后，需要使用 resp.url 作为 Base URL
list_page_resp_url = ""
# 列表页面源文件
# 1、用于提取文章网址
list_page_html = ""
@staticmethod
def from_dict(dictionary: dict):
url_picking_request = UrlPickingRequest()
# 将字典值，设置到当前对象
for key in dictionary:
setattr(url_picking_request, key, dictionary[key])
return url_picking_request
def to_dict(self):
# 转换为字典对象：
# 1、序列化为JSON时，需要调用此方法
# 2、转换为JSON字符串：json.dumps(extraction_result, default=ExtractionResult.to_dict)
data = {}
# 借助内部的 __dict__ 对象
# 1、将内部的 __dict__ 对象，更新到新的字典对象中
data.update(self.__dict__)
return data
souhunews_comm/requirements.txt 0 → 100644
pip install langid -i https://mirrors.aliyun.com/pypi/simple/
pip install redis==4.3.5 -i https://pypi.douban.com/simple
pip install kafka-python==2.0.2 -i https://pypi.douban.com/simple
pip install PyMySQL -i https://pypi.douban.com/simple
pip install gne==0.3.0 -i https://pypi.douban.com/simple
pip install selenium==4.9.1 -i https://pypi.douban.com/simple
pip install logbook -i https://pypi.douban.com/simple
pip install tqdm -i https://pypi.douban.com/simple
pip install goose3 -i https://mirrors.aliyun.com/pypi/simple
pip install Beautifulsoup4 -i https://mirrors.aliyun.com/pypi/simple
pip install langid -i https://mirrors.aliyun.com/pypi/simple/
pip install jieba -i https://mirrors.aliyun.com/pypi/simple
selenium==3.141.0
selenium-wire==5.1.0
pip install --upgrade selenium
pip install --upgrade urllib3
pip3 uninstall urllib3
ImportError: urllib3 v2.0 only supports OpenSSL 1.1.1+, currently the 'ssl' module is compiled with 'OpenSSL 1.1.0i 14 Aug 2018'. See: https://github.com/urllib3/urllib3/issues/2168
\ No newline at end of file
souhunews_comm/souhunewstaskJob_comm.py 0 → 100644
# -*- coding: utf-8 -*-
"""
任务集成测试
1、连接redis做取出
2、连接kafka做信息的获取，与存储
"""
import time
import redis
from kafka import KafkaProducer
from kafka import KafkaConsumer
import json
import itertools
from souhunewspider import SouhunewsSpider
import concurrent.futures
from baseCore import BaseCore
from queue import Queue
import configparser
class SouhunewsTaskJob(object):
def __init__(self):
# 创建ConfigParser对象
self.config = configparser.ConfigParser()
# 读取配置文件
self.config.read('config.ini')
self.r = redis.Redis(host=self.config.get('redis', 'host'),
port=self.config.get('redis', 'port'),
password=self.config.get('redis', 'pass'), db=0)
def getkafka(self):
# Kafka集群的地址
bootstrap_servers = self.config.get('kafka', 'bootstrap_servers')
# 要订阅的主题
topic = self.config.get('kafka', 'topic')
groupId=self.config.get('kafka', 'groupId')
consumer = KafkaConsumer(topic, group_id=groupId,
bootstrap_servers=[bootstrap_servers],
value_deserializer=lambda m: json.loads(m.decode('utf-8')))
try:
for record in consumer:
try:
logger.info("value:",record.value)
keymsg=record.value
if keymsg:
break
else:
continue
#print("%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value))
except Exception as e:
logger.info("msg.value error:",e)
except KeyboardInterrupt as e:
keymsg={}
finally:
consumer.close()
return keymsg
def getkeyFromredis(self,codeid):
kvalue=self.r.get('KEY_WORDS_TO_REDIS::'+codeid)
kvalue=kvalue.decode('utf-8')
kvalue=json.loads(kvalue)
return kvalue
def getkeywords(self,keywords):
kwList=[]
if ')+(' in keywords:
k1List=keywords.split('+')
kk2=[]
for k2 in k1List:
k2=k2.strip("()")
k2List=k2.split('|')
kk2.append(k2List)
if len(kk2)==2:
result = list(itertools.product(kk2[0], kk2[1]))
elif len(kk2)==3:
result = list(itertools.product(kk2[0], kk2[1],kk2[2]))
elif len(kk2)==4:
result = list(itertools.product(kk2[0], kk2[1],kk2[2],kk2[3]))
for res in result:
kwstr=''
for kw in res:
kwstr+=kw+"+"
kwList.append(kwstr.strip('+'))
elif '+(' in keywords:
k1List=keywords.split('+')
kk2=[]
for k2 in k1List:
k2=k2.strip("()")
k2List=k2.split('|')
kk2.append(k2List)
if len(kk2)==2:
result = list(itertools.product(kk2[0], kk2[1]))
for res in result:
kwstr=''
for kw in res:
kwstr+=kw+"+"
kwList.append(kwstr.strip('+'))
else:
k3=keywords.split("|")
kwList=k3
return kwList
def paserKeyMsg(self,keymsg):
logger.info('----------')
wordsCode=keymsg['wordsCode']
id=keymsg['id']
try:
searchEngines=keymsg['searchEngines']
except Exception as e:
searchEngines=[]
kwList=[]
if searchEngines:
if '3' in searchEngines:
keyword=keymsg['keyWord']
keymsglist=self.getkeywords(keyword)
for kw in keymsglist:
kwmsg={
'kw':kw,
'wordsCode':wordsCode,
'sid':id
}
kwList.append(kwmsg)
else:
pass
# logger.info('+++++')
# keyword=keymsg['keyWord']
# keymsglist=self.getkeywords(keyword)
# for kw in keymsglist:
# kwmsg={
# 'kw':kw,
# 'wordsCode':wordsCode,
# 'sid':id
# }
# kwList.append(kwmsg)
return kwList
# def runSpider(self,kwmsg):
# try:
# searchkw=kwmsg['kw']
# wordsCode=kwmsg['wordsCode']
# sid=kwmsg['sid']
#
# baiduSpider=BaiduSpider(searchkw,wordsCode,sid)
# baiduSpider.get_page_html()
# baiduSpider.get_detail_html()
# except Exception as e:
# logger.info('百度搜索异常'+searchkw)
# finally:
# baiduSpider.driver.quit()
# logger.info("关键词采集结束！"+searchkw)
def runSpider(self,kwmsg):
searchkw=kwmsg['kw']
wordsCode=kwmsg['wordsCode']
sid=kwmsg['sid']
souhunewsSpider=SouhunewsSpider(searchkw,wordsCode,sid)
try:
souhunewsSpider.get_page_html()
except Exception as e:
try:
souhunewsSpider.get_page_html()
except Exception as e:
logger.info('搜狗搜索异常'+searchkw)
if souhunewsSpider.detailList.qsize() != 0:
try:
souhunewsSpider.get_detail_html()
except Exception as e:
logger.info('详情解析异常'+searchkw)
logger.info("关键词采集结束！"+searchkw)
if __name__ == '__main__':
# ss='道地西洋参+(销售市场|交易市场|直播带货|借助大会平台|网店|微商|电商|农民博主|推介宣传|高品质定位|西洋参产品经营者加盟|引进龙头企业|西洋参冷风库|建设农旅中心|农产品展销中心|精品民宿|温泉)'
# keymsglist=getkeywords(ss)
# print(keymsglist)
# 创建Redis连接
souhunewsTaskJob=SouhunewsTaskJob()
baseCore=BaseCore()
logger=baseCore.getLogger()
print('---------------')
while True:
try:
try:
keymsg=souhunewsTaskJob.getkafka()
kwList=souhunewsTaskJob.paserKeyMsg(keymsg)
except Exception as e:
logger.info("从kafka拿取信息失败！")
time.sleep(5)
continue
if kwList:
# 创建一个线程池，指定线程数量为4
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
# 提交任务给线程池，每个任务处理一个数据
results = [executor.submit(souhunewsTaskJob.runSpider, data) for data in kwList]
# 获取任务的执行结果
for future in concurrent.futures.as_completed(results):
try:
result = future.result()
# 处理任务的执行结果
logger.info(f"任务执行结束: {result}")
except Exception as e:
# 处理任务执行过程中的异常
logger.info(f"任务执行exception: {e}")
except Exception as e:
logger.info('采集异常')
souhunews_comm/start.bat 0 → 100644
title baidu_comm
chcp 65001
cd /d %~dp0
python baidutaskJob.py
\ No newline at end of file
toutiaonews_comm/config.ini 0 → 100644
[redis]
host=114.115.236.206
port=6379
pass=clbzzsn
[mysql]
host=114.115.159.144
username=caiji
password=zzsn9988
database=caiji
url=jdbc:mysql://114.115.159.144:3306/caiji?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&useSSL=false
[kafka]
bootstrap_servers=114.115.159.144:9092
topic=keyWordsInfo
groupId=python_sougou
[selenium]
chrome_driver=C:\Users\WIN10\DataspellProjects\crawlerProjectDemo\tmpcrawler\cmd100\chromedriver.exe
binary_location=D:\crawler\baidu_crawler\tool\Google\Chrome\Application\chrome.exe
toutiaonews_comm/entity.py 0 → 100644
# -*- coding: utf-8 -*-
# 智能采集请求
# 1、考虑：请求智能采集时，不再使用实体类
# a. 仍使用：通过HTTP的 raw 请求体，直接传递HTML源文件，通过query参数传递 lang-code、link-text 参数
# b. 原因：在 postman 中，不方便进行测试，无法使用粘贴后的HTML源文件
# 2、不考虑：使用实体类，利大于弊
# a. 使用实体类，方便扩展参数字段
# b. 方便展示接口文档：调用 json_parameter_utility.get_json_parameters 函数，可显示请求实体类
class ExtractionRequest:
# 语言代码
# 1、采集“非中文”的文章时，需要用到语言代码
lang_code = ""
# 链接文本
# 1、用于采集标题，如果不提供，标题的准确度会下降
link_text = ""
# 文章页面源文件
# 1、用于采集标题、发布时间、内容等
article_html = ""
@staticmethod
def from_dict(dictionary: dict):
extraction_request = ExtractionRequest()
# 尝试方法：
# 1、将字典，更新到内部的 __dict__ 对象
# extraction_request.__dict__.update(dictionary)
# 将字典值，设置到当前对象
for key in dictionary:
setattr(extraction_request, key, dictionary[key])
return extraction_request
def to_dict(self):
# 转换为字典对象：
# 1、序列化为JSON时，需要调用此方法
# 2、转换为JSON字符串：json.dumps(extraction_result, default=ExtractionResult.to_dict)
data = {}
# 借助内部的 __dict__ 对象
# 1、将内部的 __dict__ 对象，更新到新的字典对象中
data.update(self.__dict__)
return data
# 采集结果
class ExtractionResult:
# 标题
title = ""
# 发布日期
publish_date = ""
# 正文（保留所有HTML标记，如：br、img）
text = ""
# URL
url = ""
# 摘要
meta_description = ""
# 干净正文（不带HTML）
cleaned_text = ""
# 来源（目前只支持采集中文网站中的“来源”）
# source = ""
# 顶部图片（top_image：采集不到任何内容，不再使用此属性）
# top_image = ""
def to_dict(self):
# 转换为字典对象：
# 1、序列化为JSON时，需要调用此方法
# 2、转换为JSON字符串：json.dumps(extraction_result, default=ExtractionResult.to_dict)
data = {}
# 借助内部的 __dict__ 对象
# 1、将内部的 __dict__ 对象，更新到新的字典对象中
data.update(self.__dict__)
return data
class UrlPickingRequest:
# 列表页面的响应URL
# 1、作为Base URL，用于拼接提取到的相对URL
# 2、Base URL：必须使用响应URL
# 3、示例：在 Python中，通过 requests.get(url) 请求URL后，需要使用 resp.url 作为 Base URL
list_page_resp_url = ""
# 列表页面源文件
# 1、用于提取文章网址
list_page_html = ""
@staticmethod
def from_dict(dictionary: dict):
url_picking_request = UrlPickingRequest()
# 将字典值，设置到当前对象
for key in dictionary:
setattr(url_picking_request, key, dictionary[key])
return url_picking_request
def to_dict(self):
# 转换为字典对象：
# 1、序列化为JSON时，需要调用此方法
# 2、转换为JSON字符串：json.dumps(extraction_result, default=ExtractionResult.to_dict)
data = {}
# 借助内部的 __dict__ 对象
# 1、将内部的 __dict__ 对象，更新到新的字典对象中
data.update(self.__dict__)
return data
toutiaonews_comm/requirements.txt 0 → 100644
pip install langid -i https://mirrors.aliyun.com/pypi/simple/
pip install redis==4.3.5 -i https://pypi.douban.com/simple
pip install kafka-python==2.0.2 -i https://pypi.douban.com/simple
pip install PyMySQL -i https://pypi.douban.com/simple
pip install gne==0.3.0 -i https://pypi.douban.com/simple
pip install selenium==4.9.1 -i https://pypi.douban.com/simple
pip install logbook -i https://pypi.douban.com/simple
pip install tqdm -i https://pypi.douban.com/simple
pip install goose3 -i https://mirrors.aliyun.com/pypi/simple
pip install Beautifulsoup4 -i https://mirrors.aliyun.com/pypi/simple
pip install langid -i https://mirrors.aliyun.com/pypi/simple/
pip install jieba -i https://mirrors.aliyun.com/pypi/simple
selenium==3.141.0
selenium-wire==5.1.0
pip install --upgrade selenium
pip install --upgrade urllib3
pip3 uninstall urllib3
ImportError: urllib3 v2.0 only supports OpenSSL 1.1.1+, currently the 'ssl' module is compiled with 'OpenSSL 1.1.0i 14 Aug 2018'. See: https://github.com/urllib3/urllib3/issues/2168
\ No newline at end of file
toutiaonews_comm/start.bat 0 → 100644
title baidu_comm
chcp 65001
cd /d %~dp0
python baidutaskJob.py
\ No newline at end of file
toutiaonews_comm/toutiaonewstaskJob_comm.py 0 → 100644
# -*- coding: utf-8 -*-
"""
任务集成测试
1、连接redis做取出
2、连接kafka做信息的获取，与存储
"""
import time
import redis
from kafka import KafkaProducer
from kafka import KafkaConsumer
import json
import itertools
from toutiaonewspider import QQnewsSpider
import concurrent.futures
from baseCore import BaseCore
from queue import Queue
import configparser
class QQnewsTaskJob(object):
def __init__(self):
# 创建ConfigParser对象
self.config = configparser.ConfigParser()
# 读取配置文件
self.config.read('config.ini')
self.r = redis.Redis(host=self.config.get('redis', 'host'),
port=self.config.get('redis', 'port'),
password=self.config.get('redis', 'pass'), db=0)
def getkafka(self):
# Kafka集群的地址
bootstrap_servers = self.config.get('kafka', 'bootstrap_servers')
# 要订阅的主题
topic = self.config.get('kafka', 'topic')
groupId=self.config.get('kafka', 'groupId')
consumer = KafkaConsumer(topic, group_id=groupId,
bootstrap_servers=[bootstrap_servers],
value_deserializer=lambda m: json.loads(m.decode('utf-8')))
try:
for record in consumer:
try:
logger.info("value:",record.value)
keymsg=record.value
if keymsg:
break
else:
continue
#print("%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value))
except Exception as e:
logger.info("msg.value error:",e)
except KeyboardInterrupt as e:
keymsg={}
finally:
consumer.close()
return keymsg
def getkeyFromredis(self,codeid):
kvalue=self.r.get('KEY_WORDS_TO_REDIS::'+codeid)
kvalue=kvalue.decode('utf-8')
kvalue=json.loads(kvalue)
return kvalue
def getkeywords(self,keywords):
kwList=[]
if ')+(' in keywords:
k1List=keywords.split('+')
kk2=[]
for k2 in k1List:
k2=k2.strip("()")
k2List=k2.split('|')
kk2.append(k2List)
if len(kk2)==2:
result = list(itertools.product(kk2[0], kk2[1]))
elif len(kk2)==3:
result = list(itertools.product(kk2[0], kk2[1],kk2[2]))
elif len(kk2)==4:
result = list(itertools.product(kk2[0], kk2[1],kk2[2],kk2[3]))
for res in result:
kwstr=''
for kw in res:
kwstr+=kw+"+"
kwList.append(kwstr.strip('+'))
elif '+(' in keywords:
k1List=keywords.split('+')
kk2=[]
for k2 in k1List:
k2=k2.strip("()")
k2List=k2.split('|')
kk2.append(k2List)
if len(kk2)==2:
result = list(itertools.product(kk2[0], kk2[1]))
for res in result:
kwstr=''
for kw in res:
kwstr+=kw+"+"
kwList.append(kwstr.strip('+'))
else:
k3=keywords.split("|")
kwList=k3
return kwList
def paserKeyMsg(self,keymsg):
logger.info('----------')
wordsCode=keymsg['wordsCode']
id=keymsg['id']
try:
searchEngines=keymsg['searchEngines']
except Exception as e:
searchEngines=[]
kwList=[]
if searchEngines:
if '3' in searchEngines:
keyword=keymsg['keyWord']
keymsglist=self.getkeywords(keyword)
for kw in keymsglist:
kwmsg={
'kw':kw,
'wordsCode':wordsCode,
'sid':id
}
kwList.append(kwmsg)
else:
pass
# logger.info('+++++')
# keyword=keymsg['keyWord']
# keymsglist=self.getkeywords(keyword)
# for kw in keymsglist:
# kwmsg={
# 'kw':kw,
# 'wordsCode':wordsCode,
# 'sid':id
# }
# kwList.append(kwmsg)
return kwList
# def runSpider(self,kwmsg):
# try:
# searchkw=kwmsg['kw']
# wordsCode=kwmsg['wordsCode']
# sid=kwmsg['sid']
#
# baiduSpider=BaiduSpider(searchkw,wordsCode,sid)
# baiduSpider.get_page_html()
# baiduSpider.get_detail_html()
# except Exception as e:
# logger.info('百度搜索异常'+searchkw)
# finally:
# baiduSpider.driver.quit()
# logger.info("关键词采集结束！"+searchkw)
def runSpider(self,kwmsg):
searchkw=kwmsg['kw']
wordsCode=kwmsg['wordsCode']
sid=kwmsg['sid']
qqnewsSpider=QQnewsSpider(searchkw,wordsCode,sid)
try:
qqnewsSpider.get_page_html()
except Exception as e:
try:
qqnewsSpider.get_page_html()
except Exception as e:
logger.info('搜狗搜索异常'+searchkw)
finally:
qqnewsSpider.driver.quit()
if qqnewsSpider.detailList.qsize() != 0:
try:
qqnewsSpider.get_detail_html()
except Exception as e:
logger.info('详情解析异常'+searchkw)
finally:
qqnewsSpider.driver.quit()
logger.info("关键词采集结束！"+searchkw)
if __name__ == '__main__':
# ss='道地西洋参+(销售市场|交易市场|直播带货|借助大会平台|网店|微商|电商|农民博主|推介宣传|高品质定位|西洋参产品经营者加盟|引进龙头企业|西洋参冷风库|建设农旅中心|农产品展销中心|精品民宿|温泉)'
# keymsglist=getkeywords(ss)
# print(keymsglist)
# 创建Redis连接
qqTaskJob=QQnewsTaskJob()
baseCore=BaseCore()
logger=baseCore.getLogger()
print('---------------')
while True:
try:
try:
keymsg=qqTaskJob.getkafka()
kwList=qqTaskJob.paserKeyMsg(keymsg)
except Exception as e:
logger.info("从kafka拿取信息失败！")
time.sleep(5)
continue
if kwList:
# 创建一个线程池，指定线程数量为4
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
# 提交任务给线程池，每个任务处理一个数据
results = [executor.submit(qqTaskJob.runSpider, data) for data in kwList]
# 获取任务的执行结果
for future in concurrent.futures.as_completed(results):
try:
result = future.result()
# 处理任务的执行结果
logger.info(f"任务执行结束: {result}")
except Exception as e:
# 处理任务执行过程中的异常
logger.info(f"任务执行exception: {e}")
except Exception as e:
logger.info('采集异常')
百度采集/baidu_comm/baiduSpider.py
#coding=utf-8
#coding=utf-8
......@@ -402,6 +402,7 @@ class BaiduSpider(object):
# 使用langid.py判断文本的语言
lang, confidence = langid.classify(text)
return lang
# 获取详情页
def get_detail_html(self):
# 获取当前窗口的句柄
......
百度采集/baidu_comm/baidutaskJob_loc.py
# -*- coding: utf-8 -*-
# -*- coding: utf-8 -*-
......@@ -190,7 +190,7 @@ if __name__ == '__main__':
while True:
try:
codeList=[]
codeList.append('KW-20230818-0003')
codeList.append('KW-20230925-0002')
for codeid in codeList:
try:
# keymsg=baiduTaskJob.getkafka()
......@@ -207,7 +207,7 @@ if __name__ == '__main__':
continue
if kwList:
# 创建一个线程池，指定线程数量为4
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
# 提交任务给线程池，每个任务处理一个数据
results = [executor.submit(baiduTaskJob.runSpider, data) for data in kwList]
# 获取任务的执行结果
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论