实用脚本能自动查询MongoDB数据吗？

wen 实用脚本 2026-07-21 2

本文目录导读：

Shell脚本 + mongosh（推荐）
Python脚本（功能最强）
Node.js脚本（适合全栈开发者）
实用的自动化脚本模式
远程执行和定时任务
使用建议

是的,完全可以通过实用脚本来自动查询MongoDB数据，根据你的具体需求（定时查询、条件筛选、结果处理等），有多种实现方式，以下是几种主流方案及其示例：

Shell脚本 + mongosh（推荐）

基础查询脚本

#!/bin/bash
# query_mongo.sh
# MongoDB连接信息
HOST="localhost"
PORT="27017"
DATABASE="mydb"
COLLECTION="users"
# 执行查询
mongosh --host $HOST --port $PORT $DATABASE --eval "
  db.getCollection('$COLLECTION').find(
    { status: 'active', age: { \$gte: 18 } },
    { name: 1, email: 1, age: 1 }
  ).limit(10)
" > query_result.json 2>&1
echo "查询完成，结果已保存到 query_result.json"

带参数和格式化的脚本

#!/bin/bash
# 支持参数传递
MIN_AGE=${1:-18}
STATUS=${2:-active}
mongosh --quiet --host localhost:27017 mydb --eval "
  const results = db.users.find(
    { status: '$STATUS', age: { \$gte: $MIN_AGE } }
  ).toArray();
  // 格式化输出
  print(JSON.stringify(results, null, 2));
"

Python脚本（功能最强）

使用 pymongo 库

#!/usr/bin/env python3
# query_mongo.py
from pymongo import MongoClient
from datetime import datetime, timedelta
import json
def query_users_by_criteria():
    # 连接MongoDB
    client = MongoClient('mongodb://localhost:27017/')
    db = client['mydb']
    collection = db['users']
    # 构建查询条件
    pipeline = [
        {
            '$match': {
                'status': 'active',
                'created_at': {
                    '$gte': datetime.now() - timedelta(days=7)
                }
            }
        },
        {
            '$group': {
                '_id': '$department',
                'count': {'$sum': 1},
                'avg_age': {'$avg': '$age'}
            }
        },
        {'$sort': {'count': -1}}
    ]
    # 执行聚合查询
    results = list(collection.aggregate(pipeline))
    # 保存到文件
    with open('query_result.json', 'w') as f:
        json.dump(results, f, default=str, indent=2)
    print(f"查询完成，共 {len(results)} 条记录")
    return results
if __name__ == "__main__":
    query_users_by_criteria()

带定时任务的脚本

#!/usr/bin/env python3
# scheduled_query.py
import schedule
import time
from pymongo import MongoClient
from datetime import datetime
def daily_report():
    client = MongoClient('mongodb://localhost:27017/')
    db = client['analytics']
    # 查询昨日数据
    yesterday = datetime.now() - timedelta(days=1)
    results = db.events.find({
        'timestamp': {
            '$gte': yesterday.replace(hour=0, minute=0, second=0),
            '$lt': datetime.now().replace(hour=0, minute=0, second=0)
        }
    })
    # 处理数据并发送邮件/保存报告
    with open(f'report_{datetime.now().strftime("%Y%m%d")}.json', 'w') as f:
        json.dump(list(results), f, default=str)
# 每天凌晨1点执行
schedule.every().day.at("01:00").do(daily_report)
while True:
    schedule.run_pending()
    time.sleep(60)

Node.js脚本（适合全栈开发者）

#!/usr/bin/env node
// query_mongo.js
const { MongoClient } = require('mongodb');
const fs = require('fs').promises;
async function queryData() {
    const uri = 'mongodb://localhost:27017';
    const client = new MongoClient(uri);
    try {
        await client.connect();
        const database = client.db('mydb');
        const collection = database.collection('orders');
        // 复杂的聚合查询
        const pipeline = [
            {
                '$match': {
                    'status': 'completed',
                    'total': { '$gt': 100 }
                }
            },
            {
                '$lookup': {
                    'from': 'customers',
                    'localField': 'customer_id',
                    'foreignField': '_id',
                    'as': 'customer_info'
                }
            },
            { '$unwind': '$customer_info' },
            {
                '$project': {
                    'order_id': 1,
                    'total': 1,
                    'customer_name': '$customer_info.name',
                    'customer_email': '$customer_info.email',
                    'order_date': 1
                }
            }
        ];
        const results = await collection.aggregate(pipeline).toArray();
        // 保存结果
        await fs.writeFile(
            'orders_report.json',
            JSON.stringify(results, null, 2)
        );
        console.log(`查询完成，获取到 ${results.length} 条订单`);
    } finally {
        await client.close();
    }
}
queryData().catch(console.error);

实用的自动化脚本模式

带配置文件和环境变量的脚本

#!/bin/bash
# config_loader.sh
# 从.env文件加载配置
source .env
# 执行灵活的查询
mongosh --quiet "$MONGODB_URI" --eval "
  const config = {
    collection: '${COLLECTION:-users}',
    condition: ${QUERY_CONDITION:-'{status: \"active\"}'}
  };
  print('正在查询集合: ' + config.collection);
  const db = db.getSiblingDB('${DATABASE:-mydb}');
  const results = db.getCollection(config.collection).find(
    JSON.parse(config.condition)
  ).limit(50).toArray();
  printjson(results);
"

组合多种操作的脚本

#!/usr/bin/env python3
# complex_query.py
import argparse
import json
from pymongo import MongoClient
from datetime import datetime
def main():
    parser = argparse.ArgumentParser(description='MongoDB查询工具')
    parser.add_argument('--collection', required=True, help='集合名称')
    parser.add_argument('--fields', nargs='+', help='需要返回的字段')
    parser.add_argument('--limit', type=int, default=100, help='限制返回条数')
    parser.add_argument('--output', default='output.json', help='输出文件')
    parser.add_argument('--sort', help='排序字段')
    parser.add_argument('--group', help='分组字段')
    args = parser.parse_args()
    client = MongoClient('mongodb://localhost:27017/')
    db = client['mydb']
    collection = db[args.collection]
    # 构建查询
    projection = {field: 1 for field in args.fields} if args.fields else {}
    cursor = collection.find(
        {},  # 条件留空或从参数传入
        projection
    )
    if args.sort:
        cursor = cursor.sort(args.sort, -1)  # 降序
    if args.limit:
        cursor = cursor.limit(args.limit)
    results = list(cursor)
    # 输出
    with open(args.output, 'w') as f:
        json.dump(results, f, default=str, indent=2)
    print(f"成功导出 {len(results)} 条记录到 {args.output}")
if __name__ == "__main__":
    main()

远程执行和定时任务

使用 cron 定时执行（Linux/Mac）

# crontab -e 添加以下行
# 每天凌晨2点执行数据查询
0 2 * * * /path/to/query_mongo.sh >> /var/log/mongo_query.log 2>&1
# 每小时检查一次数据
0 * * * * /path/to/check_data.py >> /var/log/check.log

使用 Windows 任务计划程序

# Windows PowerShell 脚本
$action = New-ScheduledTaskAction -Execute "C:\Python39\python.exe" -Argument "C:\scripts\query_mongo.py"
$trigger = New-ScheduledTaskTrigger -Daily -At "03:00AM"
Register-ScheduledTask -TaskName "MongoDB_Query" -Action $action -Trigger $trigger

使用建议

连接管理：始终使用连接池，避免频繁建立/关闭连接
错误处理：添加重试机制和错误日志
性能优化：使用索引、限制查询结果大小
安全性：避免硬编码凭据，使用环境变量或配置文件
日志记录：记录查询时间、结果数量等信息

这些脚本可以直接复制使用,根据你的具体需求调整参数和查询逻辑，如果你有特定的查询场景（如复杂的聚合管道、跨集合查询等），可以提供更多细节，我可以给出更精确的脚本示例。