# -*- coding: utf-8 -*-
import json
import requests,time,pymysql
import jieba
import sys

from kafka import KafkaProducer

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='root', password='f7s0&7qqtK', db='dbScore', charset='utf8mb4')
cursor= cnx.cursor()

cnx_ = baseCore.cnx
cursor_ = baseCore.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'
}
taskType = '企业动态/天眼查'
def beinWork(tyc_code, social_code):
    start_time = time.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
                # 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,title,summary,content,publish_date,source_address,origin,author,type,lang) values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)'''
                # 动态信息列表
                up_okCount = up_okCount + 1

                list_info = [
                    social_code,
                    title,
                    info_page['abstracts'],  # 摘要
                    contentText,  # 正文
                    time_format,  # 发布时间
                    link,
                    '天眼查',
                    source,
                    '2',
                    'zh'
                ]
                cursor.execute(insert_sql, tuple(list_info))
                cnx.commit()
                # 采集一条资讯记录一条，记录该企业采到了多少的资讯
                log.info(f'{social_code}----{link}:新增一条')
                sel_sql = "select article_id from brpa_source_article where source_address = %s and social_credit_code = %s"
                cursor.execute(sel_sql, (link, social_code))
                row = cursor.fetchone()
                id = row[0]
                time_now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                # todo:插入一条数据，并传入kafka
                dic_news = {
                    'attachmentIds': id,
                    'author': '',
                    'content': contentText,
                    'contentWithTag': contentText,
                    'createDate': time_now,
                    'deleteFlag': '0',
                    'id': '',
                    'keyWords': '',
                    'lang': 'zh',
                    'origin': '天眼查',
                    'publishDate': time_format,
                    'sid': '1684032033495392257',
                    'sourceAddress': link,  # 原文链接
                    'summary': info_page['abstracts'],
                    'title': contentText,
                    '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:
        # 获取企业信息
        query = "SELECT * FROM Tfbs_bak where col3 is not null and length(col3)>3 and col6 not like '%HK%' and col3 not like 'ZZSN%'   and  state3 is null  limit 1 "
        # 兴业银行
        # query = "SELECT * FROM Tfbs where col3 is not null and length(col3)>3  and col3 not like 'ZZSN%' and col5='兴业银行'"
        cursor_.execute(query)
        row = cursor_.fetchone()
        if row:
            pass
        else:
            print('没有数据了，结束脚本')
            break

        tycid = row[16]
        com_name = row[6]
        xydm = row[4]
        code = row[7]
        count = 0
        time_now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        #0 表示拿取数据
        updateBeginSql = f"update Tfbs_bak set state3='0',date3='{time_now}' where col3='{xydm}' "
        # print(updateBeginSql)
        cursor_.execute(updateBeginSql)
        cnx_.commit()
        log.info(f"{id}---{xydm}----{tycid}----开始处理")
        start_time = time.time()

        # 开始采集企业动态
        retData = beinWork(tycid, xydm)
        # 信息采集完成后将该企业的采集次数更新
        runType = 'NewsRunCount'
        count += 1
        # baseCore.updateRun(xydm, 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}")
        # 200 表示成功
        updateBeginSql = f"update Tfbs_bak set state3='200',date3='{time_now}' where col3='{xydm}' "
        # print(updateBeginSql)
        cursor_.execute(updateBeginSql)
        cnx_.commit()
    cursor.close()
    cnx.close()
    # 释放资源
    baseCore.close()

# Press the green button in the gutter to run the script.

#state3
if __name__ == '__main__':
    doJob()



