<commit_before>#!/usr/bin/env python

import copy
import errno
import glob
import logging
import os
import json
import string
import sys
import datetime
import time
import urllib
import fnmatch
import sqlite3

import couchbaseConstants
import pump

CBB_VERSION = [2004, 2014, 2015] # sqlite pragma user version.

DDOC_FILE_NAME = "design.json"
INDEX_FILE_NAME = "index.json"
FTS_FILE_NAME = "fts_index.json"

class BFD:
    """Mixin for backup-file/directory EndPoint helper methods."""
    NUM_VBUCKET = 1024

    def bucket_name(self):
        return self.source_bucket['name']

    def node_name(self):
        return self.source_node['hostname']

    @staticmethod
    def get_file_path(spec, bucket_name, file):
        bucket_name = urllib.quote_plus(bucket_name).encode('ascii')
        bucket_path = os.path.join(os.path.normpath(spec), "bucket-" + bucket_name)
        if os.path.isdir(bucket_path):
            return os.path.join(bucket_path, file)
        else:
            path, dirs = BFD.find_latest_dir(spec, None)
            if path:
                path, dirs = BFD.find_latest_dir(path, None)
                if path:
                    return os.path.join(path, "bucket-" + bucket_name, file)
        return os.path.join(bucket_path, file)

    @staticmethod
    def construct_dir(parent, bucket_name, node_name):
        return os.path.join(parent,
                    "bucket-" + urllib.quote_plus(bucket_name).encode('ascii'),
                    "node-" + urllib.quote_plus(node_name).encode('ascii'))

    @staticmethod
    def check_full_dbfiles(parent_dir):
        return glob.glob(os.path.join(parent_dir, "data-*.cbb"))

    @staticmethod
    def get_predecessors(parent_dir):
        try:
            json_file = open(os.path.join(parent_dir, "meta.json"), "r")
            json_data = json.load(json_file)
            json_file.close()
            return json_data["pred"]
        except IOError:
            return []

    @staticmethod
    def get_server_version(parent_dir):
        try:
            json_file = open(os.path.join(parent_dir, "meta.json"), "r")
            json_data = json.load(json_file)
            json_file.close()
            if "version" in json_data:
                return json_data["version"]
            return "0.0.0"
        except IOError:
            return "0.0.0"

    @staticmethod
    def get_failover_log(parent_dir):
        try:
            filepath = os.path.join(parent_dir, "failover.json")
            if os.path.isfile(filepath):
                json_file = open(filepath, "r")
                json_data = json.load(json_file)
                json_file.close()
                return json_data
            else:
                return {}
        except IOError:
            return {}

    @staticmethod
    def list_files(opts, spec, bucket, node, pattern):
        file_list = []
        prec_list = []

    @staticmethod
    def write_json_file(parent_dir, filename, output_data):
        filepath = os.path.join(parent_dir, filename)
        json_data = {}
        if os.path.isfile(filepath):
            #load into existed meta data generated from previous batch run
            try:
                json_file = open(filepath, "r")
                json_data = json.load(json_file)
                json_file.close()
            except IOError:
                pass

        for i in range(BFD.NUM_VBUCKET):
            if output_data.get(i):
                str_index = str(i)
                historic_data_exist = json_data.get(str_index)
                if historic_data_exist:
                    #Historic data will share same data type as incoming ones.
                    #It will be sufficient to only check type for historic data
                    if isinstance(json_data[str_index], int):
                        #For seqno, we want to keep the highest seqno
                        if json_data[str_index] < output_data[i]:
                            json_data[str_index] = output_data[i]
                    elif isinstance(json_data[str_index], list):
                        #For each vbucket, we want to get the superset of its references
                        if len(json_data[str_index]) < len(output_data[i]):
                            json_data[str_index] = output_data[i]
                else:
                    #Bookkeeping the incoming one.
                    json_data[str_index] = output_data[i]

        json_file = open(filepath, "w")
        json.dump(json_data, json_file, ensure_ascii=False)
        json_file.close()

    @staticmethod
    def db_dir(spec, bucket_name, node_name, tmstamp=None, mode=None, new_session=False):
        parent_dir = os.path.normpath(spec) + \
                        '/bucket-' + urllib.quote_plus(bucket_name).encode('ascii') + \
                        '/node-' + urllib.quote_plus(node_name).encode('ascii')
        if os.path.isdir(parent_dir):
            return parent_dir

        #check 3.0 directory structure
        if not tmstamp:
            tmstamp = time.strftime("%Y-%m-%dT%H%M%SZ", time.gmtime())
        parent_dir = os.path.normpath(spec)
        rootpath, dirs = BFD.find_latest_dir(parent_dir, None)
        if not rootpath or not mode or mode == "full":
            # no any backup roots exists
            path = os.path.join(parent_dir, tmstamp, tmstamp+"-full")
            return BFD.construct_dir(path, bucket_name, node_name)

        #check if any full backup exists
        path, dirs = BFD.find_latest_dir(rootpath, "full")
        if not path:
            path = os.path.join(rootpath, tmstamp+"-full")
            return BFD.construct_dir(path, bucket_name, node_name)
        else:
            #further check full backup for this bucket and node
            path = BFD.construct_dir(path, bucket_name, node_name)
            if not os.path.isdir(path):
                return path

        if mode.find("diff") >= 0:
            path, dirs = BFD.find_latest_dir(rootpath, "diff")
            if not path or new_session:
                path = os.path.join(rootpath, tmstamp+"-diff")
                return BFD.construct_dir(path, bucket_name, node_name)
            else:
                path = BFD.construct_dir(path, bucket_name, node_name)
                if not os.path.isdir(path):
                    return path
                else:
                    path = os.path.join(rootpath, tmstamp+"-diff")
                    return BFD.construct_dir(path, bucket_name, node_name)
        elif mode.find("accu") >= 0:
            path, dirs = BFD.find_latest_dir(rootpath, "accu")
            if not path or new_session:
                path = os.path.join(rootpath, tmstamp+"-accu")
                return BFD.construct_dir(path, bucket_name, node_name)
            else:
                path = BFD.construct_dir(path, bucket_name, node_name)
                if not os.path.isdir(path):
                    return path
                else:
                    path = os.path.join(rootpath, tmstamp+"-accu")
                    return BFD.construct_dir(path, bucket_name, node_name)
        else:
            return parent_dir

    @staticmethod
    def find_latest_dir(parent_dir, mode):
        all_subdirs = []
        latest_dir = None
        if not os.path.isdir(parent_dir):
            return latest_dir, all_subdirs
        for d in os.listdir(parent_dir):
            if not mode or d.find(mode) >= 0:
                bd = os.path.join(parent_dir, d)
                if os.path.isdir(bd):
                    all_subdirs.append(bd)
        if all_subdirs:
            all_subdirs = sorted(all_subdirs,key=os.path.getmtime, reverse=True)
            latest_dir = all_subdirs[0]
        return latest_dir, all_subdirs

    @staticmethod
    def find_seqno(opts, spec, bucket_name, node_name, mode):
        seqno = {}
        dep = {}
        dep_list = []
        failover_log = {}
        snapshot_markers = {}
        for i in range(BFD.NUM_VBUCKET):
            seqno[str(i)] = 0
            dep[i] = None
            failover_log[i] = None
            snapshot_markers[i] = None

        file_list = []
        failoverlog_list = []
        snapshot_list = []
        seqno_list = []
        parent_dir = os.path.normpath(spec)

        if mode == "full":
            return seqno, dep_list, failover_log, snapshot_markers
        timedir,latest_dirs = BFD.find_latest_dir(parent_dir, None)
        if not timedir:
            return seqno, dep_list, failover_log, snapshot_markers
        fulldir, latest_dirs = BFD.find_latest_dir(timedir, "full")
        if not fulldir:
            return seqno, dep_list, failover_log, snapshot_markers

        path = BFD.construct_dir(fulldir, bucket_name, node_name)
        if not os.path.isdir(path):
            return seqno, dep_list, failover_log, snapshot_markers

        last_backup = BFD.construct_dir(fulldir, bucket_name, node_name)
        file_list.extend(recursive_glob(path, 'data-*.cbb'))
        failoverlog_list.extend(recursive_glob(path, 'failover.json'))
        snapshot_list.extend(recursive_glob(path, 'snapshot_markers.json'))
        seqno_list.extend(recursive_glob(path, 'seqno.json'))

        accudir, accu_dirs = BFD.find_latest_dir(timedir, "accu")
        if accudir:
            last_backup = accudir
            path = BFD.construct_dir(accudir, bucket_name, node_name)
            if os.path.isdir(path):
                file_list.extend(recursive_glob(path, 'data-*.cbb'))
                failoverlog_list.extend(recursive_glob(path, 'failover.json'))
                snapshot_list.extend(recursive_glob(path, 'snapshot_markers.json'))
                seqno_list.extend(recursive_glob(path, 'seqno.json'))
        if mode.find("diff") >= 0:
            diffdir, diff_dirs = BFD.find_latest_dir(timedir, "diff")
            if diff_dirs:
                last_backup = BFD.construct_dir(diffdir, bucket_name, node_name)
                for dir in diff_dirs:
                    path = BFD.construct_dir(dir, bucket_name, node_name)
                    if os.path.isdir(path):
                        file_list.extend(recursive_glob(path, 'data-*.cbb'))
                        failoverlog_list.extend(recursive_glob(path, 'failover.json'))
                        snapshot_list.extend(recursive_glob(path, 'snapshot_markers.json'))
                        seqno_list.extend(recursive_glob(path, 'seqno.json'))

        if BFD.get_server_version(last_backup) < "5.0.0":
            return dict(), list(), dict(), dict()

        for x in sorted(seqno_list):
            try:
                json_file = open(x, "r")
                json_data = json.load(json_file)
                json_file.close()

                for vbid, seq in json_data.iteritems():
                    if not seq:
                        continue
                    if seqno.get(vbid) < seq:
                        seqno[vbid] = seq
            except IOError:
                pass

        for log_file in sorted(failoverlog_list):
            try:
                json_file = open(log_file, "r")
                json_data = json.load(json_file)
                json_file.close()

                for vbid, flogs in json_data.iteritems():
                    if not flogs:
                        continue
                    elif vbid not in failover_log.keys():
                        failover_log[vbid] = flogs
                    else:
                        for logpair in flogs:
                            if not failover_log[vbid]:
                                failover_log[vbid] = [logpair]
                            elif logpair not in failover_log[vbid]:
                                failover_log[vbid].append(logpair)
            except IOError:
                pass

        for snapshot in sorted(snapshot_list):
            try:
                json_file = open(snapshot, "r")
                json_data = json.load(json_file)
                json_file.close()

                for vbid, markers in json_data.iteritems():
                    snapshot_markers[vbid] = markers
            except IOError:
                pass

        for i in range(BFD.NUM_VBUCKET):
            if dep[i] and dep[i] not in dep_list:
                dep_list.append(dep[i])
        return seqno, dep_list, failover_log, snapshot_markers

# --------------------------------------------------

class BFDSource(BFD, pump.Source):
    """Can read from backup-file/directory layout."""

    def __init__(self, opts, spec, source_bucket, source_node,
                 source_map, sink_map, ctl, cur):
        super(BFDSource, self).__init__(opts, spec, source_bucket, source_node,
                                        source_map, sink_map, ctl, cur)
        self.done = False
        self.files = None
        self.cursor_db = None

    @staticmethod
    def can_handle(opts, spec):
        if os.path.isdir(spec):
            dirs = glob.glob(spec + "/bucket-*/node-*/data-*.cbb")
            if dirs:
                return True
            path, dirs = BFD.find_latest_dir(spec, None)
            if path:
                path, dirs = BFD.find_latest_dir(path, "full")
                if path:
                    return glob.glob(path + "*/bucket-*/node-*/data-*.cbb")
        return False

    @staticmethod
    def check(opts, spec):
        spec = os.path.normpath(spec)
        if not os.path.isdir(spec):
            return "error: backup_dir is not a directory: " + spec, None

        buckets = []

        bucket_dirs = glob.glob(spec + "/bucket-*")
        if not bucket_dirs:
            #check 3.0 directory structure
            path, dirs = BFD.find_latest_dir(spec, None)
            if not path:
                return "error: no backup directory found: " + spec, None
            latest_dir, dir = BFD.find_latest_dir(path, "full")
            if not latest_dir:
                return "error: no valid backup directory found: " + path, None
            bucket_dirs = glob.glob(latest_dir + "/bucket-*")

        for bucket_dir in sorted(bucket_dirs):
            if not os.path.isdir(bucket_dir):
                return "error: not a bucket directory: " + bucket_dir, None

            bucket_name = os.path.basename(bucket_dir)[len("bucket-"):].strip()
            bucket_name = urllib.unquote_plus(bucket_name).encode('ascii')
            if not bucket_name:
                return "error: bucket_name too short: " + bucket_dir, None

            bucket = { 'name': bucket_name, 'nodes': [] }
            buckets.append(bucket)

            node_dirs = glob.glob(bucket_dir + "/node-*")
            for node_dir in sorted(node_dirs):
                if not os.path.isdir(node_dir):
                    return "error: not a node directory: " + node_dir, None

                node_name = os.path.basename(node_dir)[len("node-"):].strip()
                node_name = urllib.unquote_plus(node_name).encode('ascii')
                if not node_name:
                    return "error: node_name too short: " + node_dir, None

                bucket['nodes'].append({ 'hostname': node_name })

        return 0, { 'spec': spec,
                    'buckets': buckets }

    @staticmethod
    def provide_design(opts, source_spec, source_bucket, source_map):
        return BFDSource.read_index_file(source_spec, source_bucket, DDOC_FILE_NAME)

    @staticmethod
    def provide_index(opts, source_spec, source_bucket, source_map):
        return BFDSource.read_index_file(source_spec, source_bucket, INDEX_FILE_NAME)

    @staticmethod
    def provide_fts_index(opts, source_spec, source_bucket, source_map):
        return BFDSource.read_index_file(source_spec, source_bucket, FTS_FILE_NAME)

    @staticmethod
    def read_index_file(source_spec, source_bucket, file_name):
        path = BFD.get_file_path(source_spec, source_bucket['name'], file_name)
        if os.path.exists(path):
            try:
                f = open(path, 'r')
                d = f.read()
                f.close()
                return 0, d
            except IOError, e:
                return ("error: could not read %s; exception: %s") % (path, e), None
        return 0, None

    def get_conflict_resolution_type(self):
        rv, files = BFDSource.list_files(self.opts, self.source_map['spec'],
                                         self.source_bucket['name'],
                                         self.source_node['hostname'],
                                         "meta.json")
        if rv != 0:
            return "seqno"
        try:
            json_file = open(files[0], "r")
            json_data = json.load(json_file)
            json_file.close()
            if "conflict_resolution_type" in json_data:
                return json_data["conflict_resolution_type"]
            return "seqno"
        except IOError:
            return "seqno"

    def provide_batch(self):
        if self.done:
            return 0, None

        batch = pump.Batch(self)

        batch_max_size = self.opts.extra['batch_max_size']
        batch_max_bytes = self.opts.extra['batch_max_bytes']

        s = ["SELECT cmd, vbucket_id, key, flg, exp, cas, meta, val FROM cbb_msg",
             "SELECT cmd, vbucket_id, key, flg, exp, cas, meta, val, seqno, dtype, meta_size FROM cbb_msg",
             "SELECT cmd, vbucket_id, key, flg, exp, cas, meta, val, seqno, dtype, meta_size, conf_res FROM cbb_msg"]

        if self.files is None: # None != [], as self.files will shrink to [].<commit_msg>MB-24361: Construct the meta.json path instead of using list_file()<commit_after>#!/usr/bin/env python

import copy
import errno
import glob
import logging
import os
import json
import string
import sys
import datetime
import time
import urllib
import fnmatch
import sqlite3

import couchbaseConstants
import pump

CBB_VERSION = [2004, 2014, 2015] # sqlite pragma user version.

DDOC_FILE_NAME = "design.json"
INDEX_FILE_NAME = "index.json"
FTS_FILE_NAME = "fts_index.json"

class BFD:
    """Mixin for backup-file/directory EndPoint helper methods."""
    NUM_VBUCKET = 1024

    def bucket_name(self):
        return self.source_bucket['name']

    def node_name(self):
        return self.source_node['hostname']

    @staticmethod
    def get_file_path(spec, bucket_name, file):
        bucket_name = urllib.quote_plus(bucket_name).encode('ascii')
        bucket_path = os.path.join(os.path.normpath(spec), "bucket-" + bucket_name)
        if os.path.isdir(bucket_path):
            return os.path.join(bucket_path, file)
        else:
            path, dirs = BFD.find_latest_dir(spec, None)
            if path:
                path, dirs = BFD.find_latest_dir(path, None)
                if path:
                    return os.path.join(path, "bucket-" + bucket_name, file)
        return os.path.join(bucket_path, file)

    @staticmethod
    def construct_dir(parent, bucket_name, node_name):
        return os.path.join(parent,
                    "bucket-" + urllib.quote_plus(bucket_name).encode('ascii'),
                    "node-" + urllib.quote_plus(node_name).encode('ascii'))

    @staticmethod
    def check_full_dbfiles(parent_dir):
        return glob.glob(os.path.join(parent_dir, "data-*.cbb"))

    @staticmethod
    def get_predecessors(parent_dir):
        try:
            json_file = open(os.path.join(parent_dir, "meta.json"), "r")
            json_data = json.load(json_file)
            json_file.close()
            return json_data["pred"]
        except IOError:
            return []

    @staticmethod
    def get_server_version(parent_dir):
        try:
            json_file = open(os.path.join(parent_dir, "meta.json"), "r")
            json_data = json.load(json_file)
            json_file.close()
            if "version" in json_data:
                return json_data["version"]
            return "0.0.0"
        except IOError:
            return "0.0.0"

    @staticmethod
    def get_failover_log(parent_dir):
        try:
            filepath = os.path.join(parent_dir, "failover.json")
            if os.path.isfile(filepath):
                json_file = open(filepath, "r")
                json_data = json.load(json_file)
                json_file.close()
                return json_data
            else:
                return {}
        except IOError:
            return {}

    @staticmethod
    def list_files(opts, spec, bucket, node, pattern):
        file_list = []
        prec_list = []

    @staticmethod
    def write_json_file(parent_dir, filename, output_data):
        filepath = os.path.join(parent_dir, filename)
        json_data = {}
        if os.path.isfile(filepath):
            #load into existed meta data generated from previous batch run
            try:
                json_file = open(filepath, "r")
                json_data = json.load(json_file)
                json_file.close()
            except IOError:
                pass

        for i in range(BFD.NUM_VBUCKET):
            if output_data.get(i):
                str_index = str(i)
                historic_data_exist = json_data.get(str_index)
                if historic_data_exist:
                    #Historic data will share same data type as incoming ones.
                    #It will be sufficient to only check type for historic data
                    if isinstance(json_data[str_index], int):
                        #For seqno, we want to keep the highest seqno
                        if json_data[str_index] < output_data[i]:
                            json_data[str_index] = output_data[i]
                    elif isinstance(json_data[str_index], list):
                        #For each vbucket, we want to get the superset of its references
                        if len(json_data[str_index]) < len(output_data[i]):
                            json_data[str_index] = output_data[i]
                else:
                    #Bookkeeping the incoming one.
                    json_data[str_index] = output_data[i]

        json_file = open(filepath, "w")
        json.dump(json_data, json_file, ensure_ascii=False)
        json_file.close()

    @staticmethod
    def db_dir(spec, bucket_name, node_name, tmstamp=None, mode=None, new_session=False):
        parent_dir = os.path.normpath(spec) + \
                        '/bucket-' + urllib.quote_plus(bucket_name).encode('ascii') + \
                        '/node-' + urllib.quote_plus(node_name).encode('ascii')
        if os.path.isdir(parent_dir):
            return parent_dir

        #check 3.0 directory structure
        if not tmstamp:
            tmstamp = time.strftime("%Y-%m-%dT%H%M%SZ", time.gmtime())
        parent_dir = os.path.normpath(spec)
        rootpath, dirs = BFD.find_latest_dir(parent_dir, None)
        if not rootpath or not mode or mode == "full":
            # no any backup roots exists
            path = os.path.join(parent_dir, tmstamp, tmstamp+"-full")
            return BFD.construct_dir(path, bucket_name, node_name)

        #check if any full backup exists
        path, dirs = BFD.find_latest_dir(rootpath, "full")
        if not path:
            path = os.path.join(rootpath, tmstamp+"-full")
            return BFD.construct_dir(path, bucket_name, node_name)
        else:
            #further check full backup for this bucket and node
            path = BFD.construct_dir(path, bucket_name, node_name)
            if not os.path.isdir(path):
                return path

        if mode.find("diff") >= 0:
            path, dirs = BFD.find_latest_dir(rootpath, "diff")
            if not path or new_session:
                path = os.path.join(rootpath, tmstamp+"-diff")
                return BFD.construct_dir(path, bucket_name, node_name)
            else:
                path = BFD.construct_dir(path, bucket_name, node_name)
                if not os.path.isdir(path):
                    return path
                else:
                    path = os.path.join(rootpath, tmstamp+"-diff")
                    return BFD.construct_dir(path, bucket_name, node_name)
        elif mode.find("accu") >= 0:
            path, dirs = BFD.find_latest_dir(rootpath, "accu")
            if not path or new_session:
                path = os.path.join(rootpath, tmstamp+"-accu")
                return BFD.construct_dir(path, bucket_name, node_name)
            else:
                path = BFD.construct_dir(path, bucket_name, node_name)
                if not os.path.isdir(path):
                    return path
                else:
                    path = os.path.join(rootpath, tmstamp+"-accu")
                    return BFD.construct_dir(path, bucket_name, node_name)
        else:
            return parent_dir

    @staticmethod
    def find_latest_dir(parent_dir, mode):
        all_subdirs = []
        latest_dir = None
        if not os.path.isdir(parent_dir):
            return latest_dir, all_subdirs
        for d in os.listdir(parent_dir):
            if not mode or d.find(mode) >= 0:
                bd = os.path.join(parent_dir, d)
                if os.path.isdir(bd):
                    all_subdirs.append(bd)
        if all_subdirs:
            all_subdirs = sorted(all_subdirs,key=os.path.getmtime, reverse=True)
            latest_dir = all_subdirs[0]
        return latest_dir, all_subdirs

    @staticmethod
    def find_seqno(opts, spec, bucket_name, node_name, mode):
        seqno = {}
        dep = {}
        dep_list = []
        failover_log = {}
        snapshot_markers = {}
        for i in range(BFD.NUM_VBUCKET):
            seqno[str(i)] = 0
            dep[i] = None
            failover_log[i] = None
            snapshot_markers[i] = None

        file_list = []
        failoverlog_list = []
        snapshot_list = []
        seqno_list = []
        parent_dir = os.path.normpath(spec)

        if mode == "full":
            return seqno, dep_list, failover_log, snapshot_markers
        timedir,latest_dirs = BFD.find_latest_dir(parent_dir, None)
        if not timedir:
            return seqno, dep_list, failover_log, snapshot_markers
        fulldir, latest_dirs = BFD.find_latest_dir(timedir, "full")
        if not fulldir:
            return seqno, dep_list, failover_log, snapshot_markers

        path = BFD.construct_dir(fulldir, bucket_name, node_name)
        if not os.path.isdir(path):
            return seqno, dep_list, failover_log, snapshot_markers

        last_backup = BFD.construct_dir(fulldir, bucket_name, node_name)
        file_list.extend(recursive_glob(path, 'data-*.cbb'))
        failoverlog_list.extend(recursive_glob(path, 'failover.json'))
        snapshot_list.extend(recursive_glob(path, 'snapshot_markers.json'))
        seqno_list.extend(recursive_glob(path, 'seqno.json'))

        accudir, accu_dirs = BFD.find_latest_dir(timedir, "accu")
        if accudir:
            last_backup = accudir
            path = BFD.construct_dir(accudir, bucket_name, node_name)
            if os.path.isdir(path):
                file_list.extend(recursive_glob(path, 'data-*.cbb'))
                failoverlog_list.extend(recursive_glob(path, 'failover.json'))
                snapshot_list.extend(recursive_glob(path, 'snapshot_markers.json'))
                seqno_list.extend(recursive_glob(path, 'seqno.json'))
        if mode.find("diff") >= 0:
            diffdir, diff_dirs = BFD.find_latest_dir(timedir, "diff")
            if diff_dirs:
                last_backup = BFD.construct_dir(diffdir, bucket_name, node_name)
                for dir in diff_dirs:
                    path = BFD.construct_dir(dir, bucket_name, node_name)
                    if os.path.isdir(path):
                        file_list.extend(recursive_glob(path, 'data-*.cbb'))
                        failoverlog_list.extend(recursive_glob(path, 'failover.json'))
                        snapshot_list.extend(recursive_glob(path, 'snapshot_markers.json'))
                        seqno_list.extend(recursive_glob(path, 'seqno.json'))

        if BFD.get_server_version(last_backup) < "5.0.0":
            return dict(), list(), dict(), dict()

        for x in sorted(seqno_list):
            try:
                json_file = open(x, "r")
                json_data = json.load(json_file)
                json_file.close()

                for vbid, seq in json_data.iteritems():
                    if not seq:
                        continue
                    if seqno.get(vbid) < seq:
                        seqno[vbid] = seq
            except IOError:
                pass

        for log_file in sorted(failoverlog_list):
            try:
                json_file = open(log_file, "r")
                json_data = json.load(json_file)
                json_file.close()

                for vbid, flogs in json_data.iteritems():
                    if not flogs:
                        continue
                    elif vbid not in failover_log.keys():
                        failover_log[vbid] = flogs
                    else:
                        for logpair in flogs:
                            if not failover_log[vbid]:
                                failover_log[vbid] = [logpair]
                            elif logpair not in failover_log[vbid]:
                                failover_log[vbid].append(logpair)
            except IOError:
                pass

        for snapshot in sorted(snapshot_list):
            try:
                json_file = open(snapshot, "r")
                json_data = json.load(json_file)
                json_file.close()

                for vbid, markers in json_data.iteritems():
                    snapshot_markers[vbid] = markers
            except IOError:
                pass

        for i in range(BFD.NUM_VBUCKET):
            if dep[i] and dep[i] not in dep_list:
                dep_list.append(dep[i])
        return seqno, dep_list, failover_log, snapshot_markers

# --------------------------------------------------

class BFDSource(BFD, pump.Source):
    """Can read from backup-file/directory layout."""

    def __init__(self, opts, spec, source_bucket, source_node,
                 source_map, sink_map, ctl, cur):
        super(BFDSource, self).__init__(opts, spec, source_bucket, source_node,
                                        source_map, sink_map, ctl, cur)
        self.done = False
        self.files = None
        self.cursor_db = None

    @staticmethod
    def can_handle(opts, spec):
        if os.path.isdir(spec):
            dirs = glob.glob(spec + "/bucket-*/node-*/data-*.cbb")
            if dirs:
                return True
            path, dirs = BFD.find_latest_dir(spec, None)
            if path:
                path, dirs = BFD.find_latest_dir(path, "full")
                if path:
                    return glob.glob(path + "*/bucket-*/node-*/data-*.cbb")
        return False

    @staticmethod
    def check(opts, spec):
        spec = os.path.normpath(spec)
        if not os.path.isdir(spec):
            return "error: backup_dir is not a directory: " + spec, None

        buckets = []

        bucket_dirs = glob.glob(spec + "/bucket-*")
        if not bucket_dirs:
            #check 3.0 directory structure
            path, dirs = BFD.find_latest_dir(spec, None)
            if not path:
                return "error: no backup directory found: " + spec, None
            latest_dir, dir = BFD.find_latest_dir(path, "full")
            if not latest_dir:
                return "error: no valid backup directory found: " + path, None
            bucket_dirs = glob.glob(latest_dir + "/bucket-*")

        for bucket_dir in sorted(bucket_dirs):
            if not os.path.isdir(bucket_dir):
                return "error: not a bucket directory: " + bucket_dir, None

            bucket_name = os.path.basename(bucket_dir)[len("bucket-"):].strip()
            bucket_name = urllib.unquote_plus(bucket_name).encode('ascii')
            if not bucket_name:
                return "error: bucket_name too short: " + bucket_dir, None

            bucket = { 'name': bucket_name, 'nodes': [] }
            buckets.append(bucket)

            node_dirs = glob.glob(bucket_dir + "/node-*")
            for node_dir in sorted(node_dirs):
                if not os.path.isdir(node_dir):
                    return "error: not a node directory: " + node_dir, None

                node_name = os.path.basename(node_dir)[len("node-"):].strip()
                node_name = urllib.unquote_plus(node_name).encode('ascii')
                if not node_name:
                    return "error: node_name too short: " + node_dir, None

                bucket['nodes'].append({ 'hostname': node_name })

        return 0, { 'spec': spec,
                    'buckets': buckets }

    @staticmethod
    def provide_design(opts, source_spec, source_bucket, source_map):
        return BFDSource.read_index_file(source_spec, source_bucket, DDOC_FILE_NAME)

    @staticmethod
    def provide_index(opts, source_spec, source_bucket, source_map):
        return BFDSource.read_index_file(source_spec, source_bucket, INDEX_FILE_NAME)

    @staticmethod
    def provide_fts_index(opts, source_spec, source_bucket, source_map):
        return BFDSource.read_index_file(source_spec, source_bucket, FTS_FILE_NAME)

    @staticmethod
    def read_index_file(source_spec, source_bucket, file_name):
        path = BFD.get_file_path(source_spec, source_bucket['name'], file_name)
        if os.path.exists(path):
            try:
                f = open(path, 'r')
                d = f.read()
                f.close()
                return 0, d
            except IOError, e:
                return ("error: could not read %s; exception: %s") % (path, e), None
        return 0, None

    def get_conflict_resolution_type(self):
        dir = BFD.construct_dir(self.source_map['spec'],
                                self.source_bucket['name'],
                                self.source_node['hostname'])
        metafile = os.path.join(dir, "meta.json")
        try:
            json_file = open(metafile, "r")
            json_data = json.load(json_file)
            json_file.close()
            if "conflict_resolution_type" in json_data:
                return json_data["conflict_resolution_type"]
            return "seqno"
        except IOError:
            return "seqno"

    def provide_batch(self):
        if self.done:
            return 0, None

        batch = pump.Batch(self)

        batch_max_size = self.opts.extra['batch_max_size']
        batch_max_bytes = self.opts.extra['batch_max_bytes']

        s = ["SELECT cmd, vbucket_id, key, flg, exp, cas, meta, val FROM cbb_msg",
             "SELECT cmd, vbucket_id, key, flg, exp, cas, meta, val, seqno, dtype, meta_size FROM cbb_msg",
             "SELECT cmd, vbucket_id, key, flg, exp, cas, meta, val, seqno, dtype, meta_size, conf_res FROM cbb_msg"]

        if self.files is None: # None != [], as self.files will shrink to [].
{'fraction_non_alphanumeric': 0.10092558099550104, 'fraction_numerical': 0.0024930509785941486, 'mean_word_length': 1.8026019916479281, 'pattern_counts': {'":': 6, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1006606', 'n_tokens_mistral': 10620, 'n_tokens_neox': 10177, 'n_words': 2279}