<commit_before>import datetime as dt
import logging
import sqlite3
import textwrap
import time
import sys

import click

from . import db
from .config import *
from .parser import (
    Command,
    parse_history_line,
    render_timestamp,
)


logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

stderr = logging.StreamHandler(sys.stderr)

logger.addHandler(stderr)


class PrefixWrapper(textwrap.TextWrapper):
    def __init__(self, prefix, *args, **kwargs):
        super().__init__(*args, initial_indent=prefix, subsequent_indent=prefix,
                         break_long_words=False, break_on_hyphens=False,
                         **kwargs)


def error(message):
    print('\n'.join(PrefixWrapper('error: ').wrap(message)), file=sys.stderr)


def hint(message):
    print('\n'.join(PrefixWrapper('hint: ').wrap(message)), file=sys.stderr)


class AliasedGroup(click.Group):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._aliases = {}

    def command(self, *args, **kwargs):
        aliases = kwargs.pop('aliases', [])
        def decorator(func):
            cmd = click.decorators.command(*args, **kwargs)(func)
            self.add_command(cmd, aliases=aliases)
            return cmd
        return decorator

    def add_command(self, cmd, name=None, aliases=None):
        super().add_command(cmd, name)
        if aliases:
            for alias in aliases:
                self._aliases[alias] = name or cmd.name

    def get_command(self, ctx, cmd_name):
        command = super().get_command(ctx, cmd_name)
        if command is None:
            command = super().get_command(ctx, self._aliases.get(cmd_name))
        return command


@click.group('duiker', cls=AliasedGroup)
@click.pass_context
def cli(ctx):
    DUIKER_HOME.mkdir(mode=0o700, parents=True, exist_ok=True)
    ctx.obj = {'DB': str(DUIKER_DB)}

    migrator = db.Migrator(DUIKER_DB)
    applied_migrations = set(migrator.applied_migrations)
    if not applied_migrations:
        migrator.database.user_version = 0
    migrations = db.sort_migrations()
    for migration in migrations:
        if migration not in applied_migrations:
            migration_module = db.get_migration(migration)
            logger.info('Applying migration: {}'.format(migration_module.__doc__.strip()))
            migrator.apply(migration_module)

    if HISTTIMEFORMAT:
        try:
            timestamp = dt.datetime.now().strftime(HISTTIMEFORMAT)
            dt.datetime.strptime(timestamp, HISTTIMEFORMAT)
        except ValueError:
            error('Cannot parse HISTTIMEFORMAT ({}).'.format(HISTTIMEFORMAT))
            hint('Use only standard format codes: <https://docs.python.org/3/library/datetime.html#strftime-and-strptime-behavior>.')
            sys.exit(1)



@cli.command()
@click.option('-n', '--entries', type=click.INT, default=20, show_default=True, help='recall last N commands')
@click.pass_context
def head(ctx, entries):
    """
    Show first N commands.
    """
    query = 'SELECT * FROM history ORDER BY timestamp ASC LIMIT ?'
    with sqlite3.connect(ctx.obj['DB']) as connection:
        connection.row_factory = Command.from_row
        for command in connection.execute(query, (entries,)):
            print('{:tsv}'.format(command))


@cli.command('import')
@click.argument('histfile', type=click.File(errors='replace'))
@click.option('-q', '--quiet', is_flag=True, help='do not print imported commands')
@click.pass_context
def _import(ctx, histfile, quiet):
    """
    Import Bash history output into database.

    Import from standard input:

        \bhistory | duiker import -

    Import from history output:

        \bhistory > history_file
        \bduiker import history_file
    """
    if quiet:
        logger.setLevel(logging.CRITICAL)

    with sqlite3.connect(ctx.obj['DB']) as db:
        for line in histfile:
            try:
                command = parse_history_line(line, HISTTIMEFORMAT)
            except Exception as exc:
                raise click.ClickException(exc)
            if command.timestamp is None:
                command = command._replace(timestamp=time.mktime(dt.datetime.now().timetuple()))
            db.execute('INSERT INTO history VALUES (?, ?, ?)', command)
            db.execute('INSERT INTO fts_history SELECT id, command FROM history WHERE rowid = last_insert_rowid()')
            logger.info('Imported `%s` issued %s', command.command, render_timestamp(command.timestamp))


@cli.command()
@click.pass_context
def log(ctx):
    """
    Show commands from all time.
    """
    query = 'SELECT * FROM history ORDER BY timestamp'
    with sqlite3.connect(ctx.obj['DB']) as connection:
        connection.row_factory = Command.from_row
        for command in connection.execute(query):
            print('{:tsv}'.format(command))


@cli.command()
@click.pass_context
def magic(ctx):
    """
    Print shell function that imports last command into Duiker.
    """
    print(MAGIC)


@cli.command()
@click.argument('expression')
@click.pass_context
def search(ctx, expression):
    """
    Search for a command in the history database.

    Use any SQLite full-text search (FTS) query:

        \bhttps://sqlite.org/fts3.html#full_text_index_queries
    """
    query = '''SELECT history.*
                 FROM fts_history
                 JOIN history<commit_msg>Restore magic example<commit_after>import datetime as dt
import logging
import sqlite3
import textwrap
import time
import sys

import click

from . import db
from .config import *
from .parser import (
    Command,
    parse_history_line,
    render_timestamp,
)


logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

stderr = logging.StreamHandler(sys.stderr)

logger.addHandler(stderr)


class PrefixWrapper(textwrap.TextWrapper):
    def __init__(self, prefix, *args, **kwargs):
        super().__init__(*args, initial_indent=prefix, subsequent_indent=prefix,
                         break_long_words=False, break_on_hyphens=False,
                         **kwargs)


def error(message):
    print('\n'.join(PrefixWrapper('error: ').wrap(message)), file=sys.stderr)


def hint(message):
    print('\n'.join(PrefixWrapper('hint: ').wrap(message)), file=sys.stderr)


class AliasedGroup(click.Group):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._aliases = {}

    def command(self, *args, **kwargs):
        aliases = kwargs.pop('aliases', [])
        def decorator(func):
            cmd = click.decorators.command(*args, **kwargs)(func)
            self.add_command(cmd, aliases=aliases)
            return cmd
        return decorator

    def add_command(self, cmd, name=None, aliases=None):
        super().add_command(cmd, name)
        if aliases:
            for alias in aliases:
                self._aliases[alias] = name or cmd.name

    def get_command(self, ctx, cmd_name):
        command = super().get_command(ctx, cmd_name)
        if command is None:
            command = super().get_command(ctx, self._aliases.get(cmd_name))
        return command


@click.group('duiker', cls=AliasedGroup)
@click.pass_context
def cli(ctx):
    DUIKER_HOME.mkdir(mode=0o700, parents=True, exist_ok=True)
    ctx.obj = {'DB': str(DUIKER_DB)}

    migrator = db.Migrator(DUIKER_DB)
    applied_migrations = set(migrator.applied_migrations)
    if not applied_migrations:
        migrator.database.user_version = 0
    migrations = db.sort_migrations()
    for migration in migrations:
        if migration not in applied_migrations:
            migration_module = db.get_migration(migration)
            logger.info('Applying migration: {}'.format(migration_module.__doc__.strip()))
            migrator.apply(migration_module)

    if HISTTIMEFORMAT:
        try:
            timestamp = dt.datetime.now().strftime(HISTTIMEFORMAT)
            dt.datetime.strptime(timestamp, HISTTIMEFORMAT)
        except ValueError:
            error('Cannot parse HISTTIMEFORMAT ({}).'.format(HISTTIMEFORMAT))
            hint('Use only standard format codes: <https://docs.python.org/3/library/datetime.html#strftime-and-strptime-behavior>.')
            sys.exit(1)



@cli.command()
@click.option('-n', '--entries', type=click.INT, default=20, show_default=True, help='recall last N commands')
@click.pass_context
def head(ctx, entries):
    """
    Show first N commands.
    """
    query = 'SELECT * FROM history ORDER BY timestamp ASC LIMIT ?'
    with sqlite3.connect(ctx.obj['DB']) as connection:
        connection.row_factory = Command.from_row
        for command in connection.execute(query, (entries,)):
            print('{:tsv}'.format(command))


@cli.command('import')
@click.argument('histfile', type=click.File(errors='replace'))
@click.option('-q', '--quiet', is_flag=True, help='do not print imported commands')
@click.pass_context
def _import(ctx, histfile, quiet):
    """
    Import Bash history output into database.

    Import from standard input:

        \bhistory | duiker import -

    Import from history output:

        \bhistory > history_file
        \bduiker import history_file
    """
    if quiet:
        logger.setLevel(logging.CRITICAL)

    with sqlite3.connect(ctx.obj['DB']) as db:
        for line in histfile:
            try:
                command = parse_history_line(line, HISTTIMEFORMAT)
            except Exception as exc:
                raise click.ClickException(exc)
            if command.timestamp is None:
                command = command._replace(timestamp=time.mktime(dt.datetime.now().timetuple()))
            db.execute('INSERT INTO history VALUES (?, ?, ?)', command)
            db.execute('INSERT INTO fts_history SELECT id, command FROM history WHERE rowid = last_insert_rowid()')
            logger.info('Imported `%s` issued %s', command.command, render_timestamp(command.timestamp))


@cli.command()
@click.pass_context
def log(ctx):
    """
    Show commands from all time.
    """
    query = 'SELECT * FROM history ORDER BY timestamp'
    with sqlite3.connect(ctx.obj['DB']) as connection:
        connection.row_factory = Command.from_row
        for command in connection.execute(query):
            print('{:tsv}'.format(command))


@cli.command(epilog="""Example:

    \b
{}

    \b
__prompt() {{
  history -a
  __duiker_import
  PS1="\\u@\\h:\\w$ "
}}

    \b
PROMPT_COMMAND=__prompt
    \b
""".format(MAGIC))
@click.pass_context
def magic(ctx):
    """
    Print shell function that imports last command into Duiker.
    """
    print(MAGIC)


@cli.command()
@click.argument('expression')
@click.pass_context
def search(ctx, expression):
    """
    Search for a command in the history database.

    Use any SQLite full-text search (FTS) query:

        \bhttps://sqlite.org/fts3.html#full_text_index_queries
    """
    query = '''SELECT history.*
                 FROM fts_history
                 JOIN history
{'fraction_non_alphanumeric': 0.12180190058479533, 'fraction_numerical': 0.0026498538011695905, 'mean_word_length': 2.927161822748475, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 4, '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': '3346804', 'n_tokens_mistral': 3311, 'n_tokens_neox': 3239, 'n_words': 750}