def cf_helper(path, cols, k = 5):
    df = pd.read_csv(path, sep = '\t', quoting = 3, encoding = 'utf-8', header = None, usecols=range(len(cols)))
    if df.shape[0] ==0:
        return pd.DataFrame(columns = cols)
    if df.shape[1] != len(cols):
        print(path)
        print(df.shape)
        return pd.DataFrame(columns = cols)
    df.columns = cols
    df = df.assign(key = lambda x: np.random.randint(0, high=5*k, size=x.shape[0]))
    dfs = [e[1] for e in df.groupby('key')]
    p = mp.Pool(k)
    dfs = p.map(clean_and_filter, dfs)
    p.close()
    p.join()
    return pd.concat(dfs)

def clean_and_filter_parallel(d, k = 7):
    
    indir = d['raw_local_path']
    outdir = d['clean_local_path']
    os.system("rm -rf %s" % outdir)
    os.system("mkdir %s" % outdir)
    cols = d['cols']
    files = []
    for root, dirnames, filenames in os.walk(indir):
        for filename in filenames:
            if '_0' in filename:
                files.append(os.path.join(root, filename))
                
    for i, file in enumerate(files):            
        df = cf_helper(file, cols, k = k) 
        del df['key']
        df.to_csv(os.path.join(outdir, "chunk_%d.tsv" % i), sep = '\t', index = False)
        


for d in datasets:
    print(d['raw_local_path'])
    clean_and_filter_parallel(d)

"""
Explanation: Cleaning and Filtering
End of explanation
"""


query = """
SELECT 
      *
FROM
    enwiki.block_events
"""

block_events_df = query_hive_ssh(query, '../../data/block_events.tsv', priority = True, quoting=3, delete=False)
block_events_df.columns = [c.split('.')[1] for c in block_events_df.columns]

query = """
SELECT 
      *
FROM
    enwiki.blocked_user
"""

blocked_user_df = query_hive_ssh(query, '../../data/blocked_user.tsv', priority = True, quoting=3, delete=False)
blocked_user_df.columns = [c.split('.')[1] for c in blocked_user_df.columns]

"""
Explanation: Download block_events and blocked_users
End of explanation
"""


query = """
SELECT 
      *
FROM
    enwiki.npa_warnings
"""

npa_warnings_df = query_hive_ssh(query, '../../data/npa_warnings.tsv', priority = True, quoting=3, delete=False)
npa_warnings_df.columns = [c.split('.')[1] for c in npa_warnings_df.columns]

"""
Explanation: Download NPA warnings
End of explanation
"""


query = """
SELECT
    user_text,
    COUNT(*) AS num_days
FROM
    (SELECT
        user_text,
        day
    FROM
        (SELECT 
            rev_user_text AS user_text,
            SUBSTR(rev_timestamp,0,8) AS day
        FROM
            enwiki.revision
        WHERE
            rev_user != 0
            AND rev_timestamp <= '2015-01-01'
        ) a
    GROUP BY
        user_text,
        day ) b
GROUP BY
    user_text
HAVING
    COUNT(*) > 7 
"""

long_term_users_df = query_hive_ssh(query, '../../data/long_term_users.tsv', priority = True, quoting=3, delete=False)

"""
Explanation: Download Long term Users
End of explanation
"""


## Annotate users by gender
query = """
SELECT
    user_id,
    user_name as user_text,
    up_value as gender
FROM
    enwiki.user_properties p,
    enwiki.user u
WHERE 
    p.up_user = u.user_id
    AND up_property = 'gender'
"""
d_gender = query_analytics_store(query, {})
d_gender.to_csv('../../data/genders.tsv', sep = '\t', index = False)

"""
Explanation: Download Gender
End of explanation
"""


block_events_df = pd.read_csv('../../data/block_events.tsv', sep = "\t")
block_events_df.columns = [c.split('.')[1] for c in block_events_df.columns]

nss = ['user', 'article']
rel_path = '../../data/samples'

for ns in nss:
    infile = os.path.join(rel_path, ns, 'clean', 'all_blocked_user.tsv')
    out_dir = os.path.join(rel_path, ns, 'clean', 'blocked_user_onion')
    df = pd.read_csv(infile, sep = '\t')
    users = list(set(df['user_text']))
    print(len(users))

    k_prev = 0
    ks = [5, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 150, 200, 250, 300, 500, 1000]
    dfs = {k:[] for k in ks}

    t1 = time.time()
    for i, user in enumerate(users):
        if i % 1000 ==0:
            print (i)
            print(time.time()-t1)
            t1 = time.time()
        df_user = df[df['user_text'] == user].sort_values(by='rev_timestamp')
        if df_user.shape[0] == 0:
            continue

        block_events_df_user = block_events_df[block_events_df['user_text']==user]
        seen_ids = set()


        for i,r in block_events_df_user.iterrows():
            ts = r['timestamp']
            for k in ks:
                df_user_pre = df_user[df_user['rev_timestamp'] <= ts][-k:]

                if df_user_pre.shape[0] > 0:
                    df_user_pre = df_user_pre[df_user_pre['rev_id'].apply(lambda x: x not in seen_ids )]
                    if df_user_pre.shape[0] > 0:
                        seen_ids.update(tuple(df_user_pre['rev_id']))
                        dfs[k].append(df_user_pre)

                df_user_post = df_user[df_user['rev_timestamp'] > ts][:k]
                if df_user_post.shape[0] > 0:
                    df_user_post = df_user_post[df_user_post['rev_id'].apply(lambda x: x not in seen_ids ) ]
                    if df_user_post.shape[0] > 0:
                        seen_ids.update(tuple(df_user_post['rev_id']))
                        dfs[k].append(df_user_post)

    dfs = {k: pd.concat(v) for k,v in dfs.items()}

    sizes = [(k, len(v)) for k,v in dfs.items()]
    sizes.sort(key=lambda x: x[0])
    print(sizes)


    os.system('rm -rf %s' % out_dir)
    os.system('mkdir  %s' % out_dir)

    for k, v in dfs.items():
        v.iloc[np.random.permutation(len(v))].to_csv(out_dir +'/%d.tsv' % k, sep = '\t', index=False)

"""
Explanation: Onionize all_blocked_user
We want to get the k posts before and after each block event for different values of [k1, k2, ..kn]. In order for us to grow k as we please without labeling headaches, we will create a file containing the k_i-1 through k_i posts for each block event that we have not yet labeled.
End of explanation
"""


