from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import Callable
from enum import Enum
import datetime
import calendar
from ppc_robot_lib.sklik import transformation
from ppc_robot_lib.sklik.types import Granularity
[docs]
class Op(Enum):
EQ = 'EQ'
"""Equal."""
NEQ = 'NEQ'
"""Not equal."""
GT = 'GT'
"""Greater than."""
GTE = 'GTE'
"""Greater than or equal."""
LT = 'LT'
"""Less than."""
LTE = 'LTE'
"""Less than or equal."""
CONTAINS = 'CONTAINS'
"""String contains."""
NOT_CONTAINS = 'NOT_CONTAINS'
"""String not contains."""
PHRASE = 'PHRASE'
"""Phrase match: string contains all words in the given order. It may contain another characters between them."""
NEGATIVE_PHRASE = 'NEGATIVE_PHRASE'
"""Negative phrase match."""
STARTS_WITH = 'STARTS_WITH'
"""String starts with."""
NOT_STARTS_WITH = 'NOT_STARTS_WITH'
"""String does not start with."""
ENDS_WITH = 'ENDS_WITH'
"""String ends with."""
NOT_ENDS_WITH = 'NOT_ENDS_WITH'
"""String does not end with."""
class Date:
def __init__(self, date):
if isinstance(date, str):
self.date = transformation.Date.to_internal_value(date)
else:
self.date = date
def __str__(self):
return self.date.strftime('%Y-%m-%d')
[docs]
class DuringClause(ABC):
"""
Abstract class for during clause. Use one of:
* :py:class:`DateRange`,
* :py:class:`LastNDaysRange`,
* :py:class:`LastNWeeksRange`,
* any member of :py:class:`During`.
"""
[docs]
@abstractmethod
def get_date_from(self) -> Date:
"""Gets the start date."""
pass
[docs]
@abstractmethod
def get_date_to(self) -> Date:
"""Gets the end date."""
pass
[docs]
@abstractmethod
def get_title(self) -> str:
"""Get the human readable title of the clause."""
pass
[docs]
class DateRange(DuringClause):
"""
Arbitrary date range.
"""
def __init__(self, start, end):
"""
:param start: Date from. Can be given as string in the ``YYYY-MM-DD`` format, or as a :py:class:`datetime.date`
instance.
:param end: Date to. Can be given as string in the ``YYYY-MM-DD`` format, or as a :py:class:`datetime.date`
instance.
"""
if not isinstance(start, Date):
self.start = Date(start)
else:
self.start = start
if not isinstance(end, Date):
self.end = Date(end)
else:
self.end = end
[docs]
def get_date_from(self) -> Date:
return self.start
[docs]
def get_date_to(self) -> Date:
return self.end
def __eq__(self, other):
if isinstance(other, DateRange):
return self.start == other.start and self.end == other.end
elif isinstance(other, LastRange):
return other.date_range == self
else:
return False
def __hash__(self):
return hash((self.start, self.end))
def __repr__(self):
return f'DateRange(start={str(self.start)}, end={str(self.end)})'
[docs]
def get_title(self):
return f'{str(self.start)} - {str(self.end)}'
class LastRange(DuringClause):
def __init__(self):
self.today = datetime.date.today()
self.date_range = self._create_range()
def get_date_from(self) -> Date:
# Copied for performance reasons.
new_today = datetime.date.today()
if new_today != self.today:
self.today = new_today
self.date_range = self._create_range()
return self.date_range.start
def get_date_to(self) -> Date:
# Copied for performance reasons.
new_today = datetime.date.today()
if new_today != self.today:
self.today = new_today
self.date_range = self._create_range()
return self.date_range.end
def __eq__(self, other):
# Copied for performance reasons.
new_today = datetime.date.today()
if new_today != self.today:
self.today = new_today
self.date_range = self._create_range()
if isinstance(other, LastRange):
return self.date_range == other.date_range
elif isinstance(other, DateRange):
return self.date_range == other
else:
return False
def __hash__(self):
return hash(self.date_range)
@abstractmethod
def _create_range(self) -> DateRange:
pass
def __repr__(self):
# Copied for performance reasons.
new_today = datetime.date.today()
if new_today != self.today:
self.today = new_today
self.date_range = self._create_range()
return f'LastRange(date_range={repr(self.date_range)})'
[docs]
class LastNDaysRange(LastRange):
"""
Last N days range.
"""
def __init__(self, n_days: int, offset: int = 1, title=None):
"""
:param n_days: Number of days to go back from the offset. Will be used to compute start date.
:param offset: Number of days to go back from today. Will be used to compute end date. The default value of
1 means yesterday.
"""
self.n_days = n_days
self.offset = offset
self._title = title
super().__init__()
def _create_range(self) -> DateRange:
yesterday = self.today - datetime.timedelta(days=self.offset)
start_date = self.today - datetime.timedelta(days=self.n_days)
return DateRange(start_date, yesterday)
[docs]
def get_title(self) -> str:
if self._title:
return self._title
else:
return f'{self.n_days} days'
[docs]
class LastNWeeksRange(LastRange):
"""
Last N weeks range. This range will use only full weeks starting from the previous week (always relative
to current date). Weeks begin on monday.
"""
def __init__(self, n_weeks: int, title=None):
"""
:param n_weeks: Number of weeks, shall be 1 or greater.
"""
self.n_weeks = n_weeks
self._title = title
super().__init__()
def _create_range(self) -> DateRange:
sunday = self.today - datetime.timedelta(days=self.today.isoweekday())
start_date = sunday - datetime.timedelta(days=(7 * self.n_weeks) - 1)
return DateRange(start_date, sunday)
[docs]
def get_title(self):
if self._title:
return self._title
else:
return f'{self.n_weeks} weeks'
class LastNMonthsRange(LastRange):
"""
Last N months range. This range will use only full month starting from the previous month (always relative
to current date) and optionally include the current month.
"""
def __init__(self, n_months: int, include_this_month: bool = False):
"""
:param n_months: Number of months, shall be 1 or greater.
:param include_this_month: Include this month?
"""
self.n_months = n_months
self.include_this_month = include_this_month
super().__init__()
def _create_range(self):
today = datetime.date.today()
first_of_this_month = today.replace(day=1)
if self.include_this_month:
end = today
else:
end = first_of_this_month - datetime.timedelta(days=1)
start = self._month_add(first_of_this_month, -self.n_months)
return DateRange(start, end)
@staticmethod
def _month_add(date: datetime.date, delta):
m, y = (date.month + delta) % 12, date.year + (date.month + delta - 1) // 12
if not m:
m = 12
d = min(date.day, calendar.monthrange(y, m)[1])
return date.replace(day=d, month=m, year=y)
def get_title(self):
return f'{self.n_months} months'
class FunctionalRange(LastRange):
def __init__(self, range_fn: Callable[[datetime.date], DateRange], title: str = ''):
self.range_fn = range_fn
self._title = title
super().__init__()
def _create_range(self) -> DateRange:
return self.range_fn(self.today)
def get_title(self) -> str:
return self._title
class ThisMonthToYdayRange(LastRange):
def get_title(self):
return f'1st of Month to Yesterday'
def __repr__(self):
return f'<ThisMonthToYdayRange()[{str(self.date_range)}] at {id(self)}>'
def _create_range(self):
yday = self.today - datetime.timedelta(days=1)
month_start = yday.replace(year=yday.year, month=yday.month, day=1)
return DateRange(month_start, yday)
class LastYearRange(LastRange):
def __init__(self, current_range: LastRange | DateRange, title: str = None):
self._current = current_range
self._title = title
super().__init__()
def get_title(self):
if self._title:
return self._title
else:
return f'{self._current.get_title()} Previous Year'
def __repr__(self):
return f'<PreviousYearRange(current={repr(self._current)})[{str(self.date_range)}] at {id(self)}>'
def _create_range(self):
start = self._current.get_date_from().date
end = self._current.get_date_to().date
if start.month != 2 or start.day != 29:
new_start = start.replace(year=start.year - 1)
else:
new_start = start.replace(year=start.year - 1, day=28)
if end.month != 2 or end.day != 29:
new_end = end.replace(year=end.year - 1)
else:
new_end = end.replace(year=end.year - 1, day=28)
return DateRange(new_start, new_end)
class LastMonthRange(LastRange):
def __init__(self, current_range: LastRange | DateRange, title: str = None):
self._current = current_range
self._title = title
super().__init__()
def get_title(self):
if self._title:
return self._title
else:
return f'{self._current.get_title()} Previous Month'
def __repr__(self):
return f'<PreviousMonthRange(current={repr(self._current)})[{str(self.date_range)}] at {id(self)}>'
def _create_range(self):
start = self._current.get_date_from().date
end = self._current.get_date_to().date
return DateRange(
self.shift_to_prev_month(start),
self.shift_to_prev_month(end),
)
@staticmethod
def shift_to_prev_month(date: datetime.date):
new_year = date.year
new_month = date.month - 1
new_day = date.day
if new_month == 0:
new_year = new_year - 1
new_month = 12
_, month_days = calendar.monthrange(new_year, new_month)
if new_day > month_days:
new_day = month_days
return datetime.date(new_year, new_month, new_day)
def this_week_sun_today(today: datetime.date):
return DateRange(today - datetime.timedelta(days=today.isoweekday()), today)
def this_week_mon_today(today: datetime.date):
return DateRange(today - datetime.timedelta(days=today.isoweekday() - 1), today)
def this_month(today: datetime.date):
return DateRange(today.replace(day=1), today)
def last_business_week(today: datetime.date):
monday = today - datetime.timedelta(days=today.isoweekday() + 6)
friday = monday + datetime.timedelta(days=4)
return DateRange(monday, friday)
def last_week_sun_sat(today: datetime.date):
if today.isoweekday() == 7:
saturday = today - datetime.timedelta(days=1)
else:
saturday = today - datetime.timedelta(days=today.isoweekday() + 1)
sunday = saturday - datetime.timedelta(days=6)
return DateRange(sunday, saturday)
def last_month(today: datetime.date):
prev_month = today.month - 1
if prev_month < 1:
prev_month = 12
year = today.year - 1
else:
year = today.year
_, days = calendar.monthrange(year, prev_month)
return DateRange(datetime.date(year, prev_month, 1), datetime.date(year, prev_month, days))
[docs]
class During:
TODAY = LastNDaysRange(0, offset=0, title='Today')
"""Today."""
YESTERDAY = LastNDaysRange(1, title='Yesterday')
"""Yesterday."""
LAST_7_DAYS = LastNDaysRange(7)
"""Last 7 days, ends with yesterday (always whole days)."""
THIS_WEEK_SUN_TODAY = FunctionalRange(this_week_sun_today, title='This Week (starting Sunday)')
"""This week, begins on sunday (partial)."""
THIS_WEEK_MON_TODAY = FunctionalRange(this_week_mon_today, title='This Week (starting Monday)')
"""This week, begins on monday (partial)."""
LAST_WEEK = LastNWeeksRange(1, title='Last Week')
"""Last week, begins on monday (always whole week)."""
LAST_14_DAYS = LastNDaysRange(14)
"""Last 14 days, ends with yesterday (always 14 whole days)."""
LAST_30_DAYS = LastNDaysRange(30)
"""Last 30 days, ends with yesterday (always 30 whole days)."""
LAST_BUSINESS_WEEK = FunctionalRange(last_business_week, title='Last Business Week')
"""Business days in the last week (always 5 whole days)."""
LAST_WEEK_SUN_SAT = FunctionalRange(last_week_sun_sat, title='Last Week (starting Sunday)')
"""Last week, begins on sunday (always whole week)."""
THIS_MONTH = FunctionalRange(this_month, title='this month')
"""This month up to today (partial)."""
LAST_MONTH = FunctionalRange(last_month, title='last month')
"""Last month (always whole month)."""
LAST_8_14_DAYS = LastNDaysRange(14, 8, title='last 8 - 14 days')
"""Last 8 - 14 days, ends with 8 days before today (always 7 whole days)."""
LAST_90_DAYS = LastNDaysRange(90)
"""Last 90 days, ends with yesterday (always 90 whole days)."""
LAST_180_DAYS = LastNDaysRange(180)
"""Last 180 days, ends with yesterday (always 180 whole days)."""
LAST_365_DAYS = LastNDaysRange(365)
"""Last 365 days, ends with yesterday (always 365 whole days)."""
single_value_type = str | int | float | Date | Enum
value_type = single_value_type | list[single_value_type]
[docs]
class Condition:
"""
Filter condition.
"""
def __init__(self, column: str, op: Op, value: value_type):
"""
:param column: Column name.
:param op: Operator. See Sklik API docs for allowed operators, some fields does not use operator and checks
only for equivalence.
:param value: Value or list of values to match.
"""
self.column = column
self.op = op
self.value = value
def __repr__(self):
return f'Condition(column={repr(self.column)}, op={repr(self.op)}, value={repr(self.value)})'
[docs]
class Query:
"""
Query for report.
"""
def __init__(
self,
select: list[str],
from_report: str,
where: list[Condition] = None,
during: DuringClause = None,
granularity: Granularity = Granularity.TOTAL,
):
"""
:param select: List of columns to select.
:param from_report: Name of the report, see :py:mod:`ppc_robot_lib.sklik.reports` module for available reports.
:param where: Filter conditions.
:param during: Time range for the report.
:param granularity: Granularity.
"""
self._select = select
self._from_report = from_report
self._where = where if where is not None else []
self._during = during if during is not None else During.LAST_30_DAYS
self._granularity = granularity
def get_select(self) -> list[str]:
return self._select
def select(self, select: list[str]) -> Query:
self._select = select
return self
def add_select(self, *select) -> Query:
self._select.extend(select)
return self
def get_from_report(self) -> str:
return self._from_report
def from_report(self, from_report: str) -> Query:
self._from_report = from_report
return self
def get_during(self) -> DuringClause:
return self._during
def during(self, during: DuringClause) -> Query:
self._during = during
return self
def during_range(self, start, end) -> Query:
if not isinstance(start, Date):
start = Date(start)
if not isinstance(end, Date):
end = Date(end)
self._during = DateRange(start, end)
return self
def get_where(self) -> list[Condition]:
return self._where
def and_where(self, column: str, op: Op, value: value_type) -> Query:
self._where.append(Condition(column, op, value))
return self
def add_condition(self, condition: Condition):
self._where.append(condition)
return self
def get_granularity(self):
return self._granularity
def granularity(self, granularity: Granularity) -> Query:
self._granularity = granularity
return self
def __repr__(self) -> str:
return (
f'Query(select={repr(self._select)}, from_report={repr(self._from_report)}, where={repr(self._where)}, '
f'during={repr(self._during)}, granularity={repr(self._granularity)})'
)