267 lines
8.3 KiB
Python
267 lines
8.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
ripgrep (rg.exe) 封装：日志 stdin 过滤、目录内文件搜索。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import List, Optional, Tuple
|
||
|
||
_RG_NAMES = ("rg.exe", "rg")
|
||
|
||
|
||
def find_rg_executable() -> Optional[Path]:
|
||
"""解析 rg 可执行文件；找不到返回 None。"""
|
||
env = (os.environ.get("RG_PATH") or "").strip()
|
||
if env:
|
||
p = Path(env)
|
||
if p.is_file():
|
||
return p
|
||
if getattr(sys, "frozen", False):
|
||
base = Path(sys.executable).resolve().parent
|
||
for name in _RG_NAMES:
|
||
cand = base / name
|
||
if cand.is_file():
|
||
return cand
|
||
# resources/bin/rg.exe（开发目录）
|
||
here = Path(__file__).resolve().parent.parent
|
||
project_root = here.parent
|
||
for name in _RG_NAMES:
|
||
cand = project_root / "resources" / "bin" / name
|
||
if cand.is_file():
|
||
return cand
|
||
w = shutil.which("rg")
|
||
if w:
|
||
return Path(w)
|
||
return None
|
||
|
||
|
||
def _build_rg_regex_pattern(pattern: str, whole_word: bool) -> str:
|
||
if whole_word:
|
||
return rf"\b(?:{pattern})\b"
|
||
return pattern
|
||
|
||
|
||
def _utf8_byte_range_to_qt_positions(s: str, start_byte: int, end_byte: int) -> Tuple[int, int]:
|
||
"""
|
||
将 UTF-8 字节区间 [start_byte, end_byte) 转为 QTextCursor 使用的 UTF-16 偏移（与 Qt 一致）。
|
||
"""
|
||
b = s.encode("utf-8")
|
||
if start_byte < 0 or end_byte > len(b) or start_byte > end_byte:
|
||
raise ValueError("invalid utf-8 byte range for document")
|
||
prefix_start = b[:start_byte].decode("utf-8")
|
||
prefix_end = b[:end_byte].decode("utf-8")
|
||
qt_start = len(prefix_start.encode("utf-16-le")) // 2
|
||
qt_end = len(prefix_end.encode("utf-16-le")) // 2
|
||
return qt_start, qt_end
|
||
|
||
|
||
def rg_match_spans_qt(
|
||
full_text: str,
|
||
pattern: str,
|
||
case_sensitive: bool,
|
||
use_regex: bool,
|
||
whole_word: bool,
|
||
max_spans: int = 3000,
|
||
) -> Tuple[Optional[List[Tuple[int, int]]], Optional[str]]:
|
||
"""
|
||
使用 ``rg --json`` 得到与 ripgrep 完全一致的匹配区间，并转为 Qt 文档坐标（UTF-16）。
|
||
|
||
返回 (spans, error)。error 非空表示未调用 rg 或失败；spans 为 ``(start, end)`` 列表，按从左到右顺序。
|
||
"""
|
||
exe = find_rg_executable()
|
||
if not exe:
|
||
return None, "未找到 rg.exe"
|
||
q = (pattern or "").strip()
|
||
if not q:
|
||
return [], None
|
||
args: List[str] = [str(exe), "--json", "--color", "never"]
|
||
if not case_sensitive:
|
||
args.append("-i")
|
||
if not use_regex:
|
||
args.append("--fixed-strings")
|
||
if whole_word:
|
||
args.append("-w")
|
||
pat_arg = q
|
||
else:
|
||
pat_arg = _build_rg_regex_pattern(q, whole_word)
|
||
args.extend(["--", pat_arg, "-"])
|
||
try:
|
||
r = subprocess.run(
|
||
args,
|
||
input=full_text.encode("utf-8"),
|
||
capture_output=True,
|
||
timeout=120,
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||
if sys.platform == "win32"
|
||
else 0,
|
||
)
|
||
except Exception as e:
|
||
return None, str(e)
|
||
err_txt = (r.stderr or b"").decode("utf-8", errors="replace").strip()
|
||
if r.returncode == 2:
|
||
return None, err_txt or "rg 正则解析失败"
|
||
if r.returncode not in (0, 1):
|
||
return None, err_txt or f"rg 退出码 {r.returncode}"
|
||
out = (r.stdout or b"").decode("utf-8", errors="replace")
|
||
spans: List[Tuple[int, int]] = []
|
||
for line in out.splitlines():
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
obj = json.loads(line)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if obj.get("type") != "match":
|
||
continue
|
||
data = obj.get("data") or {}
|
||
abs_off = data.get("absolute_offset")
|
||
if abs_off is None:
|
||
continue
|
||
for sm in data.get("submatches") or []:
|
||
try:
|
||
sb = abs_off + int(sm["start"])
|
||
eb = abs_off + int(sm["end"])
|
||
qt_s, qt_e = _utf8_byte_range_to_qt_positions(full_text, sb, eb)
|
||
except (KeyError, TypeError, ValueError):
|
||
continue
|
||
spans.append((qt_s, qt_e))
|
||
if len(spans) >= max_spans:
|
||
return spans, None
|
||
return spans, None
|
||
|
||
|
||
def compile_python_span_pattern(
|
||
pattern: str,
|
||
case_sensitive: bool,
|
||
use_regex: bool,
|
||
whole_word: bool,
|
||
) -> Optional[re.Pattern]:
|
||
"""与界面选项一致；在 rg 不可用时用于高亮与跳转（Python re）。"""
|
||
if not (pattern or "").strip():
|
||
return None
|
||
q = pattern.strip()
|
||
flags = 0 if case_sensitive else re.IGNORECASE
|
||
try:
|
||
if use_regex:
|
||
src = _build_rg_regex_pattern(q, whole_word)
|
||
return re.compile(src, flags)
|
||
inner = re.escape(q)
|
||
if whole_word:
|
||
inner = rf"\b{inner}\b"
|
||
return re.compile(inner, flags)
|
||
except re.error:
|
||
return None
|
||
|
||
|
||
def rg_filter_lines(
|
||
full_text: str,
|
||
pattern: str,
|
||
case_sensitive: bool,
|
||
use_regex: bool,
|
||
whole_word: bool,
|
||
) -> Tuple[Optional[str], Optional[str]]:
|
||
"""
|
||
将全文经 stdin 交给 rg，返回仅匹配行（与 rg 默认输出一致）。
|
||
|
||
- use_regex=False：``--fixed-strings``，可选 ``-w`` 全词。
|
||
- use_regex=True：Rust 正则语法（与 Python 略有差异）；全词时包 ``\\b(?:... )\\b``。
|
||
返回 (result, error)；error 非空表示未调用 rg 或失败。
|
||
"""
|
||
exe = find_rg_executable()
|
||
if not exe:
|
||
return None, "未找到 rg.exe，请将 rg.exe 放在 Server Manager 目录或设置环境变量 RG_PATH"
|
||
q = (pattern or "").strip()
|
||
if not q:
|
||
return full_text, None
|
||
args: List[str] = [str(exe), "--color", "never", "--no-heading"]
|
||
if not case_sensitive:
|
||
args.append("-i")
|
||
if not use_regex:
|
||
args.append("--fixed-strings")
|
||
if whole_word:
|
||
args.append("-w")
|
||
pat_arg = q
|
||
else:
|
||
pat_arg = _build_rg_regex_pattern(q, whole_word)
|
||
args.extend(["--", pat_arg, "-"])
|
||
try:
|
||
r = subprocess.run(
|
||
args,
|
||
input=full_text.encode("utf-8"),
|
||
capture_output=True,
|
||
timeout=120,
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||
if sys.platform == "win32"
|
||
else 0,
|
||
)
|
||
except Exception as e:
|
||
return None, str(e)
|
||
if r.returncode not in (0, 1):
|
||
err = (r.stderr or b"").decode("utf-8", errors="replace").strip()
|
||
return None, err or f"rg 退出码 {r.returncode}"
|
||
out = (r.stdout or b"").decode("utf-8", errors="replace")
|
||
if r.returncode == 1:
|
||
return "", None
|
||
return out.rstrip("\n"), None
|
||
|
||
|
||
def rg_search_directory(
|
||
search_root: Path,
|
||
pattern: str,
|
||
case_sensitive: bool,
|
||
use_regex: bool,
|
||
whole_word: bool,
|
||
glob_globs: Optional[List[str]] = None,
|
||
) -> Tuple[str, Optional[str]]:
|
||
"""在目录下递归搜索，返回 rg 标准输出（含文件名与行号）。"""
|
||
exe = find_rg_executable()
|
||
if not exe:
|
||
return "", "未找到 rg.exe"
|
||
q = (pattern or "").strip()
|
||
if not q:
|
||
return "", "请输入搜索内容"
|
||
if not search_root.is_dir():
|
||
return "", f"目录不存在: {search_root}"
|
||
args: List[str] = [str(exe), "-n", "-S", "--color", "never", "--heading"]
|
||
if not case_sensitive:
|
||
args.append("-i")
|
||
if not use_regex:
|
||
args.append("--fixed-strings")
|
||
if whole_word:
|
||
args.append("-w")
|
||
pat_arg = q
|
||
else:
|
||
pat_arg = _build_rg_regex_pattern(q, whole_word)
|
||
if glob_globs:
|
||
for g in glob_globs:
|
||
args.extend(["--glob", g])
|
||
args.extend(["--", pat_arg, str(search_root.resolve())])
|
||
try:
|
||
r = subprocess.run(
|
||
args,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=180,
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||
if sys.platform == "win32"
|
||
else 0,
|
||
)
|
||
except Exception as e:
|
||
return "", str(e)
|
||
out = (r.stdout or "").strip()
|
||
err = (r.stderr or "").strip()
|
||
if r.returncode not in (0, 1):
|
||
return "", err or f"rg 退出码 {r.returncode}"
|
||
if r.returncode == 1 and not out:
|
||
return "(无匹配)", None
|
||
return out, None
|