""" Modular search builder and query generators for Aether. """ import logging from sqlalchemy import text log = logging.getLogger(__name__) def sql_limit_offset_part(limit: int, offset: int = 0) -> str|bool: """Creates a partial SQL string for LIMIT and OFFSET.""" if limit >= 0 and offset >= 0: log.info(f'Creating partial SQL string for LIMIT and OFFSET. Limit: {limit}; Offset: {offset}') return f'LIMIT {limit} OFFSET {offset}' else: return False def sql_and_like_part(and_like_dict_obj: dict) -> tuple[str, dict]|bool: """Creates a partial SQL string for AND LIKE queries.""" data = {} if and_like_dict_obj and isinstance(and_like_dict_obj, dict): log.info('Creating partial SQL string for additional AND LIKE queries.') clauses = [] for key, value in and_like_dict_obj.items(): clauses.append(f"{key} LIKE :and_like_{key}") data[f'and_like_{key}'] = value return f"AND ({' AND '.join(clauses)})", data return False def sql_or_like_part(or_like_dict_obj: dict) -> tuple[str, dict]|bool: """Creates a partial SQL string for OR LIKE queries.""" data = {} if or_like_dict_obj and isinstance(or_like_dict_obj, dict): log.info('Creating partial SQL string for additional OR LIKE queries.') clauses = [] for key, value in or_like_dict_obj.items(): clauses.append(f"{key} LIKE :or_like_{key}") data[f'or_like_{key}'] = value return f"AND ({' OR '.join(clauses)})", data return False def sql_and_in_dict_li_part(and_in_dict_li_dict_obj: dict) -> tuple[str, dict]|bool: """Creates a partial SQL string for AND IN queries.""" data = {} if and_in_dict_li_dict_obj and isinstance(and_in_dict_li_dict_obj, dict): log.info('Creating partial SQL string for additional AND IN queries.') clauses = [] for key, value in and_in_dict_li_dict_obj.items(): clauses.append(f"{key} IN :and_in_{key}") data[f'and_in_{key}'] = value return f"AND ({' AND '.join(clauses)})", data return False def sql_and_qry_part(and_qry_dict_obj: dict) -> tuple[str, dict]|bool: """Creates a partial SQL string for additional AND queries (equals).""" data = {} if and_qry_dict_obj and isinstance(and_qry_dict_obj, dict): log.info('Creating partial SQL string for additional AND queries.') clauses = [] for key, value in and_qry_dict_obj.items(): clauses.append(f"{key} = :and_{key}") data[f'and_{key}'] = value return f"AND ({' AND '.join(clauses)})", data return False def sql_fulltext_qry_part(fulltext_qry_dict: dict) -> tuple[str, dict]|bool: """Creates a partial SQL string for fulltext search.""" data = {} if fulltext_qry_dict and isinstance(fulltext_qry_dict, dict): log.info('Creating partial SQL string for fulltext search.') clauses = [] for key, value in fulltext_qry_dict.items(): clauses.append(f"MATCH( {key} ) AGAINST( :ft_{key} IN BOOLEAN MODE )") data[f'ft_{key}'] = value return f"AND ({' OR '.join(clauses)})", data return False def sql_enable_part(table_name: str, enabled: str) -> tuple[str, bool|None]|bool: """Handles enabled/disabled status filtering with schema check.""" from app.db_sql import db if not table_name: return False if enabled in ['enabled', 'disabled', 'all']: if enabled == 'all': return '', None try: db.execute(text(f"SELECT enable FROM `{table_name}` LIMIT 0")) except: log.warning(f"Table '{table_name}' missing 'enable' column. Skipping filter.") return '', None val = (enabled == 'enabled') return f"AND `{table_name}`.enable = {str(val).lower()}", val return False def sql_hidden_part(table_name: str, hidden: str) -> tuple[str, bool|None]|bool: """Handles hidden status filtering with schema check.""" from app.db_sql import db if not table_name: return False if hidden in ['hidden', 'not_hidden', 'all']: if hidden == 'all': return '', None try: db.execute(text(f"SELECT hide FROM `{table_name}` LIMIT 0")) except: log.warning(f"Table '{table_name}' missing 'hide' column. Skipping filter.") return '', None if hidden == 'hidden': return f"AND `{table_name}`.hide = true", True return f"AND (`{table_name}`.hide = false OR `{table_name}`.hide IS NULL)", False return False def sql_where_qry_part(qry_dict_li: list) -> tuple[str, dict]|bool: """Standard v2 style WHERE clause builder.""" data = {} if qry_dict_li and isinstance(qry_dict_li, list): log.info('Creating partial SQL string for WHERE queries.') clauses = [] for qry in qry_dict_li: field = qry.get('field') op = qry.get('operator') val = qry.get('value') type_ = qry.get('type', 'AND') or 'AND' if op == 'MATCH': clauses.append(f'{type_} MATCH( {field} ) AGAINST( :{field} IN BOOLEAN MODE )') else: clauses.append(f'{type_} {field} {op} :{field}') data[field] = val return ' '.join(clauses), data return False