import os
import glob
import psycopg2
import pandas as pd
from sql_queries import *


def process_song_file(cur, filepath):
    """
    Description: This function processes song files. It reads the file in the filepath (data/song_data) to get information about
    songs and artists and populates songs and artists dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
   # open song file
    df = pd.read_json(filepath, lines=True)
    songcols = df[['song_id', 'title', 'artist_id', 'year', 'duration']]
    song = songcols.values[0]
    
    # insert song record
    song_data = song.tolist()
    cur.execute(song_table_insert, song_data)
    
    # insert artist record
    artistcols = df[['artist_id', 'artist_name', 'artist_location', 'artist_latitude', 'artist_longitude']]
    artist = artistcols.values[0]
    artist_data = artist.tolist()
    cur.execute(artist_table_insert, artist_data)


def process_log_file(cur, filepath):
    """
    Description: This function processes log files. It reads the file in the filepath (data/log_data) to get information about
    time and users and populates time and users dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    # open log file
    df = pd.read_json(filepath, lines=True)

    # filter by NextSong action
    df = df.loc[df['page'] == 'NextSong']

    # convert timestamp column to datetime
    df['t'] = pd.to_datetime(df['ts'], unit='ms')
    
    # insert time data records
    df['timestamp'] = df['ts']
    df['hour'] = pd.to_datetime(df['t']).dt.hour
    df['day'] = pd.to_datetime(df['t']).dt.day
    df['week of a year'] = pd.to_datetime(df['t']).dt.week
    df['month'] = pd.to_datetime(df['t']).dt.month
    df['year'] = pd.to_datetime(df['t']).dt.year
    df['weekday'] = pd.to_datetime(df['t']).dt.day_name()
    time_df = df[[ 'timestamp', 'hour', 'day', 'week of a year', 'month', 'year', 'weekday']] 

    for i, row in time_df.iterrows():
        cur.execute(time_table_insert, list(row))

    # load user table
    user_df = df[['userId', 'firstName', 'lastName', 'gender', 'level']]

    # insert user records
    for i, row in user_df.iterrows():
        cur.execute(user_table_insert, row)

    # insert songplay records
    for index, row in df.iterrows():
        
        # get songid and artistid from song and artist tables
        cur.execute(song_select, (row.song, row.artist, row.length))
        results = cur.fetchone()
        
        if results:
            songid, artistid = results
        else:
            songid, artistid = None, None

        # insert songplay record
        songplay_data = (row.ts, row.userId, row.level, songid, artistid, row.sessionId, row.location, row.userAgent)
        cur.execute(songplay_table_insert, songplay_data)


def process_data(cur, conn, filepath, func):
    """
    Description: This function processes all the data. It walks through the filepath and gets all the files. It prints all the files
    found. Once the files are found, the function iterates over these files and processes them. In the end, number of processed files
    is printed.

    Arguments:
        cur: the cursor object 
        conn: connection to the database
        filepath: song data file path
        func: function
    """

    # get all files matching extension from directory
    all_files = []
    for root, dirs, files in os.walk(filepath):
        files = glob.glob(os.path.join(root,'*.json'))
        for f in files :
            all_files.append(os.path.abspath(f))

    # get total number of files found
    num_files = len(all_files)
    print('{} files found in {}'.format(num_files, filepath))

    # iterate over files and process
    for i, datafile in enumerate(all_files, 1):
        func(cur, datafile)
        conn.commit()
        print('{}/{} files processed.'.format(i, num_files))


def main():
    """
    Description: This function creates a connection to a database sparkifydb and creates the cursor object. It processes all song files and all log files. 

    Once all files are processed, the connection is closed. 
    """
    conn = psycopg2.connect("host=127.0.0.1 dbname=sparkifydb user=student password=student")
    cur = conn.cursor()

    process_data(cur, conn, filepath='data/song_data', func=process_song_file)
    process_data(cur, conn, filepath='data/log_data', func=process_log_file)

    conn.close()


if __name__ == "__main__":
    main()
da517ec90ddf8e9f71899590793f668bcf26d2fc
<mask token>


def process_song_file(cur, filepath):
    """
    Description: This function processes song files. It reads the file in the filepath (data/song_data) to get information about
    songs and artists and populates songs and artists dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    df = pd.read_json(filepath, lines=True)
    songcols = df[['song_id', 'title', 'artist_id', 'year', 'duration']]
    song = songcols.values[0]
    song_data = song.tolist()
    cur.execute(song_table_insert, song_data)
    artistcols = df[['artist_id', 'artist_name', 'artist_location',
        'artist_latitude', 'artist_longitude']]
    artist = artistcols.values[0]
    artist_data = artist.tolist()
    cur.execute(artist_table_insert, artist_data)


def process_log_file(cur, filepath):
    """
    Description: This function processes log files. It reads the file in the filepath (data/log_data) to get information about
    time and users and populates time and users dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    df = pd.read_json(filepath, lines=True)
    df = df.loc[df['page'] == 'NextSong']
    df['t'] = pd.to_datetime(df['ts'], unit='ms')
    df['timestamp'] = df['ts']
    df['hour'] = pd.to_datetime(df['t']).dt.hour
    df['day'] = pd.to_datetime(df['t']).dt.day
    df['week of a year'] = pd.to_datetime(df['t']).dt.week
    df['month'] = pd.to_datetime(df['t']).dt.month
    df['year'] = pd.to_datetime(df['t']).dt.year
    df['weekday'] = pd.to_datetime(df['t']).dt.day_name()
    time_df = df[['timestamp', 'hour', 'day', 'week of a year', 'month',
        'year', 'weekday']]
    for i, row in time_df.iterrows():
        cur.execute(time_table_insert, list(row))
    user_df = df[['userId', 'firstName', 'lastName', 'gender', 'level']]
    for i, row in user_df.iterrows():
        cur.execute(user_table_insert, row)
    for index, row in df.iterrows():
        cur.execute(song_select, (row.song, row.artist, row.length))
        results = cur.fetchone()
        if results:
            songid, artistid = results
        else:
            songid, artistid = None, None
        songplay_data = (row.ts, row.userId, row.level, songid, artistid,
            row.sessionId, row.location, row.userAgent)
        cur.execute(songplay_table_insert, songplay_data)


<mask token>


def main():
    """
    Description: This function creates a connection to a database sparkifydb and creates the cursor object. It processes all song files and all log files. 

    Once all files are processed, the connection is closed. 
    """
    conn = psycopg2.connect(
        'host=127.0.0.1 dbname=sparkifydb user=student password=student')
    cur = conn.cursor()
    process_data(cur, conn, filepath='data/song_data', func=process_song_file)
    process_data(cur, conn, filepath='data/log_data', func=process_log_file)
    conn.close()


<mask token>

<mask token>


def process_song_file(cur, filepath):
    """
    Description: This function processes song files. It reads the file in the filepath (data/song_data) to get information about
    songs and artists and populates songs and artists dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    df = pd.read_json(filepath, lines=True)
    songcols = df[['song_id', 'title', 'artist_id', 'year', 'duration']]
    song = songcols.values[0]
    song_data = song.tolist()
    cur.execute(song_table_insert, song_data)
    artistcols = df[['artist_id', 'artist_name', 'artist_location',
        'artist_latitude', 'artist_longitude']]
    artist = artistcols.values[0]
    artist_data = artist.tolist()
    cur.execute(artist_table_insert, artist_data)


def process_log_file(cur, filepath):
    """
    Description: This function processes log files. It reads the file in the filepath (data/log_data) to get information about
    time and users and populates time and users dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    df = pd.read_json(filepath, lines=True)
    df = df.loc[df['page'] == 'NextSong']
    df['t'] = pd.to_datetime(df['ts'], unit='ms')
    df['timestamp'] = df['ts']
    df['hour'] = pd.to_datetime(df['t']).dt.hour
    df['day'] = pd.to_datetime(df['t']).dt.day
    df['week of a year'] = pd.to_datetime(df['t']).dt.week
    df['month'] = pd.to_datetime(df['t']).dt.month
    df['year'] = pd.to_datetime(df['t']).dt.year
    df['weekday'] = pd.to_datetime(df['t']).dt.day_name()
    time_df = df[['timestamp', 'hour', 'day', 'week of a year', 'month',
        'year', 'weekday']]
    for i, row in time_df.iterrows():
        cur.execute(time_table_insert, list(row))
    user_df = df[['userId', 'firstName', 'lastName', 'gender', 'level']]
    for i, row in user_df.iterrows():
        cur.execute(user_table_insert, row)
    for index, row in df.iterrows():
        cur.execute(song_select, (row.song, row.artist, row.length))
        results = cur.fetchone()
        if results:
            songid, artistid = results
        else:
            songid, artistid = None, None
        songplay_data = (row.ts, row.userId, row.level, songid, artistid,
            row.sessionId, row.location, row.userAgent)
        cur.execute(songplay_table_insert, songplay_data)


def process_data(cur, conn, filepath, func):
    """
    Description: This function processes all the data. It walks through the filepath and gets all the files. It prints all the files
    found. Once the files are found, the function iterates over these files and processes them. In the end, number of processed files
    is printed.

    Arguments:
        cur: the cursor object 
        conn: connection to the database
        filepath: song data file path
        func: function
    """
    all_files = []
    for root, dirs, files in os.walk(filepath):
        files = glob.glob(os.path.join(root, '*.json'))
        for f in files:
            all_files.append(os.path.abspath(f))
    num_files = len(all_files)
    print('{} files found in {}'.format(num_files, filepath))
    for i, datafile in enumerate(all_files, 1):
        func(cur, datafile)
        conn.commit()
        print('{}/{} files processed.'.format(i, num_files))


def main():
    """
    Description: This function creates a connection to a database sparkifydb and creates the cursor object. It processes all song files and all log files. 

    Once all files are processed, the connection is closed. 
    """
    conn = psycopg2.connect(
        'host=127.0.0.1 dbname=sparkifydb user=student password=student')
    cur = conn.cursor()
    process_data(cur, conn, filepath='data/song_data', func=process_song_file)
    process_data(cur, conn, filepath='data/log_data', func=process_log_file)
    conn.close()


<mask token>

<mask token>


def process_song_file(cur, filepath):
    """
    Description: This function processes song files. It reads the file in the filepath (data/song_data) to get information about
    songs and artists and populates songs and artists dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    df = pd.read_json(filepath, lines=True)
    songcols = df[['song_id', 'title', 'artist_id', 'year', 'duration']]
    song = songcols.values[0]
    song_data = song.tolist()
    cur.execute(song_table_insert, song_data)
    artistcols = df[['artist_id', 'artist_name', 'artist_location',
        'artist_latitude', 'artist_longitude']]
    artist = artistcols.values[0]
    artist_data = artist.tolist()
    cur.execute(artist_table_insert, artist_data)


def process_log_file(cur, filepath):
    """
    Description: This function processes log files. It reads the file in the filepath (data/log_data) to get information about
    time and users and populates time and users dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    df = pd.read_json(filepath, lines=True)
    df = df.loc[df['page'] == 'NextSong']
    df['t'] = pd.to_datetime(df['ts'], unit='ms')
    df['timestamp'] = df['ts']
    df['hour'] = pd.to_datetime(df['t']).dt.hour
    df['day'] = pd.to_datetime(df['t']).dt.day
    df['week of a year'] = pd.to_datetime(df['t']).dt.week
    df['month'] = pd.to_datetime(df['t']).dt.month
    df['year'] = pd.to_datetime(df['t']).dt.year
    df['weekday'] = pd.to_datetime(df['t']).dt.day_name()
    time_df = df[['timestamp', 'hour', 'day', 'week of a year', 'month',
        'year', 'weekday']]
    for i, row in time_df.iterrows():
        cur.execute(time_table_insert, list(row))
    user_df = df[['userId', 'firstName', 'lastName', 'gender', 'level']]
    for i, row in user_df.iterrows():
        cur.execute(user_table_insert, row)
    for index, row in df.iterrows():
        cur.execute(song_select, (row.song, row.artist, row.length))
        results = cur.fetchone()
        if results:
            songid, artistid = results
        else:
            songid, artistid = None, None
        songplay_data = (row.ts, row.userId, row.level, songid, artistid,
            row.sessionId, row.location, row.userAgent)
        cur.execute(songplay_table_insert, songplay_data)


def process_data(cur, conn, filepath, func):
    """
    Description: This function processes all the data. It walks through the filepath and gets all the files. It prints all the files
    found. Once the files are found, the function iterates over these files and processes them. In the end, number of processed files
    is printed.

    Arguments:
        cur: the cursor object 
        conn: connection to the database
        filepath: song data file path
        func: function
    """
    all_files = []
    for root, dirs, files in os.walk(filepath):
        files = glob.glob(os.path.join(root, '*.json'))
        for f in files:
            all_files.append(os.path.abspath(f))
    num_files = len(all_files)
    print('{} files found in {}'.format(num_files, filepath))
    for i, datafile in enumerate(all_files, 1):
        func(cur, datafile)
        conn.commit()
        print('{}/{} files processed.'.format(i, num_files))


def main():
    """
    Description: This function creates a connection to a database sparkifydb and creates the cursor object. It processes all song files and all log files. 

    Once all files are processed, the connection is closed. 
    """
    conn = psycopg2.connect(
        'host=127.0.0.1 dbname=sparkifydb user=student password=student')
    cur = conn.cursor()
    process_data(cur, conn, filepath='data/song_data', func=process_song_file)
    process_data(cur, conn, filepath='data/log_data', func=process_log_file)
    conn.close()


if __name__ == '__main__':
    main()

import os
import glob
import psycopg2
import pandas as pd
from sql_queries import *


def process_song_file(cur, filepath):
    """
    Description: This function processes song files. It reads the file in the filepath (data/song_data) to get information about
    songs and artists and populates songs and artists dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    df = pd.read_json(filepath, lines=True)
    songcols = df[['song_id', 'title', 'artist_id', 'year', 'duration']]
    song = songcols.values[0]
    song_data = song.tolist()
    cur.execute(song_table_insert, song_data)
    artistcols = df[['artist_id', 'artist_name', 'artist_location',
        'artist_latitude', 'artist_longitude']]
    artist = artistcols.values[0]
    artist_data = artist.tolist()
    cur.execute(artist_table_insert, artist_data)


def process_log_file(cur, filepath):
    """
    Description: This function processes log files. It reads the file in the filepath (data/log_data) to get information about
    time and users and populates time and users dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    df = pd.read_json(filepath, lines=True)
    df = df.loc[df['page'] == 'NextSong']
    df['t'] = pd.to_datetime(df['ts'], unit='ms')
    df['timestamp'] = df['ts']
    df['hour'] = pd.to_datetime(df['t']).dt.hour
    df['day'] = pd.to_datetime(df['t']).dt.day
    df['week of a year'] = pd.to_datetime(df['t']).dt.week
    df['month'] = pd.to_datetime(df['t']).dt.month
    df['year'] = pd.to_datetime(df['t']).dt.year
    df['weekday'] = pd.to_datetime(df['t']).dt.day_name()
    time_df = df[['timestamp', 'hour', 'day', 'week of a year', 'month',
        'year', 'weekday']]
    for i, row in time_df.iterrows():
        cur.execute(time_table_insert, list(row))
    user_df = df[['userId', 'firstName', 'lastName', 'gender', 'level']]
    for i, row in user_df.iterrows():
        cur.execute(user_table_insert, row)
    for index, row in df.iterrows():
        cur.execute(song_select, (row.song, row.artist, row.length))
        results = cur.fetchone()
        if results:
            songid, artistid = results
        else:
            songid, artistid = None, None
        songplay_data = (row.ts, row.userId, row.level, songid, artistid,
            row.sessionId, row.location, row.userAgent)
        cur.execute(songplay_table_insert, songplay_data)


def process_data(cur, conn, filepath, func):
    """
    Description: This function processes all the data. It walks through the filepath and gets all the files. It prints all the files
    found. Once the files are found, the function iterates over these files and processes them. In the end, number of processed files
    is printed.

    Arguments:
        cur: the cursor object 
        conn: connection to the database
        filepath: song data file path
        func: function
    """
    all_files = []
    for root, dirs, files in os.walk(filepath):
        files = glob.glob(os.path.join(root, '*.json'))
        for f in files:
            all_files.append(os.path.abspath(f))
    num_files = len(all_files)
    print('{} files found in {}'.format(num_files, filepath))
    for i, datafile in enumerate(all_files, 1):
        func(cur, datafile)
        conn.commit()
        print('{}/{} files processed.'.format(i, num_files))


def main():
    """
    Description: This function creates a connection to a database sparkifydb and creates the cursor object. It processes all song files and all log files. 

    Once all files are processed, the connection is closed. 
    """
    conn = psycopg2.connect(
        'host=127.0.0.1 dbname=sparkifydb user=student password=student')
    cur = conn.cursor()
    process_data(cur, conn, filepath='data/song_data', func=process_song_file)
    process_data(cur, conn, filepath='data/log_data', func=process_log_file)
    conn.close()


if __name__ == '__main__':
    main()

import os
import glob
import psycopg2
import pandas as pd
from sql_queries import *


def process_song_file(cur, filepath):
    """
    Description: This function processes song files. It reads the file in the filepath (data/song_data) to get information about
    songs and artists and populates songs and artists dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
   # open song file
    df = pd.read_json(filepath, lines=True)
    songcols = df[['song_id', 'title', 'artist_id', 'year', 'duration']]
    song = songcols.values[0]
    
    # insert song record
    song_data = song.tolist()
    cur.execute(song_table_insert, song_data)
    
    # insert artist record
    artistcols = df[['artist_id', 'artist_name', 'artist_location', 'artist_latitude', 'artist_longitude']]
    artist = artistcols.values[0]
    artist_data = artist.tolist()
    cur.execute(artist_table_insert, artist_data)


def process_log_file(cur, filepath):
    """
    Description: This function processes log files. It reads the file in the filepath (data/log_data) to get information about
    time and users and populates time and users dimension tables. 

    Arguments:
        cur: the cursor object. 
        filepath: song data file path. 

    Returns:
        None
    """
    # open log file
    df = pd.read_json(filepath, lines=True)

    # filter by NextSong action
    df = df.loc[df['page'] == 'NextSong']

    # convert timestamp column to datetime
    df['t'] = pd.to_datetime(df['ts'], unit='ms')
    
    # insert time data records
    df['timestamp'] = df['ts']
    df['hour'] = pd.to_datetime(df['t']).dt.hour
    df['day'] = pd.to_datetime(df['t']).dt.day
    df['week of a year'] = pd.to_datetime(df['t']).dt.week
    df['month'] = pd.to_datetime(df['t']).dt.month
    df['year'] = pd.to_datetime(df['t']).dt.year
    df['weekday'] = pd.to_datetime(df['t']).dt.day_name()
    time_df = df[[ 'timestamp', 'hour', 'day', 'week of a year', 'month', 'year', 'weekday']] 

    for i, row in time_df.iterrows():
        cur.execute(time_table_insert, list(row))

    # load user table
    user_df = df[['userId', 'firstName', 'lastName', 'gender', 'level']]

    # insert user records
    for i, row in user_df.iterrows():
        cur.execute(user_table_insert, row)

    # insert songplay records
    for index, row in df.iterrows():
        
        # get songid and artistid from song and artist tables
        cur.execute(song_select, (row.song, row.artist, row.length))
        results = cur.fetchone()
        
        if results:
            songid, artistid = results
        else:
            songid, artistid = None, None

        # insert songplay record
        songplay_data = (row.ts, row.userId, row.level, songid, artistid, row.sessionId, row.location, row.userAgent)
        cur.execute(songplay_table_insert, songplay_data)


def process_data(cur, conn, filepath, func):
    """
    Description: This function processes all the data. It walks through the filepath and gets all the files. It prints all the files
    found. Once the files are found, the function iterates over these files and processes them. In the end, number of processed files
    is printed.

    Arguments:
        cur: the cursor object 
        conn: connection to the database
        filepath: song data file path
        func: function
    """

    # get all files matching extension from directory
    all_files = []
    for root, dirs, files in os.walk(filepath):
        files = glob.glob(os.path.join(root,'*.json'))
        for f in files :
            all_files.append(os.path.abspath(f))

    # get total number of files found
    num_files = len(all_files)
    print('{} files found in {}'.format(num_files, filepath))

    # iterate over files and process
    for i, datafile in enumerate(all_files, 1):
        func(cur, datafile)
        conn.commit()
        print('{}/{} files processed.'.format(i, num_files))


def main():
    """
    Description: This function creates a connection to a database sparkifydb and creates the cursor object. It processes all song files and all log files. 

    Once all files are processed, the connection is closed. 
    """
    conn = psycopg2.connect("host=127.0.0.1 dbname=sparkifydb user=student password=student")
    cur = conn.cursor()

    process_data(cur, conn, filepath='data/song_data', func=process_song_file)
    process_data(cur, conn, filepath='data/log_data', func=process_log_file)

    conn.close()


if __name__ == "__main__":
    main()