Spaces:
Paused
Paused
File size: 2,708 Bytes
046723b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
import re
import pluggy
from price_parser import Price
from loguru import logger
hookimpl = pluggy.HookimplMarker("changedetectionio_conditions")
@hookimpl
def register_operators():
def starts_with(_, text, prefix):
return text.lower().strip().startswith(str(prefix).strip().lower())
def ends_with(_, text, suffix):
return text.lower().strip().endswith(str(suffix).strip().lower())
def length_min(_, text, strlen):
return len(text) >= int(strlen)
def length_max(_, text, strlen):
return len(text) <= int(strlen)
# ✅ Custom function for case-insensitive regex matching
def contains_regex(_, text, pattern):
"""Returns True if `text` contains `pattern` (case-insensitive regex match)."""
return bool(re.search(pattern, str(text), re.IGNORECASE))
# ✅ Custom function for NOT matching case-insensitive regex
def not_contains_regex(_, text, pattern):
"""Returns True if `text` does NOT contain `pattern` (case-insensitive regex match)."""
return not bool(re.search(pattern, str(text), re.IGNORECASE))
return {
"!contains_regex": not_contains_regex,
"contains_regex": contains_regex,
"ends_with": ends_with,
"length_max": length_max,
"length_min": length_min,
"starts_with": starts_with,
}
@hookimpl
def register_operator_choices():
return [
("starts_with", "Text Starts With"),
("ends_with", "Text Ends With"),
("length_min", "Length minimum"),
("length_max", "Length maximum"),
("contains_regex", "Text Matches Regex"),
("!contains_regex", "Text Does NOT Match Regex"),
]
@hookimpl
def register_field_choices():
return [
("extracted_number", "Extracted number after 'Filters & Triggers'"),
# ("meta_description", "Meta Description"),
# ("meta_keywords", "Meta Keywords"),
("page_filtered_text", "Page text after 'Filters & Triggers'"),
#("page_title", "Page <title>"), # actual page title <title>
]
@hookimpl
def add_data(current_watch_uuid, application_datastruct, ephemeral_data):
res = {}
if 'text' in ephemeral_data:
res['page_filtered_text'] = ephemeral_data['text']
# Better to not wrap this in try/except so that the UI can see any errors
price = Price.fromstring(ephemeral_data.get('text'))
if price and price.amount != None:
# This is slightly misleading, it's extracting a PRICE not a Number..
res['extracted_number'] = float(price.amount)
logger.debug(f"Extracted number result: '{price}' - returning float({res['extracted_number']})")
return res
|