133 lines
4.6 KiB
Python
133 lines
4.6 KiB
Python
import urllib.request
|
|
import json
|
|
import ssl
|
|
import time
|
|
import threading
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
IME_TOKEN = "eyJhbGciOiJIUzUxMiIsInR5cCI6IkpXVCJ9.eyJodHRwOi8vc2NoZW1hcy54bWxzb2FwLm9yZy93cy8yMDA1LzA1L2lkZW50aXR5L2NsYWltcy9uYW1lIjoiMDkzNjMxNDAyNjIiLCJodHRwOi8vc2NoZW1hcy54bWxzb2FwLm9yZy93cy8yMDA1LzA1L2lkZW50aXR5L2NsYWltcy9uYW1laWRlbnRpZmllciI6IjE0ODgiLCJleHAiOjE3ODgzMjQ5MzN9.9r-XUKmRUBQ6uEOZGq09l474ZaDDPazNVUDHuXY2_QmBtS2RU78xJLt753GaGU2SZtpndzNYKWDYaoc58SNyAA"
|
|
BASE_URL = "https://api.yektazob.com"
|
|
|
|
_cache_categories = {
|
|
"data": None,
|
|
"last_fetched": 0
|
|
}
|
|
_cache_symbols = {}
|
|
_lock = threading.Lock()
|
|
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
HEADERS = {
|
|
"Authorization": f"Bearer {IME_TOKEN}",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
|
|
}
|
|
|
|
def get_ime_categories(ttl_seconds: int = 3600) -> List[Dict[str, Any]]:
|
|
now = time.time()
|
|
with _lock:
|
|
if _cache_categories["data"] and (now - _cache_categories["last_fetched"]) < ttl_seconds:
|
|
return _cache_categories["data"]
|
|
|
|
url = f"{BASE_URL}/Symbols/IME/Categories/GetAll"
|
|
payload = json.dumps({"PageNumber": 1, "PageSize": 1000}).encode('utf-8')
|
|
req = urllib.request.Request(url, data=payload, headers=HEADERS)
|
|
|
|
with urllib.request.urlopen(req, context=ctx, timeout=12) as resp:
|
|
raw_cats = json.loads(resp.read().decode('utf-8'))
|
|
|
|
cleaned = []
|
|
for root in raw_cats:
|
|
root_title = root.get("title", "")
|
|
root_id = root.get("id")
|
|
subs = []
|
|
for sub in root.get("children", []):
|
|
sub_id = sub.get("id")
|
|
sub_title = sub.get("title", "")
|
|
symbols = []
|
|
for s in sub.get("symbols", []):
|
|
s_title = s.get("title", "")
|
|
parts = [p.strip() for p in s_title.split(" - ") if p.strip()]
|
|
product_name = s.get("productName") or (parts[0] if len(parts) > 0 else s_title)
|
|
symbol_code = s.get("symbol") or (parts[1] if len(parts) > 1 else "")
|
|
manufacturer = s.get("manufacturer") or (parts[2] if len(parts) > 2 else "")
|
|
|
|
symbols.append({
|
|
"id": s.get("id"),
|
|
"title": s_title,
|
|
"product_name": product_name,
|
|
"symbol_code": symbol_code,
|
|
"manufacturer": manufacturer,
|
|
"is_active": s.get("isActive", True),
|
|
"updated_at": s.get("updatedAt", "")
|
|
})
|
|
|
|
subs.append({
|
|
"id": sub_id,
|
|
"title": sub_title,
|
|
"symbols_count": len(symbols),
|
|
"symbols": symbols
|
|
})
|
|
|
|
total_symbols = sum(s["symbols_count"] for s in subs)
|
|
cleaned.append({
|
|
"id": root_id,
|
|
"title": root_title,
|
|
"subcategories_count": len(subs),
|
|
"total_symbols": total_symbols,
|
|
"subcategories": subs
|
|
})
|
|
|
|
with _lock:
|
|
_cache_categories["data"] = cleaned
|
|
_cache_categories["last_fetched"] = now
|
|
|
|
return cleaned
|
|
|
|
|
|
def get_ime_symbol_detail(symbol_id: int, ttl_seconds: int = 1800) -> Dict[str, Any]:
|
|
now = time.time()
|
|
with _lock:
|
|
cached = _cache_symbols.get(symbol_id)
|
|
if cached and (now - cached["time"]) < ttl_seconds:
|
|
return cached["data"]
|
|
|
|
url = f"{BASE_URL}/Symbols/IME/Symbols/Get/{symbol_id}"
|
|
req = urllib.request.Request(url, headers=HEADERS)
|
|
|
|
with urllib.request.urlopen(req, context=ctx, timeout=12) as resp:
|
|
raw_detail = json.loads(resp.read().decode('utf-8'))
|
|
|
|
raw_prices = raw_detail.get("symbolPrices", []) or []
|
|
sorted_prices = sorted(raw_prices, key=lambda p: p.get("date", ""))
|
|
|
|
clean_prices = []
|
|
for p in sorted_prices:
|
|
d_str = p.get("date", "")
|
|
date_iso = d_str.split("T")[0] if "T" in d_str else d_str
|
|
clean_prices.append({
|
|
"id": p.get("id"),
|
|
"date": date_iso,
|
|
"high": p.get("high"),
|
|
"low": p.get("low"),
|
|
"mid": p.get("mid")
|
|
})
|
|
|
|
result = {
|
|
"id": raw_detail.get("id"),
|
|
"cat_id": raw_detail.get("catID"),
|
|
"title": raw_detail.get("title"),
|
|
"order_no": raw_detail.get("orderNo"),
|
|
"small_desc": raw_detail.get("smallDesc"),
|
|
"total_points": len(clean_prices),
|
|
"latest_price": clean_prices[-1] if clean_prices else None,
|
|
"prices": clean_prices
|
|
}
|
|
|
|
with _lock:
|
|
_cache_symbols[symbol_id] = {"time": now, "data": result}
|
|
|
|
return result
|