209 lines
7.3 KiB
Python
209 lines
7.3 KiB
Python
import urllib.request
|
|
import json
|
|
import ssl
|
|
import time
|
|
import re
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
IME_TOKEN = "eyJhbGciOiJIUzUxMiIsInR5cCI6IkpXVCJ9.eyJodHRwOi8vc2NoZW1hcy54bWxzb2FwLm9yZy93cy8yMDA1LzA1L2lkZW50aXR5L2NsYWltcy9uYW1lIjoiMDkzNjMxNDAyNjIiLCJodHRwOi8vc2NoZW1hcy54bWxzb2FwLm9yZy93cy8yMDA1LzA1L2lkZW50aXR5L2NsYWltcy9uYW1laWRlbnRpZmllciI6IjE0ODgiLCJleHAiOjE3ODgzMjQ5MzN9.9r-XUKmRUBQ6uEOZGq09l474ZaDDPazNVUDHuXY2_QmBtS2RU78xJLt753GaGU2SZtpndzNYKWDYaoc58SNyAA"
|
|
BASE_URL = "https://api.yektazob.com"
|
|
|
|
# 6 hours cache
|
|
CACHE_TTL = 21600
|
|
|
|
_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 clean_persian_title(title: str) -> tuple[str, str]:
|
|
"""
|
|
Removes English symbol codes (e.g. FTKC-BSG017OO-00, KHSS-DRIBRI-00)
|
|
and returns (clean_title, manufacturer).
|
|
"""
|
|
if not title:
|
|
return "", ""
|
|
parts = [p.strip() for p in title.split(" - ") if p.strip()]
|
|
if len(parts) >= 3:
|
|
if re.search(r'^[A-Za-z0-9_-]+$', parts[1]):
|
|
product = parts[0]
|
|
manufacturer = parts[2]
|
|
return f"{product} ({manufacturer})", manufacturer
|
|
elif len(parts) == 2:
|
|
if re.search(r'^[A-Za-z0-9_-]+$', parts[1]):
|
|
return parts[0], ""
|
|
return f"{parts[0]} ({parts[1]})", parts[1]
|
|
|
|
cleaned = re.sub(r'\b[A-Za-z0-9]{3,}-[A-Za-z0-9_-]+\b', '', title)
|
|
cleaned = re.sub(r'\s*-\s*-+\s*', ' - ', cleaned).strip(" -")
|
|
return cleaned or title, ""
|
|
|
|
def fetch_sub_symbols(sub_id: int) -> tuple[int, List[Dict[str, Any]]]:
|
|
try:
|
|
url = f"{BASE_URL}/Symbols/IME/Symbols/GetAll"
|
|
payload = json.dumps({"PageNumber": 1, "PageSize": 1000, "CategoryId": sub_id}).encode('utf-8')
|
|
req = urllib.request.Request(url, data=payload, headers=HEADERS)
|
|
with urllib.request.urlopen(req, context=ctx, timeout=10) as resp:
|
|
data = json.loads(resp.read().decode('utf-8'))
|
|
return sub_id, data if isinstance(data, list) else []
|
|
except:
|
|
return sub_id, []
|
|
|
|
def get_ime_categories(ttl_seconds: int = CACHE_TTL) -> 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'))
|
|
|
|
# Collect subcategories for parallel price enrichment
|
|
sub_ids = []
|
|
for root in raw_cats:
|
|
for sub in root.get("children", []):
|
|
sub_ids.append(sub.get("id"))
|
|
|
|
# Enrich symbols with their latest prices in parallel
|
|
price_map: Dict[int, Dict[str, Any]] = {}
|
|
with ThreadPoolExecutor(max_workers=14) as executor:
|
|
results = list(executor.map(fetch_sub_symbols, sub_ids))
|
|
|
|
for s_id, sym_list in results:
|
|
for s in sym_list:
|
|
s_num_id = s.get("id")
|
|
prices = s.get("symbolPrices") or []
|
|
if prices and s_num_id:
|
|
last_p = prices[-1]
|
|
price_map[s_num_id] = {
|
|
"id": last_p.get("id"),
|
|
"date": last_p.get("date", "").split("T")[0],
|
|
"high": last_p.get("high"),
|
|
"low": last_p.get("low"),
|
|
"mid": last_p.get("mid")
|
|
}
|
|
|
|
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_id = s.get("id")
|
|
raw_title = s.get("title", "")
|
|
clean_title, mfr = clean_persian_title(raw_title)
|
|
product_name = s.get("productName") or clean_title
|
|
latest_p = price_map.get(s_id)
|
|
|
|
symbols.append({
|
|
"id": s_id,
|
|
"title": clean_title,
|
|
"product_name": product_name,
|
|
"manufacturer": s.get("manufacturer") or mfr,
|
|
"latest_price": latest_p,
|
|
"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 = CACHE_TTL) -> 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 = []
|
|
last_valid_mid = None
|
|
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
|
|
mid_val = p.get("mid") or p.get("high") or p.get("low") or 0
|
|
if mid_val <= 0 and last_valid_mid is not None:
|
|
mid_val = last_valid_mid
|
|
else:
|
|
last_valid_mid = mid_val
|
|
|
|
clean_prices.append({
|
|
"id": p.get("id"),
|
|
"date": date_iso,
|
|
"high": p.get("high") or mid_val,
|
|
"low": p.get("low") or mid_val,
|
|
"mid": mid_val
|
|
})
|
|
|
|
raw_title = raw_detail.get("title", "")
|
|
clean_title, mfr = clean_persian_title(raw_title)
|
|
|
|
result = {
|
|
"id": raw_detail.get("id"),
|
|
"cat_id": raw_detail.get("catID"),
|
|
"title": clean_title,
|
|
"manufacturer": mfr,
|
|
"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
|
|
|
|
# Pre-warm cache on boot
|
|
threading.Thread(target=get_ime_categories, daemon=True).start()
|