import threading
from collections.abc import Iterable, Iterator
from functools import cached_property
from itertools import chain, zip_longest
from operator import methodcaller
from typing import TYPE_CHECKING, Any, ClassVar, TypeVar, cast
import regex as re
from dateparser.timezone_parser import is_timezone_token
from dateparser.utils import normalize_unicode
if TYPE_CHECKING:
from dateparser.conf import Settings
_T = TypeVar("_T")
_CacheKey = tuple[Any, ...]
PARSER_HARDCODED_TOKENS = [":", ".", " ", "-", "/"]
PARSER_KNOWN_TOKENS = ["am", "pm", "UTC", "GMT", "Z", "early_morning"]
ALWAYS_KEEP_TOKENS = ["+", *PARSER_HARDCODED_TOKENS]
KNOWN_WORD_TOKENS = [
"monday",
"tuesday",
"wednesday",
"thursday",
"friday",
"saturday",
"sunday",
"january",
"february",
"march",
"april",
"may",
"june",
"july",
"august",
"september",
"october",
"november",
"december",
"century",
"decade",
"fortnight",
"year",
"month",
"week",
"day",
"hour",
"minute",
"second",
"ago",
"in",
"am",
"pm",
"early_morning",
"morning",
"afternoon",
"evening",
"night",
]
PARENTHESES_PATTERN = re.compile(r"[\(\)]")
NUMERAL_PATTERN = re.compile(r"(\d+)")
KEEP_TOKEN_PATTERN = re.compile(r"^.*[^\W_].*$", flags=re.U)
_WEEKDAY_TOKENS = KNOWN_WORD_TOKENS[0:7]
_MONTH_TOKENS = KNOWN_WORD_TOKENS[7:19]
[docs]
class UnknownTokenError(Exception):
pass
def _parse_bool(value: object) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.strip().lower() in ("true", "1", "yes")
return False
[docs]
class Dictionary:
"""
Class that modifies and stores translations and handles splitting of date string.
:param locale_info:
Locale info (translation data) of the locale.
:type language_info: dict
:param settings:
Configure customized behavior using settings defined in :mod:`dateparser.conf.Settings`.
:type settings: dict
:return: a Dictionary instance.
"""
_split_regex_cache: ClassVar[dict[_CacheKey, dict[str, re.Pattern[str]]]] = {}
_sorted_words_cache: ClassVar[dict[_CacheKey, dict[str, list[str]]]] = {}
_split_relative_regex_cache: ClassVar[
dict[_CacheKey, dict[str, re.Pattern[str]]]
] = {}
_sorted_relative_strings_cache: ClassVar[dict[_CacheKey, dict[str, list[str]]]] = {}
_match_relative_regex_cache: ClassVar[
dict[_CacheKey, dict[str, re.Pattern[str]]]
] = {}
# The caches above are shared across all Dictionary instances and threads.
# The lock keeps each check-populate-evict-read sequence atomic, so a
# concurrent eviction cannot drop an entry between its check and its read.
_cache_lock = threading.RLock()
def __init__(
self, locale_info: dict[str, Any], settings: "Settings | None" = None
) -> None:
dictionary: dict[str, str | None] = {}
self._settings = cast("Settings", settings)
self.info = locale_info
if "skip" in locale_info:
skip = map(methodcaller("lower"), locale_info["skip"])
dictionary.update(dict.fromkeys(skip))
if "pertain" in locale_info:
pertain = map(methodcaller("lower"), locale_info["pertain"])
dictionary.update(dict.fromkeys(pertain))
meanings: dict[str, list[str]] = {}
for word in KNOWN_WORD_TOKENS:
if word in locale_info:
translations = list(map(methodcaller("lower"), locale_info[word]))
dictionary.update(zip_longest(translations, [], fillvalue=word))
for translation in translations:
meanings.setdefault(translation, []).append(word)
self._meanings = meanings
# Some locales (mostly Romance languages) reuse the same abbreviation for a
# weekday and a month (e.g. Italian "mar" is both "martedì" and "marzo").
# The loop above always resolves those in favor of the month, since months
# are listed after weekdays in KNOWN_WORD_TOKENS. Keep track of the weekday
# each conflicting token could also mean, and of which tokens are
# unambiguously months, so a caller with extra context (e.g. another,
# unambiguous month elsewhere in the same string) can recover the weekday
# reading.
weekday_translations = {}
for word in _WEEKDAY_TOKENS:
if word in locale_info:
for token in map(methodcaller("lower"), locale_info[word]):
weekday_translations[token] = word
self._month_translations: set[str] = set()
for word in _MONTH_TOKENS:
if word in locale_info:
self._month_translations.update(
map(methodcaller("lower"), locale_info[word])
)
self._weekday_month_conflicts = {
token: weekday
for token, weekday in weekday_translations.items()
if token in self._month_translations
}
dictionary.update(zip_longest(ALWAYS_KEEP_TOKENS, ALWAYS_KEEP_TOKENS))
dictionary.update(
zip_longest(
map(methodcaller("lower"), PARSER_KNOWN_TOKENS), PARSER_KNOWN_TOKENS
)
)
relative_type = locale_info.get("relative-type", {})
for key, value in relative_type.items():
relative_translations = map(methodcaller("lower"), value)
dictionary.update(zip_longest(relative_translations, [], fillvalue=key))
self._dictionary = dictionary
no_word_spacing = locale_info.get("no_word_spacing", False)
self._no_word_spacing = _parse_bool(no_word_spacing)
relative_type_regex = locale_info.get("relative-type-regex", {})
self._relative_strings = list(chain.from_iterable(relative_type_regex.values()))
def _is_unambiguous_month(self, token: str) -> bool:
"""
Whether ``token`` (already lowercased) names a month and, unlike the
tokens in :attr:`_weekday_month_conflicts`, cannot also mean a weekday.
"""
return (
token in self._month_translations
and token not in self._weekday_month_conflicts
)
def __contains__(self, key: object) -> bool:
if key in self._settings.SKIP_TOKENS:
return True
return self._dictionary.__contains__(key)
def __getitem__(self, key: str) -> str | None:
if key in self._settings.SKIP_TOKENS:
return None
return self._dictionary.__getitem__(key)
def __iter__(self) -> Iterator[str]:
return chain(self._settings.SKIP_TOKENS, iter(self._dictionary))
def _get_meanings(self, key: str) -> list[str]:
"""Return every meaning of *key*, starting with the one that indexing
returns, if any."""
default = self[key]
if default is None:
return []
others = [m for m in self._meanings.get(key, ()) if m != default]
return [default, *others]
[docs]
def are_tokens_valid(self, tokens: Iterable[str]) -> bool:
"""
Check if tokens are valid tokens for the locale.
:param tokens:
a list of string tokens.
:type tokens: list
:return: True if tokens are valid, False otherwise.
"""
has_only_keep_tokens = not set(tokens) - set(ALWAYS_KEEP_TOKENS)
if has_only_keep_tokens:
return False
match_relative_regex = self._get_match_relative_regex_cache()
return all(
self._is_known_token(token, match_relative_regex) for token in tokens
)
def _is_known_token(
self, token: str, match_relative_regex: re.Pattern[str]
) -> bool:
"""Whether ``token`` is recognised by this locale: a number, a relative
expression, or a dictionary word (the same per-token check that
:meth:`are_tokens_valid` applies)."""
return bool(
token.isdigit() or match_relative_regex.match(token) or token in self
)
def _strip_unknown_edge_tokens(self, tokens: list[str]) -> list[str]:
"""Return ``tokens`` without the leading and trailing tokens that this
locale does not recognise (neither digits, relative expressions, nor
dictionary words), along with any whitespace those removals leave at
the edges. Interior tokens are left untouched. Used when the
``IGNORE_SURROUNDING_TEXT`` setting is enabled.
A timezone abbreviation that is the final token (e.g. a trailing
``"EST"``) is kept even though the locale dictionary does not list it,
so that the offset is applied rather than silently discarded. It
survives translation and is popped later by
:func:`~dateparser.timezone_parser.pop_tz_offset_from_string`. The
timezone is kept only when it is the last token: if further
unrecognized text follows it, the tokenizer merges the two and the
timezone is dropped together with that text, like any other edge noise.
This timezone exception is deliberately made at the trailing edge
only: a timezone conventionally follows the date it qualifies
(``"... 1:21 PM EST"``) and does not precede it, so the leading edge
keeps stripping unrecognized tokens unconditionally."""
match_relative_regex = self._get_match_relative_regex_cache()
def is_extra_text(token: str) -> bool:
return token.isspace() or not self._is_known_token(
token, match_relative_regex
)
start, end = 0, len(tokens)
# Leading edge: drop every unrecognized token. A timezone does not
# precede the date it qualifies, so no timezone exception is made here.
while start < end and is_extra_text(tokens[start]):
start += 1
# Trailing edge: drop unrecognized tokens, but keep a recognized
# timezone abbreviation (e.g. ``" est"``) so its offset is applied.
while (
end > start
and is_extra_text(tokens[end - 1])
and not is_timezone_token(tokens[end - 1])
):
end -= 1
return tokens[start:end]
[docs]
def split(self, string: str, keep_formatting: bool = False) -> list[str]:
"""
Split the date string using translations in locale info.
:param string:
Date string to be splitted.
:type string:
str
:param keep_formatting:
If True, retain formatting of the date string.
:type keep_formatting: bool
:return: A list of string tokens formed after splitting the date string.
"""
if not string:
return []
split_relative_regex = self._get_split_relative_regex_cache()
match_relative_regex = self._get_match_relative_regex_cache()
tokens: list[Any] = split_relative_regex.split(string)
for i, token in enumerate(tokens):
if match_relative_regex.match(token):
tokens[i] = [token]
continue
tokens[i] = self._split_by_known_words(token, keep_formatting)
return list(filter(bool, chain.from_iterable(tokens)))
@cached_property
def _cache_key(self) -> _CacheKey:
# The cached values only depend on the words of the dictionary.
return (type(self), *self._settings.SKIP_TOKENS)
def _add_to_cache(self, value: _T, cache: dict[_CacheKey, dict[str, _T]]) -> None:
with self._cache_lock:
key = self._cache_key
entry = cache.pop(key, {})
entry[self.info["name"]] = value
cache[key] = entry
if (
self._settings.CACHE_SIZE_LIMIT
and len(cache) > self._settings.CACHE_SIZE_LIMIT
):
cache.pop(next(iter(cache)))
def _split_by_known_words(self, string: str, keep_formatting: bool) -> list[str]:
regex = self._get_split_regex_cache()
splitted: list[str] = []
unknown = string
while unknown:
match = regex.match(string)
if not match:
curr_split = (
self._split_by_numerals(string, keep_formatting)
if self._should_capture(string, keep_formatting)
else []
)
unknown = ""
else:
unparsed, known, unknown = match.groups()
curr_split = (
[known] if self._should_capture(known, keep_formatting) else []
)
if unparsed and self._should_capture(unparsed, keep_formatting):
curr_split = (
self._split_by_numerals(unparsed, keep_formatting) + curr_split
)
if unknown:
string = unknown if string != unknown else ""
for token in curr_split:
if (
splitted
and splitted[-1].isdigit()
and token in {"st", "nd", "rd", "th"}
):
continue
splitted.append(token)
return splitted
def _split_by_numerals(self, string: str, keep_formatting: bool) -> list[str]:
return [
token
for token in NUMERAL_PATTERN.split(string)
if self._should_capture(token, keep_formatting)
]
def _should_capture(
self, token: str, keep_formatting: bool
) -> bool | re.Match[str] | None:
return (
keep_formatting
or token in ALWAYS_KEEP_TOKENS
or KEEP_TOKEN_PATTERN.match(token)
)
def _get_sorted_words_from_cache(self) -> list[str]:
with self._cache_lock:
if (
self._cache_key not in self._sorted_words_cache
or self.info["name"] not in self._sorted_words_cache[self._cache_key]
):
self._add_to_cache(
cache=self._sorted_words_cache,
value=sorted(self, key=len, reverse=True),
)
return self._sorted_words_cache[self._cache_key][self.info["name"]]
def _get_split_regex_cache(self) -> re.Pattern[str]:
with self._cache_lock:
if (
self._cache_key not in self._split_regex_cache
or self.info["name"] not in self._split_regex_cache[self._cache_key]
):
self._construct_split_regex()
return self._split_regex_cache[self._cache_key][self.info["name"]]
def _construct_split_regex(self) -> None:
known_words_group = "|".join(
map(re.escape, self._get_sorted_words_from_cache())
)
if self._no_word_spacing:
regex = rf"^(.*?)({known_words_group})(.*)$"
else:
regex = rf"^(.*?(?:\A|\W|_|\d))({known_words_group})((?:\Z|\W|_|\d).*)$"
self._add_to_cache(
cache=self._split_regex_cache,
value=re.compile(regex, re.UNICODE | re.IGNORECASE),
)
def _get_sorted_relative_strings_from_cache(self) -> list[str]:
with self._cache_lock:
if (
self._cache_key not in self._sorted_relative_strings_cache
or self.info["name"]
not in self._sorted_relative_strings_cache[self._cache_key]
):
self._add_to_cache(
cache=self._sorted_relative_strings_cache,
value=sorted(
[
PARENTHESES_PATTERN.sub("", key)
for key in self._relative_strings
],
key=len,
reverse=True,
),
)
return self._sorted_relative_strings_cache[self._cache_key][
self.info["name"]
]
def _get_split_relative_regex_cache(self) -> re.Pattern[str]:
with self._cache_lock:
if (
self._cache_key not in self._split_relative_regex_cache
or self.info["name"]
not in self._split_relative_regex_cache[self._cache_key]
):
self._construct_split_relative_regex()
return self._split_relative_regex_cache[self._cache_key][self.info["name"]]
def _construct_split_relative_regex(self) -> None:
known_relative_strings_group = "|".join(
self._get_sorted_relative_strings_from_cache()
)
if self._no_word_spacing:
regex = f"({known_relative_strings_group})"
else:
regex = (
f"(?<=(?:\\A|\\W|_))({known_relative_strings_group})(?=(?:\\Z|\\W|_))"
)
self._add_to_cache(
cache=self._split_relative_regex_cache,
value=re.compile(regex, re.UNICODE | re.IGNORECASE),
)
def _get_match_relative_regex_cache(self) -> re.Pattern[str]:
with self._cache_lock:
if (
self._cache_key not in self._match_relative_regex_cache
or self.info["name"]
not in self._match_relative_regex_cache[self._cache_key]
):
self._construct_match_relative_regex()
return self._match_relative_regex_cache[self._cache_key][self.info["name"]]
def _construct_match_relative_regex(self) -> None:
known_relative_strings_group = "|".join(
self._get_sorted_relative_strings_from_cache()
)
regex = f"^({known_relative_strings_group})$"
self._add_to_cache(
cache=self._match_relative_regex_cache,
value=re.compile(regex, re.UNICODE | re.IGNORECASE),
)
[docs]
class NormalizedDictionary(Dictionary):
def __init__(
self, locale_info: dict[str, Any], settings: "Settings | None" = None
) -> None:
super().__init__(locale_info, settings)
self._normalize()
def _normalize(self) -> None:
new_dict: dict[str, str | None] = {}
conflicting_keys = []
for key, value in self._dictionary.items():
normalized = normalize_unicode(key)
if key != normalized and normalized in self._dictionary:
conflicting_keys.append(key)
else:
new_dict[normalized] = value
for key in conflicting_keys:
normalized = normalize_unicode(key)
if key in (self.info.get("skip", []) + self.info.get("pertain", [])):
new_dict[normalized] = self._dictionary[key]
self._dictionary = new_dict
meanings: dict[str, list[str]] = {}
for key, key_meanings in self._meanings.items():
normalized_meanings = meanings.setdefault(normalize_unicode(key), [])
normalized_meanings.extend(
m for m in key_meanings if m not in normalized_meanings
)
self._meanings = meanings
self._relative_strings = list(map(normalize_unicode, self._relative_strings))