-
Notifications
You must be signed in to change notification settings - Fork 0
/
arg_parser.py
153 lines (117 loc) · 11.2 KB
/
arg_parser.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
import argparse
def get_args():
# set/parse cli arguments
parser = argparse.ArgumentParser(description="eth ETL program")
subparsers = parser.add_subparsers(dest="job", help="Choose a job to execute.")
# Export interacting contract addresses given the blocknumber and the "from" contract.
cf_parser = subparsers.add_parser("contracts_from", help="export contracts that are called from the input address")
cf_parser.add_argument("--blocknumber", "-b", type=str, default='latest', help="Block number of interest")
cf_parser.add_argument("--addr", "-a", type=str, nargs='+', required=True, help="Contract address of interest (from)")
cf_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
# Export interacting contract addresses given the blocknumber and the "to" contract.
ct_parser = subparsers.add_parser("contracts_to", help="export contracts that make calls to the input address (for inter-contract calls, use traces_* jobs)")
ct_parser.add_argument("--blocknumber", "-b", type=str, default='latest', help="Block number of interest")
ct_parser.add_argument("--addr", "-a",type=str, nargs='+',required=True, help="Contract address of interest (to)")
ct_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
# Export transactions given the blocknumber and the "from" contract.
txf_parser = subparsers.add_parser("txs_from", help="export transactions that were from the input address ")
txf_parser.add_argument("--blocknumber","-b", type=str, default='latest', help="Block number of interest")
txf_parser.add_argument("--addr", "-a",type=str, nargs='+',required=True, help="Contract address of interest (from)")
txf_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
# Export transactions given the blocknumber and the "to" contract.
txt_parser = subparsers.add_parser("txs_to", help="export transactions that were making calls to the input address (for inter-contract calls, use traces_* jobs)")
txt_parser.add_argument("--blocknumber", "-b",type=str, default='latest', help="Block number of interest")
txt_parser.add_argument("--addr", "-a",type=str, nargs='+',required=True, help="Contract address of interest (to)")
txt_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
# Export traces given the blocknnumber and the "from" contract.
trf_parser = subparsers.add_parser("traces_from", help="export traces of transactions that were from the input address")
trf_parser.add_argument("--blocknumber", "-b",type=str, default='latest', help="Block number of interest")
trf_parser.add_argument("--addr", "-a",type=str, nargs='+',required=True, help="Contract address of interest (from)")
trf_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
# Export traces given the blocknnumber and the "to" contract.
trt_parser = subparsers.add_parser("traces_to", help="export traces of the transactions that were making calls to the input address")
trt_parser.add_argument("--blocknumber", "-b",type=str, default='latest', help="Block number of interest")
trt_parser.add_argument("--addr", "-a",type=str, nargs='+',required=True, help="Contract address of interest (to)")
trt_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
# apply filter (if supported by the rpc provider) to get addrs/transactions/traces
filter_parser = subparsers.add_parser("apply_filter", help="apply filter on the range of blocknumbers to query addrs/transactions/traces")
filter_parser.add_argument("--job_name", "-jn", type=str, required=True, help="job to run for each match in the filter. check help (-h) instructions")
filter_parser.add_argument("--start_block", "-sb", type=str, required=True, help="starting blocknumber")
filter_parser.add_argument("--end_block", '-eb', type=str, required=True, help="ending blocknumber")
filter_parser.add_argument("--addr", "-a", type=str, nargs='+', required=True,help="Contract address of interest")
filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
filter_parser.add_argument("--save_blocklist", "-sbl", type=bool, default=True, help="flag for saving interim block list")
# get logs (2k range limit)
filter_parser = subparsers.add_parser("get_logs", help="apply filter on the range of blocknumbers to get event logs")
filter_parser.add_argument("--start_block", "-sb", type=str, required=True, help="starting blocknumber")
filter_parser.add_argument("--end_block", '-eb', type=str, required=True, help="ending blocknumber")
filter_parser.add_argument("--addr", "-a", type=str, nargs='+', required=True,help="Contract address of interest")
filter_parser.add_argument("--topics", "-t", type=str, nargs='+', required=False,help="topics of interest (optional, if not specified, get_logs returns all topics)")
filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
# traces by applying filters (2k range limit)
filter_parser = subparsers.add_parser("trace_filter", help="apply trace filter on the range of blocknumbers to get traces")
filter_parser.add_argument("--start_block", "-sb", type=str, required=True, help="starting blocknumber")
filter_parser.add_argument("--end_block", '-eb', type=str, required=True, help="ending blocknumber")
filter_parser.add_argument("--addr", "-a", type=str, nargs='+', required=True,help="Contract address of interest")
filter_parser.add_argument("--pos", "-p", type=str, required=True,help="Contract address position")
filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
# export all traces given a blocknumber and a target transaction position
filter_parser = subparsers.add_parser("trace_out", help="full export of a trace given a blocknumber and transaction position")
filter_parser.add_argument("--blocknumber", "-b", type=str, required=True, help="target blocknumber")
filter_parser.add_argument("--tx_pos", "-p", type=str, nargs='+', required=True,help="target transaction position")
filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
### decoding ###
# after trace_out job was executed, decode exported inputs with a specific search keyword for function names
filter_parser = subparsers.add_parser("decode_trace", help="decode exported traces (from trace_out job) with a specific search keyword (function name)")
filter_parser.add_argument("--search_keyword", "-s", type=str, required=True, help="search keyword for function names")
filter_parser.add_argument("--exported_file", "-e", type=str, required=True,help="exported traces (csv file)")
filter_parser.add_argument("--transfer_func_patterns", "-p", type=str, default = False, help="if set it to True, the job will try to parse transfer functions using predefined patterns. It will fail if it finds a new pattern")
filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
filter_parser = subparsers.add_parser("decode_logs", help="decode exported logs (from get_logs job)")
filter_parser.add_argument("--exported_file", "-e", type=str, required=True,help="exported traces (csv file)")
filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
### price fetching ###
filter_parser = subparsers.add_parser("price_current", help="fetch the most recent token price")
filter_parser.add_argument("--source", "-s", type=str, required=True, help="source for the price (e.g., defillama, coingecko)")
filter_parser.add_argument("--token", "-t", type=str, nargs = '+', required=True,help="token address(es) or path to a csv file")
filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id when running multiple jobs")
# wip
# filter_parser = subparsers.add_parser("price_historical", help="fetch price time series data")
# filter_parser.add_argument("--source", "-s", type=str, required=True, help="source for the price (e.g., defillama, coingecko)")
# filter_parser.add_argument("--token", "-t", type=str, nargs = '+', required=True,help="token address(es) or path to a csv file")
# filter_parser.add_argument("--start_block", "-sb", type=str, required=True,help="start block")
# filter_parser.add_argument("--end_block", "-eb", type=str, required=True,help="last block")
# filter_parser.add_argument("--interval", "-i", type=int, required=False, default = 1, help="interval for fetch prices (in days, defaults to 1 day)")
# filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id when running multiple jobs")
return parser.parse_args()
# fetch the right pipeline for the given job name
def job_parser(args):
etl_jobs = ['contracts_from', 'contracts_to', 'txs_to', 'traces_to', 'txs_from', 'traces_from', 'apply_filter', 'get_logs', 'trace_filter', 'trace_out']
decode_jobs = ['decode_trace', 'decode_logs']
price_fetch_jobs = ['price_current', 'price_historical']
if(args.job in etl_jobs):
pipeline = 'eth_etl'
elif(args.job in decode_jobs):
pipeline = 'decode'
elif(args.job in price_fetch_jobs):
pipeline = 'price_fetch'
else:
raise ValueError(f'job alias : {args.job} not recognized by the parser')
return pipeline
# debugging
"""
import argparse
filter_parser = argparse.ArgumentParser()
# traces by applying filters (2k range limit)
filter_parser.add_argument("--source", "-s", type=str, required=True, help="source for the price (e.g., defillama, coingecko)")
filter_parser.add_argument("--token", "-t", type=str, nargs = '+', required=True,help="token address(es) or path to a csv file")
filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id when running multiple jobs")
args = argparse.Namespace(job= 'price_current', source= 'defillama', token = '/home/takim/work/blockchain-graph-analytics/intermediate_results/transfer_balance/230923/unique_tokens.csv', job_id= '0')
filter_parser = argparse.ArgumentParser()
# traces by applying filters (2k range limit)
filter_parser.add_argument("--search_keyword", "-s", type=str, required=True, help="search keyword for function names")
filter_parser.add_argument("--exported_file", "-e", type=str, required=True,help="exported traces (csv file)")
filter_parser.add_argument("--transfer_func_patterns", "-p", type=str, default = False, help="if set it to True, the job will try to parse transfer functions using predefined patterns. It will fail if it finds a new pattern")
filter_parser.add_argument("--job_id", "-j", type=str, default='0', help="job id for running multiple jobs")
args = argparse.Namespace(job= 'decode_trace', search_keyword= 'transfer', exported_file = '/home/takim/work/eth-tracker/output/160224/trace_out/19240945/traced_out.csv', transfer_func_patterns = 'True', job_id='0')
"""