import os
import sys
import shutil
from io import FileIO, StringIO
import httplib2
import datetime
from subprocess import Popen, PIPE
import json

import pandas as pd
from apiclient.http import MediaIoBaseDownload
from apiclient.discovery import build
from apiclient.errors import HttpError
from oauth2client.client import flow_from_clientsecrets
from oauth2client.file import Storage
from oauth2client.tools import argparser, run_flow

sys.path.append("/home/lyin/Documents/datalytics/Modules/Snowflake") #rbox
sys.path.append("/home/lyin/Documents/datalytics/Modules")
from Snowflake import sfq, s3_2_table, create_table
import s3
import projects.youtube_reports_api.config as config

"""
Some notes about Airflow:
ti = kwargs['ti']
gets the task instance -- a uniquely named (per DAG/subDAG), that is the main 
model that DOES STUFF in Airflow.

The task instance stores variables accessed through an xcom_pull, which
is accesses a key value pair indexed within a task id.

All functions can be recycled outside airflow by adding params for each key 
accessed through the xcom pull.

The key value pairs are binded to a task instance when a task has values 
returned or explicitly pushed with a key value pair (see build_paths)

TODO: quality checks in staging table...
Count rows to similar dates?
Make sure same columns?
"""

##############################################################################
########################### GENERAL FUNCTIONS ################################
##############################################################################
def gzip(filename):
    """
    Uses subprocess' Popen to use the PIGZ program to compress the filename.
    The old file.csv is replaced by file.csv.gz
    """
    process = Popen([config.pigz,'--best',filename],
                    stdout=PIPE, stderr=PIPE)
    
    stdout, stderr = process.communicate()

def pull_context(task_instance, key, task_ids):
    """
    Performs an xcom_pull on the task instance that built the context.
    
    To learn more about the Task Instance model:
    https://airflow.incubator.apache.org/code.html?highlight=task%20instance#airflow.models.TaskInstance

    To read more about xcom_pull check out the example:
    https://github.com/apache/incubator-airflow/blob/master/airflow/example_dags/example_xcom.py
    """
    return task_instance.xcom_pull(key=key,task_ids=task_ids)

def clear_stage(content_owner):
    """
    Deletes all files in the staging directory for a Content Owner.
    """
    directory = os.path.join(
        config.STAGE, config.CMS_DICT.get(content_owner,content_owner)
    )

    files_to_clear = os.listdir(directory)
    
    for file in files_to_clear:
        os.remove(os.path.join(directory,file))

        
def local_clear(context):
    """
    After download_report task fails, this deletes a the partially downloaded 
    file.
    
    This function just shows how a failiure function works, and is not 
    necessary.
    """
    local_target = context['ti'].xcom_pull(
        key= 'local_target',
        task_ids= context['task_id'])
    
    print('local download failed on {}'.format(local_target))
    return

##############################################################################
########################### Initiate Context #################################
##############################################################################
def get_job_name(full_name):
    """
    Cleans up a raw job name and returns a string.
    """
    job_name = full_name[:-11].replace('_','').replace('a2','').replace('a1','')
    return config.v1_reports.get(job_name, job_name)

def build_paths(job_id, job_name, content_owner, time_delta=0, **kwargs):
    '''
    String splicing to create the paths for files to land locally, on s3 and 
    in Snowflake.

    The dictionary of these paths are put in the Task Instance as a key-val 
    pair.

    All reports are one week behind the date the script is scheduled.

    Execution date is determined from run_dags_for_yt_api.py
    it's the current date - days_ago.
    '''
    start_date = kwargs['execution_date'] - datetime.timedelta(days=time_delta)
    today = datetime.datetime.now().strftime("%Y-%m-%d")
    year = start_date.strftime("%Y")
    month =  start_date.strftime("%m")
    start_date_str = start_date.strftime("%Y-%m-%d")

    content_owner_name = config.CMS_DICT.get(content_owner,content_owner)

    # normalize names for v1.1 reports
    job_name = get_job_name(job_name)

    print("Building filepaths for {} {} with rel. exec. date {} on {}".format(
        job_name, start_date_str, 
        kwargs['execution_date'].strftime("%Y-%m-%d"), today))

    s3_target = os.path.join(
        config.S3_PATH, job_name, year, month
        ,'R' + start_date_str, content_owner_name, 'DL' + today, 
        job_name + '_R{}_DL{}.csv.gz'.format(start_date_str, today))
    
    local_target = os.path.join(
        config.STAGE, content_owner_name,
        s3_target.split('/')[-1].replace('.gz',''))
    
    snowflake_target = '.'.join([config.PROD_DB, config.PROD_SCHEMA, job_name]) + '_INTERMEDIATE'
        
    responseBody = {
        "s3_target":            s3_target,
        "job_id":               job_id,
        "local_target":         local_target,
        "snowflake_target":     snowflake_target,
        "content_owner":        content_owner,
        "content_owner_name":   content_owner_name,
        "execution_date":       start_date
    }

    for (key, value) in responseBody.items():
        """
        The responseBody gets copied over
        to the Task Instance (ti).
        """
        kwargs['ti'].xcom_push(key=key, value=value)
            
def kick_off(content_owner):
    """
    Reads a json file of job metadata and returns a list of dictionaries 
    filtered by content-owner.

    jobs = [
        {"content_owner":"J8vAyKuNSYBIN_9RIdxggQ","job_name":"asset_cards_2016-02-01",'job_id':'0ac214a3-269f-4e01-aed6-a51b825055e6'},
        {"content_owner":"J8vAyKuNSYBIN_9RIdxggQ","job_name":"asset_estimated_revenue_a1_2016-08-23","job_id":"11521cfd-b885-4555-9092-1362db8a535d"},
        {"content_owner":"n293L2XLLmtg5wlKBao5Uw","job_name":"asset_annotations_2016-02-01","job_id":"ddfd1521-d74c-4ca9-a397-9ca0ab3eb5bf"},
        {"content_owner":"n293L2XLLmtg5wlKBao5Uw","job_name":"content_owner_demographics_a1","job_id":"c62098b7-8bb7-4b70-9ecb-9b32818bc87d"}
    ]
    """
    with open(config.job_meta) as f:
        jobs = json.loads(f.read())

    return [job for job in jobs if job['content_owner'] == content_owner]

##############################################################################
########################### YOUTUBE API CALLS ################################
##############################################################################
def get_authenticated_services(SCOPES):
    """
    Before using the Youtube API, we must be be authenticated.
    The authentication process is as follows:
    1. request access token using your CLIENT_SECRETS_FILE,
       to perfom tasks in YOUTUBE_SCOPES.
    2. store access token in local directory.
    3. check for valid credentials from the access token.
    4. Based on the actions wished to perform, a different youtube object
       is built for reporting or querying.
    """
    # 1
    flow = flow_from_clientsecrets(
        config.CLIENT_SECRETS_FILE,
        scope = " ".join(SCOPES),
        message = config.MISSING_CLIENT_SECRETS_MESSAGE)
 
    # 2
    storage = Storage(config.TOKEN)
    credentials = storage.get()

    # 3
    if credentials is None or credentials.invalid:
        credentials = run_flow(flow, storage, 
                               argparser.parse_args(args=[]))

    http = credentials.authorize(httplib2.Http())
  
    # 4
    youtube_reports = build(config.YOUTUBE_REPORTING_API_SERVICE_NAME, 
                            config.YOUTUBE_REPORTING_API_VERSION,
                            http = http)
    return youtube_reports

##############################################################################
############################# TARGET PRACTICE ################################
##############################################################################
def get_url(content_owner, job_id, start_date):
    """
    Makes an HTTP API call to the YT API.
    Checks if an report exists upstream, and returns a DL url.

    If the report does not exist, it returns an empty response.

    Read more about the function here:
    https://developers.google.com/youtube/reporting/v1/reference/rest/v1/jobs.reports/list
    """
    start_time_search = start_date.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    end_time_search = (start_date
        + datetime.timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    try:
        youtube_reporting = get_authenticated_services(
            config.YOUTUBE_REPORT_SCOPES)
    except Exception as e:
        print("Couldn't get authentication", e)
        return

    target_report = youtube_reporting.jobs().reports().list(
        onBehalfOfContentOwner=content_owner, jobId=job_id,
        startTimeAtOrAfter=start_time_search,
        startTimeBefore=end_time_search).execute()
    if target_report:
        return target_report['reports'][0]['downloadUrl']
    else:
        print("report not available!")

def exists_in_table(table, content_owner, start_date):
    """
    Reads logfile into a Pandas dataframe via log_dataframe.
    List comp through report_date -- checks that the current download's date
    is not in that list.
    returns True is the date is already in the table, else False.
    """
    delivered = False

    wh = config.WH

    query = """
    SELECT * FROM {} 
    WHERE CONTENT_OWNER = '{}' 
    AND DATE = TO_DATE('{}')
    LIMIT 1""".format(
        table,
        config.CMS_DICT.get(content_owner,content_owner),
        start_date.strftime('%Y-%m-%d'))

    print(query)

    df = sfq(query, wh=wh)

    if isinstance(df,pd.DataFrame):
        if not df.empty:
            delivered = True
    
    return delivered

def count_rows(table, start_date, content_owner=False):
    """
    Reads logfile into a Pandas dataframe via log_dataframe.
    List comp through report_date -- checks that the current download's date
    is not in that list.
    returns True is the date is already in the table, else False.
    """
    rows = -1

    wh = config.WH
    if content_owner:
        query = """
        SELECT COUNT(*) FROM {} 
        WHERE CONTENT_OWNER = '{}'
        AND DATE = TO_DATE('{}')""".format(
            table,
            config.CMS_DICT.get(content_owner,content_owner),
            start_date.strftime('%Y-%m-%d'))
    else:
        query = """
        SELECT COUNT(*) FROM {} 
        where DATE = TO_DATE('{}')""".format(
            table,
            start_date.strftime('%Y-%m-%d'))

    print(query)

    df = sfq(query, wh=wh)

    if isinstance(df,pd.DataFrame):
        # if df['COUNT(*)'] != 0:
        rows = df['COUNT(*)'].iloc[0]
    
    return rows

def check_upstream(task_id, **kwargs):
    """
    Checks if the report exists upstream in the YT API.
    For a given date and content owner.
    """
    ti = kwargs['ti']
    job_id = ti.xcom_pull(task_id, key='job_id')
    content_owner = ti.xcom_pull(task_id, key='content_owner')
    start_date = ti.xcom_pull(task_id, key='execution_date')

    exists = False

    if get_url(content_owner, job_id, start_date):
      exists = True

    return exists


def check_downstream(task_id, **kwargs):
    """
    Checks if the report is already in the Snowflake database
    for a given date and content owner.
    """
    ti = kwargs['ti']
    snowflake_target = ti.xcom_pull(task_id, key='snowflake_target')
    content_owner = ti.xcom_pull(task_id, key='content_owner')
    start_date = ti.xcom_pull(task_id, key='execution_date')

    not_exists = True
    if count_rows(snowflake_target, start_date, content_owner) > 1:
        not_exists = False
    
    return not_exists

##############################################################################
########################### MOVING DATA'ROUND ################################
##############################################################################
def download_report(task_id, **kwargs):
    """
    Creates a youtube_reporting token and a media HTTP request.
    Sets the object's uri to the given url
    Creates file object from Filename to write contents of HTTP request.
    Downloads the request chunkwise to file, returns status.
    """

    ti = kwargs['ti']
    start_date = ti.xcom_pull(task_id, key='execution_date')
    job_id = ti.xcom_pull(task_id, key='job_id')
    local_target = ti.xcom_pull(task_id, key='local_target')
    content_owner = ti.xcom_pull(task_id, key='content_owner')
    s3_target = ti.xcom_pull(task_id, key='s3_target')

    print("downloading report for {} on date {} for content owner {}.".format(
        task_id.replace('build_paths_',''),
        start_date.strftime('%Y-%m-%d'),
        content_owner))

    youtube_reporting = get_authenticated_services(
        config.YOUTUBE_REPORT_SCOPES)

    url = get_url(content_owner, job_id, start_date)

    request = youtube_reporting.media().download(resourceName=" ")
    request.uri = url
    
    with FileIO(local_target, mode='wb') as fh:

        downloader = MediaIoBaseDownload(fh, request, chunksize=2e8)

        done = False
        while done is False:
            status, done = downloader.next_chunk(num_retries=10)
        
        # gzip it
        gzip(local_target)
        
        #return local_target
        # added this from upload report 2017-01-04
        local_file = local_target + '.gz'

        resp = s3.disk_2_s3(local_file, s3_target)
        if resp == 'Write Permissions Denied':
            raise ValueError("{} failed to upload to {}".format(
                local_file, s3_target))
        
        os.remove(local_file)

        return resp

def upload_report(task_id, **kwargs):
    """
    Uploads a local file to s3.
    removes the local file from disk.
    """
    ti = kwargs['ti']
    local_target = ti.xcom_pull(task_id, key='local_target')
    s3_target = ti.xcom_pull(task_id, key='s3_target')
    local_file = local_target + '.gz'

    resp = s3.disk_2_s3(local_file, s3_target)
    if resp == 'Write Permissions Denied':
        raise ValueError("{} failed to upload to {}".format(
            local_file, s3_target))
    
    os.remove(local_file)

    return resp

def insert_to_staging(fmt, task_id,  **kwargs):
    """
    Uploads an s3 file to a staging Snowflake table.
    The staging table is generated from the s3 file.

    The stagin table is then inserted into a production table
    if the prod table exists. Otherwise the prod table is generated.
    The prod table has all the same columns with the addition of
    CONTENT_OWNER.
    """
    ti = kwargs['ti']
    s3_target = ti.xcom_pull(task_id, key='s3_target')
    snowflake_target = ti.xcom_pull(task_id, key='snowflake_target')
    content_owner_name = ti.xcom_pull(task_id, key='content_owner_name')

    prod = snowflake_target
    dev = snowflake_target + '_dev_' + content_owner_name
    wh = config.WH

    # set up dev table.
    resp0 = create_table(s3_target, dev, fmt, wh=wh, delim=',')
    resp1 = s3_2_table(s3_target, dev, fmt, wh=wh)
    
    return({"dev":resp0, "ingestion":resp1})

def insert_to_prod(fmt, task_id,  **kwargs):
    """
    Uploads an s3 file to a staging Snowflake table.
    The staging table is generated from the s3 file.

    The stagin table is then inserted into a production table
    if the prod table exists. Otherwise the prod table is generated.
    The prod table has all the same columns with the addition of
    CONTENT_OWNER.
    """
    ti = kwargs['ti']
    s3_target = ti.xcom_pull(task_id, key='s3_target')
    snowflake_target = ti.xcom_pull(task_id, key='snowflake_target')
    content_owner_name = ti.xcom_pull(task_id, key='content_owner_name')
    execution_date = ti.xcom_pull(task_id, key='execution_date')

    prod = snowflake_target
    dev = snowflake_target + '_dev_' + content_owner_name
    wh = config.WH

    create_prod = """CREATE TABLE IF NOT EXISTS {} AS 
        SELECT *,'{}' AS CONTENT_OWNER, TO_DATE('{}') AS PROCESS_DATE FROM {}"""

    insert_prod = """INSERT INTO {} SELECT *,
        '{}' AS CONTENT_OWNER, TO_DATE('{}') AS PROCESS_DATE FROM {}"""

    # create or insert into prod table
    resp2 = sfq(create_prod.format(prod, content_owner_name, dev, 
                execution_date), wh=wh) 
    
    # this might change one day!.
    if resp2 == '{} already exists, statement succeeded.'.format(prod):
        resp3 = "no insert needed."
    else:
        resp3 = sfq(insert_prod.format(prod, content_owner_name, dev, 
                    execution_date), wh=wh)

    sfq("DROP TABLE {}".format(dev),wh=wh)
    
    return({"prod":resp2, "insert":resp3})

def clear_prod(task_id, **kwargs):
    """
    Deletes files from the same content owner and date.
    """
    ti = kwargs['ti']
    # s3_target = ti.xcom_pull(task_id, key='s3_target')
    snowflake_target = ti.xcom_pull(task_id, key='snowflake_target')
    # content_owner = ti.xcom_pull(task_id, key='content_owner')
    start_date = ti.xcom_pull(task_id, key='execution_date')
    content_owner_name = ti.xcom_pull(task_id, key='content_owner_name')
    # content_owner_name = config.CMS_DICT.get(content_owner,content_owner)
    prod = snowflake_target
    wh = config.WH

    delete_prod = """DELETE FROM {} 
    WHERE DATE = TO_DATE('{}') 
    and CONTENT_OWNER = '{}'"""

    # create or insert into prod table
    resp2 = sfq(delete_prod.format(prod, start_date, content_owner_name), wh=wh) 
    
    return({"prod":resp2})

def qc_columns(task_id, **kwargs):
    """
    Make sure the staging table has the same column names as the production table.
    """
    ti = kwargs['ti']
    snowflake_target = ti.xcom_pull(task_id, key='snowflake_target')
    # content_owner = ti.xcom_pull(task_id, key='content_owner')
    content_owner_name = ti.xcom_pull(task_id, key='content_owner_name')
    # content_owner_name = config.CMS_DICT.get(content_owner,content_owner)
    prod = snowflake_target
    dev = snowflake_target + '_dev_' + content_owner_name
    wh = config.WH

    prod_cols = [col for col in sfq("desc table {}".format(prod), wh=wh)['name']\
                 if col != "CONTENT_OWNER"]
    dev_cols = sfq("desc table {}".format(dev), wh=wh)['name']

    if len(set(prod_cols).intersection(dev_cols)) == len(prod_cols):
        return True
    else:
        return False

def qc_rows(task_id, **kwargs):
    """
    Make sure the number of rows sounds right.
    40% deviation is acceptable.
    """
    ti = kwargs['ti']
    start_date = ti.xcom_pull(task_id, key='execution_date')
    content_owner = ti.xcom_pull(task_id, key='content_owner')
    snowflake_target = ti.xcom_pull(task_id, key='snowflake_target')
    content_owner_name = ti.xcom_pull(task_id, key='content_owner_name')
    # content_owner_name = config.CMS_DICT.get(content_owner,content_owner)
    prod = snowflake_target
    dev = snowflake_target + '_dev_' + content_owner_name
    wh = config.WH

    row_count_dev = count_rows(dev, start_date)
    row_count_prod = count_rows(prod, datetime.datetime(2016,11,1), content_owner)

    score = ((row_count_dev - row_count_prod) / row_count_prod) * 100
    print("rows in dev: {}\nrows in prod: {}\nscore: {}".format(
        row_count_dev, row_count_prod, abs(score)))

    if  abs(score) <= 40:
        return True
    else:
        return False