GenReport.py

Go to the documentation of this file.

import ssl
import os
import csv
import re
import pyshorteners
import sys, traceback
from datetime import datetime
from pymongo import MongoClient
from dotenv import load_dotenv
from botocore.exceptions import ClientError
from SendEmail import send_report
from UploadReports import upload_files_s3
from ReportSnapshot import create_snapshot
f_path = __file__
index = f_path.find("report_generation/")
f_path = f_path[:index]
sys.path.insert(1, f_path+'importer')
from MongoConnection import mongo_connection
load_dotenv()
 
 
mongo_conn = mongo_connection()
 
collection = mongo_conn.mongo_conn_sales()
 
 
ROOT_DIR = os.path.expanduser('~')
 
path = ROOT_DIR+"/Reports/"
if not os.path.exists(path):
    os.mkdir(path)
 
def get_month(date):
    """
    Generate the filename according to date in 'MonthYear' format.
 
    Args:
        date (str): accounting date, i.e processing files date
    Returns: full_date (str) 
    """
    m = re.findall(r'-\d{2}-', date)[0]
    m = m.replace("-", "")
    month_dic = {"01": "Jan", "02": "Feb", "03": "Mar", "04": "Apr", "05": "May", "06": "Jun", "07": "Jul", "08": "Aug", "09": "Sep", "10": "Oct", "11": "Nov", "12": "Dec"}
    month = month_dic[m]
    year = re.findall(r'20\d{2}-', date)[0]
    year = year.replace("-", "")
    full_date = month+year
    return full_date
 
def build_csv(event):
    """
    Generates the compressed csv file according to sales which matches with 'accounting_date' field
 
    Args:
        event (dict): It contains necessary info related to sales and client
    Returns: Nothing.
    """
    date = event["date"]
    bucket = event["bucket"] 
    # path to upload files, each month it's a different path
    s3_path = "{}/monthly_reports/{}/".format(event["path"], date)
    file = get_month(date)
    filename = f"{file}.csv"
    print(f"Creating {filename}")
    date = datetime.strptime(date, "%Y-%m-%d")
    csv_columns = ["quantity", "isrc_id", "release_id", "release_title", "track_title", "artists", "label_id", "territory_code", "type", "service_id", "date", 
                    "accounting_date", "foreign_currency", "total_foreign", "exchange_rate", "currency_period","local_currency", "total_local"]
    try:
        with open(path+filename, 'w', encoding='utf-8') as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=csv_columns)
            writer.writeheader()
            fields = {"_id":0, "quantity": 1, "isrc_id": 1, "release_id": 1, "release_title": 1, "track_title": 1, "artists": 1, "label_id": 1, "territory_code": 1, 
            "type": 1, "service_id": 1, "date": 1, "accounting_date": 1, "foreign_currency": 1, "total_foreign": 1, "exchange_rate": 1, "currency_period":1, "local_currency": 1, "total_local":1}
            template_sales = collection.find({"accounting_date": {"$in": [date]}, "client_id": event["client_id"], "file": {"$in":event["files"]}},fields)
            for post in template_sales:
                writer.writerow(post)
 
        if os.path.isfile(f"{path+filename}.gz"): # Delete file if it exists to avoid duplicate files
            os.system("rm {}.gz".format(path+filename))
        command = f"gzip {path+filename}"
        os.system(command) # Compress file to gzip (client requirement)
        filename = filename+".gz"
        url_to_download, final_path = upload_files_s3(bucket, s3_path, path, filename)
        create_snapshot(event, final_path)
        print(f"{filename} is ready!")
        if url_to_download == None:
            raise Exception("Link not generated")
        # print(send_report(event["client_email"], url_to_download, file))
        os.system("rm -r {}".format(path+filename))
        return url_to_download
    except Exception as e:
        raise Exception("An error occurred: {}".format(e))
        # print(sys.exc_info()[2])
        # print(traceback.format_exc())