data_quality_server/client_mongo_mysql_liantong.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 bson import ObjectId

ReluClient = MongoDBInterface(ReluMongodb)

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


# 开始评估
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_liantong (mongoid,toptype,subtype, site, spidercode, channel,comeintime, area, city, district, score, error_type, create_time,
                agency,agencyperson,agencytel,bidamount,bidendtime,bidopentime,bidstarttime,bidway,budget,buyer,buyerperson,buyertel,com_package,docendtime,
                docstarttime,est_purchase_time,href,projectcode,projectname,publishtime,s_winner,title,winnerorder,winnerperson,winnertel) 
               VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%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_liantong 
        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 has_non_empty_qa(data):
    # 获取data字典
    data_dict = data.get('data', {})

    # 遍历所有键值对
    for key, value in data_dict.items():
        # 检查键以'_qa'结尾且值不为空
        if key.endswith('_qa') and value:  # value不为None、空字典、空列表等
            return True
    return False

def parse_timestamp(timestamp):
    if not timestamp:
        return None
    try:
        return datetime.fromtimestamp(int(timestamp))
    except (ValueError, TypeError):
        return None
def insert_bid_statistics (col,conn,query,batch_id):
    # 定义来源  1标讯简版2拟在建3新闻4预算5专项债
    data_source = 1
    # 使用聚合管道进行多条件统计
    pipeline = [
        {"$match": query},
        {"$facet": {
            "总量": [{"$count": "count"}],  # 添加总量统计
            "招标": [{"$match": {"toptype": "招标"}}, {"$count": "count"}],
            "预告": [{"$match": {"toptype": "预告"}}, {"$count": "count"}],
            "结果": [{"$match": {"toptype": "结果"}}, {"$count": "count"}],
            "其它": [{"$match": {"toptype": "其它"}}, {"$count": "count"}],
            "other": [{"$match": {"toptype": {"$nin": ["招标", "预告", "结果", "其它"]}}}, {"$count": "count"}]
        }}
    ]

    result = list(col.aggregate(pipeline))[0]

    # 提取统计结果
    count_total = result["总量"][0]["count"] if result["总量"] else 0
    count_zhaobiao = result["招标"][0]["count"] if result["招标"] else 0
    count_yugao = result["预告"][0]["count"] if result["预告"] else 0
    count_jieguo = result["结果"][0]["count"] if result["结果"] else 0
    count_qita = result["其它"][0]["count"] if result["其它"] else 0
    count_other = result["other"][0]["count"] if result["other"] else 0
    count_jieguo = count_jieguo + count_qita

    print(f"招标数量: {count_zhaobiao}")
    print(f"预告数量: {count_yugao}")
    print(f"结果数量: {count_jieguo}")
    print(f"其它类型数量: {count_other}")
    sql_query = """INSERT IGNORE INTO bid_statistics_liantong (zhaobiao_count,yugao_count,jieguo_count,other_count, total_count, batch_id,data_source) 
                   VALUES ( %s, %s, %s, %s, %s,%s,%s)"""
    params = (count_zhaobiao,count_yugao,count_jieguo,count_other,count_total,batch_id,data_source)
    MysqlUtil.insert_data(conn, sql_query, params)

def batch_load_data():
    """
    批量数据质量检查
    """
    # 获取今天的日期（字符串格式）
    today_date = datetime.now().strftime("%Y-%m-%d")
    yesterday_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
    # 获取昨天 00:00:00 的时间戳
    start_date = int(datetime.strptime(f"{yesterday_date} 00:00:00", "%Y-%m-%d %H:%M:%S").timestamp())
    # print("start_date",start_date)
    # 获取今天 00:00:00 的时间戳
    end_date = int(datetime.strptime(f"{today_date} 00:00:00", "%Y-%m-%d %H:%M:%S").timestamp())
    # print("end_date", end_date)
    # 规则查询,根据必要条件 公司名称（用户ID）、版本号
    rules_id = get_rule("中国联通简版-标讯", "v1.4")
    print(rules_id)
    # 初始化mysql
    conn = MysqlUtil.connect_to_mysql(host='172.20.45.129', port='4000', user='root', password='=PDT49#80Z!RVv52_z',database='quality')

    max_id = ObjectId("0" * 24)
    # max_id = ObjectId("688363ebf0c6ad8b095e2245")
    # 查询条件：_id >= max_id, appid匹配，且 createtime 在 [start_date, end_date] 之间
    query = {
        "_id": {"$gte": max_id},
        # "_id": max_id,
        "appid": "jyGQ1XQQsEAwNeSENOFR9D",
        "createtime": {"$gte": start_date, "$lte": end_date}
    }
    # mongo_client = MongoClient('mongodb://127.0.0.1:27087/', unicode_decode_error_handler="ignore",directConnection=True)  # 修改为你的连接地址
    mongo_client = MongoClient('mongodb://172.20.17.61:27080/', unicode_decode_error_handler="ignore",directConnection=True)   # 正式环境
    client = mongo_client.jyqyfw
    coll_user = client["usermail"]
    #存入数据总量，计算缺失率
    insert_bid_statistics(coll_user,conn,query,today_date)
    while True:
        try:
            for item in coll_user.find(query).sort("_id", 1):
                print("------数据处理开始--------")
                max_id = item["_id"]
                item["_id"] = str(item["_id"])
                print(f"正在处理数据: {max_id}")
                # 质量检查逻辑
                result = start_quality(item, rules_id, a2s_ip, topic, timeout)
                print(result)
                code = result.get("code")
                if code != 200:
                    # 数据出错，跳过
                    continue
                #只将有错误的数据存库
                if has_non_empty_qa(result):
                    data = result.get("data", {})

                    # 数据插入到 MySQL
                    toptype = item.get("toptype", "")
                    subtype = item.get("subtype", "")
                    site = item.get("site", "")
                    spidercode = item.get("spidercode", "")
                    channel = item.get("channel", "")
                    comeintime = item.get("comeintime", "")
                    comeintime = datetime.fromtimestamp(comeintime)
                    area = item.get("area", "")
                    city = item.get("city", "")
                    district = item.get("district", "")
                    #---
                    agency = item.get("agency", "")
                    agencyperson = item.get("agencyperson", "")
                    agencytel = item.get("agencytel", "")
                    bidamount = item.get("bidamount", "")
                    bidendtime = item.get("bidendtime", "")
                    bidendtime = parse_timestamp(bidendtime)
                    bidopentime = item.get("bidopentime", "")
                    bidopentime = parse_timestamp(bidopentime)
                    bidstarttime = item.get("bidstarttime", "")
                    bidstarttime = parse_timestamp(bidstarttime)
                    bidway = item.get("bidway", "")
                    budget = item.get("budget", "")
                    buyer = item.get("buyer", "")
                    buyerperson = item.get("buyerperson", "")
                    buyertel = item.get("buyertel", "")
                    com_package = item.get("com_package", "")  #json串
                    com_package =json.dumps(com_package)
                    docendtime = item.get("docendtime", "")
                    docendtime = parse_timestamp(docendtime)
                    docstarttime = item.get("docstarttime", "")
                    docstarttime = parse_timestamp(docstarttime)
                    est_purchase_time = item.get("est_purchase_time", "")
                    est_purchase_time = parse_timestamp(est_purchase_time)
                    href = item.get("href", "")
                    projectcode = item.get("projectcode", "")
                    projectname = item.get("projectname", "")
                    publishtime = item.get("publishtime", "")
                    publishtime = parse_timestamp(publishtime)
                    s_winner = item.get("s_winner", "")
                    title = item.get("title", "")
                    winnerorder = item.get("winnerorder", "")#json串
                    winnerorder = json.dumps(winnerorder)
                    winnerperson = item.get("winnerperson", "")
                    winnertel = item.get("winnertel", "")

                    #---
                    score = data.get("score", "")
                    error_type_data = json.dumps(data)
                    create_time = today_date

                    params = (item["_id"], toptype, subtype, site, spidercode,channel, comeintime, area, city, district, score, error_type_data,create_time,
                              agency,agencyperson,agencytel,bidamount,bidendtime,bidopentime,bidstarttime,bidway,budget,buyer,buyerperson,buyertel,com_package,docendtime,
                              docstarttime,est_purchase_time,href,projectcode,projectname,publishtime,s_winner,title,winnerorder,winnerperson,winnertel)
                    insert_batch_data(conn, params)
            print("---- 数据处理完成 ----")
            break
        except Exception as e:
            print(f"错误: {e}")
            import traceback
            traceback.print_exc()  # 打印完整堆栈信息
            time.sleep(10)
        finally:
            # 确保在循环结束后关闭连接
            conn.close()  # 关闭MySQL连接
            mongo_client.close()  # 关闭MongoDB连接


if __name__ == '__main__':
    batch_load_data()