MongoDB如何查询产品最后更新时间 用Python筛选指定周期未更新商品

前置依赖 #

需要先安装pymongo依赖包：
pip install pymongo
如果需要精确计算自然月间隔，可额外安装dateutil工具包：
pip install python-dateutil

核心逻辑说明 #

我们的筛选目标是：商品的价格历史item_price_history中，不存在符合指定时间范围的记录。实现上直接用MongoDB的聚合查询完成计算，避免全量拉取数据到本地处理，性能更高。

完整代码实现 #

from pymongo import MongoClient
from datetime import datetime, timedelta, timezone
# 若使用精确自然月计算，取消下面注释
# from dateutil.relativedelta import relativedelta

# -------------------------- 配置部分 请根据实际情况修改 --------------------------
# MongoDB连接配置
MONGO_URI = "mongodb://localhost:27017/"
DB_NAME = "你的数据库名"
COLL_NAME = "你的商品集合名"

# 筛选规则配置，二选一即可
# 规则1：筛选近N个月未更新价格的商品，此处N设为2
USE_RULE = "recent_months"
MONTHS_THRESHOLD = 2
# 注：如果需要精确的自然月计算，可替换下方threshold_time的计算逻辑为relativedelta(months=MONTHS_THRESHOLD)

# 规则2：筛选指定日期范围内无价格更新的商品，时间格式为YYYY-MM-DD
# USE_RULE = "custom_range"
# START_DATE = "2024-01-01"
# END_DATE = "2024-03-01"
# -----------------------------------------------------------------------------

# 初始化MongoDB连接
client = MongoClient(MONGO_URI)
coll = client[DB_NAME][COLL_NAME]

# 生成查询时间条件
if USE_RULE == "recent_months":
    # 计算N个月前的UTC时间（和MongoDB存储的ISODate时区保持一致）
    threshold_time = datetime.now(timezone.utc) - timedelta(days=MONTHS_THRESHOLD * 30)
    # 聚合查询：取最新价格更新时间早于阈值的商品
    pipeline = [
        # 提取每个商品的最新价格更新时间
        {
            "$addFields": {
                "last_update_time": {"$max": "$item_price_history.date"}
            }
        },
        # 筛选最新更新时间早于阈值，或无价格历史的商品
        {
            "$match": {
                "$or": [
                    {"last_update_time": {"$lt": threshold_time}},
                    {"last_update_time": {"$exists": False}}
                ]
            }
        },
        # 可选：返回需要的字段，不需要的可以调整
        {
            "$project": {
                "market": 1,
                "product": 1,
                "last_update_time": 1,
                "_id": 0
            }
        }
    ]
else:
    # 转换自定义日期为UTC时间
    start_dt = datetime.strptime(START_DATE, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    end_dt = datetime.strptime(END_DATE, "%Y-%m-%d").replace(tzinfo=timezone.utc)
    # 聚合查询：筛选指定时间范围内无任何价格记录的商品
    pipeline = [
        {
            "$match": {
                "item_price_history": {
                    "$not": {
                        "$elemMatch": {
                            "date": {"$gte": start_dt, "$lte": end_dt}
                        }
                    }
                }
            }
        },
        # 可选：返回需要的字段
        {
            "$project": {
                "market": 1,
                "product": 1,
                "_id": 0
            }
        }
    ]

# 执行查询获取结果
result = list(coll.aggregate(pipeline))

# 结果处理示例：打印符合条件的商品
print(f"共找到{len(result)}个符合条件的商品：")
for item in result:
    print(item)

验证方法 #

如果你需要在Robo 3T中验证查询结果，直接把上面代码中对应场景的pipeline数组内容复制到Robo 3T的聚合查询编辑器中执行即可，结果和Python脚本返回的完全一致。

注意事项 #

• 如果你的MongoDB存储的时间不是UTC时区，需要把代码中生成时间的时区调整为和数据库一致，避免时间偏移导致筛选结果错误
• 如果商品数据量极大，可以给item_price_history.date加索引提升查询性能

内容的提问来源于stack exchange，提问作者juanguit