恢复但仅表上丢失的数据

我使用golang来解决这个问题：

tl;dw:

1. 读取所有电台表
2. 如果两者都存在，请选择源和目标的每个日期回顾 ( DATE(submitted_at), COUNT(1))
3. 如果目标的编号与源的编号不同，则插入到目标

    var src, dest *Pg.RDBMS
    {
        opt := `host=127.0.0.1 user=x password=x dbname=x sslmode=disable` 
        conn := sqlx.MustConnect(`postgres`, opt)
        conn.DB.SetMaxOpenConns(61) 
        dest = &Pg.RDBMS{
            Name:    `destination`,
            Adapter: conn,
        }
    }

    {
        opt := `host=127.0.0.1 user=y password=y dbname=y sslmode=disable` 
        conn := sqlx.MustConnect(`postgres`, opt)
        src = &Pg.RDBMS{
            Name:    `source`,
            Adapter: conn,
        }
    }


    srcStations := src.QStrMapMap(`imei`, `SELECT * FROM stations`)
    destStations := dest.QStrMapMap(`imei`, `SELECT * FROM stations`)

    // compare stations
    removeUnecessary := func(m M.SX) {
        delete(m, `last_level_sensor`)
        delete(m, `last_submitted_at`)
        delete(m, `updated_at`)
        delete(m, `updated_by`)
        delete(m, `name`)
        delete(m, `lat`)
        delete(m, `long`)
    }

    for imei := range srcStations {
        ss := srcStations.GetMSX(imei)
        removeUnecessary(ss)
        ds := destStations.GetMSX(imei)
        removeUnecessary(ds)

        // check the diff of stations table
        if len(ds) == 0 {
            fmt.Println(`missing from destination: `+imei, ss)
            // TODO: insert on destination
            continue
        } else if !reflect.DeepEqual(ss, ds) {
            fmt.Println(`different: ` + imei)
            diffs := pretty.Diff(ss, ds)
            for _, diff := range diffs {
                fmt.Println(diff)
            }
        }

        // if source different from destination, there might be one more than station created not with same order after moving server
        srcStationId := fmt.Sprint(ss[`id`])
        dstStationId := fmt.Sprint(ds[`id`])
        if srcStationId != dstStationId {
            fmt.Println(`source station id not the same as destination`, srcStationId, dstStationId)
        }

        // check whether tables exists
        if !src.TableExists(`station_logs_` + srcStationId) {
            fmt.Println(`missing source table station_logs_` + srcStationId)
            continue
        }
        if !dest.TableExists(`station_logs_` + dstStationId) {
            fmt.Println(`missing destination table station_logs_` + dstStationId)
            continue
        }

        srcSelect1 := fmt.Sprintf(`SELECT DATE(submitted_at)::TEXT, COUNT(1) FROM station_logs_%s GROUP BY 1`, srcStationId)
        srcDtCount := src.QStrIntMap(srcSelect1)

        dstSelect1 := fmt.Sprintf(`SELECT DATE(submitted_at)::TEXT, COUNT(1) FROM station_logs_%s GROUP BY 1`, dstStationId)
        dstDtCount := dest.QStrIntMap(dstSelect1)

        fmt.Println(`comparing station_logs_`, srcStationId, dstStationId)

        for submittedAt, srcCount := range srcDtCount {
            dstCount := dstDtCount[submittedAt]
            if srcCount != dstCount {
                fmt.Println(`different count: `, srcStationId, submittedAt, srcCount, dstCount)

                query2 := fmt.Sprintf(`SELECT * FROM station_logs_%s WHERE DATE(submitted_at) = '%s'`, srcStationId, submittedAt)

                srcLogs := src.QStrMapMap(`id`, query2)

                dstInsertLogs := fmt.Sprintf(`
INSERT INTO station_logs_%s (
    created_at, updated_at, submitted_at, sequence, level_sensor, accel_x, accel_y, accel_z, power_current, ip_address, log_type, station_id, power_voltage, data, is_deleted, temperature, wind_speed, soil_moisture, wind_direction, raindrop, humidity, barometric_pressure, wind_speed_average, wind_gust, wind_direction_average, rain_rate
) VALUES (
            $1,         $2,        $3,           $4,           $5,         $6,      $7,      $8,      $9,            $10,       $11,      $12,        $13,           $14,   $15,        $16,         $17,        $18,           $19,         $20,      $21,                 $22,                 $23,       $24,                  $25,        $26 
)`, dstStationId)
                query3 := fmt.Sprintf(`
SELECT COUNT(1) FROM station_logs_%s WHERE submitted_at = $1`, dstStationId)

                skipped := 0
                dest.DoTransaction(func(tx *Pg.Tx) string {
                    for id := range srcLogs {
                        log := srcLogs.GetMSX(id)
                        if tx.QInt(query3, log[`submitted_at`]) > 0 {
                            skipped++
                            continue
                        }

                        tx.DoExec(dstInsertLogs,
                            log[`created_at`],
                            log[`updated_at`],
                            log[`submitted_at`],
                            log[`sequence`],
                            log[`level_sensor`],
                            log[`accel_x`],
                            log[`accel_y`],
                            log[`accel_z`],
                            log[`power_current`],
                            log[`ip_address`],
                            log[`log_type`],
                            log[`station_id`],
                            log[`power_voltage`],
                            log[`data`],
                            log[`is_deleted`],
                            log[`temperature`],
                            log[`wind_speed`],
                            log[`soil_moisture`],
                            log[`wind_direction`],
                            log[`raindrop`],
                            log[`humidity`],
                            log[`barometric_pressure`],
                            log[`wind_speed_average`],
                            log[`wind_gust`],
                            log[`wind_direction_average`],
                            log[`rain_rate`],
                        )
                    }
                    return ``
                })

                fmt.Println(`inserted: `, srcStationId, submittedAt, len(srcLogs), `skipped`, skipped)

            }

        }
   }