通过 mgo 程序化导入 MongoDB 备份：BSON 与 JSON 策略

package main

import (
    "fmt"
    "os"
    "os/exec"
    "log"
)

// RestoreMongoDBBackup 使用 mongorestore 工具恢复 MongoDB 备份
func RestoreMongoDBBackup(dbName, backupPath string) error {
    // 构建 mongorestore 命令
    // --db 参数指定要恢复到的数据库名称
    // backupPath 通常指向 mongodump 生成的数据库目录，例如 /path/to/dump/your_database_name
    cmd := exec.Command("mongorestore", "--db", dbName, backupPath)

    // 可选：如果需要认证，可以添加 --username, --password, --authenticationDatabase 等参数
    // 例如：cmd = exec.Command("mongorestore", "--db", dbName, "--username", "admin", "--password", "pwd", "--authenticationDatabase", "admin", backupPath)

    // 执行命令并捕获标准输出和标准错误
    output, err := cmd.CombinedOutput()
    if err != nil {
        log.Printf("执行 mongorestore 失败: %v\n", err)
        log.Printf("mongorestore 输出: \n%s\n", output)
        return fmt.Errorf("mongorestore 命令执行失败: %v, 输出: %s", err, output)
    }

    fmt.Printf("mongorestore 成功完成。恢复到数据库: %s\n", dbName)
    fmt.Printf("mongorestore 输出:\n%s\n", output)
    return nil
}

func main() {
    targetDB := "my_new_database"
    // 假设 mongodump 备份文件位于 /tmp/dump/my_old_database 目录下
    // 请替换为实际的备份路径
    pathToBackup := "/tmp/dump/my_old_database" 

    // 检查 mongorestore 工具是否存在
    if _, err := exec.LookPath("mongorestore"); err != nil {
        log.Fatalf("错误: 找不到 mongorestore 工具。请确保已安装 MongoDB 数据库工具并配置 PATH 环境变量。")
    }

    fmt.Printf("开始恢复 MongoDB 备份到数据库 '%s'...\n", targetDB)
    err := RestoreMongoDBBackup(targetDB, pathToBackup)
    if err != nil {
        log.Fatalf("MongoDB 备份恢复失败: %v", err)
    }
    fmt.Println("MongoDB 备份恢复成功！")
}

package main

import (
    "bufio"
    "encoding/json"
    "fmt"
    "io"
    "log"
    "os"

    "gopkg.in/mgo.v2"
    "gopkg.in/mgo.v2/bson" // mgo 提供的 bson.M 类型，等同于 map[string]interface{}
)

// ImportJSONCollection 从 JSON 文件导入数据到 MongoDB 集合
func ImportJSONCollection(session *mgo.Session, dbName, collectionName, jsonFilePath string) error {
    file, err := os.Open(jsonFilePath)
    if err != nil {
        return fmt.Errorf("打开 JSON 文件失败: %v", err)
    }
    defer file.Close()

    c := session.DB(dbName).C(collectionName)
    bulk := c.Bulk() // 使用 mgo 的批量操作以提高性能

    scanner := bufio.NewScanner(file)
    importedCount := 0

    for scanner.Scan() {
        line := scanner.Bytes()
        if len(line) == 0 {
            continue // 跳过空行
        }

        var doc bson.M // 使用 bson.M (map[string]interface{}) 避免定义固定结构体
        err := json.Unmarshal(line, &doc)
        if err != nil {
            log.Printf("警告: 解析 JSON 行失败，跳过。行内容: %s, 错误: %v", string(line), err)
            continue
        }

        // 注意：如果 JSON 中包含 {$oid: "..."} 或 {$date: "..."} 等扩展 JSON 格式，
        // 默认的 json.Unmarshal 不会将其转换为 mgo.bson.ObjectId 或 time.Time。
        // 在这种情况下，你需要手动转换这些字段，或者编写自定义的 Unmarshaler。
        // 例如，对于 "$oid" 字段，你可能需要:
        // if oidStr, ok := doc["_id"].(map[string]interface{})["$oid"].(string); ok {
        //     doc["_id"] = bson.ObjectIdHex(oidStr)
        // }

        bulk.Insert(doc)
        importedCount++
    }

    if err := scanner.Err(); err != nil && err != io.EOF {
        return fmt.Errorf("读取 JSON 文件时发生错误: %v", err)
    }

    // 执行批量插入
    result, err := bulk.Run()
    if err != nil {
        return fmt.Errorf("批量插入文档失败: %v", err)
    }

    fmt.Printf("成功导入 %d 个文档到 %s.%s (批量插入成功 %d 次)\n", importedCount, dbName, collectionName, result.Inserted)
    return nil
}

func main() {
    session, err := mgo.Dial("localhost:27017") // 替换为你的 MongoDB 连接字符串
    if err != nil {
        log.Fatalf("连接 MongoDB 失败: %v", err)
    }
    defer session.Close()

    targetDB := "my_new_database"
    targetCollection := "my_collection"
    // 假设 mongoexport 导出的文件为 /tmp/my_collection.json
    // 请替换为实际的 JSON 文件路径
    jsonFilePath := "/tmp/my_collection.json" 

    fmt.Printf("开始从 '%s' 导入数据到 %s.%s...\n", jsonFilePath, targetDB, targetCollection)
    err = ImportJSONCollection(session, targetDB, targetCollection, jsonFilePath)
    if err != nil {
        log.Fatalf("JSON 数据导入失败: %v", err)
    }
    fmt.Println("JSON 数据导入成功！")
}