<commit_before>from __future__ import absolute_import, print_function, unicode_literals
from builtins import dict, str, int

from sys import version_info, exit
from math import ceil
from datetime import datetime
from functools import wraps
if version_info.major is not 3:
    msg = "Python 3.x is required to use this module."
    if __name__ == '__main__':
        print(msg)
        exit()
    else:
        raise ImportError(msg)

import csv
import time
import tarfile
import zlib
import logging
import xml.etree.ElementTree as ET
import re
import os
import multiprocessing as mp
import pickle
import sys
from os import path, remove
from ftplib import FTP
from io import BytesIO

logger = logging.getLogger('content_manager')
if __name__ == '__main__':
    # NOTE: PEP8 will complain about this, however having the args parsed up
    # here prevents a long wait just to fined out you entered a command wrong.
    from argparse import ArgumentParser
    parser = ArgumentParser(
        description='Manage content on INDRA\'s database.'
        )
    parser.add_argument(
        choices=['upload', 'update'],
        dest='task',
        help=('Choose whether you want to perform an initial upload or update '
              'the existing content on the database.')
        )
    parser.add_argument(
        '-c', '--continue',
        dest='continuing',
        action='store_true',
        help='Continue uploading or updating, picking up where you left off.'
        )
    parser.add_argument(
        '-n', '--num_procs',
        dest='num_procs',
        type=int,
        default=1,
        help=('Select the number of processors to use during this operation. '
              'Default is 1.')
        )
    parser.add_argument(
        '-d', '--debug',
        dest='debug',
        action='store_true',
        help='Run with debugging level output.'
        )
    parser.add_argument(
        '-t', '--test',
        action='store_true',
        help='Run tests using one of the designated test databases.'
        )
    args = parser.parse_args()
    if args.debug:
        logger.setLevel(logging.DEBUG)
        from indra.db.database_manager import logger as db_logger
        db_logger.setLevel(logging.DEBUG)

    if not args.continuing and args.task == 'upload':
        print("#"*63)
        print(
            "# You are about to wipe the database completely clean and     #\n"
            "# load it from scratch, which could take hours, and eliminate #\n"
            "# any updates, and potentially interfere with other users.    #\n"
            "#                                                             #\n"
            "# If you wish to continue an earlier upload, add the -c       #\n"
            "# option, and if you wish to update the database, specify     #\n"
            "# `update` in your command. For mor details, use --help.      #"
            )
        print("#"*63)
        resp = input("Are you sure you want to continue? [yes/NO]: ")
        if resp == 'no':
            print ("Aborting...")
            exit()
from indra.util import zip_string, unzip_string
from indra.util import UnicodeXMLTreeBuilder as UTB
from indra.literature.pmc_client import id_lookup
from indra.literature import pubmed_client

from indra.db.util import get_primary_db, get_defaults
from indra.db.database_manager import texttypes, formats, DatabaseManager
from indra.db.database_manager import sql_expressions as sql_exp


try:
    from psycopg2 import DatabaseError
except ImportError:
    class IntegrityError(object):
        "Using this in a try-except will catch nothing. (That's the point.)"
        pass


ftp_blocksize = 33554432  # Chunk size recommended by NCBI
BATCH_SIZE = 10000


THIS_DIR = path.dirname(path.abspath(__file__))


class UploadError(Exception):
    pass


class NihFtpClient(object):
    """High level access to the NIH FTP repositories.

    Parameters
    ----------
    my_path : str
        The path to the subdirectory around which this client operates.
    ftp_url : str
        The url to the ftp site. May be a local directory (see `local`). By
        default this is `'ftp.ncbi.nlm.nih.gov'`.
    local : bool
        These methods may be run on a local directory (intended for testing).
        (default is `False`).
    """
    def __init__(self, my_path, ftp_url='ftp.ncbi.nlm.nih.gov', local=False):
        self.my_path = my_path
        self.is_local = local
        self.ftp_url = ftp_url

    def _path_join(self, *args):
        joined_str = path.join(*args)
        part_list = joined_str.split('/')
        for part in part_list[1:]:
            if part == '..':
                idx = part_list.index(part) - 1
                part_list.pop(idx)
                part_list.pop(idx)
        ret = path.join(*part_list)
        if part_list[0] == '':
            ret = '/' + ret
        return ret

    def get_ftp_connection(self, ftp_path=None):
        if ftp_path is None:
            ftp_path = self.my_path
        # Get an FTP connection
        ftp = FTP(self.ftp_url)
        ftp.login()
        # Change to the manuscripts directory
        ftp.cwd(ftp_path)
        return ftp

    def get_xml_file(self, xml_file):
        "Get the content from an xml file as an ElementTree."
        logger.info("Downloading %s" % (xml_file))
        xml_bytes = self.get_uncompressed_bytes(xml_file, force_str=False)
        logger.info("Parsing XML metadata")
        return ET.XML(xml_bytes, parser=UTB())

    def get_csv_as_dict(self, csv_file, cols=None, header=None):
        "Get the content from a csv file as a list of dicts."
        csv_str = self.get_file(csv_file)
        lst = [row for row in csv.reader(csv_str.splitlines())]
        if cols is None:
            if header is not None:
                cols = lst[header]
                lst = lst[header+1:]
            else:
                cols = list(range(len(lst[0])))
        return [dict(zip(cols, row)) for row in lst]

    def ret_file(self, f_path, buf):
        "Load the content of a file into the given buffer."
        full_path = self._path_join(self.my_path, f_path)
        if not self.is_local:
            with self.get_ftp_connection() as ftp:
                ftp.retrbinary('RETR /%s' % full_path,
                               callback=lambda s: buf.write(s),
                               blocksize=ftp_blocksize)
                buf.flush()
        else:
            with open(self._path_join(self.ftp_url, full_path), 'rb') as f:
                buf.write(f.read())
                buf.flush()
        return

    def download_file(self, f_path, dest=None):
        "Download a file into a file given by f_path."
        name = path.basename(f_path)
        if dest is not None:
            name = path.join(dest, name)
        with open(name, 'wb') as gzf:
            self.ret_file(f_path, gzf)
        return name

    def get_file(self, f_path, force_str=True, decompress=True):
        "Get the contents of a file as a string."
        gzf_bytes = BytesIO()
        self.ret_file(f_path, gzf_bytes)
        ret = gzf_bytes.getvalue()
        if f_path.endswith('.gz') and decompress:
            ret = zlib.decompress(ret, 16+zlib.MAX_WBITS)
        if force_str and isinstance(ret, bytes):
            ret = ret.decode('utf8')
        return ret

    def get_uncompressed_bytes(self, f_path, force_str=True):
        "Get a file that is gzipped, and return the unzipped string."
        return self.get_file(f_path, force_str=force_str, decompress=True)

    def ftp_ls_timestamped(self, ftp_path=None):
        "Get all contents and metadata in mlsd format from the ftp directory."
        if ftp_path is None:
            ftp_path = self.my_path
        else:
            ftp_path = self._path_join(self.my_path, ftp_path)

        if not self.is_local:
            with self.get_ftp_connection(ftp_path) as ftp:
                raw_contents = ftp.mlsd()
                contents = [(k, meta['modify']) for k, meta in raw_contents
                            if not k.startswith('.')]
        else:
            dir_path = self._path_join(self.fpt_url, ftp_path)
            raw_contents = os.listdir(dir_path)
            contents = [(fname, path.getmtime(path.join(dir_path, fname)))
                        for fname in raw_contents]
        return contents

    def ftp_ls(self, ftp_path=None):
        "Get a list of the contents in the ftp directory."
        if ftp_path is None:
            ftp_path = self.my_path
        else:
            ftp_path = self._path_join(self.my_path, ftp_path)
        if not self.is_local:
            with self.get_ftp_connection(ftp_path) as ftp:
                contents = ftp.nlst()
        else:
            contents = os.listdir(self._path_join(self.ftp_url, ftp_path))
        return contents


class ContentManager(object):
    """Abstract class for all upload/update managers.

    This abstract class provides the api required for any object that is
    used to manage content between the database and the content.
    """
    my_source = NotImplemented
    tr_cols = NotImplemented
    err_patt = re.compile('.*?constraint "(.*?)".*?Key \((.*?)\)=\((.*?)\).*?',
                          re.DOTALL)

    def __init__(self):
        self.review_fname = None
        return

    def copy_into_db(self, db, tbl_name, data, cols=None, retry=True):
        "Wrapper around the db.copy feature, pickels args upon exception."
        vile_data = None
        try:
            db.copy(tbl_name, data, cols=cols)
        except DatabaseError as e:
            db.grab_session()
            db.session.rollback()
            logger.warning('Failed in a copy.')

            # Try to extract the failed record and try again.
            m = self.err_patt.match(e.args[0])
            if m is not None and retry:
                constraint, id_types, ids = m.groups()
                logger.info('Constraint %s was violated by (%s)=(%s).'
                            % (constraint, id_types, ids))
                id_dict = dict(zip(id_types.split(', '), ids.split(', ')))

                # Primary keys are returned from the database as integers.
                for id_type, id_val in id_dict.copy().items():
                    if id_type in ['text_ref_id', 'text_content_id']:
                        id_dict[id_type] = int(id_val)

                # Now look for matches and remove offending entries.
                new_data = set()
                vile_data = set()
                for datum in data:
                    # Determine if this record is a match using best available
                    # means.
                    if cols is not None:
                        is_match = all([
                            datum[cols.index(id_type)] == id_val
                            for id_type, id_val in id_dict.items()
                            ])
                    else:
                        is_match = all([
                            id_val in datum for id_val in id_dict.values()
                            ])

                    # Now decide to ad to the new data or log the error.
                    if is_match:
                        logger.debug("Found offending data: %s." % str(datum))
                        self.add_to_review('conflict in copy',
                                           'Entry violated \"%s\": %s'
                                           % (constraint, str(datum)))
                        vile_data.add(datum)
                    else:
                        new_data.add(datum)

                if not len(vile_data):
                    logger.error("Failed to find erroneous data.")
                    raise e

                logger.info('Resubmitting copy without the offending ids.')
                more_vile_data = self.copy_into_db(db, tbl_name, new_data,
                                                   cols=cols, retry=True)

                if more_vile_data is not None:
                    vile_data = vile_data.union(more_vile_data)
            else:
                pkl_file_fmt = "copy_failure_%d.pkl"
                i = 0
                while os.path.exists(pkl_file_fmt % i):
                    i += 1
                with open(pkl_file_fmt % i, 'wb') as f:
                    pickle.dump((e, tbl_name, data, cols), f, protocol=3)
                logger.error('Could not resubmit, not a handled error. '
                             'Pickling the data in %s.' % (pkl_file_fmt % i))
                logger.exception(e)
                raise e
        logger.debug('Finished copying.')
        return vile_data

    def make_text_ref_str(self, tr):
        """Make a string from a text ref using tr_cols."""
        return str([getattr(tr, id_type) for id_type in self.tr_cols])

    def add_to_review(self, desc, msg):
        """Add an entry to the review document."""
        # NOTE: If this is ever done on AWS or through a
        # container, the review file MUST be loaded somewhere
        # it won't disappear. (such as s3). Perhaps these could
        # be logged on the database?
        logger.warning("Found \"%s\"! Check %s."
                       % (desc, self.review_fname))
        with open(self.review_fname, 'a+') as f:
            f.write(msg + '\n')
        return

    def filter_text_refs(self, db, tr_data_set, primary_id_types=None):
        """Try to reconcile the data we have with what's already on the db.

        Note that this method is VERY slow in general, and therefore should
        be avoided whenever possible.

        The process can be sped up considerably by multiple orders of
        magnitude if you specify a limited set of id types to query to get
        text refs. This does leave some possibility of missing relevant refs.
        """
        logger.info("Beginning to filter %d text refs..." % len(tr_data_set))

        # This is a helper for accessing the data tuples we create
        def id_idx(id_type):
            return self.tr_cols.index(id_type)

        # If there are not actual refs to work with, don't waste time.
        N = len(tr_data_set)
        if not N:
            return set(), []

        # Get all text refs that match any of the id data we have. This means
        # each text ref WILL find a match in the data we have (unless something
        # is seriously broken.
        logger.debug("Getting list of existing text refs...")
        or_list = []
        if primary_id_types is not None:
            match_id_types = primary_id_types
        else:
            match_id_types = self.tr_cols
        for id_type in match_id_types:
            id_list = [entry[id_idx(id_type)] for entry in tr_data_set
                       if entry[id_idx(id_type)] is not None]
            if id_list:
                or_list.append(getattr(db.TextRef, id_type).in_(id_list))
        if len(or_list) == 1:
            tr_list = db.select_all(db.TextRef, or_list[0])
        else:
            tr_list = db.select_all(db.TextRef, sql_exp.or_(*or_list))
        logger.debug("Found %d potentially relevent text refs." % len(tr_list))

        # Create an index of tupled data entries for quick lookups by any id
        # type, for example tr_data_idx_dict['pmid'][<a pmid>] will get the
        # tuple with all the id data. This avoids several iterations through
        # the list of text ref data dicts on for each text ref, at the cost of
        # using extra memory.
        logger.debug("Building index of new data...")
        tr_data_idx_dict = {id_type: {e[id_idx(id_type)]: e
                                      for e in tr_data_set
                                      if e[id_idx(id_type)] is not None}
                            for id_type in self.tr_cols}

        # Look for updates to the existing text refs
        logger.debug("Beginning to iterate over text refs...")
        tr_data_match_list = []
        flawed_tr_data = []
        multi_match_records = set()
        update_dict = {}

        def add_to_found_record_list(record):
            if record not in tr_data_match_list:
                tr_data_match_list.append(record)
                added = True
            else:
                self.add_to_review(
                    "tr matching input record matched to another tr",
                    "Input record %s already matched. Matched again to %s."
                    % (record, self.make_text_ref_str(tr))
                    )
                flawed_tr_data.append(('over_match_db', record))
                multi_match_records.add(record)
                added = False
            return added

        for tr in tr_list:
            match_set = set()

            # Find the matches in the data. We continue looking after finding
            # one match in case there are any inconsistencies which need to be
            # considered.
            for id_type, tr_data_idx in tr_data_idx_dict.items():
                candidate = tr_data_idx.get(getattr(tr, id_type))
                if candidate is not None:
                    match_set.add(candidate)

            # As per the process of getting the tr_list, every tr MUST have a
            # match, or else something is broken.
            assert match_set, "No matches found, which is impossible."

            # Assuming we found exactly one matched data entry, we can now look
            # for new id data (for example manuscript ids) and update the tr
            # objects. These changes are transmuted to the db via the commit
            # command below.
            if len(match_set) == 1:
                tr_new = match_set.pop()

                # This is how we tell what doesn't need to be added to the db.
                if not add_to_found_record_list(tr_new):
                    continue

                # Go through all the id_types
                all_good = True
                id_updates = {}
                for i, id_type in enumerate(self.tr_cols):
                    # Check if the text ref is missing that id.
                    if getattr(tr, id_type) is None:
                        # If so, and if our new data does have that id, update
                        # the text ref.
                        if tr_new[i] is not None:
                            logger.debug("Will update text ref for %s: %s."
                                         % (id_type, tr_new[i]))
                            id_updates[id_type] = tr_new[i]
                    else:
                        # Check to see that all the ids agree. If not, report
                        # it in the review.txt file.
                        if tr_new[i] is not None \
                         and tr_new[i] != getattr(tr, id_type).strip().upper():
                            self.add_to_review(
                                'conflicting ids',
                                'Got conflicting %s: in db %s vs %s.'
                                % (id_type, self.make_text_ref_str(tr), tr_new)
                                )
                            flawed_tr_data.append((id_type, tr_new))
                            all_good = False

                if all_good and len(id_updates):
                    update_dict[tr.id] = (tr, id_updates, tr_new)
            else:
                # These still matched something in the db, so they shouldn't be
                # uploaded as new refs.
                for tr_new in match_set:
                    add_to_found_record_list(tr_new)
                    flawed_tr_data.append(('over_match_input', tr_new))

                # This condition only occurs if the records we got are
                # internally inconsistent. This is rare, but it can happen.
                self.add_to_review(
                    "multiple matches in records for tex ref",
                    'Multiple matches for %s from %s: %s.'
                    % (self.make_text_ref_str(tr), self.my_source, match_set))

        # Remove unhealthy text refs.
        logger.info("Applying %d updates." % len(update_dict))
        for tr, id_updates, record in update_dict.values():
            if record not in multi_match_records:
                for id_type, id_val in id_updates.items():
                    setattr(tr, id_type, id_val)
            else:
                logger.warning("Skipping update of text ref %d with %s due "
                               "to multiple matches to record %s."
                               % (tr.id, id_updates, record))

        # This applies all the changes made to the text refs to the db.
        logger.debug("Committing changes...")
        db.commit("Failed to update with new ids.")

        # Now update the text refs with any new refs that were found
        filtered_tr_records = tr_data_set - set(tr_data_match_list) \
            - multi_match_records

        logger.debug("Filtering complete! %d records remaining."
                     % len(filtered_tr_records))
        return filtered_tr_records, flawed_tr_data

    @classmethod
    def _principal_action(cls, func):
        @wraps(func)
        def take_action(self, db, *args, **kwargs):
            time_string = datetime.now().strftime('%Y%m%d-%H%M%S')
            self.review_fname = path.join(
                THIS_DIR,
                "review_%s_%s_%s.txt" % (func.__name__, self.my_source,
                                         time_string)
                )
            logger.info("Creating review file %s." % self.review_fname)
            open(self.review_fname, 'w').close()
            completed = func(self, db, *args, **kwargs)
            if completed:
                db.insert('updates', init_upload=(func.__name__ == 'populate'),
                          source=self.my_source, datetime=datetime.utcnow())
            return completed
        return take_action

    def populate(self, db):
        "A stub for the method used to initially populate the database."
        raise NotImplementedError(
            "`Populate` not implemented for `%s`." % self.__class__.__name__
            )

    def update(self, db):
        "A stub for the method used to update the content on the database."
        raise NotImplementedError(
            "`Update` not implemented for `%s`." % self.__class__.__name__
            )


class NihManager(ContentManager):
    """Abstract class for all the managers that use the NIH FTP service.

    See `NihFtpClient` for parameters.
    """
    my_path = NotImplemented

    def __init__(self, *args, **kwargs):<commit_msg>Make the review file continue if continuing.<commit_after>from __future__ import absolute_import, print_function, unicode_literals
from builtins import dict, str, int

from sys import version_info, exit
from math import ceil
from datetime import datetime
from functools import wraps
if version_info.major is not 3:
    msg = "Python 3.x is required to use this module."
    if __name__ == '__main__':
        print(msg)
        exit()
    else:
        raise ImportError(msg)

import csv
import time
import tarfile
import zlib
import logging
import xml.etree.ElementTree as ET
import re
import os
import multiprocessing as mp
import pickle
import sys
from os import path, remove
from ftplib import FTP
from io import BytesIO

logger = logging.getLogger('content_manager')
if __name__ == '__main__':
    # NOTE: PEP8 will complain about this, however having the args parsed up
    # here prevents a long wait just to fined out you entered a command wrong.
    from argparse import ArgumentParser
    parser = ArgumentParser(
        description='Manage content on INDRA\'s database.'
        )
    parser.add_argument(
        choices=['upload', 'update'],
        dest='task',
        help=('Choose whether you want to perform an initial upload or update '
              'the existing content on the database.')
        )
    parser.add_argument(
        '-c', '--continue',
        dest='continuing',
        action='store_true',
        help='Continue uploading or updating, picking up where you left off.'
        )
    parser.add_argument(
        '-n', '--num_procs',
        dest='num_procs',
        type=int,
        default=1,
        help=('Select the number of processors to use during this operation. '
              'Default is 1.')
        )
    parser.add_argument(
        '-d', '--debug',
        dest='debug',
        action='store_true',
        help='Run with debugging level output.'
        )
    parser.add_argument(
        '-t', '--test',
        action='store_true',
        help='Run tests using one of the designated test databases.'
        )
    args = parser.parse_args()
    if args.debug:
        logger.setLevel(logging.DEBUG)
        from indra.db.database_manager import logger as db_logger
        db_logger.setLevel(logging.DEBUG)

    if not args.continuing and args.task == 'upload':
        print("#"*63)
        print(
            "# You are about to wipe the database completely clean and     #\n"
            "# load it from scratch, which could take hours, and eliminate #\n"
            "# any updates, and potentially interfere with other users.    #\n"
            "#                                                             #\n"
            "# If you wish to continue an earlier upload, add the -c       #\n"
            "# option, and if you wish to update the database, specify     #\n"
            "# `update` in your command. For mor details, use --help.      #"
            )
        print("#"*63)
        resp = input("Are you sure you want to continue? [yes/NO]: ")
        if resp == 'no':
            print ("Aborting...")
            exit()
from indra.util import zip_string, unzip_string
from indra.util import UnicodeXMLTreeBuilder as UTB
from indra.literature.pmc_client import id_lookup
from indra.literature import pubmed_client

from indra.db.util import get_primary_db, get_defaults
from indra.db.database_manager import texttypes, formats, DatabaseManager
from indra.db.database_manager import sql_expressions as sql_exp


try:
    from psycopg2 import DatabaseError
except ImportError:
    class IntegrityError(object):
        "Using this in a try-except will catch nothing. (That's the point.)"
        pass


ftp_blocksize = 33554432  # Chunk size recommended by NCBI
BATCH_SIZE = 10000


THIS_DIR = path.dirname(path.abspath(__file__))


class UploadError(Exception):
    pass


class NihFtpClient(object):
    """High level access to the NIH FTP repositories.

    Parameters
    ----------
    my_path : str
        The path to the subdirectory around which this client operates.
    ftp_url : str
        The url to the ftp site. May be a local directory (see `local`). By
        default this is `'ftp.ncbi.nlm.nih.gov'`.
    local : bool
        These methods may be run on a local directory (intended for testing).
        (default is `False`).
    """
    def __init__(self, my_path, ftp_url='ftp.ncbi.nlm.nih.gov', local=False):
        self.my_path = my_path
        self.is_local = local
        self.ftp_url = ftp_url

    def _path_join(self, *args):
        joined_str = path.join(*args)
        part_list = joined_str.split('/')
        for part in part_list[1:]:
            if part == '..':
                idx = part_list.index(part) - 1
                part_list.pop(idx)
                part_list.pop(idx)
        ret = path.join(*part_list)
        if part_list[0] == '':
            ret = '/' + ret
        return ret

    def get_ftp_connection(self, ftp_path=None):
        if ftp_path is None:
            ftp_path = self.my_path
        # Get an FTP connection
        ftp = FTP(self.ftp_url)
        ftp.login()
        # Change to the manuscripts directory
        ftp.cwd(ftp_path)
        return ftp

    def get_xml_file(self, xml_file):
        "Get the content from an xml file as an ElementTree."
        logger.info("Downloading %s" % (xml_file))
        xml_bytes = self.get_uncompressed_bytes(xml_file, force_str=False)
        logger.info("Parsing XML metadata")
        return ET.XML(xml_bytes, parser=UTB())

    def get_csv_as_dict(self, csv_file, cols=None, header=None):
        "Get the content from a csv file as a list of dicts."
        csv_str = self.get_file(csv_file)
        lst = [row for row in csv.reader(csv_str.splitlines())]
        if cols is None:
            if header is not None:
                cols = lst[header]
                lst = lst[header+1:]
            else:
                cols = list(range(len(lst[0])))
        return [dict(zip(cols, row)) for row in lst]

    def ret_file(self, f_path, buf):
        "Load the content of a file into the given buffer."
        full_path = self._path_join(self.my_path, f_path)
        if not self.is_local:
            with self.get_ftp_connection() as ftp:
                ftp.retrbinary('RETR /%s' % full_path,
                               callback=lambda s: buf.write(s),
                               blocksize=ftp_blocksize)
                buf.flush()
        else:
            with open(self._path_join(self.ftp_url, full_path), 'rb') as f:
                buf.write(f.read())
                buf.flush()
        return

    def download_file(self, f_path, dest=None):
        "Download a file into a file given by f_path."
        name = path.basename(f_path)
        if dest is not None:
            name = path.join(dest, name)
        with open(name, 'wb') as gzf:
            self.ret_file(f_path, gzf)
        return name

    def get_file(self, f_path, force_str=True, decompress=True):
        "Get the contents of a file as a string."
        gzf_bytes = BytesIO()
        self.ret_file(f_path, gzf_bytes)
        ret = gzf_bytes.getvalue()
        if f_path.endswith('.gz') and decompress:
            ret = zlib.decompress(ret, 16+zlib.MAX_WBITS)
        if force_str and isinstance(ret, bytes):
            ret = ret.decode('utf8')
        return ret

    def get_uncompressed_bytes(self, f_path, force_str=True):
        "Get a file that is gzipped, and return the unzipped string."
        return self.get_file(f_path, force_str=force_str, decompress=True)

    def ftp_ls_timestamped(self, ftp_path=None):
        "Get all contents and metadata in mlsd format from the ftp directory."
        if ftp_path is None:
            ftp_path = self.my_path
        else:
            ftp_path = self._path_join(self.my_path, ftp_path)

        if not self.is_local:
            with self.get_ftp_connection(ftp_path) as ftp:
                raw_contents = ftp.mlsd()
                contents = [(k, meta['modify']) for k, meta in raw_contents
                            if not k.startswith('.')]
        else:
            dir_path = self._path_join(self.fpt_url, ftp_path)
            raw_contents = os.listdir(dir_path)
            contents = [(fname, path.getmtime(path.join(dir_path, fname)))
                        for fname in raw_contents]
        return contents

    def ftp_ls(self, ftp_path=None):
        "Get a list of the contents in the ftp directory."
        if ftp_path is None:
            ftp_path = self.my_path
        else:
            ftp_path = self._path_join(self.my_path, ftp_path)
        if not self.is_local:
            with self.get_ftp_connection(ftp_path) as ftp:
                contents = ftp.nlst()
        else:
            contents = os.listdir(self._path_join(self.ftp_url, ftp_path))
        return contents


class ContentManager(object):
    """Abstract class for all upload/update managers.

    This abstract class provides the api required for any object that is
    used to manage content between the database and the content.
    """
    my_source = NotImplemented
    tr_cols = NotImplemented
    err_patt = re.compile('.*?constraint "(.*?)".*?Key \((.*?)\)=\((.*?)\).*?',
                          re.DOTALL)

    def __init__(self):
        self.review_fname = None
        return

    def copy_into_db(self, db, tbl_name, data, cols=None, retry=True):
        "Wrapper around the db.copy feature, pickels args upon exception."
        vile_data = None
        try:
            db.copy(tbl_name, data, cols=cols)
        except DatabaseError as e:
            db.grab_session()
            db.session.rollback()
            logger.warning('Failed in a copy.')

            # Try to extract the failed record and try again.
            m = self.err_patt.match(e.args[0])
            if m is not None and retry:
                constraint, id_types, ids = m.groups()
                logger.info('Constraint %s was violated by (%s)=(%s).'
                            % (constraint, id_types, ids))
                id_dict = dict(zip(id_types.split(', '), ids.split(', ')))

                # Primary keys are returned from the database as integers.
                for id_type, id_val in id_dict.copy().items():
                    if id_type in ['text_ref_id', 'text_content_id']:
                        id_dict[id_type] = int(id_val)

                # Now look for matches and remove offending entries.
                new_data = set()
                vile_data = set()
                for datum in data:
                    # Determine if this record is a match using best available
                    # means.
                    if cols is not None:
                        is_match = all([
                            datum[cols.index(id_type)] == id_val
                            for id_type, id_val in id_dict.items()
                            ])
                    else:
                        is_match = all([
                            id_val in datum for id_val in id_dict.values()
                            ])

                    # Now decide to ad to the new data or log the error.
                    if is_match:
                        logger.debug("Found offending data: %s." % str(datum))
                        self.add_to_review('conflict in copy',
                                           'Entry violated \"%s\": %s'
                                           % (constraint, str(datum)))
                        vile_data.add(datum)
                    else:
                        new_data.add(datum)

                if not len(vile_data):
                    logger.error("Failed to find erroneous data.")
                    raise e

                logger.info('Resubmitting copy without the offending ids.')
                more_vile_data = self.copy_into_db(db, tbl_name, new_data,
                                                   cols=cols, retry=True)

                if more_vile_data is not None:
                    vile_data = vile_data.union(more_vile_data)
            else:
                pkl_file_fmt = "copy_failure_%d.pkl"
                i = 0
                while os.path.exists(pkl_file_fmt % i):
                    i += 1
                with open(pkl_file_fmt % i, 'wb') as f:
                    pickle.dump((e, tbl_name, data, cols), f, protocol=3)
                logger.error('Could not resubmit, not a handled error. '
                             'Pickling the data in %s.' % (pkl_file_fmt % i))
                logger.exception(e)
                raise e
        logger.debug('Finished copying.')
        return vile_data

    def make_text_ref_str(self, tr):
        """Make a string from a text ref using tr_cols."""
        return str([getattr(tr, id_type) for id_type in self.tr_cols])

    def add_to_review(self, desc, msg):
        """Add an entry to the review document."""
        # NOTE: If this is ever done on AWS or through a
        # container, the review file MUST be loaded somewhere
        # it won't disappear. (such as s3). Perhaps these could
        # be logged on the database?
        logger.warning("Found \"%s\"! Check %s."
                       % (desc, self.review_fname))
        with open(self.review_fname, 'a+') as f:
            f.write(msg + '\n')
        return

    def filter_text_refs(self, db, tr_data_set, primary_id_types=None):
        """Try to reconcile the data we have with what's already on the db.

        Note that this method is VERY slow in general, and therefore should
        be avoided whenever possible.

        The process can be sped up considerably by multiple orders of
        magnitude if you specify a limited set of id types to query to get
        text refs. This does leave some possibility of missing relevant refs.
        """
        logger.info("Beginning to filter %d text refs..." % len(tr_data_set))

        # This is a helper for accessing the data tuples we create
        def id_idx(id_type):
            return self.tr_cols.index(id_type)

        # If there are not actual refs to work with, don't waste time.
        N = len(tr_data_set)
        if not N:
            return set(), []

        # Get all text refs that match any of the id data we have. This means
        # each text ref WILL find a match in the data we have (unless something
        # is seriously broken.
        logger.debug("Getting list of existing text refs...")
        or_list = []
        if primary_id_types is not None:
            match_id_types = primary_id_types
        else:
            match_id_types = self.tr_cols
        for id_type in match_id_types:
            id_list = [entry[id_idx(id_type)] for entry in tr_data_set
                       if entry[id_idx(id_type)] is not None]
            if id_list:
                or_list.append(getattr(db.TextRef, id_type).in_(id_list))
        if len(or_list) == 1:
            tr_list = db.select_all(db.TextRef, or_list[0])
        else:
            tr_list = db.select_all(db.TextRef, sql_exp.or_(*or_list))
        logger.debug("Found %d potentially relevent text refs." % len(tr_list))

        # Create an index of tupled data entries for quick lookups by any id
        # type, for example tr_data_idx_dict['pmid'][<a pmid>] will get the
        # tuple with all the id data. This avoids several iterations through
        # the list of text ref data dicts on for each text ref, at the cost of
        # using extra memory.
        logger.debug("Building index of new data...")
        tr_data_idx_dict = {id_type: {e[id_idx(id_type)]: e
                                      for e in tr_data_set
                                      if e[id_idx(id_type)] is not None}
                            for id_type in self.tr_cols}

        # Look for updates to the existing text refs
        logger.debug("Beginning to iterate over text refs...")
        tr_data_match_list = []
        flawed_tr_data = []
        multi_match_records = set()
        update_dict = {}

        def add_to_found_record_list(record):
            if record not in tr_data_match_list:
                tr_data_match_list.append(record)
                added = True
            else:
                self.add_to_review(
                    "tr matching input record matched to another tr",
                    "Input record %s already matched. Matched again to %s."
                    % (record, self.make_text_ref_str(tr))
                    )
                flawed_tr_data.append(('over_match_db', record))
                multi_match_records.add(record)
                added = False
            return added

        for tr in tr_list:
            match_set = set()

            # Find the matches in the data. We continue looking after finding
            # one match in case there are any inconsistencies which need to be
            # considered.
            for id_type, tr_data_idx in tr_data_idx_dict.items():
                candidate = tr_data_idx.get(getattr(tr, id_type))
                if candidate is not None:
                    match_set.add(candidate)

            # As per the process of getting the tr_list, every tr MUST have a
            # match, or else something is broken.
            assert match_set, "No matches found, which is impossible."

            # Assuming we found exactly one matched data entry, we can now look
            # for new id data (for example manuscript ids) and update the tr
            # objects. These changes are transmuted to the db via the commit
            # command below.
            if len(match_set) == 1:
                tr_new = match_set.pop()

                # This is how we tell what doesn't need to be added to the db.
                if not add_to_found_record_list(tr_new):
                    continue

                # Go through all the id_types
                all_good = True
                id_updates = {}
                for i, id_type in enumerate(self.tr_cols):
                    # Check if the text ref is missing that id.
                    if getattr(tr, id_type) is None:
                        # If so, and if our new data does have that id, update
                        # the text ref.
                        if tr_new[i] is not None:
                            logger.debug("Will update text ref for %s: %s."
                                         % (id_type, tr_new[i]))
                            id_updates[id_type] = tr_new[i]
                    else:
                        # Check to see that all the ids agree. If not, report
                        # it in the review.txt file.
                        if tr_new[i] is not None \
                         and tr_new[i] != getattr(tr, id_type).strip().upper():
                            self.add_to_review(
                                'conflicting ids',
                                'Got conflicting %s: in db %s vs %s.'
                                % (id_type, self.make_text_ref_str(tr), tr_new)
                                )
                            flawed_tr_data.append((id_type, tr_new))
                            all_good = False

                if all_good and len(id_updates):
                    update_dict[tr.id] = (tr, id_updates, tr_new)
            else:
                # These still matched something in the db, so they shouldn't be
                # uploaded as new refs.
                for tr_new in match_set:
                    add_to_found_record_list(tr_new)
                    flawed_tr_data.append(('over_match_input', tr_new))

                # This condition only occurs if the records we got are
                # internally inconsistent. This is rare, but it can happen.
                self.add_to_review(
                    "multiple matches in records for tex ref",
                    'Multiple matches for %s from %s: %s.'
                    % (self.make_text_ref_str(tr), self.my_source, match_set))

        # Remove unhealthy text refs.
        logger.info("Applying %d updates." % len(update_dict))
        for tr, id_updates, record in update_dict.values():
            if record not in multi_match_records:
                for id_type, id_val in id_updates.items():
                    setattr(tr, id_type, id_val)
            else:
                logger.warning("Skipping update of text ref %d with %s due "
                               "to multiple matches to record %s."
                               % (tr.id, id_updates, record))

        # This applies all the changes made to the text refs to the db.
        logger.debug("Committing changes...")
        db.commit("Failed to update with new ids.")

        # Now update the text refs with any new refs that were found
        filtered_tr_records = tr_data_set - set(tr_data_match_list) \
            - multi_match_records

        logger.debug("Filtering complete! %d records remaining."
                     % len(filtered_tr_records))
        return filtered_tr_records, flawed_tr_data

    @classmethod
    def _principal_action(cls, func):
        @wraps(func)
        def take_action(self, db, *args, **kwargs):
            review_fmt = "review_%s_%s_%%s.txt" % (func.__name__,
                                                   self.my_source)
            self.review_fname = path.join(THIS_DIR, review_fmt % 'in_progress')
            logger.info("Creating review file %s." % self.review_fname)
            open(self.review_fname, 'a+').close()
            completed = func(self, db, *args, **kwargs)
            if completed:
                utcnow = datetime.utcnow()
                db.insert('updates', init_upload=(func.__name__ == 'populate'),
                          source=self.my_source, datetime=utcnow)
                os.rename(self.review_fname,
                          review_fmt % utcnow.strftime('%Y%m%d-%H%M%S'))
            return completed
        return take_action

    def populate(self, db):
        "A stub for the method used to initially populate the database."
        raise NotImplementedError(
            "`Populate` not implemented for `%s`." % self.__class__.__name__
            )

    def update(self, db):
        "A stub for the method used to update the content on the database."
        raise NotImplementedError(
            "`Update` not implemented for `%s`." % self.__class__.__name__
            )


class NihManager(ContentManager):
    """Abstract class for all the managers that use the NIH FTP service.

    See `NihFtpClient` for parameters.
    """
    my_path = NotImplemented

    def __init__(self, *args, **kwargs):
{'fraction_non_alphanumeric': 0.08544184499390177, 'fraction_numerical': 0.001729681782902761, 'mean_word_length': 1.694711682103376, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 16}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 48, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7528446', 'n_tokens_mistral': 12952, 'n_tokens_neox': 12342, 'n_words': 3999}