wting_autojump/bin/autojump

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
  Copyright © 2008-2012 Joel Schaerer
  Copyright © 2012-2016 William Ting

  * This program is free software; you can redistribute it and/or modify
  it under the terms of the GNU General Public License as published by
  the Free Software Foundation; either version 3, or (at your option)
  any later version.

  * This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU General Public License for more details.

  * You should have received a copy of the GNU General Public License
  along with this program; if not, write to the Free Software Foundation, Inc.,
  51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""
from __future__ import print_function

import os
import sys
import subprocess
from itertools import chain
from math import sqrt
from operator import attrgetter
from operator import itemgetter

if sys.version_info[0] == 3:
    ifilter = filter
    imap = map
    os.getcwdu = os.getcwd
else:
    from itertools import ifilter
    from itertools import imap

# Vendorized argparse for Python 2.6 support
from autojump_argparse import ArgumentParser

# autojump is not a standard python package but rather installed as a mixed
# shell + Python app with no outside dependencies (except Python). As a
# consequence we use relative imports and depend on file prefixes to prevent
# module conflicts.
from autojump_data import dictify
from autojump_data import entriefy
from autojump_data import Entry
from autojump_data import load
from autojump_data import save
from autojump_match import match_anywhere
from autojump_match import match_consecutive
from autojump_match import match_fuzzy
from autojump_utils import first
from autojump_utils import get_pwd
from autojump_utils import get_tab_entry_info
from autojump_utils import has_uppercase
from autojump_utils import is_autojump_sourced
from autojump_utils import is_osx
from autojump_utils import is_msys2
from autojump_utils import is_windows
from autojump_utils import last
from autojump_utils import print_entry
from autojump_utils import print_local
from autojump_utils import print_tab_menu
from autojump_utils import sanitize
from autojump_utils import take
from autojump_utils import unico

VERSION = '22.5.3'
FUZZY_MATCH_THRESHOLD = 0.6
TAB_ENTRIES_COUNT = 9
TAB_SEPARATOR = '__'

def convert_path(path, option="-u"):
    """
    使用 cygpath 转换路径格式

    参数:
    - path: 要转换的路径（字符串）
    - option: 转换选项
      - '-u': Windows 路径 → Unix 路径（默认）
      - '-w': Unix 路径 → Windows 路径
    """
    if is_msys2():
        try:
            result = subprocess.run(
                ['cygpath', option, path],
                capture_output=True,
                text=True,
                check=True
            )
            return result.stdout.strip()  # 返回转换后的路径
        except subprocess.CalledProcessError as e:
            print(f"Error: {e.stderr}")
            return path  # 转换失败时返回原始路径
    else:
        return path;

def set_defaults():
    config = {}

    if is_osx():
        data_home = os.path.join(os.path.expanduser('~'), 'Library')
    elif is_windows() and not is_msys2():
        data_home = os.getenv('APPDATA')
    else:
        data_home = os.getenv(
            'XDG_DATA_HOME',
            os.path.join(
                os.getenv('HOME'),
                '.local',
                'share',
            ),
        )
    config['data_path'] = os.path.join(data_home, 'autojump', 'autojump.txt')
    config['backup_path'] = os.path.join(data_home, 'autojump', 'autojump.txt.bak')

    return config


def parse_arguments():
    parser = ArgumentParser(
        description='Automatically jump to directory passed as an argument.',
        epilog='Please see autojump(1) man pages for full documentation.',
    )
    parser.add_argument(
        'directory', metavar='DIRECTORY', nargs='*', default='',
        help='directory to jump to',
    )
    parser.add_argument(
        '-a', '--add', metavar='DIRECTORY',
        help='add path',
    )
    parser.add_argument(
        '-i', '--increase', metavar='WEIGHT', nargs='?', type=int,
        const=10, default=False,
        help='increase current directory weight',
    )
    parser.add_argument(
        '-d', '--decrease', metavar='WEIGHT', nargs='?', type=int,
        const=15, default=False,
        help='decrease current directory weight',
    )
    parser.add_argument(
        '--complete', action='store_true', default=False,
        help='used for tab completion',
    )
    parser.add_argument(
        '--purge', action='store_true', default=False,
        help='remove non-existent paths from database',
    )
    parser.add_argument(
        '-s', '--stat', action='store_true', default=False,
        help='show database entries and their key weights',
    )
    parser.add_argument(
        '-v', '--version', action='version', version='%(prog)s v' +
        VERSION, help='show version information',
    )

    return parser.parse_args()


def add_path(data, path, weight=10):
    """
    Add a new path or increment an existing one.

    os.path.realpath() is not used because it's preferable to use symlinks
    with resulting duplicate entries in the database than a single canonical
    path.
    """
    path = unico(path).rstrip(os.sep)
    if path == os.path.expanduser('~'):
        return data, Entry(path, 0)

    data[path] = sqrt((data.get(path, 0) ** 2) + (weight ** 2))

    return data, Entry(path, data[path])


def decrease_path(data, path, weight=15):
    """Decrease or zero out a path."""
    path = unico(path).rstrip(os.sep)
    data[path] = max(0, data.get(path, 0) - weight)
    return data, Entry(path, data[path])


def detect_smartcase(needles):
    """
    If any needles contain an uppercase letter then use case sensitive
    searching. Otherwise use case insensitive searching.
    """
    return not any(imap(has_uppercase, needles))


def find_matches(entries, needles, check_entries=True):
    """Return an iterator to matching entries."""
    # TODO(wting|2014-02-24): replace assertion with unit test
    assert isinstance(needles, list), 'Needles must be a list.'
    ignore_case = detect_smartcase(needles)

    try:
        pwd = os.getcwdu()
    except OSError:
        pwd = None

    # using closure to prevent constantly hitting hdd
    def is_cwd(entry):
        return os.path.realpath(entry.path) == pwd

    if check_entries:
        path_exists = lambda entry: os.path.exists(entry.path)
    else:
        path_exists = lambda _: True

    data = sorted(
        entries,
        key=attrgetter('weight', 'path'),
        reverse=True,
    )

    return ifilter(
        lambda entry: not is_cwd(entry) and path_exists(entry),
        chain(
            match_consecutive(needles, data, ignore_case),
            match_fuzzy(needles, data, ignore_case),
            match_anywhere(needles, data, ignore_case),
        ),
    )


def handle_tab_completion(needle, entries):
    tab_needle, tab_index, tab_path = get_tab_entry_info(needle, TAB_SEPARATOR)

    if tab_path:
        print_local(tab_path)
    elif tab_index:
        get_ith_path = lambda i, iterable: last(take(i, iterable)).path
        print_local(get_ith_path(
            tab_index,
            find_matches(entries, [tab_needle], check_entries=False),
        ))
    elif tab_needle:
        # found partial tab completion entry
        print_tab_menu(
            tab_needle,
            take(
                TAB_ENTRIES_COUNT, find_matches(
                    entries,
                    [tab_needle],
                    check_entries=False,
                ),
            ),
            TAB_SEPARATOR,
        )
    else:
        print_tab_menu(
            needle,
            take(
                TAB_ENTRIES_COUNT, find_matches(
                    entries,
                    [needle],
                    check_entries=False,
                ),
            ),
            TAB_SEPARATOR,
        )


def purge_missing_paths(entries):
    """Remove non-existent paths from a list of entries."""
    exists = lambda entry: os.path.exists(entry.path)
    return ifilter(exists, entries)


def print_stats(data, data_path):
    for path, weight in sorted(data.items(), key=itemgetter(1)):
        print_entry(Entry(path, weight))

    print('________________________________________\n')
    print('%d:\t total weight' % sum(data.values()))
    print('%d:\t number of entries' % len(data))

    try:
        print_local(
            '%.2f:\t current directory weight' % data.get(os.getcwdu(), 0),
        )
    except OSError:
        # current directory no longer exists
        pass

    print('\ndata:\t %s' % data_path)


def main(args):  # noqa
    if not is_autojump_sourced() and not is_windows():
        print("Please source the correct autojump file in your shell's")
        print('startup file. For more information, please reinstall autojump')
        print('and read the post installation instructions.')
        return 1

    config = set_defaults();

    # all arguments are mutually exclusive
    if args.add:
        args.add = convert_path(args.add);
        save(config, first(add_path(load(config), args.add)))
    elif args.complete:
        handle_tab_completion(
            needle=first(chain(sanitize(args.directory), [''])),
            entries=entriefy(load(config)),
        )
    elif args.decrease:
        data, entry = decrease_path(load(config), get_pwd(), args.decrease)
        save(config, data)
        print_entry(entry)
    elif args.increase:
        data, entry = add_path(load(config), get_pwd(), args.increase)
        save(config, data)
        print_entry(entry)
    elif args.purge:
        old_data = load(config)
        new_data = dictify(purge_missing_paths(entriefy(old_data)))
        save(config, new_data)
        print('Purged %d entries.' % (len(old_data) - len(new_data)))
    elif args.stat:
        print_stats(load(config), config['data_path'])
    elif not args.directory:
        # Return best match.
        entries = entriefy(load(config))
        print_local(first(chain(
            imap(attrgetter('path'), find_matches(entries, [''])),
            # always return a path to calling shell functions
            ['.'],
        )))
    else:
        entries = entriefy(load(config))
        needles = sanitize(args.directory)
        tab_needle, tab_index, tab_path = \
            get_tab_entry_info(first(needles), TAB_SEPARATOR)

        # Handle `j foo__`, assuming first index.
        if not tab_path and not tab_index \
                and tab_needle and needles[0] == tab_needle + TAB_SEPARATOR:
            tab_index = 1

        if tab_path:
            print_local(tab_path)
        elif tab_index:
            get_ith_path = lambda i, iterable: last(take(i, iterable)).path
            print_local(
                get_ith_path(
                    tab_index,
                    find_matches(entries, [tab_needle]),
                ),
            )
        else:
            print_local(first(chain(
                imap(attrgetter('path'), find_matches(entries, needles)),
                # always return a path to calling shell functions
                ['.'],
            )))

    return 0


if __name__ == '__main__':
    if is_msys2():
        class BinaryStdout:
            '''
            print(xxx, end="\n")
            '''
            def __init__(self, stream):
                self.stream = stream

            def write(self, s):
                if isinstance(s, str):
                    s = s.encode('utf-8')
                self.stream.buffer.write(s)

            def flush(self):
                self.stream.buffer.flush()

            def __getattr__(self, attr):
                return getattr(self.stream, attr)


        sys.stdout = BinaryStdout(sys.stdout)
    sys.exit(main(parse_arguments()))