<commit_before>#!/usr/bin/env python
# encoding: utf-8
from __future__ import print_function
"""
optimize_path
"""
from collections import OrderedDict, Counter
import json
import logging
import os
import re
import subprocess

log = logging.getLogger()

CHECKSUM_BIN='md5sum'

def sbin_policy(iterable):
    """
    # TODO: static optimizations
    """
    for s in sorted(
        (
            (
                (int('/sbin' not in x)*2) +
                (int('/usr/local' in x)*3) +
                (int('/home' in x)*4)
            ),
            len(x),
            x)
            for x in iterable ):
        #logging.debug(s)
        yield s[-1]


def optimize_path(iterable, strip=None, sortfunc=sbin_policy):
    """
    optimize paths
    """
    if strip:
        pattern = re.compile( strip )
        log.debug((pattern, strip, ))
        keys = (p.strip() for p in iterable if not pattern.match(p))
    else:
        keys = (p.strip() for p in iterable)

    paths = OrderedDict.fromkeys(keys)

    if sortfunc:
        return sortfunc(paths.iterkeys())
    else:
        return paths.iterkeys()


def get_checksum(path, hashbin=CHECKSUM_BIN):
    logging.debug((hashbin,path))
    if os.path.isfile(path):
        if os.access(path, os.R_OK):
            return (subprocess.check_output( (hashbin, path) ).rstrip() +
                    (os.path.islink(path) and " -> %s" % os.path.realpath(path) or '') )
    return ('?  %s' % path)


def detect_duplicates(iterable):
    paths = OrderedDict()
    for path in iterable:
        for f in os.listdir(path):
            basename = os.path.basename(f)
            basename_paths = paths.get(basename,[])
            basename_paths.append(path)
            paths[basename] = basename_paths


    annotated_paths = (
        (k, [get_checksum( os.path.join(p,k) ) for p in v] )
            for (k,v) in paths.iteritems() if len(v) > 1)

    def label(v):
        counts = Counter(l.split()[0] for l in v)
        #logging.debug(counts)
        if len(v) in counts.itervalues():
            return {'type':'duplicates', 'files': sorted(v)}
        else:
            return {'type':'hashes', 'files': v}

    return OrderedDict(
        (k, label(v)) for (k,v) in annotated_paths
    )


def check_difference(iterable):
    raise NotImplementedError()
    import difflib

    iterable = list(iterable)
    regular = list( optimize_path( iterable, sortfunc=None) )
    optimized = list( optimize_path( iterable, sortfunc=sbin_policy ) )

    logging.debug(regular)
    logging.debug(optimized)

    return '\n'.join( difflib.ndiff(regular, optimized) )
    # detect colli

import unittest
import sys
from StringIO import StringIO

class ShellTestCase(unittest.TestCase):
    __stdout = sys.stdout
    __stderr = sys.stderr
    def _run_command(self, _cmd):
        # TODO: not monkey-patch sys.stderr
        try:
            #process = subprocess.Popen((here,) + cmd,
            #    stdout=subprocess.PIPE,
            #    stderr=subprocess.PIPE)
            #stdout,stderr = process.communicate()
            #logging.debug(_cmd)
            logging.info("### TEST CASE: %r" % _cmd)

            stdout = sys.stdout = StringIO()
            stderr = sys.stderr = StringIO()

            sys.argv = _cmd
            main()

            _stderr = stderr.read()
            _stdout = stdout.read() # ..
        finally:
            sys.stdout = self.__stdout
            sys.stderr = self.__stderr

            logging.debug('# >2\n%s',_stderr)
            logging.debug('# >1\n%s', _stdout)
            return _stderr, _stdout

class Test_optimize_path(ShellTestCase):
    def test_optimize_path(self):
        print( '\n'.join( optimize_path(os.environ['PATH'].split(':')) ) )

    def test_script(self):
        cmds = (
         ('-p', ),
         ('-c', ),
         #('-w','hashes', ),
         #('-w','duplicates', ),
        )
        here = os.path.abspath(__file__)

        for cmd in cmds:
            _cmd = [here, ] + list(cmd)
            stderr, stdout = self._run_command(_cmd)
            self.assertEqual(len(stderr), 0)<commit_msg>BUG: Optimizepath.py stdout/stderr redirection<commit_after>#!/usr/bin/env python
# encoding: utf-8
from __future__ import print_function
"""
optimize_path
"""
from collections import OrderedDict, Counter
import json
import logging
import os
import re
import subprocess

log = logging.getLogger()

CHECKSUM_BIN='md5sum'

def sbin_policy(iterable):
    """
    # TODO: static optimizations
    """
    for s in sorted(
        (
            (
                (int('/sbin' not in x)*2) +
                (int('/usr/local' in x)*3) +
                (int('/home' in x)*4)
            ),
            len(x),
            x)
            for x in iterable ):
        #logging.debug(s)
        yield s[-1]


def optimize_path(iterable, strip=None, sortfunc=sbin_policy):
    """
    optimize paths
    """
    if strip:
        pattern = re.compile( strip )
        log.debug((pattern, strip, ))
        keys = (p.strip() for p in iterable if not pattern.match(p))
    else:
        keys = (p.strip() for p in iterable)

    paths = OrderedDict.fromkeys(keys)

    if sortfunc:
        return sortfunc(paths.iterkeys())
    else:
        return paths.iterkeys()


def get_checksum(path, hashbin=CHECKSUM_BIN):
    logging.debug((hashbin,path))
    if os.path.isfile(path):
        if os.access(path, os.R_OK):
            return (subprocess.check_output( (hashbin, path) ).rstrip() +
                    (os.path.islink(path) and " -> %s" % os.path.realpath(path) or '') )
    return ('?  %s' % path)


def detect_duplicates(iterable):
    paths = OrderedDict()
    for path in iterable:
        for f in os.listdir(path):
            if not os.path.exists(path):
                continue  # TODO
            basename = os.path.basename(f)
            basename_paths = paths.get(basename,[])
            basename_paths.append(path)
            paths[basename] = basename_paths


    annotated_paths = (
        (k, [get_checksum( os.path.join(p,k) ) for p in v] )
            for (k,v) in paths.iteritems() if len(v) > 1)

    def label(v):
        counts = Counter(l.split()[0] for l in v)
        #logging.debug(counts)
        if len(v) in counts.itervalues():
            return {'type':'duplicates', 'files': sorted(v)}
        else:
            return {'type':'hashes', 'files': v}

    return OrderedDict(
        (k, label(v)) for (k,v) in annotated_paths
    )


def check_difference(iterable):
    raise NotImplementedError()
    import difflib

    iterable = list(iterable)
    regular = list( optimize_path( iterable, sortfunc=None) )
    optimized = list( optimize_path( iterable, sortfunc=sbin_policy ) )

    logging.debug(regular)
    logging.debug(optimized)

    return '\n'.join( difflib.ndiff(regular, optimized) )
    # detect colli

import unittest
import sys
from StringIO import StringIO

class ShellTestCase(unittest.TestCase):
    __stdout = sys.stdout
    __stderr = sys.stderr
    def _run_command(self, _cmd):
        # TODO: not monkey-patch sys.stderr
        try:
            #process = subprocess.Popen((here,) + cmd,
            #    stdout=subprocess.PIPE,
            #    stderr=subprocess.PIPE)
            #stdout,stderr = process.communicate()
            #logging.debug(_cmd)
            logging.info("### TEST CASE: %r" % _cmd)

            stdout = sys.stdout = StringIO()
            stderr = sys.stderr = StringIO()

            sys.argv = _cmd
            main()

            _stderr = stderr.read()
            _stdout = stdout.read() # ..

            logging.debug('# >2\n%s',_stderr)
            logging.debug('# >1\n%s', _stdout)
            return _stderr, _stdout

        finally:
            sys.stdout = self.__stdout
            sys.stderr = self.__stderr



class Test_optimize_path(ShellTestCase):
    def test_optimize_path(self):
        print( '\n'.join( optimize_path(os.environ['PATH'].split(':')) ) )

    def test_script(self):
        cmds = (
         ('-p', ),
         ('-c', ),
         #('-w','hashes', ),
         #('-w','duplicates', ),
        )
        here = os.path.abspath(__file__)

        for cmd in cmds:
            _cmd = [here, ] + list(cmd)
            stderr, stdout = self._run_command(_cmd)
            self.assertEqual(len(stderr), 0)
{'fraction_non_alphanumeric': 0.13175594523180398, 'fraction_numerical': 0.002642325246216671, 'mean_word_length': 2.0636497424576894, 'pattern_counts': {'":': 0, '<': 3, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4200527', 'n_tokens_mistral': 2700, 'n_tokens_neox': 2602, 'n_words': 595}