# -*- 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']
        #     if 'java.util.ArrayList' in searchEngines:
        #         searchEngines=searchEngines[1]
        # except Exception as e:
        #     searchEngines=[]
        kwList=[]
        searchEngines=['3']
        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:
                logger.info('+++++')
        else:
            logger.info('+++++searchEngines为空')
            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):
        searchkw=kwmsg['kw']
        wordsCode=kwmsg['wordsCode']
        sid=kwmsg['sid']
        qqnewsSpider=QQnewsSpider(searchkw,wordsCode,sid)
        try:
            qqnewsSpider.get_page_html()
        except Exception as e:
            logger.info('搜狗搜索异常'+searchkw)

        if qqnewsSpider.detailList.qsize() != 0:
            try:
                qqnewsSpider.get_detail_html()
            except Exception as e:
                logger.info('详情解析异常'+searchkw)
        logger.info("关键词采集结束！"+searchkw)
import random
if __name__ == '__main__':
    qqnewsTaskJob=QQnewsTaskJob()
    baseCore=BaseCore()
    logger=baseCore.getLogger()
    # ss='(中国机床工具工业协会|中国内燃机工业协会|中国机电工业价格协会|中国机械电子兵器船舶工业档案学会|中国仪器仪表行业协会|中国工程机械工业协会|中国文化办公设备制造行业协会|中国机械工业金属切削刀具技术协会|中国机械工业教育协会|中国汽车工业协会|中国机械通用零部件工业协会|中国环保机械行业协会|中国模具工业协会|中国机械工业勘察设计协会|中国机械制造工艺协会|中国机械工业审计学会|中国轴承工业协会|中国机电一体化技术应用协会|中国机械工程学会|中国液压气动密封件工业协会|中国铸造协会|中国通用机械工业协会|中国锻压协会|中国制冷空调工业协会|中国热处理行业协会|中国电工技术学会|中国仪器仪表学会|中国石油和石油化工设备工业协会|中国表面工程协会|中国食品和包装机械工业协会|中国焊接协会|中国汽车工程学会|中国塑料机械工业协会|中国机械工业企业管理协会|中国印刷及设备器材工业协会|中国机械工业质量管理协会|中国电器工业协会|中国机械工业安全卫生协会|中国重型机械工业协会|中国机械工业标准化技术协会|中国机械工业职工思想政治工作研究会|中国农业机械工业协会|中国机电装备维修与改造技术协会 |机械工业信息研究院|机械工业教育发展中心|机械工业经济管理研究院|机械工业信息中心|机械工业人才开发服务中心|机械工业北京电工技术经济研究所|机械工业技术发展基金会|机械工业哈尔滨焊接技术培训中心|机械工业仪器仪表综合技术经济研究所)+(私收会费|私吞|肆意牟利|损失浪费|索贿|贪财|贪官污吏|贪污|违背组织原则|违法|违纪|为官不廉|为政擅权|窝案|舞弊|泄露国家机密|信鬼神|性关系|虚假信息|虚假招标|隐瞒不报|隐瞒真相|营私|鬻爵|主动投案|资产流失|钻空子|钻漏洞|被调查|被双开|不担当|不老实|不良影响|不正当|不作为|超标准建设|超标准装修|吃空饷|吃拿卡要|渎职|对党不忠诚|非法批地|腐败|腐虫|腐化堕落|公车私用|公费开销|公款吃喝|公款出境|公款旅游|勾结|官迷心窍|好色|回扣|贿赂|挤占挪用|纪律审查|监察调查|监守自盗|践踏法律|接受审查调查|截留克扣|开除党籍|开除公职|抗议|利欲熏心|敛财|乱摊派|乱作为|落马|落网|买官|买卖审批权限|卖官|谋取暴利|谋取私利|目无法纪|幕后交易|弄虚作假|挪用公款|骗取|钱色交易|潜规则|侵害权益|侵吞公款|侵占挪用|圈子文化|权利扭曲|权钱交易|权色交易|山头主义|涉案|生活糜烂|生活奢靡|失察|失管|收送|受贿|双规|双开|私分|私人会所|私设小金库|负面|下降|违规|不利|亏损|上诉|不法|不良名单|停职|公开谴责|公诉|内幕交易|刑事拘留|刑事责任|刑拘|判决|判刑|判赔|司法处置|合同纠纷|处分|处罚|强制执行|仲裁|伪造|伪造公章|投案|投诉|拘留|接受调查|控诉|查封|涉嫌|涉诉监察调查|纠纷|经营异常名录|缉捕|罚单|罚款|罚金|罪犯|自首|获刑|行贿|警示函|贪腐|违约金|追究刑责|造假|逮捕|非法|非法集资判决书|申诉|纠纷|通报|开除|留党察看|追债|逃债|资产负债率|情色交易|搞权钱|曝光|黑料|重罚|虚假报告|侵犯)'
    # keymsglist=baiduTaskJob.getkeywords(ss)
    # print(keymsglist)
    # 创建Redis连接

    print('---------------')
    while True:
        try:
            codeList=[]
            codeList.append('KW-20220602-0003')
            for codeid in codeList:
                try:
                    keymsg=qqnewsTaskJob.getkeyFromredis(codeid)
                    kwList=qqnewsTaskJob.paserKeyMsg(keymsg)
                    if len(kwList)<1:
                        continue
                    logger.info(f"需要搜索的关键词:{kwList}")
                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(qqnewsTaskJob.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('采集异常')

