98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
"""Full-text search over a loaded document's glyph bounds.
|
|
|
|
Extracted verbatim from ``routers.documents.search_document`` so the router stays
|
|
a thin HTTP wrapper. The router remains responsible for the engine-availability /
|
|
document-existence checks and for translating exceptions into HTTP errors.
|
|
"""
|
|
|
|
from app.schemas.search import SearchMatch, SearchRect
|
|
|
|
|
|
def compute_search_matches(
|
|
doc,
|
|
q: str,
|
|
case_sensitive: bool = False,
|
|
whole_words: bool = False,
|
|
) -> list[SearchMatch]:
|
|
def _is_word_char(ch: str) -> bool:
|
|
return ch.isalnum() or ch == "_"
|
|
|
|
def _find_all(haystack: str, needle: str) -> list[int]:
|
|
"""Return start indices of all non-overlapping occurrences of needle in haystack."""
|
|
results: list[int] = []
|
|
start = 0
|
|
needle_len = len(needle)
|
|
while True:
|
|
pos = haystack.find(needle, start)
|
|
if pos == -1:
|
|
break
|
|
if whole_words:
|
|
before_ok = pos == 0 or not _is_word_char(haystack[pos - 1])
|
|
after_ok = (pos + needle_len) >= len(haystack) or not _is_word_char(
|
|
haystack[pos + needle_len]
|
|
)
|
|
if before_ok and after_ok:
|
|
results.append(pos)
|
|
else:
|
|
results.append(pos)
|
|
start = pos + 1
|
|
return results
|
|
|
|
matches = []
|
|
search_needle = q if case_sensitive else q.lower()
|
|
query_len = len(search_needle)
|
|
|
|
for page_idx in range(doc.page_count):
|
|
page = doc.get_page(page_idx)
|
|
glyphs = page.extract_text_with_bounds()
|
|
if not glyphs:
|
|
continue
|
|
|
|
text_str = ""
|
|
char_to_glyph: list[int] = []
|
|
for i, g in enumerate(glyphs):
|
|
s = g.get("text", "")
|
|
start_len = len(text_str)
|
|
text_str += s
|
|
for _ in range(len(text_str) - start_len):
|
|
char_to_glyph.append(i)
|
|
|
|
search_text = text_str if case_sensitive else text_str.lower()
|
|
|
|
for idx in _find_all(search_text, search_needle):
|
|
if idx + query_len - 1 >= len(char_to_glyph):
|
|
continue
|
|
|
|
start_glyph_idx = char_to_glyph[idx]
|
|
end_glyph_idx = char_to_glyph[idx + query_len - 1]
|
|
|
|
rects = []
|
|
current_rect = None
|
|
|
|
for g_idx in range(start_glyph_idx, end_glyph_idx + 1):
|
|
g = glyphs[g_idx]
|
|
dom_y = page.height - (g["y"] + g["h"])
|
|
|
|
if current_rect is None:
|
|
current_rect = {"x": g["x"], "y": dom_y, "w": g["w"], "h": g["h"]}
|
|
else:
|
|
if abs(dom_y - current_rect["y"]) < g.get("fontSize", 12) * 0.5:
|
|
max_x = max(current_rect["x"] + current_rect["w"], g["x"] + g["w"])
|
|
current_rect["w"] = max_x - current_rect["x"]
|
|
current_rect["y"] = min(current_rect["y"], dom_y)
|
|
current_rect["h"] = max(current_rect["h"], g["h"])
|
|
else:
|
|
rects.append(SearchRect(**current_rect))
|
|
current_rect = {"x": g["x"], "y": dom_y, "w": g["w"], "h": g["h"]}
|
|
|
|
if current_rect:
|
|
rects.append(SearchRect(**current_rect))
|
|
|
|
matches.append(
|
|
SearchMatch(
|
|
pageIndex=page_idx, rects=rects, text=text_str[idx : idx + query_len]
|
|
)
|
|
)
|
|
|
|
return matches
|