276 lines
10 KiB
Python
276 lines
10 KiB
Python
"""
|
|
Create pseudonyms map as follows :
|
|
table, field,value,enc,filter
|
|
"""
|
|
import pandas as pd
|
|
import numpy as np
|
|
from google.oauth2 import service_account
|
|
from google.cloud import bigquery as bq
|
|
import json
|
|
import threading
|
|
import sys
|
|
import os
|
|
import itertools
|
|
|
|
DATASET_SUFFIX = '_pseudo'
|
|
PSEUDO_TABLENAME = 'map'
|
|
|
|
SYS_ARGS = {'context':''}
|
|
if len(sys.argv) > 1:
|
|
|
|
N = len(sys.argv)
|
|
for i in range(1,N):
|
|
value = None
|
|
if sys.argv[i].startswith('--'):
|
|
key = sys.argv[i].replace('-','')
|
|
SYS_ARGS[key] = 1
|
|
if i + 1 < N:
|
|
value = sys.argv[i + 1] = sys.argv[i+1].strip()
|
|
if key and value:
|
|
SYS_ARGS[key] = value
|
|
|
|
|
|
i += 2
|
|
|
|
|
|
class void :
|
|
pass
|
|
|
|
class pseudonym :
|
|
@staticmethod
|
|
def meta(**args) :
|
|
"""
|
|
:key Bigquery private key (service account)
|
|
:dataset dataset of the input table
|
|
:table table name
|
|
:filter optional filter (SQL statement)
|
|
"""
|
|
credentials = service_account.Credentials.from_service_account_file(args['key'])
|
|
SQL = ["SELECT * FROM :dataset.:table"]
|
|
# if 'filter' in args :
|
|
# SQL += ['WHERE',args['filter']]
|
|
dataset = args['dataset']
|
|
table = args['table']
|
|
SQL = " ".join(SQL+["LIMIT 1"]).replace(":dataset",dataset).replace(":table",table)
|
|
|
|
df = pd.read_gbq(SQL,credentials=credentials,dialect='standard')
|
|
return df.columns
|
|
@staticmethod
|
|
def apply(**args):
|
|
"""
|
|
This function applies the
|
|
"""
|
|
columns = pseudonym.meta(**args)
|
|
#
|
|
# we need to make the schema here
|
|
client = bq.Client.from_service_account_json(args['key'])
|
|
datasets = list(client.list_datasets())
|
|
dataset_name = args['dataset']+DATASET_SUFFIX
|
|
if np.sum( [ 1*(item.dataset_id == dataset_name) for item in datasets]) == 0:
|
|
#-- make the target dataset
|
|
|
|
dataset = bq.Dataset(client.dataset(dataset_name))
|
|
client.create_dataset(dataset)
|
|
|
|
for name in columns :
|
|
p = dict(args,**{"field":name})
|
|
p['filter'] = '' if 'filter' not in args else args['filter']
|
|
pseudonym.post(**p)
|
|
|
|
|
|
#
|
|
# let us submit the query
|
|
pass
|
|
@staticmethod
|
|
def post(**args) :
|
|
"""
|
|
This function will submit a query to bigquery for insertion
|
|
"""
|
|
SQL = " ".join(['SELECT DISTINCT CAST(',args['field']," AS STRING) AS values, COUNT(*) as counts FROM :dataset.:table :filter"]).replace(':dataset',args['dataset'])
|
|
SQL = SQL.replace(':table',args['table'])
|
|
if args['filter'].strip() != '' :
|
|
SQL = SQL.replace(":filter", "WHERE "+args['filter'])
|
|
else:
|
|
SQL = SQL.replace(":filter","")
|
|
SQL += " ".join(['GROUP BY ',args['field'],'ORDER BY 1 '])
|
|
TABLE_NAME = ".".join([args['dataset'],args['table']])
|
|
credentials = service_account.Credentials.from_service_account_file(args['key'])
|
|
df = pd.read_gbq(SQL,credentials=credentials,dialect='standard')
|
|
df['table'] = args['table']
|
|
df['field'] = args['field']
|
|
# df['filter']= args['filter']
|
|
N = df.shape[0] + 10000
|
|
beg = np.random.randint(11,200)
|
|
df['encoded'] = np.random.choice(np.arange(beg,N),df.shape[0],replace=False)
|
|
df = df[['table','field','values','counts','encoded']]
|
|
# print (df.head()[:5])
|
|
# sys.stdout.flush()
|
|
TABLE_NAME = ".".join([args['dataset']+DATASET_SUFFIX,PSEUDO_TABLENAME])
|
|
df.to_gbq(TABLE_NAME,credentials=credentials,if_exists='append',chunksize=10000)
|
|
# df.to_gbq(TABLE_NAME.replace('.','_pseudo.'),credentials=credentials,if_exists='append')
|
|
|
|
class Builder :
|
|
"""
|
|
This class will build a dataset from encoded values
|
|
"""
|
|
def encode(self,**args):
|
|
"""
|
|
This function will create pseudonyms for a given table from the mapping tables
|
|
"""
|
|
SQL = "SELECT * FROM :dataset.:table limit 1".replace(':dataset',args['dataset']).replace(":table",args['table'])
|
|
credentials = service_account.Credentials.from_service_account_file(args['key'])
|
|
columns = pd.read_gbq(SQL,credentials=credentials,dialect='standard').columns.tolist()
|
|
TEMPLATE = ['(SELECT encoded FROM :dataset'+DATASET_SUFFIX+'.'+PSEUDO_TABLENAME,"WHERE table=':table' AND field = ':name' AND CAST(values AS STRING)=CAST(:table.:name AS STRING ) ) as :name"]
|
|
SQL = ["SELECT"]
|
|
FIELDS = []
|
|
FILTER = args['filter'] if 'filter' in args else ""
|
|
for field in columns :
|
|
FIELDS += [" ".join(TEMPLATE).replace(":name",field)]
|
|
# if field in FILTER :
|
|
# FILTER = FILTER.replace(field,'values')
|
|
|
|
SQL += [",\n\t".join(FIELDS)]
|
|
SQL += ['FROM :dataset.:table']
|
|
if FILTER != "" :
|
|
SQL += ["WHERE ",FILTER]
|
|
return ("\n".join(SQL).replace(":dataset",args['dataset']).replace(':table',args['table']) )
|
|
|
|
def process(self,**args):
|
|
"""
|
|
:dataset
|
|
:table
|
|
:key
|
|
"""
|
|
pseudonym.apply(**args)
|
|
def decode(self,**args):
|
|
"""
|
|
This function should be able to take a pseudonymized data frame and convert it to original values
|
|
...
|
|
"""
|
|
|
|
pass
|
|
class Binary :
|
|
"""
|
|
This is a utility class to import and export a data to/from a binary matrix
|
|
"""
|
|
def __stream(self,column) :
|
|
"""
|
|
This function will convert a column into a binary matrix with the value-space representing each column of the resulting matrix
|
|
:column a column vector i.e every item is a row
|
|
"""
|
|
# values = np.unique(column)
|
|
|
|
values = column.dropna().unique()
|
|
values.sort()
|
|
#
|
|
# Let's treat the case of missing values i.e nulls
|
|
#
|
|
row_count,col_count = column.size,values.size
|
|
|
|
matrix = [ np.zeros(col_count) for i in np.arange(row_count)]
|
|
#
|
|
# let's create a binary matrix of the feature that was passed in
|
|
# The indices of the matrix are inspired by classical x,y axis
|
|
|
|
if col_count > 0 and values.size > 1:
|
|
|
|
for yi in np.arange(row_count) :
|
|
value = column[yi]
|
|
if value not in values :
|
|
continue
|
|
xi = np.where(values == value)
|
|
xi = xi[0][0] #-- column index
|
|
matrix[yi][xi] = 1
|
|
|
|
return matrix
|
|
def Export(self,df) :
|
|
"""
|
|
This function will convert a data-frame to a binary matrix
|
|
:return _map,matrix
|
|
"""
|
|
#
|
|
# This will give us a map of how each column was mapped to a bitstream
|
|
|
|
_map = df.fillna(np.nan).apply(lambda column: self.__stream(column),axis=0)
|
|
|
|
#
|
|
# We will merge this to have a healthy matrix
|
|
_matrix = _map.apply(lambda row: list(list(itertools.chain(*row.values.tolist()))),axis=1)
|
|
_matrix = np.matrix([list(item) for item in _matrix])
|
|
#
|
|
# let's format the map so we don't have an unreasonable amount of data
|
|
#
|
|
columns = _map.columns.tolist()
|
|
beg = 0
|
|
end = 0
|
|
_map = _map.loc[0]
|
|
_m = {}
|
|
for name in columns :
|
|
end += _map[name].size
|
|
_m[name] = {"start":beg,"end":end}
|
|
beg = end
|
|
|
|
return _m,_matrix.astype(np.float32)
|
|
|
|
def Import(self,df,values,_map):
|
|
"""
|
|
This function will convert a binary stream into a
|
|
:values original/pseudonymed values
|
|
:_map field map of the binary matrix
|
|
"""
|
|
r = pd.DataFrame(None,columns=_map.keys())
|
|
for key in _map:
|
|
i = np.arange(_map[key]['start'],_map[key]['end'])
|
|
columns = values[key]
|
|
r[key] = df[i].apply(lambda row: np.array( columns)[row==1][0], axis=1 )
|
|
return r
|
|
pass
|
|
|
|
# has_basic = 'dataset' in SYS_ARGS.keys() and 'table' in SYS_ARGS.keys() and 'key' in SYS_ARGS.keys()
|
|
# has_action= 'export' in SYS_ARGS.keys() or 'pseudo' in SYS_ARGS.keys()
|
|
# df = pd.DataFrame({"fname":['james','james','steve','kevin','kevin'],"lname":["bond","dean","nyemba",'james','johnson']})
|
|
# df['age'] = (np.random.sample(df.shape[0]) * 100).astype(np.int32)
|
|
if __name__ == '__main__' :
|
|
"""
|
|
Run the program from the command line passing the following mandatory arguments
|
|
python bridge.py <[--pseudo|--export <PATH>]> --dataset <dataset> --table <tablename> [--filter <table filter>]
|
|
--pseudo will create pseudonyms for a given
|
|
--export will export data to a specified location
|
|
"""
|
|
has_basic = 'dataset' in SYS_ARGS.keys() and 'table' in SYS_ARGS.keys() and 'key' in SYS_ARGS.keys()
|
|
has_action= 'export' in SYS_ARGS.keys() or 'pseudo' in SYS_ARGS.keys()
|
|
if has_basic and has_action :
|
|
builder = Builder()
|
|
if 'export' in SYS_ARGS :
|
|
print ()
|
|
print ("exporting ....")
|
|
if not os.path.exists(SYS_ARGS['export']) :
|
|
os.mkdir(SYS_ARGS['export'])
|
|
SQL = builder.encode(**SYS_ARGS)
|
|
#
|
|
# Assuming the user wants to filter the records returned :
|
|
#
|
|
|
|
credentials = service_account.Credentials.from_service_account_file(SYS_ARGS['key'])
|
|
df = pd.read_gbq(SQL,credentials =credentials,dialect='standard')
|
|
FILENAME = os.sep.join([SYS_ARGS['export'],SYS_ARGS['table']+'.csv'])
|
|
#
|
|
# This would allow us to export it to wherever we see fit
|
|
print (FILENAME)
|
|
df.to_csv(FILENAME,index=False)
|
|
f = open(FILENAME.replace('.csv','.sql'),'w+')
|
|
f.write(SQL)
|
|
f.close()
|
|
elif 'pseudo' in SYS_ARGS :
|
|
builder.process(**SYS_ARGS)
|
|
else:
|
|
print ("")
|
|
print (SYS_ARGS.keys())
|
|
print ("has basic ",has_basic)
|
|
print ("has action ",has_action)
|
|
# pseudonym.apply(table='person',dataset='wgan_original',key='./curation-test-2.json')
|
|
# args = {"dataset":"wgan_original","table":"observation","key":"./curation-test-2.json"}
|
|
# builder = Builder()
|
|
# # builder.encode(dataset='wgan_original',table='person',key='./curation-test-2.json')
|
|
# builder.process(**args)
|