quality_tools/tools/test.py

import pandas as pd
from pymongo import MongoClient
from bson import ObjectId, json_util
import json

# 连接到 MongoDB
client = MongoClient('172.20.45.129', 27002)
db = client['data_quality']
collection = db['standard_sample_data']


def insert_without_escape(collection, record):
    """插入数据并确保引号不被转义"""
    try:
        # 1. 将 ObjectId 转为字符串（临时解决序列化问题）
        if '_id' in record and isinstance(record['_id'], ObjectId):
            record['_id'] = str(record['_id'])

        # 2. 手动处理引号（确保不被转义）
        if 'projectname' in record:
            record['projectname'] = record['projectname'].replace('\\"', '"')

        # 3. 使用 json_util 处理特殊类型
        record_json = json_util.dumps(record)
        record_for_db = json_util.loads(record_json)

        # 4. 还原 ObjectId
        if '_id' in record_for_db and isinstance(record['_id'], str):
            record_for_db['_id'] = ObjectId(record['_id'])

        # 5. 插入数据
        collection.replace_one(
            {"_id": record_for_db.get("_id", ObjectId())},
            record_for_db,
            upsert=True
        )
        return True
    except Exception as e:
        print(f"插入失败: {str(e)}")
        return False


# 示例记录
record = {
    "_id": ObjectId("6864eb6dd5d8e4081f7a6a79"),
    "projectname": '中职学校"物理、化学、生物探索实践工坊"实训室改造项目',
    "site": "甘肃省阳光招标采购平台",
    "budget": 138295.32
}

# 插入数据
if insert_without_escape(collection, record):
    print("✅ 插入成功")

    # 验证存储结果"_id": object_id
    doc = collection.find_one({"_id": ObjectId(record["_id"])})
    print("MongoDB 实际存储内容:")
    print(doc["projectname"])  # 应输出：中职学校"物理、化学、生物探索实践工坊"实训室改造项目
else:
    print("❌ 插入失败")