Shiji ReviewPro - Getting started in Peliqan

# Required Imports

from datetime import date, datetime
from dateutil.relativedelta import relativedelta
import time
import calendar


# Required helper functions
def get_month_end_date(date):
    month_end_date = calendar.monthrange(date.year, date.month)[1]
    return f"{date.year}-{date.month}-{month_end_date}"

def get_start_date(connection_name):
    connections = pq.list_connections()
    for connection in connections:
        if connection['name'] == connection_name:
            return connection['param1']



# Required Inputs
connection_name = 'Shiji ReviewPro' 
source_schema_name = 'shiji reviewpro'
source_tables = 'available_hotels'
END_DATE = date.today()

# setting start date
START_DATE = get_start_date(connection_name)  # fetching from connect-form

# required data objects
monthly_avg_indexed_ratings, weekly_avg_indexed_ratings, monthly_avg_rating_reviews  = [], [], []


# Required Connections
source_conn = pq.connect(connection_name)
dbconn = pq.dbconnect(pq.DW_NAME)

# fetching all the available Hotels
st.text('Fetching Hotels....')
hotels_list = dbconn.fetch(pq.DW_NAME, source_schema_name, source_tables)


# Fecthing monthly data: 
st.text(f"Syncing Streams from {START_DATE} ....")

for hotel in hotels_list:

    # monthly_avg_indexed_ratings - monthly_avg_rating_reviews 
    date_iter = datetime.strptime(START_DATE, '%Y-%m-%d').date()
    while date_iter <= date.today():   # '=' considering same start_date and current_date | YYYY-MM-DD
 
        body = {
            'pid': hotel['id'],
            'start_date': str(date_iter),
            'end_date': get_month_end_date(date_iter)    
        }
        # Monthly average index rating : 
        index_rating_result = source_conn.get('average_index_rating', body)
        index_rating_result['year'] = date_iter.year
        index_rating_result['month'] = date_iter.strftime("%B").lower()
        index_rating_result[ 'external_id'] = str( index_rating_result['pid'] ) + "_" + str(date_iter.year) + "_" + date_iter.strftime("%B").lower() 
        monthly_avg_indexed_ratings.append(index_rating_result)

        time.sleep(0.5)    # to avoid rate limiting

        # Monthly average review rating
        review_rating_result = source_conn.get('average_review_rating_distribution', body)
        review_rating_result['year'] = date_iter.year
        review_rating_result['month'] = date_iter.strftime("%B").lower() 
        review_rating_result[ 'external_id'] = str(review_rating_result['pid']) + "_" + str(date_iter.year) + "_" +  date_iter.strftime("%B").lower() 
        monthly_avg_rating_reviews.append(review_rating_result)

        date_iter += relativedelta(months=1)          # moving interation cursor
        time.sleep(0.5)

    # weekly_avg_indexed_ratings
    date_iter = datetime.strptime(START_DATE, '%Y-%m-%d').date()
    while date_iter <= date.today():  
        
        body = {
            'pid': hotel['id'],
            'start_date': str(date_iter),
            'end_date': date_iter + relativedelta(weeks=1)   
        }
        weekly_index_result = source_conn.get('average_index_rating', body)
        weekly_index_result['year'] = date_iter.year
        weekly_index_result['week'] = date_iter.strftime("%U")
        weekly_index_result[ 'external_id'] =  str(weekly_index_result['pid']) + "_" + str(date_iter.year) + "_" + date_iter.strftime("%U")
        weekly_avg_indexed_ratings.append(weekly_index_result)

        # moving interation cursor
        date_iter += relativedelta(weeks=1)
        time.sleep(0.5)


# writing data to our datawarehouse

for stream_name, data in {'monthly_avg_indexed_ratings': monthly_avg_indexed_ratings, 'weekly_avg_indexed_ratings': weekly_avg_indexed_ratings, 'monthly_avg_rating_reviews': monthly_avg_rating_reviews}.items():

    write_result = dbconn.write(source_schema_name, stream_name, records = data , pk='external_id')
    if write_result['status'] == 'success' :
        st.text(f"Data Synced for stream {stream_name}! ")
    else:
        st.text(f"Error while syncing stream {stream_name}! ")
        st.json(write_result)


exit(0)