1. refactor python vim module; 1.1. experiment with fast select based on popup in vim, and multi threaded app in python; 1.2. TODO, figure out some thread safe way to call vim.command from python side threads; 1.3. update pydantic validate params;
244 lines
4.8 KiB
Python
244 lines
4.8 KiB
Python
import functools
|
|
import configparser
|
|
import collections
|
|
import asyncio
|
|
import threading
|
|
import re
|
|
import inspect
|
|
import pathlib
|
|
import logging
|
|
import fnmatch
|
|
import vim
|
|
|
|
from typing import (
|
|
Optional,
|
|
ClassVar,
|
|
Self,
|
|
Any,
|
|
Callable,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
logging.basicConfig(level=logging.WARNING)
|
|
|
|
|
|
MODULE_NAME = 'online_fxreader_pr34_vim'
|
|
|
|
def f1():
|
|
t1 = vim.current.window
|
|
t2 = t1.width
|
|
vim.command('vnew')
|
|
t3 = t2 // 3
|
|
vim.command('vertical resize %d' % t3)
|
|
vim.current.window = t1
|
|
|
|
|
|
def f2():
|
|
context = {k: vim.options['splitright'] for k in ['splitright']}
|
|
try:
|
|
current_window = vim.current.window
|
|
vim.options['splitright'] = True
|
|
vim.command('vnew')
|
|
vim.command('r! tmux show-buffer')
|
|
vim.current.window = current_window
|
|
finally:
|
|
for k, v in context.items():
|
|
vim.options[k] = v
|
|
|
|
|
|
def f5_1(pattern, flags, info):
|
|
import subprocess
|
|
import io
|
|
import re
|
|
import tempfile
|
|
import traceback
|
|
import logging
|
|
|
|
# print([pattern, flags, info])
|
|
completed_process = None
|
|
|
|
options = dict(
|
|
recursive=False,
|
|
ext=[],
|
|
)
|
|
|
|
# print('fuck')
|
|
if b'r' in flags:
|
|
while True:
|
|
ext_m = re.compile(r'^.([^\,]+),(.*)$').match(pattern)
|
|
|
|
if pattern[:3] in [r'\r,']:
|
|
options['recursive'] = True
|
|
pattern = pattern[3:]
|
|
elif not ext_m is None:
|
|
options['ext'].append(ext_m[1])
|
|
pattern = ext_m[2]
|
|
else:
|
|
break
|
|
|
|
print(
|
|
[
|
|
flags,
|
|
pattern,
|
|
options,
|
|
]
|
|
)
|
|
|
|
try:
|
|
git_cmd = [
|
|
'git',
|
|
'grep',
|
|
'-n',
|
|
]
|
|
|
|
if options['recursive']:
|
|
git_cmd.append('--recurse-submodules')
|
|
|
|
git_cmd.extend(['-P', pattern])
|
|
|
|
if len(options['ext']) > 0:
|
|
git_cmd.extend(['--', *['**/*%s' % o for o in options['ext']]])
|
|
|
|
completed_process = subprocess.run(
|
|
git_cmd,
|
|
capture_output=True,
|
|
)
|
|
assert completed_process.returncode == 0 or (
|
|
completed_process.stdout == b''
|
|
# completed_process.stdout == b'' and
|
|
# completed_process.stderr == b''
|
|
)
|
|
t1 = completed_process.stdout
|
|
except:
|
|
logging.error(
|
|
''.join(
|
|
[
|
|
traceback.format_exc(),
|
|
getattr(completed_process, 'stdout', b'').decode('utf-8'),
|
|
getattr(completed_process, 'stderr', b'').decode('utf-8'),
|
|
]
|
|
)
|
|
)
|
|
t1 = b''
|
|
|
|
def watch(data):
|
|
with tempfile.NamedTemporaryFile(suffix='.txt') as f:
|
|
with io.open(f.name, 'wb') as f2:
|
|
f2.write(data)
|
|
vim.command('!less %s' % f.name)
|
|
|
|
# watch(t1)
|
|
|
|
t2 = []
|
|
for o in t1.splitlines():
|
|
try:
|
|
# watch(o.encode('utf-8'))
|
|
t3 = o.decode('utf-8')
|
|
t4 = re.compile(r'^([^\:\=]+)[\:\=](\d+)[\:\=](.*)$').match(t3)
|
|
if not t4 is None:
|
|
t2.append(
|
|
dict(
|
|
name=t4[3].strip(),
|
|
filename=t4[1],
|
|
cmd=t4[2],
|
|
)
|
|
)
|
|
except:
|
|
pass
|
|
# print(t2)
|
|
|
|
# return [{'name': 'blah', 'filename': 'docker-compose.yml', 'cmd': '23'}]
|
|
return t2
|
|
|
|
|
|
class EditorConfigModeline:
|
|
_instance: ClassVar[Optional['EditorConfigModeline']] = None
|
|
|
|
def __init__(self) -> None:
|
|
self.configs: dict[
|
|
pathlib.Path,
|
|
dict[str, str],
|
|
] = dict()
|
|
|
|
@classmethod
|
|
def singleton(cls) -> Self:
|
|
if cls._instance is None:
|
|
cls._instance = cls()
|
|
|
|
return cls._instance
|
|
|
|
def load_config(self) -> Optional[dict[str, str]]:
|
|
cwd = pathlib.Path.cwd()
|
|
|
|
if not cwd in self.configs:
|
|
config_path = cwd / '.editorconfig'
|
|
|
|
if not config_path.exists():
|
|
return None
|
|
|
|
parser = configparser.ConfigParser()
|
|
parser.optionxform = str # keep case
|
|
parser.read(str(config_path))
|
|
|
|
config: dict[str, str] = dict()
|
|
|
|
for section in parser.sections():
|
|
logger.info(dict(section=section))
|
|
|
|
if len(section) > 0:
|
|
# pattern = section[1:-1]
|
|
pattern = section
|
|
if not parser[section].get('vim_modeline') is None:
|
|
config[pattern] = parser[section].get('vim_modeline')
|
|
self.validate_modeline(config[pattern])
|
|
|
|
self.configs[cwd] = config
|
|
|
|
return self.configs[cwd]
|
|
|
|
@classmethod
|
|
def validate_modeline(cls, modeline: str) -> None:
|
|
pattern = re.compile(r'^set(\s+(noet|sts|ts|et|ai|ci|noai|noci|sw)(\=\w)?)+$')
|
|
assert pattern.match(modeline), 'invalid modeline %s' % modeline
|
|
|
|
@classmethod
|
|
def find_entry(
|
|
cls,
|
|
file_path: pathlib.Path,
|
|
config: Optional[dict[str, str]] = None,
|
|
) -> Optional[str]:
|
|
if config is None:
|
|
return None
|
|
|
|
project_root = pathlib.Path.cwd()
|
|
|
|
if file_path.is_relative_to(project_root):
|
|
rel_path = file_path.relative_to(pathlib.Path.cwd())
|
|
else:
|
|
rel_path = file_path
|
|
|
|
for pattern, modeline in config.items():
|
|
if fnmatch.fnmatch(str(rel_path), pattern):
|
|
return modeline
|
|
|
|
return None
|
|
|
|
def on_buffer(self) -> None:
|
|
config = self.load_config()
|
|
|
|
logger.info(dict(config=config))
|
|
|
|
buf_name = vim.current.buffer.name
|
|
file_path = pathlib.Path(buf_name).resolve()
|
|
|
|
entry = self.find_entry(file_path, config=config)
|
|
|
|
logger.info(dict(modeline=entry))
|
|
|
|
vim.command('silent! {}'.format(entry))
|
|
|
|
# vim.command("echo '{}'".format('applied %s' % entry))
|
|
|
|
# raise NotImplementedError
|