You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
154 lines
4.7 KiB
154 lines
4.7 KiB
"""
|
|
Advanced Search Query Parser for Boolean & Operator-based Search.
|
|
Supports:
|
|
- Exact Phrase Match: "phrase" or «phrase»
|
|
- Mandatory / Required Term: +word or +"phrase"
|
|
- Excluded / Negative Term: -word or -"phrase"
|
|
- Grouped OR Conditions: (word1 word2) or +(word1 word2)
|
|
- Excluded Grouped OR Conditions: -(word1 word2)
|
|
- Compound Expressions: +term1 +term2 +"exact phrase" +(g1 g2) -excluded
|
|
Seamlessly integrates with Arabic & Persian search normalization.
|
|
"""
|
|
|
|
import re
|
|
from typing import List, Optional, Tuple, Union
|
|
from django.db.models import Q
|
|
from utils.text_normalizer import normalize_for_search
|
|
|
|
TOKEN_REGEX = re.compile(
|
|
r'(?P<group>(?P<g_prefix>[+-]?)\s*\((?P<g_body>[^)]+)\))|'
|
|
r'(?P<phrase>(?P<p_prefix>[+-]?)\s*["«](?P<p_body>[^"»]+)["»])|'
|
|
r'(?P<word>(?P<w_prefix>[+-]?)(?P<w_body>[^\s()«»"]+))'
|
|
)
|
|
|
|
|
|
def sanitize_search_query(query: str) -> str:
|
|
"""
|
|
Cleans search query and balances unclosed quotes/parentheses.
|
|
"""
|
|
if not query:
|
|
return ""
|
|
q = query.strip()
|
|
# Balance double quotes
|
|
if q.count('"') % 2 != 0:
|
|
q += '"'
|
|
# Balance Persian/Arabic quotes
|
|
if q.count('«') > q.count('»'):
|
|
q += '»' * (q.count('«') - q.count('»'))
|
|
# Balance parentheses
|
|
if q.count('(') > q.count(')'):
|
|
q += ')' * (q.count('(') - q.count(')'))
|
|
return q
|
|
|
|
|
|
def parse_search_tokens(query: str) -> List[Tuple[str, str, Union[str, List[str]]]]:
|
|
"""
|
|
Parses a query string into structured search tokens:
|
|
- ('PHRASE', '+' | '-', phrase_text)
|
|
- ('WORD', '+' | '-', word_text)
|
|
- ('GROUP', '+' | '-', [item1, item2, ...])
|
|
"""
|
|
q = sanitize_search_query(query)
|
|
tokens = []
|
|
|
|
for m in TOKEN_REGEX.finditer(q):
|
|
if m.group('group'):
|
|
prefix = m.group('g_prefix') or '+'
|
|
body = m.group('g_body').strip()
|
|
items = []
|
|
for im in re.finditer(r'["«](?P<phrase>[^"»]+)["»]|(?P<word>[^\s"«»]+)', body):
|
|
txt = im.group('phrase') or im.group('word')
|
|
if txt and txt.strip():
|
|
items.append(txt.strip())
|
|
if items:
|
|
tokens.append(('GROUP', prefix, items))
|
|
elif m.group('phrase'):
|
|
prefix = m.group('p_prefix') or '+'
|
|
body = m.group('p_body').strip()
|
|
if body:
|
|
tokens.append(('PHRASE', prefix, body))
|
|
elif m.group('word'):
|
|
prefix = m.group('w_prefix') or '+'
|
|
body = m.group('w_body').strip()
|
|
if body and body not in ('+', '-'):
|
|
tokens.append(('WORD', prefix, body))
|
|
|
|
return tokens
|
|
|
|
|
|
def _build_single_term_match(norm_term: str, fields: List[str]) -> Q:
|
|
"""
|
|
Builds a Q condition matching a normalized term across the specified fields.
|
|
"""
|
|
term_q = Q()
|
|
for f in fields:
|
|
term_q |= Q(**{f"{f}__icontains": norm_term})
|
|
return term_q
|
|
|
|
|
|
def build_advanced_search_q(query_str: Optional[str], fields: Optional[List[str]] = None) -> Q:
|
|
"""
|
|
Converts a user search query with operators (+, -, "", ()) into a Django Q object.
|
|
|
|
:param query_str: Raw search query from the request
|
|
:param fields: List of model fields to search in (defaults to ['normalized_text'])
|
|
:return: Combined Django Q object
|
|
"""
|
|
if not query_str or not query_str.strip():
|
|
return Q()
|
|
|
|
if not fields:
|
|
fields = ['normalized_text']
|
|
|
|
tokens = parse_search_tokens(query_str)
|
|
if not tokens:
|
|
return Q()
|
|
|
|
positive_q = Q()
|
|
has_positive = False
|
|
|
|
negative_q = Q()
|
|
has_negative = False
|
|
|
|
for token_type, prefix, payload in tokens:
|
|
is_exclude = (prefix == '-')
|
|
|
|
if token_type in ('WORD', 'PHRASE'):
|
|
norm_val = normalize_for_search(payload)
|
|
if not norm_val:
|
|
continue
|
|
|
|
term_q = _build_single_term_match(norm_val, fields)
|
|
if not term_q:
|
|
continue
|
|
|
|
if is_exclude:
|
|
negative_q &= ~term_q
|
|
has_negative = True
|
|
else:
|
|
positive_q &= term_q
|
|
has_positive = True
|
|
|
|
elif token_type == 'GROUP':
|
|
group_q = Q()
|
|
for item in payload:
|
|
norm_it = normalize_for_search(item)
|
|
if norm_it:
|
|
group_q |= _build_single_term_match(norm_it, fields)
|
|
|
|
if group_q:
|
|
if is_exclude:
|
|
negative_q &= ~group_q
|
|
has_negative = True
|
|
else:
|
|
positive_q &= group_q
|
|
has_positive = True
|
|
|
|
if has_positive and has_negative:
|
|
return positive_q & negative_q
|
|
elif has_positive:
|
|
return positive_q
|
|
elif has_negative:
|
|
return negative_q
|
|
|
|
return Q()
|