File size: 4,677 Bytes
9d4bd7c |
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 |
import asyncio
import re
from datetime import datetime
import httpx
async def fetch_all_utt_data():
base_headers = {
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Language': 'en-US,en;q=0.9',
'Connection': 'keep-alive',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'Origin': 'https://www.uttamis.co.tz',
'Referer': 'https://www.uttamis.co.tz/fund-performance',
'Sec-Fetch-Dest': 'empty',
'Sec-Fetch-Mode': 'cors',
'Sec-Fetch-Site': 'same-origin',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36',
'X-Requested-With': 'XMLHttpRequest',
'sec-ch-ua': '"Google Chrome";v="135", "Not-A.Brand";v="8", "Chromium";v="135"',
'sec-ch-ua-mobile': '?0',
'sec-ch-ua-platform': '"Windows"',
}
all_data = []
start = 0
length = 4000
max_records = 5000
async with httpx.AsyncClient(timeout=30.0) as client:
# Fetch the initial page to get CSRF token and cookies
get_response = await client.get("https://www.uttamis.co.tz/fund-performance", headers=base_headers)
if get_response.status_code != 200:
print(f"Failed to fetch CSRF token: {get_response.status_code}")
return []
html_content = get_response.text
print(html_content[1000:2000])
csrf_token_match = re.search(r'<meta name="csrf-token" content="([^"]+)"', html_content)
if not csrf_token_match:
print("CSRF token not found.")
return []
csrf_token = csrf_token_match.group(1)
# Prepare POST headers with CSRF token
post_headers = base_headers.copy()
post_headers['X-CSRF-TOKEN'] = csrf_token
while start < max_records:
payload = {
"csrf-token": csrf_token,
"draw": "1",
"start": str(start),
"length": str(length),
"search[value]": "",
"search[regex]": "false",
}
# Columns configuration
for i, col in enumerate([
("DT_RowIndex", False),
("sname.name", True),
("net_asset_value", True),
("outstanding_number_of_units", True),
("nav_per_unit", True),
("sale_price_per_unit", True),
("repurchase_price_per_unit", True),
("date_valued", True),
]):
data_key = col[0].split(".")[0]
payload[f"columns[{i}][data]"] = data_key
payload[f"columns[{i}][name]"] = col[0]
payload[f"columns[{i}][searchable]"] = str(col[1]).lower()
payload[f"columns[{i}][orderable]"] = str(col[1]).lower()
payload[f"columns[{i}][search][value]"] = ""
payload[f"columns[{i}][search][regex]"] = "false"
# Make the POST request
response = await client.post(
"https://www.uttamis.co.tz/navs",
data=payload,
headers=post_headers
)
if response.status_code != 200:
print(f"Request failed at offset {start}: {response.status_code}")
break
json_data = response.json()
rows = json_data.get("data", [])
if not rows:
break
all_data.extend(rows)
# Check if there are more records
if len(rows) < length:
break
start += length
return all_data
def parse_utt_api_row(row: dict) -> dict:
def clean_number(value: str) -> float:
return float(value.replace(',', ''))
date_str = row.get("date_valued")
try:
date = datetime.strptime(date_str, "%d-%b-%Y").date()
except ValueError:
date = datetime.strptime(date_str, "%d-%m-%Y").date()
return {
"date": date,
"nav_per_unit": clean_number(row["nav_per_unit"]),
"sale_price_per_unit": clean_number(row["sale_price_per_unit"]),
"repurchase_price_per_unit": clean_number(row["repurchase_price_per_unit"]),
"outstanding_number_of_units": int(clean_number(row["outstanding_number_of_units"])),
"net_asset_value": int(clean_number(row["net_asset_value"]))
}
# Example usage
async def main():
data = await fetch_all_utt_data()
parsed_data = [parse_utt_api_row(row) for row in data]
print(parsed_data)
if __name__ == "__main__":
asyncio.run(main()) |