
import json
import requests, time, pymysql
import jieba
import sys
from kafka import KafkaProducer
from getTycId import getTycIdByXYDM
from base.BaseCore import BaseCore
from base.smart import smart_extractor
# sys.path.append('D:\\KK\\zzsn_spider\\base')
# import BaseCore
# from smart import smart_extractor
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# 初始化，设置中文分词
jieba.cut("必须加载jieba")
smart =smart_extractor.SmartExtractor('cn')
baseCore = BaseCore()
log = baseCore.getLogger()
cnx = pymysql.connect(host='114.116.44.11', user='caiji', password='f7s0&7qqtK', db='dbScore', charset='utf8mb4')
cursor = cnx.cursor()

pageSize = 10
headers = {
    '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',
    'Accept-Encoding': 'gzip, deflate, br',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
    'Connection': 'keep-alive',
    'Content-Type': 'application/json',
    'Cookie': 'jsid=SEO-BAIDU-ALL-SY-000001; TYCID=77e997401d5f11ee9e91d5a0fd3c0b89; ssuid=6450041974; _ga=GA1.2.858826166.1688800641; _gid=GA1.2.2142449376.1689575510; tyc-user-info-save-time=1689764135027; sensorsdata2015jssdkcross=%7B%22distinct_id%22%3A%22309757777%22%2C%22first_id%22%3A%22189345cb10257d-0cfee05327f673-26031d51-1327104-189345cb10375b%22%2C%22props%22%3A%7B%22%24latest_traffic_source_type%22%3A%22%E8%87%AA%E7%84%B6%E6%90%9C%E7%B4%A2%E6%B5%81%E9%87%8F%22%2C%22%24latest_search_keyword%22%3A%22%E6%9C%AA%E5%8F%96%E5%88%B0%E5%80%BC%22%2C%22%24latest_referrer%22%3A%22https%3A%2F%2Fwww.baidu.com%2Flink%22%7D%2C%22identities%22%3A%22eyIkaWRlbnRpdHlfY29va2llX2lkIjoiMTg5MzQ1Y2IxMDI1N2QtMGNmZWUwNTMyN2Y2NzMtMjYwMzFkNTEtMTMyNzEwNC0xODkzNDVjYjEwMzc1YiIsIiRpZGVudGl0eV9sb2dpbl9pZCI6IjMwOTc1Nzc3NyJ9%22%2C%22history_login_id%22%3A%7B%22name%22%3A%22%24identity_login_id%22%2C%22value%22%3A%22309757777%22%7D%2C%22%24device_id%22%3A%22189345cb10257d-0cfee05327f673-26031d51-1327104-189345cb10375b%22%7D; bannerFlag=true; Hm_lvt_e92c8d65d92d534b0fc290df538b4758=1689752829,1689821665,1689831487,1689845884; searchSessionId=1689845917.81838207; HWWAFSESID=146bb1d25b1515339d3; HWWAFSESTIME=1689858023324; Hm_lpvt_e92c8d65d92d534b0fc290df538b4758=1689859758',
    'Host': 'capi.tianyancha.com',
    'Origin': 'https://www.tianyancha.com',
    'Referer': 'https://www.tianyancha.com/',
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36 Edg/114.0.1823.51'
}
cnx_ = baseCore.cnx
cursor_ = baseCore.cursor

taskType = '企业动态/天眼查'

def beinWork(tyc_code, social_code,start_time):

    time.sleep(3)
    # retData={'up_state':False,'total':0,'up_okCount':0,'up_errorCount':0,'up_repetCount':0}
    retData = {'total': 0, 'up_okCount': 0, 'up_errorCount': 0, 'up_repetCount': 0}
    t = time.time()
    url = f'https://capi.tianyancha.com/cloud-yq-news/company/detail/publicmsg/news/webSimple?_={t}&id={tyc_code}&ps={pageSize}&pn=1&emotion=-100&event=-100'
    for m in range(0, 3):
        try:
            ip = baseCore.get_proxy()
            headers['User-Agent'] = baseCore.getRandomUserAgent()
            response = requests.get(url=url, headers=headers, proxies=ip, verify=False)
            # time.sleep(random.randint(3, 5))
            break
        except Exception as e:
            pass

    if (response.status_code == 200):
        pass
    else:
        log.error(f"{tyc_code}-----获取总数接口失败")
        e = '获取总数接口失败'
        state = 0
        takeTime = baseCore.getTimeCost(start_time, time.time())
        baseCore.recordLog(social_code, taskType, state, takeTime, url, e)
        return retData
    try:
        json_1 = json.loads(response.content.decode('utf-8'))
        total = json_1['data']['total']
    except:
        log.error(f"{tyc_code}-----获取总数失败")
        e = '获取总数失败'
        state = 0
        takeTime = baseCore.getTimeCost(start_time, time.time())
        baseCore.recordLog(social_code, taskType, state, takeTime, url, e)
        return retData
    if (total > 0):
        if (total % pageSize == 0):
            totalPage = total // pageSize
        else:
            totalPage = total // pageSize + 1
    else:
        log.error(f"{tyc_code}--------总数为0")
        retData['state'] = True
        return retData
    log.info(f"{tyc_code}-------总数：{total}----总页数:{totalPage}")

    retData['total'] = total
    up_okCount = 0
    up_errorCount = 0
    up_repetCount = 0
    for num in range(1, totalPage + 1):
        time.sleep(3)
        log.info(f"获取分页数据--{tyc_code}----分页{num}----开始")
        start_page = time.time()
        url_page = f'https://capi.tianyancha.com/cloud-yq-news/company/detail/publicmsg/news/webSimple?_={time.time()}&id={tyc_code}&ps={pageSize}&pn={num}&emotion=-100&event=-100'
        for m in range(0, 3):
            try:
                ip = baseCore.get_proxy()
                headers['User-Agent'] = baseCore.getRandomUserAgent()
                response_page = requests.get(url=url_page, headers=headers, proxies=ip, verify=False)
                # time.sleep(3)
                break
            except:
                pass

        if (response_page.status_code == 200):
            pass
        else:
            log.error(f"{tyc_code}--{num}页---获取分页数据失败")
            e = '获取分页数据失败'
            state = 0
            takeTime = baseCore.getTimeCost(start_time, time.time())
            baseCore.recordLog(social_code, taskType, state, takeTime, url_page, e)
            up_errorCount = up_errorCount + pageSize
            continue
        try:
            json_page = json.loads(response_page.content.decode('utf-8'))
            info_list_page = json_page['data']['items']
        except:
            log.error(f"{tyc_code}--{num}页---获取分页数据失败")
            e = '获取分页数据失败'
            state = 0
            takeTime = baseCore.getTimeCost(start_time, time.time())
            baseCore.recordLog(social_code, taskType, state, takeTime, url_page, e)
            up_errorCount = up_errorCount + pageSize
            continue
        pageIndex = 0
        for info_page in info_list_page:
            pageIndex = pageIndex + 1
            title = info_page['title']
            source = info_page['website']
            link = info_page['uri']
            try:
                sel_sql = '''select social_credit_code from brpa_source_article where source_address = %s and social_credit_code=%s and type='2' '''
                cursor_.execute(sel_sql, (link, social_code))
            except Exception as e:
                print(e)
            selects = cursor_.fetchone()
            if selects:
                log.info(f'{tyc_code}-----{social_code}----{link}:已经存在')

                # todo:如果该条数据存在则说明该条数据之后的都已经采集完成，就可以跳出函数，执行下一个企业

                retData['up_okCount'] = up_okCount
                retData['up_errorCount'] = up_errorCount
                retData['up_repetCount'] = up_repetCount
                return retData
            try:
                time_struct = time.localtime(int(info_page['rtm'] / 1000))  # 首先把时间戳转换为结构化时间
                time_format = time.strftime("%Y-%m-%d %H:%M:%S", time_struct)  # 把结构化时间转换为格式化时间
            except:
                time_format = baseCore.getNowTime(1)
            try:
                # 开始进行智能解析
                # lang = baseCore.detect_language(title)
                # smart = smart_extractor.SmartExtractor(lang)
                #带标签正文
                contentText = smart.extract_by_url(link).text
                #不带标签正文
                content = smart.extract_by_url(link).cleaned_text
                # time.sleep(3)
            except Exception as e:
                contentText = ''
            if contentText == '':
                log.error(f'获取正文失败：--------{tyc_code}--------{num}--------{link}')
                e = '获取正文失败'
                state = 0
                takeTime = baseCore.getTimeCost(start_time, time.time())
                baseCore.recordLog(social_code, taskType, state, takeTime, link, e)
                up_errorCount = up_errorCount + 1
                try:
                    insert_err_sql = f"insert into dt_err(xydm,`from`,url,title,pub_date,zhaiyao,create_date,state,pageNo,pageIndex) values('{social_code}','{source}','{link}','{title}','{time_format}','{info_page['abstracts']}',now(),1,{num},{pageIndex})"
                    cursor.execute(insert_err_sql)
                    cnx.commit()
                except:
                    pass
                continue
            try:
                insert_sql = '''insert into brpa_source_article(social_credit_code,source_address,origin,type,create_time) values(%s,%s,%s,%s,now())'''
                # 动态信息列表
                up_okCount = up_okCount + 1
                list_info = [
                    social_code,
                    link,
                    '天眼查',
                    '2',
                ]
                cursor_.execute(insert_sql, tuple(list_info))
                cnx_.commit()
                # 采集一条资讯记录一条，记录该企业采到了多少的资讯
                log.info(f'{social_code}----{link}:新增一条')
                time_now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                # todo:插入一条数据，并传入kafka
                dic_news = {
                    'attachmentIds': '',
                    'author': '',
                    'content': content,
                    'contentWithTag': contentText,
                    'createDate': time_now,
                    'deleteFlag': '0',
                    'id': '',
                    'keyWords': '',
                    'lang': 'zh',
                    'origin': '天眼查',
                    'publishDate': time_format,
                    'sid': '1684032033495392257',
                    'sourceAddress': link,  # 原文链接
                    'summary': info_page['abstracts'],
                    'title': title,
                    'type': 2,
                    'socialCreditCode': social_code,
                    'year': time_format[:4]
                }
            except Exception as e:
                log.info(f'传输失败:{social_code}----{link}')
                e = '数据库传输失败'
                state = 0
                takeTime = baseCore.getTimeCost(start_time, time.time())
                baseCore.recordLog(social_code, taskType, state, takeTime, link, e)
                continue
                # print(dic_news)
                # 将相应字段通过kafka传输保存
            try:
                producer = KafkaProducer(bootstrap_servers=['114.115.159.144:9092'])
                kafka_result = producer.send("researchReportTopic",
                                             json.dumps(dic_news, ensure_ascii=False).encode('utf8'))

                print(kafka_result.get(timeout=10))
                dic_result = {
                    'success': 'ture',
                    'message': '操作成功',
                    'code': '200',
                }
                log.info(dic_result)
                # 传输成功,写入日志中
                state = 1
                takeTime = baseCore.getTimeCost(start_time, time.time())
                baseCore.recordLog(social_code, taskType, state, takeTime, link, '')
                # return True
            except Exception as e:
                dic_result = {
                    'success': 'false',
                    'message': '操作失败',
                    'code': '204',
                    'e': e
                }
                log.error(dic_result)
                e = 'Kafka操作失败'
                state = 0
                takeTime = baseCore.getTimeCost(start_time, time.time())
                baseCore.recordLog(social_code, taskType, state, takeTime, link, e)

        log.info(f"获取分页数据--{tyc_code}----分页{num}，耗时{baseCore.getTimeCost(start_page, time.time())}")
    retData['up_okCount'] = up_okCount
    retData['up_errorCount'] = up_errorCount
    retData['up_repetCount'] = up_repetCount
    return retData

# 日志信息保存至现已创建好数据库中,因此并没有再对此前保存日志信息数据库进行保存
def doJob():
    while True:
        # 根据从Redis中拿到的社会信用代码,在数据库中获取对应基本信息
        # social_code = baseCore.redicPullData('NewsEnterprise:gnqy_socialCode')
        social_code = '912301001275921118'
        # 判断 如果Redis中已经没有数据，则等待
        if social_code == None:
            time.sleep(20)
            continue
        start = time.time()
        try:
            data = baseCore.getInfomation(social_code)
            if len(data) != 0:
                pass
            else:
                #数据重新塞入redis
                baseCore.rePutIntoR('NewsEnterprise:gnqy_socialCode',social_code)
                continue
            id = data[0]
            xydm = data[2]
            tycid = data[11]
            if tycid == None or tycid == '':
                try:
                    retData = getTycIdByXYDM(xydm)
                    if retData['tycData'] and retData['reput']:
                        tycid = retData['tycData']['id']
                        # todo:写入数据库
                        updateSql = f"update EnterpriseInfo set TYCID = '{tycid}' where SocialCode = '{xydm}'"
                        cursor_.execute(updateSql)
                        cnx_.commit()
                    elif not retData['tycData'] and retData['reput']:
                        state = 0
                        takeTime = baseCore.getTimeCost(start, time.time())
                        baseCore.recordLog(social_code, taskType, state, takeTime, '', '获取天眼查id失败')
                        log.info(f'======={social_code}====重新放入redis====')
                        baseCore.rePutIntoR('NewsEnterprise:gnqy_socialCode', social_code)
                        continue
                    elif not retData['reput'] and not retData['tycData']:
                        continue
                except:
                    state = 0
                    takeTime = baseCore.getTimeCost(start, time.time())
                    baseCore.recordLog(social_code, taskType, state, takeTime, '', '获取天眼查id失败')
                    baseCore.rePutIntoR('NewsEnterprise:gnqy_socialCode', social_code)
                    continue
            count = data[17]
            log.info(f"{id}---{xydm}----{tycid}----开始处理")
            start_time = time.time()
            # 开始采集企业动态
            retData = beinWork(tycid, xydm,start_time)
            # 信息采集完成后将该企业的采集次数更新
            runType = 'NewsRunCount'
            count += 1
            baseCore.updateRun(social_code, runType, count)
            total = retData['total']
            up_okCount = retData['up_okCount']
            up_errorCount = retData['up_errorCount']
            up_repetCount = retData['up_repetCount']
            log.info(
                f"{id}---{xydm}----{tycid}----结束处理，耗时{baseCore.getTimeCost(start_time, time.time())}---总数:{total}---成功数:{up_okCount}----失败数:{up_errorCount}--重复数:{up_repetCount}")
        except Exception as e:
            log.info(f'==={social_code}=====获取企业信息失败====')
            #重新塞入redis
            baseCore.rePutIntoR('NewsEnterprise:gnqy_socialCode',social_code)
            state = 0
            takeTime = baseCore.getTimeCost(start, time.time())
            baseCore.recordLog(social_code, taskType, state, takeTime, '', f'获取企业信息失败--{e}')
            time.sleep(5)

    cursor.close()
    cnx.close()
    # 释放资源
    baseCore.close()

if __name__ == '__main__':

    doJob()

