Instagram Post Metadata Parser

PostParser.py

import csv
import glob
import json
import os
import re
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime

import numpy as np
import pandas as pd
import polars as pl
from tqdm import tqdm

from CreateSQL import *

# import random
# import shutil

def insta_parser(data, file):

    # extract information from parsed JSON file
    # Poster ID
    try:
        poster_id = data['owner']['username']
    except:
        poster_id = None
    
    # Poster ID Numeric
    try:
        poster_id_num = data['owner']['id']
    except:
        poster_id_num = None
    
    # Post ID (convert ID to JSON format)
    try:
        post_id = data['id'] + '.info'   
    except:
        post_id = None

    # File ID
    try:
        file_id = file
    except:
        file_id = None

    # create datachunk and write it as row
    datachunk = (
        poster_id,
        poster_id_num,
        post_id,
        file_id
    )

    # create keys for each value
    keys = ['PosterID',
            'PosterIDNumeric',
            'PostID',
            'FileID'
            ]
    
    # create dictionary
    dictionary = dict(zip(keys, datachunk))

    return dictionary


def data_merge(path, files, pbar):

    # specify the directory
    zip_path = path

    # create a list to store dict values
    dicts = []
    
    # create JSON parser
    for file in files:

        try:
            with open(zip_path + '/' + file, 'r', encoding = 'utf-8') as f:
                data = json.loads(f.read())

                result = insta_parser(data, file)

                # append a dictionary to a list
                dicts.append(result)

        except ValueError: 
            # there are some empty json files (e.g., academy-1704600271848757741.info)
            tqdm.write(f'{file} does not contain data. Skip this file.')

            pass

        # Update progress
        pbar.update(1)

    return dicts



def main(path, order):

    # specify the directory
    print('path finding...')

    zip_path = path

    # list all files to parse
    print(f'scanning files {order}...')

    for root, dirs, files in os.walk(zip_path): # os.walk is faster than os.listdir by using scandir
        files = files

    # files = files[0:10000] # test with 10000 files

    # too many files might cause a memory issue : split it into a half and run seperately

    # count # of all files to be processed
    tqdm.write(f'All files to be processed: {len(files)} \n Processing {order} half: {round(len(files)/2)}')

    if order == 1: # run first half if order == 1
        files = files[:round(len(files)/2)]

    elif order == 2: # run the last if order == 2
        files = files[round(len(files)/2):]


    # Create dataframe to save the results
    final_result = []
    json_dumps = []

    # Determine chunksize
    n_workers = 100
    chunksize = round(len(files) / n_workers)

    print(f'chunk size: {chunksize}')

    # Multithread using chunks
    with tqdm(total = len(files), colour = 'GREEN') as pbar:
        with ThreadPoolExecutor(n_workers) as exe:

            # Split the copy operations into chunks
            for i in range(0, len(files), chunksize):

                # select a chunk of filenames
                filenames = files[i:(i + chunksize)]

                # submit the batch copy task
                _ = exe.submit(data_merge, path, filenames, pbar)

                # Append results from seperate thread into a single list file
                final_result.append(_)
            
    # # Concat results from final_result
    for f in as_completed(final_result):
        json_dumps = json_dumps + f.result()
    
    tqdm.write(f'Summary: \n # of JSON files: {len(files)}')
    tqdm.write(f' # of JSON files w/ data: {len(json_dumps)}')
               
    return json_dumps


if __name__ == '__main__':

    tqdm.write('start processing...')

    # Run main function to get dataframe
    path = r'D:\Influencer dataset\Post metadata\info'

    # path = r'C:\Users\js223\OneDrive\Desktop\Brokerage\Insta_Analysis_Ver2\codes\Python\social_capital\info'

    # which one will be processed? first half?
    for i in range(1, 3):
        order = i

        json_result = main(path, order)

        # Save JSON dump
        ## Create connection to MySQL
        
        tqdm.write('save JSON.')

        with open(f'JSON_dump{i}.json', 'w', encoding = 'utf-8') as fout:
            json.dump(json_result, fout)

        # Save to MySQL DB - Instagram
        ## Create connection to MySQL

        df_fin = pd.json_normalize(json_result)

        ## Export to MySQL - DB: Instagram - TABLE: influencers
        connection = create_engine(db = 'organic')

        df_fin.to_sql(con = connection,
                    name = f'organicpost{i}_file',
                    if_exists = 'replace',
                    chunksize = 50000,
                    index = False)
        
        del json_result, df_fin
    



#check if it worked properly

# path = 'D:/Influencer brand dataset/json_files/json'
# with open(path + '/' + '1863762656479305326.json', 'r', encoding = 'utf-8') as f:
#     data = json.loads(f.read())
# data['edge_media_to_sponsor_user']

# path = 'D:/Influencer brand dataset/json_files/json'
# with open(path + '/' + '1863762563540801088.json', 'r', encoding = 'utf-8') as f:
#     data = json.loads(f.read())
# data['edge_media_to_sponsor_user']