Python实现聚合问答采集文章

Python与SEO 思享 535浏览
摘要:
最近在一个社区得到了一个用python聚合问答的工具,但是因为已经打包成程序了,所以无法研究,于是在网上找了一个网友分享的源代码,转载于此,供志愿者学习和后期扩展。这个工具可以通过头条搜索、百度下拉结果、搜狗下拉、百度知道搜索、新浪爱问、搜狗问问,将头条答案组合成一篇文章。python文件已经反编译完美还原源代码,亲测有效。

最近在某社区获取到了一个用python聚合问答的工具,但是因为已经打包成程序了没办法研究于是网上找了找,找到了一个网友分享的源码,这里转载过来义工以后学习拓展

该工具可以通过头条搜索、百度下拉结果、搜狗下拉、百度知道搜索、新浪爱问、搜狗问问来进行组合标题答案构成文章。

该python文件已反编译完美还原源代码,亲测有效。

python

源码

import configparser
import json
import threading
from queue import Queue
from urllib import request
from urllib.parse import quote
import os
import random
import re
import requests
import time
from lxml import etree
config = configparser.RawConfigParser()
config.read('peizhi.ini')
ZHANGHAO = config.get('KUANDAI', 'ZHANGHAO')
MIMA = config.get('KUANDAI', 'MIMA')
IP = int(config.get('KUANDAI', 'IP'))
KD_QUEUE = Queue(1000000)
web_ck = config.get('KUANDAI', 'web_ck')
FOREGROUND_RED = 0x0c  # red.
FOREGROUND_GREEN = 0x0a  # green.
FOREGROUND_BLUE = 0x09  # blue.
CHA = ''
with open('替换词库.txt', 'r', encoding='utf8') as (f):
tihuan_list = f.read().split('\n')
def connect():
cmd_str = 'rasdial %s %s %s' % ('宽带连接', ZHANGHAO, MIMA)
os.system(cmd_str)
print('拨号')
time.sleep(2)
def disconnect():
cmd_str = 'rasdial 宽带连接 /disconnect'
os.system(cmd_str)
print('断开链接')
time.sleep(2)
def get_connect():
header_baidu = {
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) '
'Chrome/69.0.3497.100 Safari/537.36'}
try:
code = requests.get('http://apps.game.qq.com/comm-htdocs/ip/get_ip.php', headers=header_baidu,
timeout=5).status_code
if code != 200:
connect()
else:
disconnect()
time.sleep(5)
connect()
except:
disconnect()
time.sleep(5)
connect()
def get_toutiao_urls(wd):
url = 'https://so.toutiao.com/search?keyword=' + wd + '&pd=question&source=search_subtab_switch&dvpf=pc&aid=4916&page_num=0'
headers = {'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'Cookie': web_ck,
'Host': 'so.toutiao.com',
'Pragma': 'no-cache',
'Referer': 'https://so.toutiao.com/search?keyword=seo&pd=question&source=search_subtab_switch&dvpf=pc&aid=4916&page_num=0',
'sec-ch-ua-mobile': '?0',
'Sec-Fetch-Dest': 'document',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'same-origin',
'Sec-Fetch-User': '?1',
'Upgrade-Insecure-Requests': '1',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
res = requests.get(url=url, headers=headers)
res.encoding = 'utf8'
title_list = re.findall('"title":"(.*?)",', res.text)
title_list = [i for i in title_list if '\\' not in i]
url_list = re.findall('"url":"(.*?)","', res.text)
url_list = [i for i in url_list if 'wukong' in i]
return [i.replace('http:', 'https:') for i in url_list], title_list
def get_wukong_content(url):
headers = {'accept-encoding': 'gzip, deflate, br',
'accept-language': 'zh-CN,zh;q=0.9,zh-TW;q=0.8,en-US;q=0.7,en;q=0.6',
'cache-control': 'no-cache',
'pragma': 'no-cache',
'sec-ch-ua-mobile': '?0',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'none',
'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
try:
res = requests.get(url=url, headers=headers)
if res.status_code == 200:
res.encoding = 'utf8'
wenzhang_list = \
json.loads(re.findall('INITIAL_STATE__=([\\s\\S]*?)</script><script>', res.text)[0])['qData']['data'][
'ans_list']
wenzhang = [i['content'] for i in wenzhang_list]
return wenzhang
return
except Exception as e:
try:
print(e)
return
finally:
e = None
del e
def wukong(wd):
res = ''
uu, tt = get_toutiao_urls(wd)
title_two = random.sample(tt, 2)
url = random.choice(uu)
contents_list = get_wukong_content(url)
contents = random.sample(contents_list, 2)
for index, i in enumerate(contents):
i = i.replace('</p>', '\n')
i = re.sub('<.*?>', '', i)
if len(i) >= 4:
res = i.startswith('{') or res + '<h2>' + title_two[index] + '</h2>' + '\n' + i + '\n'
return res
def get_xiala_bd(wd):
try:
url = 'https://sp0.baidu.com/5a1Fazu8AA54nxGko9WTAnF6hhy/su?wd=%s&json=1' % wd
text = requests.get(url).text.replace('window.baidu.sug(', '').replace(');', '')
text_json = json.loads(text)
return wd + '(%s)' % random.choice(text_json['s'])
except:
try:
res = requests.get(
url=f"http://www.baidu.com/s?wd={wd}&rsv_spt=1&rsv_iqid=0xcab0d69f000cf3cf&issp=1&f=8&rsv_bp=1&rsv_idx=2&ie=utf-8&tn=baiduhome_pg&rsv_dl=ib&rsv_enter=1&rsv_sug3=31&rsv_sug1=15&rsv_sug7=100")
list_A = re.findall('c-font-medium new-inc-rs-item" href="/[\\s\\S]*?">(.*?)</a>', res.text)
return wd + '(%s)' % random.choice(list_A)
except Exception as e:
try:
pass
finally:
e = None
del e
return wd
def get_xiala_sogou(wd):
try:
url = 'https://www.sogou.com/suggnew/ajajjson?key=%s&type=web&ori=yes&pr=web&abtestid=0&ipn=&t=1584098135513&suguuid=d74c8137-d92b-4941-a64b-00e0fcac9a0b&ip=180.110.15.100&iploc=3201&suid=43DD6EB44B238B0A5CE7ECB4000BC889&yyid=null&pid=sogou&policyno=null&mfp=null&hs=https&mp=1&prereq_a=dhahdhhad.com&sugsuv=005FC078B46EDD435CFDF9C25D63C464&sugtime=1584098144455' % wd
res = requests.get(url)
res = res.text.replace('window.sogou.sug(', '')
res = re.findall(',\\[(.*?)\\]', res.replace(')', ''))[0].split('","')
if len(res) == 0:
return wd
for x in res:
if x != '':
return wd + '(%s)' % random.choice(res).replace('"', '').replace('()', '')
return wd
except:
return wd
def get_zhidao_urls(wd):
url = f"http://zhidao.baidu.com/search?lm=0&rn=10&pn=0&fr=search&&word={wd}"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36'}
try:
res = requests.get(url=url, headers=headers, timeout=10)
res.encoding = res.apparent_encoding
if '安全验证' in res.text:
print('zhidao出验证码了~~~~')
return ''
questions_list = re.findall('zhidao.baidu.com/question/(.*?).html', res.text)
questions_list = [f"http://zhidao.baidu.com/question/{i}.html" for i in questions_list]
random.shuffle(questions_list)
return questions_list[1:3]
except:
return []
def get_zhidao_answer(url):
try:
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36'}
req = request.Request(url, headers=headers)
request2 = request.urlopen(req)
content = request2.read().decode('gbk')
request2.close()
html = etree.HTML(content)
title = re.findall('<title>(.*?)_百度知道</title>', content)[0]
contents = html.xpath(".//div[@class='bd answer']//text()")
cc = contents
contents = [re.sub('[\\s]+', '', i) for i in contents]
contents = '\n'.join([i for i in contents if len(i) >= 5])
if len(cc) <= 10:
bst_list = html.xpath("//div[@class='line content']/div[@accuse='aContent']//text()")
cc_li = []
for i in bst_list:
if i != '' and i != '\n' and len(i) >= 8:
cc_li.append(i.strip())
contents = ''.join(cc_li)
return f"<h2>{title}</h2>\n{contents}"
except:
pass
return ''
def zhidao(wd):
res = ''
uu = get_zhidao_urls(wd)
random.shuffle(uu)
for x in uu:
c = get_zhidao_answer(x)
res = res + c.strip() + '\n'
res = res.replace('你对这个回答的评价是?', '')
return res
def get_sina_urls(wd):
headers = {'pragma': 'no-cache',
'referer': f"https://iask.sina.com.cn/search?searchWord={quote(wd)}&record=1",
'sec-ch-ua-mobile': '?0',
'sec-fetch-dest': 'document',
'sec-fetch-mode': 'navigate',
'sec-fetch-site': 'same-origin',
'sec-fetch-user': '?1',
'upgrade-insecure-requests': '1',
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36'}
url = f"https://iask.sina.com.cn/search?searchWord={wd}&page=1"
for i in range(4):
res = requests.get(url=url, headers=headers, timeout=10)
res.encoding = res.apparent_encoding
if res.status_code == 200:
break
time.sleep(2)
if res.status_code == 403:
return ''
url_list = ['https://iask.sina.com.cn' + i for i in re.findall('<p class="title-text"><a href="(.*?)"', res.text)]
return url_list
def get_sina_answer(url):
headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36'}
try:
res = requests.get(url=url, headers=headers, timeout=10)
res.encoding = 'utf8'
html = etree.HTML(res.text)
title = re.findall('<title>(.*?) 爱问知识人</title>', res.text)[0]
pre_list = html.xpath("//pre[@class='list-text']/text()")
content = sorted(pre_list, key=(lambda i: len(i)), reverse=False)[(-1)]
content = f"<h2>{title}</h2>\n{content}"
except:
content = ''
return content
def sina(wd):
res = ''
t = 0
uu = get_sina_urls(wd)
random.shuffle(uu)
for x in uu:
c = get_sina_answer(x)
if len(c) >= 10:
res = res + c.strip() + '\n'
t += 1
if t >= 2:
break
res = res.replace('你对这个回答的评价是?', '')
return res
def get_sogou_urls(wd):
headers = {
'user-agent': 'Mozilla/5.0 (Linux; Android 5.0; SM-G900P Build/LRX21T) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Mobile Safari/537.36'}
url = f"https://m.sogou.com/web/searchList.jsp?keyword={wd}&insite=wenwen.sogou.com&pid=sogou-waps-fd2ae8ec902471d8&rcer=uNz_alvVqvzeAE_5"
try:
res = requests.get(url=url, headers=headers)
res.encoding = res.apparent_encoding
id_list = re.findall('&amp;url=http%3A%2F%2Fwenwen.sogou.com%2Fz%2F(.*?).htm&amp;vrid', res.text)
url_list = [f"https://wenwen.sogou.com/z/{i}.htm" for i in id_list][1:]
except:
url_list = []
return url_list
def get_sogou_answer(url):
headers = {
'user-agent': 'Mozilla/5.0 (Linux; Android 5.0; SM-G900P Build/LRX21T) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Mobile Safari/537.36'}
try:
res = requests.get(url=url, headers=headers, timeout=10)
res.encoding = 'utf8'
html = etree.HTML(res.text)
title = re.findall('<title>(.*?)</title>', res.text)[0]
pre_list = html.xpath("//h3[@class='ask-fresh-con answerContent']//text()")
pre_list2 = html.xpath("//h3[@class='ask-fresh-con']//text()")
pre_list.extend(pre_list2)
content = ''.join(pre_list)
content = content.replace('\u3000', '')
content = f"<h2>{title}</h2>\n{content}"
except:
content = ''
return content
def sogou(wd):
res = ''
t = 0
uu = get_sogou_urls(wd)
random.shuffle(uu)
for x in uu:
c = get_sogou_answer(x)
if len(c) >= 10:
res = res + c.strip() + '\n'
t += 1
if t >= 2:
break
res = res.replace('你对这个回答的评价是?', '')
return res
def result_tihuan(text):
for x in tihuan_list:
text = text.replace(x, '')
text = re.sub('[a-zA-z]+://[a-zA-Z0-9\\.\\-\\/_]+', '', text)
result = '\n'.join(['<p>' + i + '</p>' for i in text.split('\n') if len(i) >= 5])
result = result.replace('<p><h2>', '<h2>')
result = result.replace('</h2></p>', '</h2>')
return result
with open('关键词.txt', 'r', encoding='utf8') as (f):
kd_list = f.read().split('\n')
def get_baidu_pic(wd):
headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
url = 'https://image.baidu.com/search/acjson?tn=resultjson_com&logid=7941037868480343284&ipn=rj&ct=201326592&is=&fp=result&queryWord=' + wd + '&cl=2&lm=-1&ie=utf-8&oe=utf-8&adpicid=&st=&z=&ic=&hd=&latest=&copyright=&word=' + wd + '&s=&se=&tab=&width=&height=&face=&istype=&qc=&nc=1&fr=&expermode=&nojc=&pn=30&rn=30&gsm=1e&1626606839586='
res = requests.get(url=url, headers=headers)
pic_list = []
for i in res.json()['data']:
try:
pic_list.append(f"<img src='{i['thumbURL']}' alt='{wd}'>")
except:
pass
return pic_list
def get_content_main(wd):
wd = re.sub('[\\s]+', '', wd)
try:
tt = get_xiala_bd(wd)
if tt == wd:
tt = get_xiala_sogou(wd)
except:
tt = wd
tt = re.sub('[?.!!]', '', tt)
print(f"[INFO:当前生成内容:{tt}]")
rr_jieguo = ''
temp = zhidao(wd)
rr_jieguo = rr_jieguo + temp + '\n'
temp = sina(wd)
rr_jieguo = rr_jieguo + temp + '\n'
temp = sogou(wd)
rr_jieguo = rr_jieguo + temp + '\n'
try:
temp = wukong(wd)
rr_jieguo = rr_jieguo + temp + '\n'
except Exception as e:
try:
pass
finally:
e = None
del e
result_cc = result_tihuan(rr_jieguo)
try:
pic_list = get_baidu_pic(wd)
except:
pic_list = []
try:
result_cc_h2 = result_cc.split('<h2>')
result_con = ''
for index, mwy in enumerate(result_cc_h2):
if len(mwy) >= 5:
if index % 2 == 0:
result_con = result_con + '<p>' + random.choice(pic_list) + '</p>' + '\n' + '<h2>' + mwy
else:
result_con = result_con + '\n' + '<h2>' + mwy
except:
result_con = result_cc
result_con_ll = result_con.split('\n')
result_con_ll = [re.sub('\\s{3,}', '', i) for i in result_con_ll]
result_con_ll = [i for i in result_con_ll if len(i) > 3]
result_con = '\n'.join(result_con_ll)
with open((os.path.join('文章', tt + '.txt')), 'w', encoding='utf8') as (f):
f.write(result_con)
print(f"[SUCCESS:当前内容:{tt}]--------生成成功!")
def colorize(num, string, bold=False, highlight=False):
assert isinstance(num, int)
attr = []
if highlight:
num += 10
attr.append(str(num))
if bold: attr.append('1')
return '\x1b[%sm%s\x1b[0m' % (';'.join(attr), string)
def main():
for x in kd_list:
KD_QUEUE.put(x)
ci = 40
print("#### 增加自定义延迟 ####")
xiancheng = int(input('请选择线程:'))
xiancheng_m = int(input('线程等待秒:'))
real_huan_ip_ci = int(ci / xiancheng)
real = 0
while True:
if KD_QUEUE.empty():
break
if IP != 0:
if real % real_huan_ip_ci == 0:
print('该换IP了~~~~~')
get_connect()
thread_list = []
for i in range(xiancheng):
if KD_QUEUE.empty():
break
wd = KD_QUEUE.get()
t1 = threading.Thread(target=get_content_main, args=(wd,))
thread_list.append(t1)
for t in thread_list:
t.setDaemon(True)
t.start()
for t in thread_list:
t.join()
print("#### 线程等待 ####" + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
time.sleep(xiancheng_m)
real += 1
print('任务完成!')
main()
# okay decompiling __init__.pyc

推荐阅读

pycharm怎么用国内镜像安装第三方库

1、在pycharm中打开Terminal,如下图。 2、以安装pymysql库为例,输入以下命令回车即可使用镜像安装。 pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pymysql ......

win10系统安装软件错误”The installer has encountered an unexpected error…error code is 2503″

在Windows10系统中,错误“安装程序在安装此软件包时遇到意外错误。这可能表明包装有问题。错误代码是2503。安装python3.8.9时报告了。安装程序在安装此软件包时遇到了意外错误。这可能表明行李有问题。...

python报错requests.exceptions.ConnectionError

http连接数超过了最大限制。默认情况下,连接是保持活动的,因此服务器保持太多的连接以至于无法创建新的连接。请求速度太快。...