voith · August 14, 2024 13:05 · voith · Aug 9, 2024
diff --git a/chainlink_data_streams.py b/chainlink_data_streams.py
 import hmac
 import hashlib
 import binascii
 import time
 from dataclasses import dataclass
 from urllib.parse import urlparse, urlunparse, urljoin

 import requests
 from eth_abi import decode
 from eth_utils import to_hex

 API_URL = 'https://api.testnet-dataengine.chain.link'
 # replace <CHAINLINK_CLIENT_ID>
 CLIENT_ID = '<CHAINLINK_CLIENT_ID>'
 # replace <CHAINLINK_SECRET>
 USER_SECRET = '<CHAINLINK_SECRET>'

 MEEM_FEED_IDS = {
    'DOGE': '0x00032057c7f224d0266b4311a81cdc3e38145e36442713350d3300fb12e85c99',
    'SHIB': '0x0003e5dfcb9c8fda20285c89ab6b289926b17dde5e704a1c228a8b468913f56a',
    'PEPE': '0x0003c8541bf8517752d466f6efc2057b54a9e189e45f08fa37a4db73452df4d4',
    'FLOKI': '0x0003e35ba9242261003459b8c97933127cdff9b727b9492ab017de9484fdc6b9',
    'ETH': '0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782',
 }



 @dataclass
 class FeedData:
    feed_id: str
    valid_from_timestamp: int
    observations_timestamp: int
    native_fee: int
    link_fee: int
    expires_at: int
    benchmark_price: int
    bid: int
    ask: int


 def extract_full_path_with_params(url):
    # Parse the URL
    parsed_url = urlparse(url)

    # Reconstruct the URL with all components
    full_path_with_params = urlunparse((
        '',  # scheme (empty for path only)
        '',  # netloc (empty for path only)
        parsed_url.path,  # path
        parsed_url.params,  # params
        parsed_url.query,  # query
        parsed_url.fragment  # fragment
    ))

    return full_path_with_params


 def create_url(api_url, path, params):
    return requests.Request(
        'GET',
        urljoin(api_url, path),
        params=params
    ).prepare().url


 def generate_hmac(method: str, path: str, body: bytes, client_id: str, timestamp: int, user_secret: str) -> str:
    # Calculate the SHA256 hash of the body
    body_hash = hashlib.sha256(body).digest()
    body_hash_hex = binascii.hexlify(body_hash).decode('utf-8')

    # Construct the string to be signed
    server_body_hash_string = f"{method} {path} {body_hash_hex} {client_id} {timestamp}"

    # Create HMAC signature
    signed_message = hmac.new(user_secret.encode(), server_body_hash_string.encode(), hashlib.sha256)
    user_hmac = binascii.hexlify(signed_message.digest()).decode('utf-8')

    return user_hmac


 def generate_auth_headers(
        method: str,
        path: str,
        body: bytes,
        client_id: str,
        user_secret: str,
        timestamp: int
 ) -> dict:
    headers = dict()
    # Generate HMAC signature
    hmac_string = generate_hmac(method, path, body, client_id, timestamp, user_secret)

    # Add headers
    headers['Authorization'] = client_id
    headers['X-Authorization-Timestamp'] = str(timestamp)
    headers['X-Authorization-Signature-Sha256'] = hmac_string
    return headers


 def prepare_bulk_url(feed_ids: list, timestamp):
    path = "/api/v1/reports/bulk"
    return create_url(
        API_URL,
        path,
        dict(
            feedIDs=','.join(feed_ids),
            timestamp=int(timestamp/1000))
        )


 def decode_feed_data(data):
    feed_data = []
    for report in data['reports']:
        intermediate_decoded_data = decode(
            ["bytes32[3]", "bytes", "bytes32[]", "bytes32[]", "bytes32"],
            bytes.fromhex(report['fullReport'][2:])
        )

        decoded_data = decode(
            ["bytes32", "uint32", "uint32", "uint192", "uint192", "uint32", "int192", "int192", "int192"],
            intermediate_decoded_data[1]
        )
        feed_data.append(FeedData(
            feed_id=to_hex(decoded_data[0]),
            valid_from_timestamp=decoded_data[1],
            observations_timestamp=decoded_data[2],
            native_fee=decoded_data[3],
            link_fee=decoded_data[4],
            expires_at=decoded_data[5],
            benchmark_price=decoded_data[6],
            bid=decoded_data[7],
            ask=decoded_data[8],
        ))
    return feed_data


 def fetch_latest_meem_index_feeds():
    method = "GET"
    body = b""
    timestamp = int(time.time() * 1000)
    request_url = prepare_bulk_url(
        [MEEM_FEED_IDS['DOGE'], MEEM_FEED_IDS['SHIB'], MEEM_FEED_IDS['PEPE'], MEEM_FEED_IDS['FLOKI'], MEEM_FEED_IDS['ETH']],
        timestamp
    )
    path = extract_full_path_with_params(request_url)
    response = requests.get(
        request_url,
        headers=generate_auth_headers(
            method,
            path,
            body,
            CLIENT_ID,
            USER_SECRET,
            timestamp
        )
    )
    response.raise_for_status()
    data = response.json()
    return decode_feed_data(data), response

 def main():
    while True:
        print('requesting feed data...')
        feed_data, response = fetch_latest_meem_index_feeds()
        if len(feed_data) != 5:
            request = response.request
            print(f'Oh no.. received only {len(feed_data)} feeds, expected 5')
            print('Below are the details of the request')
            print(f'request url: {request.url}')
            print(f'request method: {request.method}')
            print(f'request headers: {request.headers}')
            print(f'request body: {request.body}')
            return
        print('received correct feed data...')
        time.sleep(1)

 if __name__ == "__main__":
    main()
	import hmac
	import hashlib
	import binascii
	import time
	from dataclasses import dataclass
	from urllib.parse import urlparse, urlunparse, urljoin

	import requests
	from eth_abi import decode
	from eth_utils import to_hex

	API_URL = 'https://api.testnet-dataengine.chain.link'
	# replace <CHAINLINK_CLIENT_ID>
	CLIENT_ID = '<CHAINLINK_CLIENT_ID>'
	# replace <CHAINLINK_SECRET>
	USER_SECRET = '<CHAINLINK_SECRET>'

	MEEM_FEED_IDS = {
	'DOGE': '0x00032057c7f224d0266b4311a81cdc3e38145e36442713350d3300fb12e85c99',
	'SHIB': '0x0003e5dfcb9c8fda20285c89ab6b289926b17dde5e704a1c228a8b468913f56a',
	'PEPE': '0x0003c8541bf8517752d466f6efc2057b54a9e189e45f08fa37a4db73452df4d4',
	'FLOKI': '0x0003e35ba9242261003459b8c97933127cdff9b727b9492ab017de9484fdc6b9',
	'ETH': '0x000359843a543ee2fe414dc14c7e7920ef10f4372990b79d6361cdc0dd1ba782',
	}



	@dataclass
	class FeedData:
	feed_id: str
	valid_from_timestamp: int
	observations_timestamp: int
	native_fee: int
	link_fee: int
	expires_at: int
	benchmark_price: int
	bid: int
	ask: int


	def extract_full_path_with_params(url):
	# Parse the URL
	parsed_url = urlparse(url)

	# Reconstruct the URL with all components
	full_path_with_params = urlunparse((
	'', # scheme (empty for path only)
	'', # netloc (empty for path only)
	parsed_url.path, # path
	parsed_url.params, # params
	parsed_url.query, # query
	parsed_url.fragment # fragment
	))

	return full_path_with_params


	def create_url(api_url, path, params):
	return requests.Request(
	'GET',
	urljoin(api_url, path),
	params=params
	).prepare().url


	def generate_hmac(method: str, path: str, body: bytes, client_id: str, timestamp: int, user_secret: str) -> str:
	# Calculate the SHA256 hash of the body
	body_hash = hashlib.sha256(body).digest()
	body_hash_hex = binascii.hexlify(body_hash).decode('utf-8')

	# Construct the string to be signed
	server_body_hash_string = f"{method} {path} {body_hash_hex} {client_id} {timestamp}"

	# Create HMAC signature
	signed_message = hmac.new(user_secret.encode(), server_body_hash_string.encode(), hashlib.sha256)
	user_hmac = binascii.hexlify(signed_message.digest()).decode('utf-8')

	return user_hmac


	def generate_auth_headers(
	method: str,
	path: str,
	body: bytes,
	client_id: str,
	user_secret: str,
	timestamp: int
	) -> dict:
	headers = dict()
	# Generate HMAC signature
	hmac_string = generate_hmac(method, path, body, client_id, timestamp, user_secret)

	# Add headers
	headers['Authorization'] = client_id
	headers['X-Authorization-Timestamp'] = str(timestamp)
	headers['X-Authorization-Signature-Sha256'] = hmac_string
	return headers


	def prepare_bulk_url(feed_ids: list, timestamp):
	path = "/api/v1/reports/bulk"
	return create_url(
	API_URL,
	path,
	dict(
	feedIDs=','.join(feed_ids),
	timestamp=int(timestamp/1000))
	)


	def decode_feed_data(data):
	feed_data = []
	for report in data['reports']:
	intermediate_decoded_data = decode(
	["bytes32[3]", "bytes", "bytes32[]", "bytes32[]", "bytes32"],
	bytes.fromhex(report['fullReport'][2:])
	)

	decoded_data = decode(
	["bytes32", "uint32", "uint32", "uint192", "uint192", "uint32", "int192", "int192", "int192"],
	intermediate_decoded_data[1]
	)
	feed_data.append(FeedData(
	feed_id=to_hex(decoded_data[0]),
	valid_from_timestamp=decoded_data[1],
	observations_timestamp=decoded_data[2],
	native_fee=decoded_data[3],
	link_fee=decoded_data[4],
	expires_at=decoded_data[5],
	benchmark_price=decoded_data[6],
	bid=decoded_data[7],
	ask=decoded_data[8],
	))
	return feed_data


	def fetch_latest_meem_index_feeds():
	method = "GET"
	body = b""
	timestamp = int(time.time() * 1000)
	request_url = prepare_bulk_url(
	[MEEM_FEED_IDS['DOGE'], MEEM_FEED_IDS['SHIB'], MEEM_FEED_IDS['PEPE'], MEEM_FEED_IDS['FLOKI'], MEEM_FEED_IDS['ETH']],
	timestamp
	)
	path = extract_full_path_with_params(request_url)
	response = requests.get(
	request_url,
	headers=generate_auth_headers(
	method,
	path,
	body,
	CLIENT_ID,
	USER_SECRET,
	timestamp
	)
	)
	response.raise_for_status()
	data = response.json()
	return decode_feed_data(data), response

	def main():
	while True:
	print('requesting feed data...')
	feed_data, response = fetch_latest_meem_index_feeds()
	if len(feed_data) != 5:
	request = response.request
	print(f'Oh no.. received only {len(feed_data)} feeds, expected 5')
	print('Below are the details of the request')
	print(f'request url: {request.url}')
	print(f'request method: {request.method}')
	print(f'request headers: {request.headers}')
	print(f'request body: {request.body}')
	return
	print('received correct feed data...')
	time.sleep(1)

	if __name__ == "__main__":
	main()