Challenge.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import re\n",
    "from sqlalchemy import create_engine \n",
    "import psycopg2\n",
    "from config import db_password\n",
    "import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define a variable for the directory that's holding the data\n",
    "file_dir = \"C:/Users/krumb/Classwork/Movies_ETL\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": "C:\\Users\\krumb\\Anaconda3\\envs\\PythonData\\lib\\site-packages\\IPython\\core\\interactiveshell.py:3051: DtypeWarning: Columns (10) have mixed types. Specify dtype option on import or set low_memory=False.\n  interactivity=interactivity, compiler=compiler, result=result)\n"
    }
   ],
   "source": [
    "kaggle_metadata = pd.read_csv(f\"{file_dir}/Data/movies_metadata.csv\")\n",
    "ratings = pd.read_csv(f\"{file_dir}/Data/ratings.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# To open the file in the directory instead of needing to type out entire directory every time\n",
    "# If I move files, I only need to update the file_dir rather than \n",
    "wiki_data = f\"{file_dir}wikipedia.movies.json\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create pipeline\n",
    "def pipeline(wiki_data, kaggle_metadata, ratings):\n",
    "# Load the JSON file into a list of dictionaries\n",
    "    with open(f\"{file_dir}/Data/wikipedia.movies.json\", mode=\"r\") as file:\n",
    "        wiki_movies_raw = json.load(file)\n",
    "\n",
    "    # Create a fiter expression only for movies with a director and IMDb link\n",
    "    wiki_movies = [movie for movie in wiki_movies_raw\n",
    "                if (\"Director\" in movie or \"Directed by\" in movie) \n",
    "                and \"imdb_link\" in movie\n",
    "                and \"No. of episodes\" not in movie]\n",
    "\n",
    "     # Handle alternative titles by creating an empty dictionary\n",
    "# loop through all alt. title keys\n",
    "# if they exist, remove the key-value pair and add to alt. titles dict.\n",
    "def clean_movie(movie):\n",
    "    movie = dict(movie) # create a non-destructive copy\n",
    "    alt_titles = {}\n",
    "    # combine alternate titles into one list\n",
    "    for key in ['Also known as','Arabic','Cantonese','Chinese','French',\n",
    "                'Hangul','Hebrew','Hepburn','Japanese','Literally',\n",
    "                'Mandarin','McCune–Reischauer','Original title','Polish',\n",
    "                'Revised Romanization','Romanized','Russian',\n",
    "                'Simplified','Traditional','Yiddish']:\n",
    "        if key in movie:\n",
    "            alt_titles[key] = movie[key]\n",
    "            movie.pop(key)\n",
    "    if len(alt_titles) > 0:\n",
    "        movie['alt_titles'] = alt_titles \n",
    "\n",
    "    # merger column names that mean the same thing\n",
    "    def change_column_name(old_name, new_name):\n",
    "        if old_name in movie:\n",
    "            movie[new_name] = movie.pop(old_name)\n",
    "    change_column_name('Adaptation by', 'Writer(s)')\n",
    "    change_column_name('Country of origin', 'Country')\n",
    "    change_column_name('Directed by', 'Director')\n",
    "    change_column_name('Distributed by', 'Distributor')\n",
    "    change_column_name('Edited by', 'Editor(s)')\n",
    "    change_column_name('Length', 'Running time')\n",
    "    change_column_name('Original release', 'Release date')\n",
    "    change_column_name('Music by', 'Composer(s)')\n",
    "    change_column_name('Produced by', 'Producer(s)')\n",
    "    change_column_name('Producer', 'Producer(s)')\n",
    "    change_column_name('Productioncompanies ', 'Production company(s)')\n",
    "    change_column_name('Productioncompany ', 'Production company(s)')\n",
    "    change_column_name('Released', 'Release Date')\n",
    "    change_column_name('Release Date', 'Release date')\n",
    "    change_column_name('Screen story by', 'Writer(s)')\n",
    "    change_column_name('Screenplay by', 'Writer(s)')\n",
    "    change_column_name('Story by', 'Writer(s)')\n",
    "    change_column_name('Theme music composer', 'Composer(s)')\n",
    "    change_column_name('Written by', 'Writer(s)')\n",
    "\n",
    "    return movie\n",
    "\n",
    "    # Make a list of cleanned movies with a list comprehension\n",
    "    clean_movies = [clean_movie(movie) for movie in wiki_movies]\n",
    "    wiki_movies_df = pd.DataFrame(clean_movies)\n",
    "\n",
    "\n",
    "    wiki_movies_df['imdb_id'] = wiki_movies_df['imdb_link'].str.extract(r'(tt\\d{7})')\n",
    "    wiki_movies_df.drop_duplicates(subset='imdb_id', inplace=True) \n",
    "\n",
    "    # How to get rid of columns where 90% of the values are null\n",
    "    wiki_columns_to_keep = [column for column in wiki_movies_df.columns if wiki_movies_df[column].isnull().sum() < len              (wiki_movies_df) * 0.9]\n",
    "    wiki_movies_df = wiki_movies_df[wiki_columns_to_keep]\n",
    "\n",
    "    # BOX OFFICE\n",
    "    #  Box office data should be numeric. Only want to look at rows that has defined data (aka drop missing values) \n",
    "    box_office = wiki_movies_df[\"Box office\"].dropna()\n",
    "\n",
    "    # Can see that some are stored as lists. We want to use a join() function and use a space as our joining character\n",
    "    box_office = box_office.apply(lambda x: \" \".join(x) if type(x) == list else x)\n",
    "\n",
    "    # Create a regular expression to catch all of the box office values\n",
    "    form_one = r\"\\$\\s*\\d+\\.?\\d*\\s*[mb]illi?on\"\n",
    "    form_two = r\"\\$\\s*\\d{1,3}(?:[,\\.]\\d{3})+(?!\\s[mb]illion)\"\n",
    "    box_office = box_office.str.replace(r\"\\$.*[---](?![a-z])\", \"$\", regex=True)\n",
    "\n",
    "    # Need a function to turn the extracted values into a numeric value\n",
    "    def parse_dollars(s):\n",
    "        # if s is not a string, return NaN\n",
    "        if type(s) != str:\n",
    "            return np.nan\n",
    "\n",
    "        # if input is of the form $###.# million\n",
    "        if re.match(r'\\$\\s*\\d+\\.?\\d*\\s*milli?on', s, flags=re.IGNORECASE):\n",
    "\n",
    "            # remove dollar sign and \" million\"\n",
    "            s = re.sub('\\$|\\s|[a-zA-Z]','', s)\n",
    "\n",
    "            # convert to float and multiply by a million\n",
    "            value = float(s) * 10**6\n",
    "\n",
    "            # return value\n",
    "            return value\n",
    "\n",
    "        # if input is of the form $###.# billion\n",
    "        elif re.match(r'\\$\\s*\\d+\\.?\\d*\\s*billi?on', s, flags=re.IGNORECASE):\n",
    "\n",
    "            # remove dollar sign and \" billion\"\n",
    "            s = re.sub('\\$|\\s|[a-zA-Z]','', s)\n",
    "\n",
    "            # convert to float and multiply by a billion\n",
    "            value = float(s) * 10**9\n",
    "\n",
    "            # return value\n",
    "            return value\n",
    "\n",
    "        # if input is of the form $###,###,###\n",
    "        elif re.match(r'\\$\\s*\\d{1,3}(?:[,\\.]\\d{3})+(?!\\s[mb]illion)', s, flags=re.IGNORECASE):\n",
    "\n",
    "            # remove dollar sign and commas\n",
    "            s = re.sub('\\$|,','', s)\n",
    "\n",
    "            # convert to float\n",
    "            value = float(s)\n",
    "\n",
    "            # return value\n",
    "            return value\n",
    "\n",
    "        # otherwise, return NaN\n",
    "        else:\n",
    "            return np.nan\n",
    "\n",
    "        # Now we can parse the box office values to numeric values\n",
    "    wiki_movies_df['box_office'] = box_office.str.extract(f'({form_one}|{form_two})', flags=re.IGNORECASE)[0].apply                 (parse_dollars)\n",
    "    wiki_movies_df.drop('Box office', axis=1, inplace=True)\n",
    "\n",
    "    # BUDGET\n",
    "    budget = wiki_movies_df['Budget'].dropna()\n",
    "    budget = budget.map(lambda x: ' '.join(x) if type(x) == list else x)\n",
    "    budget = budget.str.replace(r'\\$.*[-—–](?![a-z])', '$', regex=True)\n",
    "    matches_form_one = budget.str.contains(form_one, flags=re.IGNORECASE)\n",
    "    matches_form_two = budget.str.contains(form_two, flags=re.IGNORECASE)\n",
    "    budget = budget.str.replace(r'\\[\\d+\\]\\s*', '')\n",
    "    wiki_movies_df['budget'] = budget.str.extract(f'({form_one}|{form_two})', flags=re.IGNORECASE)[0].apply(parse_dollars)\n",
    "    wiki_movies_df.drop('Budget', axis=1, inplace=True)\n",
    "\n",
    "    # RELEASE DATE\n",
    "    release_date = wiki_movies_df['Release date'].dropna().apply(lambda x: ' '.join(x) if type(x) == list else x)\n",
    "    date_form_one = r\"(?:January|February|March|April|May|June|July|August|September|October|November|December)\\s[123]\\d,           \\s\\d{4}\"\n",
    "    date_form_two = r\"\\d{4}.[01]\\d.[123]\\d\"\n",
    "    date_form_three = r\"(?:January|February|March|April|May|June|July|August|September|October|November|December)\\s\\d{4}\"\n",
    "    date_form_four = r\"\\d{4}\"\n",
    "    wiki_movies_df['release_date'] = pd.to_datetime(release_date.str.extract(\n",
    "        f'({date_form_one}|{date_form_two}|{date_form_three}|{date_form_four})')[0], infer_datetime_format=True)\n",
    "\n",
    "    # RUNNING TIME\n",
    "    running_time = wiki_movies_df['Running time'].dropna().apply(lambda x: ' '.join(x) if type(x) == list else x)\n",
    "    running_time_extract = running_time.str.extract(r'(\\d+)\\s*ho?u?r?s?\\s*(\\d*)|(\\d+)\\s*m')\n",
    "    running_time_extract = running_time_extract.apply(lambda col: pd.to_numeric(col, errors='coerce')).fillna(0)\n",
    "    wiki_movies_df['running_time'] = running_time_extract.apply(lambda row: row[0]*60 + row[1] if row[2] == 0 else row[2],          axis=1)\n",
    "    wiki_movies_df.drop('Running time', axis=1, inplace=True)\n",
    "\n",
    "    # Keep rows where adult column is False, and drop adult column\n",
    "    kaggle_metadata = kaggle_metadata[kaggle_metadata['adult'] == 'False'].drop('adult',axis='columns')\n",
    "    kaggle_metadata['video'] == 'True'\n",
    "    # Assign  it back to video\n",
    "    kaggle_metadata['video'] = kaggle_metadata['video'] == 'True'\n",
    "\n",
    "    kaggle_metadata['budget'] = kaggle_metadata['budget'].astype(int)\n",
    "    kaggle_metadata['id'] = pd.to_numeric(kaggle_metadata['id'], errors='raise')\n",
    "    kaggle_metadata['popularity'] = pd.to_numeric(kaggle_metadata['popularity'], errors='raise')\n",
    "    kaggle_metadata['release_date'] = pd.to_datetime(kaggle_metadata['release_date'])\n",
    "\n",
    "    # RATINGS DATA \n",
    "    ratings.info(null_counts=True)\n",
    "    pd.to_datetime(ratings['timestamp'], unit='s')\n",
    "    ratings['timestamp'] = pd.to_datetime(ratings['timestamp'], unit='s')\n",
    "\n",
    "    # MERGING DATA\n",
    "    movies_df = pd.merge(wiki_movies_df, kaggle_metadata, on='imdb_id', suffixes=['_wiki','_kaggle'])\n",
    "\n",
    "# Print list of columns so we can see what is redundant\n",
    "# Competing data:\n",
    "# Wiki                     Movielens                Resolution\n",
    "#--------------------------------------------------------------------------\n",
    "# title_wiki               title_kaggle            Drop wikipedia\n",
    "# running_time             runtime                 Keep kaggle data, but fill in zeros with wikipedia data                \n",
    "# budget_wiki              budget_kaggle           Keep kaggle data, but fill in zeros with wikipedia data\n",
    "# box_office               revenue                 Keep kaggle data, but fill in zeros with wikipedia data\n",
    "# release_date_wiki        release_date_kaggle     Drop Wikipedia\n",
    "# Language                 original_language       Drop wikipedia\n",
    "# Production company(s)    production_companies    Drop Wikipedia\n",
    "\n",
    "    # PUT IT ALL TOGETHER\n",
    "    movies_df.drop(columns=['title_wiki','release_date_wiki','Language','Production company(s)'], inplace=True)\n",
    "\n",
    "    # Make a function that fills in missing data for a column pair, then drops the redundant column\n",
    "    def fill_missing_kaggle_data(df, kaggle_column, wiki_column):\n",
    "        df[kaggle_column] = df.apply(\n",
    "            lambda row: row[wiki_column] if row[kaggle_column] == 0 else row[kaggle_column]\n",
    "            , axis=1)\n",
    "        df.drop(columns=wiki_column, inplace=True)\n",
    "\n",
    "    # Call the new function for the columns we will be filling in that have zeros\n",
    "    fill_missing_kaggle_data(movies_df, 'runtime', 'running_time')\n",
    "    fill_missing_kaggle_data(movies_df, 'budget_kaggle', 'budget_wiki')\n",
    "    fill_missing_kaggle_data(movies_df, 'revenue', 'box_office')\n",
    "\n",
    "    # Reorder the rows\n",
    "    movies_df = movies_df[['imdb_id','id','title_kaggle','original_title','tagline','belongs_to_collection','url',                      'imdb_link','runtime','budget_kaggle','revenue','release_date_kaggle','popularity','vote_average','vote_count',\n",
    "            'genres','original_language','overview','spoken_languages','Country',\n",
    "            'production_companies','production_countries','Distributor',\n",
    "            'Producer(s)','Director','Starring','Cinematography','Editor(s)','Writer(s)','Composer(s)','Based on']]\n",
    "\n",
    "    # Rename columns\n",
    "    movies_df.rename({'id':'kaggle_id',\n",
    "                    'title_kaggle':'title',\n",
    "                    'url':'wikipedia_url',\n",
    "                    'budget_kaggle':'budget',\n",
    "                    'release_date_kaggle':'release_date',\n",
    "                    'Country':'country',\n",
    "                    'Distributor':'distributor',\n",
    "                    'Producer(s)':'producers',\n",
    "                    'Director':'director',\n",
    "                    'Starring':'starring',\n",
    "                    'Cinematography':'cinematography',\n",
    "                    'Editor(s)':'editors',\n",
    "                    'Writer(s)':'writers',\n",
    "                    'Composer(s)':'composers',\n",
    "                    'Based on':'based_on'\n",
    "                    }, axis='columns', inplace=True)\n",
    "\n",
    "    rating_counts = ratings.groupby(['movieId','rating'], as_index=False).count() \\\n",
    "                    .rename({'userId':'count'}, axis=1) \\\n",
    "                    .pivot(index='movieId',columns='rating', values='count')\n",
    "    rating_counts.columns = ['rating_' + str(col) for col in rating_counts.columns]\n",
    "\n",
    "    # Use a left merge\n",
    "    movies_with_ratings_df = pd.merge(movies_df, rating_counts, left_on='kaggle_id', right_index=True, how='left')\n",
    "    movies_with_ratings_df[rating_counts.columns] = movies_with_ratings_df[rating_counts.columns].fillna(0)\n",
    "\n",
    "    # START DATA LOAD\n",
    "    db_string = f\"postgres://postgres:{db_password}@127.0.0.1:5432/movie_data\"\n",
    "    # Create a database engine \n",
    "    engine = create_engine(db_string)\n",
    "    # IMPORT MOVIE DATA\n",
    "    movies_df.to_sql(name='movies', con=engine, if_exists=\"replace\")\n",
    "\n",
    "    # LOAD IN THE RATINGS DATA\n",
    "    # Create variable for number of rows imported\n",
    "    rows_imported = 0\n",
    "    # get the start_time from time.time()\n",
    "    start_time = time.time()\n",
    "\n",
    "    for data in pd.read_csv(f'{file_dir}/Data/ratings.csv', chunksize=100000):\n",
    "        print(f'importing rows {rows_imported} to {rows_imported + len(data)}...', end='')\n",
    "        data.to_sql(name='ratings', con=engine, if_exists='append')\n",
    "        rows_imported += len(data)\n",
    "\n",
    "        # add elapsed time to final print out\n",
    "        print(f'Done. {time.time() - start_time} total seconds elapsed')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Call the pipeline\n",
    "pipeline(wiki_data, kaggle_metadata, ratings)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.9-final"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}