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 (product_name, manufacturer). Avoids duplicate manufacturer names. """ if not title: return "", "" parts = [p.strip() for p in title.split(" - ") if p.strip()] if len(parts) >= 3: product = parts[0] manufacturer = parts[2] return product, manufacturer elif len(parts) == 2: if re.search(r'^[A-Za-z0-9_-]+$', parts[1]): return parts[0], "" return parts[0], 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()