data_quality_server/client_spider.py

# coding:utf-8
import time
from a2s.tools import json_serialize, json_deserialize
from a2s.a2s_client import a2s_execute
from docs.config import ReluMongodb
from util.mogodb_helper import MongoDBInterface
from pymongo import MongoClient
from util.mysql_tool import MysqlUtil
import json
from datetime import datetime, timedelta
from elasticsearch import Elasticsearch


ReluClient = MongoDBInterface(ReluMongodb)

# 评估服务配置
a2s_ip = "192.168.3.240:9090"
# a2s_ip = "172.17.0.11:9090"
topic = "quality_bid"
#本地测试用的主题
# topic = "test_quality_bid"
timeout = 180

# 获取当前时间
now = datetime.now()
current_datetime = now.strftime("%Y-%m-%d %H:%M:%S")
# 获取今天的日期
today = datetime.today()
# 获取昨天的日期
yesterday = today - timedelta(days=1)
# 获取昨天0点的时间
yesterday_midnight = datetime(yesterday.year, yesterday.month, yesterday.day)
# 获取今天0点的时间
today_midnight = datetime(today.year, today.month, today.day)
# 转换为Unix时间戳
start_date = int(yesterday_midnight.timestamp())
end_date = int(today_midnight.timestamp())

# ES 连接配置
es_host = "http://127.0.0.1:19800"
es_username = "jianyuGr"
es_password = "we3g8glKfe#"

# 初始化 Elasticsearch 客户端
es_client = Elasticsearch(es_host,http_auth=(es_username, es_password),retry_on_timeout=True) # 使用基本认证

# 开始评估
def start_quality(data: dict, rules_id: int, a2s_ip, topic, timeout, retry=3):
    # 本次不使用SSL，所以channel是不安全的
    row = {"data": data, "rules_id": rules_id}
    bytes_data = json_serialize(row)
    for t in range(retry):
        print("topic",topic)
        try:
            resp_data = a2s_execute(a2s_ip, topic, timeout, bytes_data)
            if resp_data is None:
                continue
            result = json_deserialize(resp_data)
            return result
        except Exception as e:
            print(e)
    return {}

# 获取规则ID
def get_rule(company, version):
    rule_id = ReluClient.find_rule_by_company(ReluMongodb["col"], company, version)
    return rule_id

def find_error_id(conn, cleaned_key, sub_value):
    """
    查找 error_dict 中的 id
    """
    query = """SELECT id FROM error_dict WHERE fields = %s AND error = %s"""
    params = (cleaned_key, sub_value)
    result = MysqlUtil.query_data(conn, query, params)
    #[(10,)]
    # 检查查询结果是否为空
    if not result:
        print(f"Error: No matching record found for fields={cleaned_key}, error={sub_value}")
        return None  # 或者返回一个默认值，根据需求而定

    record = result[0][0]
    return record

def insert_batch_data(conn, params):
    """
    执行批量插入数据
    """
    query = """INSERT IGNORE INTO bid_analysis (mongoid, site, spidercode, comeintime, area, city, district, score, error_type, spider_modified_time) 
               VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"""
    MysqlUtil.insert_data(conn, query, params)

def insert_dynamic_error_field(conn, cleaned_key, error_ids, mongoid):
    """
    动态插入 error_ids 到相应的 cleaned_key_error 字段
    """
    # 构造动态插入 SQL 语句，更新指定的 cleaned_key_error 字段
    query = f"""
        UPDATE bid_analysis 
        SET {cleaned_key}_error = %s 
        WHERE mongoid = %s
    """
    # 拼接多个 error_id，用分隔符分开
    error_ids_str = ','.join(map(str, error_ids))
    params = (error_ids_str, mongoid )

    MysqlUtil.update_data(conn, query, params)

def get_last_processed_id():
    """
    获取上次处理的最大 ID (例如从数据库或文件中读取)
    """
    # 这里假设从文件读取中断 ID，你也可以从数据库或 Redis 等存储获取
    try:
        with open('docs/last_processed_id.txt', 'r') as f:
            last_id = f.read().strip()
            if last_id:
                return last_id
            else:
                return None
    except FileNotFoundError:
        return None

def save_last_processed_id(last_id):
    """
    保存当前处理的最大 ID，用于恢复
    """
    with open('docs/last_processed_id.txt', 'w') as f:
        f.write(str(last_id))

def batch_load_data():
    """
    批量数据质量检查
    """
    # 规则查询,根据必要条件 公司名称（用户ID）、版本号
    rules_id = get_rule("北京剑鱼信息技术有限公司", "v1.2")
    print(rules_id)

    # 初始化mysql
    # conn = MysqlUtil.connect_to_mysql(host='192.168.3.14', port='4000', user='DataScBi', password='DT#Sc20221123Ht',database='quality')
    conn = MysqlUtil.connect_to_mysql(host='192.168.3.217', port='4000', user='root', password='=PDT49#80Z!RVv52_z',database='quality')
    # 初始化爬虫代码库
    collection = MongoClient(f'mongodb://{"127.0.0.1:27089"}/', unicode_decode_error_handler="ignore", directConnection=True)["editor"]["lua_logs_auditor"]

    # 获取上次处理的 ID，如果没有，则从头开始
    last_processed_id = get_last_processed_id()
    print(f"上次处理的 ID: {last_processed_id}")

    # 获取ES数据
    es_query = {
        "query": {
            "bool": {
                "filter": [
                    {
                        "range": {
                            "comeintime": {
                                "gte": start_date,
                                "lt": end_date
                            }
                        }
                    }
                ]
            }
        },
        "sort": [
            {"_id": {"order": "asc"}}  # 如果 comeintime 相同，再按 _id 排序
        ],
        "size": 100  # 每次返回的数据量
    }

    # 如果有上次处理的 ID，使用 `search_after` 进行分页
    if last_processed_id:
        es_query["search_after"] = [last_processed_id]  # 确保传入的是字符串类型的 _id

    try:
        # 使用 scroll API 来分批获取数据
        response = es_client.search(index="bidding", body=es_query, size=100)
        hits = response['hits']['hits']

        while hits:
            print(f"---- 批次开始 ----")
            max_id = None
            for hit in hits:
                item = hit["_source"]
                print("------一条数据开始--------")
                max_id = hit["_id"]
                print(f"正在处理数据: {max_id}")
                item["_id"] = str(hit["_id"])

                # 质量检查逻辑
                result = start_quality(item, rules_id, a2s_ip, topic, timeout)
                print(result)

                code = result.get("code")
                if code != 200:
                    # 数据出错，跳过
                    continue

                data = result.get("data", {})

                # 数据插入到 MySQL
                site = item.get("site", "")
                spidercode = item.get("spidercode", "")
                comeintime = item.get("comeintime", "")
                comeintime = datetime.fromtimestamp(comeintime)
                area = item.get("area", "")
                city = item.get("city", "")
                district = item.get("district", "")
                score = item.get("score", "")
                error_type_data = json.dumps(data)
                spider_modified_time = current_datetime

                info = collection.find_one({"code": spidercode})
                if info:
                    spider_modified_time = info.get("modifytime", "")
                    spider_modified_time = datetime.fromtimestamp(spider_modified_time)

                params = (item["_id"], site, spidercode, comeintime, area, city, district, score, error_type_data,spider_modified_time)
                insert_batch_data(conn, params)

                # 遍历错误原因字典并提取非空字典中的值
                for key, value in data.items():
                    error_ids = []
                    if isinstance(value, dict) and value:
                        cleaned_key = key[:-3] if key.endswith('_qa') else key  # 去除 '_qa' 后缀
                        for sub_key, sub_value in value.items():
                            error_id = find_error_id(conn, cleaned_key, sub_value)
                            if error_id:
                                error_ids.append(error_id)
                            print(f"  {sub_key}: {sub_value}")

                        # 插入错误ID到cleaned_key_error字段
                        insert_dynamic_error_field(conn, cleaned_key, error_ids, item["_id"])
                print("------一条数据结束------")
            # 保存当前批次处理的最大 ID
            if max_id:
                save_last_processed_id(max_id)
                print(f"保存当前处理的最大 ID: {max_id}")
            # 批次结束的打印信息
            print("---- 当前批次数据处理完成 ----")

            # 获取下一批数据
            search_after = hits[-1]["_id"]  # 获取当前批次最后一条数据的 _id 作为下一批的起始点
            es_query["search_after"] = [search_after]  # 保持 _id 类型一致
            response = es_client.search(index="bidding", body=es_query, scroll="1m")
            hits = response['hits']['hits']

            # 如果没有更多数据，跳出循环
            if not hits:
                print("没有更多数据，结束批次处理")
                break
        print("数据处理完成")
    except Exception as e:
        print(f"错误: {e}")
        time.sleep(10)
    finally:
        if conn.is_connected():
            conn.close()  # 确保连接关闭
            print("MySQL 连接已关闭")

if __name__ == '__main__':
    batch_load_data()