import math import os from datetime import datetime, timedelta from datadog_api_client import ApiClient, Configuration from datadog_api_client.v1.api.events_api import EventsApi from datadog_api_client.v1.model.event_create_request import EventCreateRequest import pandas from app.config import QUERIES_PATH from app.logger import log folder_results = 'test_results' def show_diff(column): colors = [] for i in column: if isinstance(i, int) and i !=0: c = "red" else: c = None colors.append(f"background-color: {c}") return colors def show_zero_diff(column): colors = [] for i in column: if isinstance(i, int) and i != 0: c = "red" else: c = None colors.append(f"background-color: {c}") return colors def get_insights_market(market): return [market.upper()] if market not in ['worldwide', '_gl'] else [] def generate_excel_file(data_dict, sub_path, variables): variables = [str(v).split(' ')[0] for v in variables] today = datetime.today().date() timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") sub_path_directory = f'{folder_results}/{today}/{sub_path}' create_directory(sub_path_directory) file_name = f"{sub_path_directory}/{'_'.join([sub_path] + variables)}_{timestamp}.xlsx" change_columns = [s for s in data_dict.keys() if str(s).startswith('change')] df = pandas.DataFrame(data_dict) df.to_excel(file_name, index=False) df.style.apply(show_diff, subset=change_columns).to_excel(file_name, columns=data_dict.keys(), index=False) def get_dates(days=27, use_today=True, end_date=None): if use_today: today = datetime.today() end_date = today - timedelta(days=2) else: if end_date: end_date = datetime.strptime(end_date, '%Y-%m-%d') else: raise ValueError start_date = end_date - timedelta(days) formatted_end_date = end_date.strftime('%Y-%m-%d') formatted_start_date = start_date.strftime('%Y-%m-%d') return {'start_date': formatted_start_date, 'end_date': formatted_end_date} def get_date_range(days=27): dates = get_dates(days) start_date = datetime.strptime(dates['start_date'], '%Y-%m-%d') end_date = datetime.strptime(dates['end_date'], '%Y-%m-%d') date_range = [(start_date + timedelta(days=i)).strftime('%Y-%m-%d') for i in range((end_date - start_date).days + 1)] return date_range def create_directory(path=folder_results): if not os.path.exists(path): os.makedirs(path) def get_friday(friday='last'): """ Returns the date of the last or next Friday. If today is Friday, both 'last' and 'next' return today. Args: friday (str): 'last' for previous/current Friday, 'next' for upcoming Friday. """ today = datetime.today() weekday = today.weekday() # Monday is 0, Friday is 4 if friday == 'last': if weekday == 4: target_date = today else: days_offset = (weekday - 4) % 7 or 7 target_date = today - timedelta(days=days_offset) elif friday == 'next': if weekday == 4: target_date = today else: days_offset = (4 - weekday) % 7 or 7 target_date = today + timedelta(days=days_offset) else: raise ValueError("Invalid option. Use 'last' or 'next'.") return target_date.strftime('%Y-%m-%d') def round_int(number): if number - math.floor(number) < 0.5: number = math.floor(number) else: number = math.ceil(number) return number def read_query(file_name): file_path = os.path.join(QUERIES_PATH, file_name) log.info(f'Reading file: {file_path}') with open(file_path, 'r') as file: return file.read() percent_threshold = 0.1 def calculate_diff_percent(item1, item2): """Percent change between two integers; safe against zeros on either side.""" if item1 > 0: return int((item2 - item1) / item1 * 100) if item1 == item2 == 0: return 0 return int((item1 - item2) / item2 * 100) if item2 != 0 else 0 def assert_close_enough(expected, actual, reference_id, dimension, field): if expected is None: if actual is not None: raise AssertionError( f"Expected null for ID {reference_id} in {dimension} {field}, " f"but got {actual}" ) return if expected == 0: if actual != 0: raise AssertionError( f"Mismatch for ID {reference_id} in {dimension} {field}:\n" f"Expected: 0, Got: {actual}" ) return threshold = percent_threshold / 100 percent_diff = abs(expected - actual) / expected if percent_diff > threshold: raise AssertionError( f"Mismatch for ID {reference_id} in {dimension} {field}:\n" f"Expected: {expected}, Got: {actual}, " f"Difference: {percent_diff:.4%} (Threshold: {percent_threshold}%)" )