import math, random def _generate_sparkline(current_price: float, pct_change: float = 0.0, n_points: int = 40) -> list: if not current_price or current_price <= 0: return [0.0] * n_points pct = (pct_change or 0.0) / 100.0 start_price = current_price / (1.0 + pct) if (1.0 + pct) > 0 else current_price points = [] trend_step = (current_price - start_price) / max(1, n_points - 1) volatility = max(abs(pct) * 0.4, 0.007) * current_price rand_seed = int(abs(current_price) * 1000) % 100000 r = random.Random(rand_seed) for i in range(n_points - 1): progress = i / float(n_points - 1) base = start_price + trend_step * i wave = math.sin(progress * math.pi * 3.5) * (volatility * 0.6) + math.cos(progress * math.pi * 5.2) * (volatility * 0.4) noise = (r.random() - 0.5) * volatility * 0.45 val = base + wave + noise points.append(round(max(val, current_price * 0.2), 2 if current_price > 100 else 4)) points.append(round(current_price, 2 if current_price > 100 else 4)) return points from fastapi import FastAPI, Query, WebSocket, HTTPException, Request, Response, APIRouter, Depends, Header from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, RedirectResponse, JSONResponse from pydantic import BaseModel, Field, field_validator from contextlib import asynccontextmanager from datetime import datetime, timedelta import asyncio import base64 import hashlib import hmac import json import ipaddress import os import re import time import auth from db import get_conn, init_db from live_stream import broadcast_loop, connect as ws_connect, get_latest as get_commodity_latest import currency_stream import world_economy import steel_stocks import bonds import reports import scrape_lme import crypto_market @asynccontextmanager async def lifespan(app: FastAPI): asyncio.create_task(broadcast_loop()) asyncio.create_task(currency_stream.scrape_loop()) asyncio.create_task(world_economy.scrape_loop()) asyncio.create_task(steel_stocks.scrape_loop()) asyncio.create_task(crypto_market.scrape_loop()) yield app = FastAPI(lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.middleware("http") async def security_headers(request, call_next): response = await call_next(request) response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate" response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "ALLOWALL" response.headers["Access-Control-Allow-Origin"] = "*" return response init_db() SESSION_COOKIE = "am_session" COOKIE_SECURE = False COOKIE_SAMESITE = "lax" def _load_api_tokens(): tokens = { "flutter": "b9660e2c174a5c76b958afdef8cc7d95f71abe23155299832b94984fcadf9f73", "didvan": "didvan_statista_key_2026", "default": "statista_secret_token_2026", } raw = os.environ.get("API_TOKENS", "").strip() if raw: for part in raw.replace("\n", ",").split(","): part = part.strip() if not part: continue if "=" in part: partner, token = part.split("=", 1) elif ":" in part: partner, token = part.split(":", 1) else: partner, token = "default", part if partner.strip() and token.strip(): tokens[partner.strip()] = token.strip() single = os.environ.get("API_TOKEN", "").strip() if single: tokens.setdefault("legacy", single) return tokens API_PARTNER_TOKENS = _load_api_tokens() def _bearer_token(authorization): if not authorization: return None authorization = authorization.strip() scheme, _, token = authorization.partition(" ") if scheme.lower() == "bearer" and token.strip(): return token.strip() return authorization def _match_api_partner(token): for partner, expected in API_PARTNER_TOKENS.items(): if hmac.compare_digest(token, expected): return partner return None def require_session(request: Request, authorization: str | None = Header(default=None)): token = request.cookies.get(SESSION_COOKIE) if token: row = auth.user_for_token(token) if row: return row bearer = _bearer_token(authorization) or request.query_params.get("api_token") or request.query_params.get("token") if bearer: partner = _match_api_partner(bearer) if partner or bearer in API_PARTNER_TOKENS.values(): return {"id": "partner", "name": partner or "flutter_client", "email": "client@steelforesight.ir", "status": "approved"} row = auth.user_for_token(bearer) if row: return row raise HTTPException(status_code=401, detail="Authentication required") @app.websocket("/ws/live") async def websocket_live(ws: WebSocket): await ws.accept() await ws_connect(ws) @app.get("/api/commodity") def get_commodity(_: object = Depends(require_session)): return get_commodity_latest() @app.get("/api/tree") def get_tree(_: object = Depends(require_session)): rows = get_conn().execute( "SELECT DISTINCT grp, category, title FROM prices ORDER BY grp, category, title" ).fetchall() tree = {} for r in rows: tree.setdefault(r["grp"], {}).setdefault(r["category"], []).append(r["title"]) return tree @app.get("/api/history") def get_history(title: str = Query(..., max_length=120), days: int = Query(0, ge=0, le=3650), _: object = Depends(require_session)): if days > 0: cutoff = (datetime.utcnow() - timedelta(days=days)).strftime("%Y-%m-%d") rows = get_conn().execute( "SELECT date, low, mid, high FROM prices WHERE title=? AND date>=? ORDER BY date LIMIT 6000", (title, cutoff) ).fetchall() else: rows = get_conn().execute( "SELECT date, low, mid, high FROM prices WHERE title=? ORDER BY date LIMIT 6000", (title,) ).fetchall() return [dict(r) for r in rows] @app.get("/api/search") def search(q: str = Query(..., max_length=64), _: object = Depends(require_session)): safe = q.replace(chr(92), chr(92)+chr(92)).replace("%", chr(92)+"%").replace("_", chr(92)+"_") rows = get_conn().execute( "SELECT DISTINCT title, grp as 'group', category FROM prices " "WHERE title LIKE ? ESCAPE '" + chr(92) + "' LIMIT 50", (f"%{safe}%",) ).fetchall() return [dict(r) for r in rows] @app.get("/api/live") def get_live(_: object = Depends(require_session)): try: rows = get_conn().execute(""" SELECT name, price, day_change, pct_change, unit, fetched_at FROM te_prices WHERE fetched_at = (SELECT MAX(fetched_at) FROM te_prices) ORDER BY name """).fetchall() return [dict(r) for r in rows] except Exception: return [] @app.get("/api/latest") def get_latest(_: object = Depends(require_session)): rows = get_conn().execute(""" SELECT p.grp as 'group', p.category, p.title, p.date, p.low, p.mid, p.high, prev.mid as prev_mid FROM prices p JOIN (SELECT title, MAX(date) as max_date FROM prices GROUP BY title) m ON p.title = m.title AND p.date = m.max_date LEFT JOIN prices prev ON prev.title = p.title AND prev.date = (SELECT MAX(date) FROM prices WHERE title=p.title AND date < p.date) """).fetchall() result = [] for r in rows: pct = round(((r["mid"]-r["prev_mid"])/r["prev_mid"]*100) if r["prev_mid"] else 0, 2) d = dict(r) d["pct"] = pct d["sparkline"] = _generate_sparkline(r["mid"] or 0, pct) result.append(d) return result @app.get("/api/prices/latest") @app.get("/api/v1/prices/latest") def get_prices_latest(_: object = Depends(require_session)): generated_at = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") return { "generated_at": generated_at, "datasets": { "metals": get_latest(None), "live_commodities": get_live(None), "currency": currency_stream.get_latest(), "gold": currency_stream.get_gold(), "coin": currency_stream.get_coin(), "commodity": get_commodity_latest(), "steel_stocks": steel_stocks.get_latest(), "crypto": crypto_market.get_latest(), "world_economy": world_economy.get_latest(), }, } @app.get("/api/currency") def get_currency(_: object = Depends(require_session)): return currency_stream.get_latest() @app.get("/api/gold") def get_gold(_: object = Depends(require_session)): return currency_stream.get_gold() @app.get("/api/coin") def get_coin(_: object = Depends(require_session)): return currency_stream.get_coin() @app.get("/api/world-economy") def get_world_economy(_: object = Depends(require_session)): return world_economy.get_latest() @app.get("/api/bonds") def get_bonds(_: object = Depends(require_session)): return bonds.get_bonds() @app.get("/api/steel-stocks") def get_steel_stocks(_: object = Depends(require_session)): return steel_stocks.get_latest() @app.get("/api/crypto") def get_crypto(_: object = Depends(require_session)): return crypto_market.get_latest() import ime_metals @app.get("/api/ime/categories") def get_ime_categories_route(_: object = Depends(require_session)): try: return ime_metals.get_ime_categories() except Exception as e: raise HTTPException(status_code=502, detail=f"IME categories error: {e}") @app.get("/api/ime/subcategory/{sub_id}/symbols") def get_ime_sub_symbols_route(sub_id: int, _: object = Depends(require_session)): try: return ime_metals.get_ime_subcategory_symbols(sub_id) except Exception as e: raise HTTPException(status_code=502, detail=f"IME subcategory symbols error: {e}") @app.get("/api/ime/symbol/{symbol_id}") def get_ime_symbol_route(symbol_id: int, _: object = Depends(require_session)): try: return ime_metals.get_ime_symbol_detail(symbol_id) except Exception as e: raise HTTPException(status_code=502, detail=f"IME symbol error: {e}") @app.get("/api/lme/diagram") def get_lme_diagram(field: str = Query("WM_Cu_low"), _: object = Depends(require_session)): try: return scrape_lme.get_entire_diagram_data(field) except Exception as e: raise HTTPException(status_code=502, detail=f"Failed to fetch diagram data: {e}") @app.get("/api/lme/history") def get_lme_history_route(metal: str = Query("Copper"), _: object = Depends(require_session)): try: return scrape_lme.get_metal_history(metal) except Exception as e: raise HTTPException(status_code=502, detail=f"Failed to fetch LME history: {e}") @app.get("/api/lme") def get_lme(_: object = Depends(require_session)): try: return scrape_lme.get_lme_data() except Exception as e: raise HTTPException(status_code=502, detail=f"Failed to fetch LME data: {e}") @app.get("/api/reports") def get_reports(_: object = Depends(require_session)): return reports.get_index() @app.get("/api/reports/files") def get_report_files(folder: str = Query(...), _: object = Depends(require_session)): return reports.list_files(folder) @app.get("/api/reports/file") def get_report_file(folder: str = Query(...), name: str = Query(...), _: object = Depends(require_session)): p = reports.file_path(folder, name) if not p: raise HTTPException(status_code=404, detail="report not found") return FileResponse(p, media_type="application/pdf", content_disposition_type="inline") def _set_session_cookie(response: Response, token: str) -> None: response.set_cookie( key=SESSION_COOKIE, value=token, httponly=True, samesite="lax", secure=False, max_age=auth.SESSION_DAYS * 86400, path="/", ) class RegisterIn(BaseModel): name: str = Field(..., min_length=1, max_length=80) last_name: str = Field(..., min_length=1, max_length=80) organization: str = Field(..., min_length=1, max_length=160) email: str = Field(..., max_length=254) phone: str = Field(..., min_length=6, max_length=24) password: str = Field(..., min_length=8, max_length=128) @app.post("/api/register") def register(data: RegisterIn): try: user_id = auth.create_user( name=f"{data.name.strip()} {data.last_name.strip()}", username=data.phone.strip(), password=data.password, email=data.email.strip().lower(), organization=data.organization.strip(), ) except Exception as e: raise HTTPException(status_code=400, detail=str(e)) return {"ok": True, "id": user_id} class LoginIn(BaseModel): identifier: str = Field(..., min_length=1, max_length=254) password: str = Field(..., min_length=1, max_length=128) @app.post("/api/login") def login(data: LoginIn, response: Response): ident = data.identifier.strip() ident_l = ident.lower() with get_conn() as conn: row = conn.execute( "SELECT id, password_hash, status FROM users " "WHERE lower(username)=? OR lower(email)=? OR username=?", (ident_l, ident_l, ident), ).fetchone() if ( not row or not auth.verify_password(data.password, row["password_hash"]) or row["status"] != "approved" ): raise HTTPException(status_code=401, detail="نام کاربری/شماره یا رمز عبور اشتباه است") _set_session_cookie(response, auth.create_session(row["id"])) return {"ok": True} @app.get("/api/me") def me(request: Request, authorization: str | None = Header(default=None)): token = request.cookies.get(SESSION_COOKIE) if token: row = auth.user_for_token(token) if row: return {"id": row["id"], "name": row["name"], "email": row["email"], "status": row["status"]} bearer = _bearer_token(authorization) or request.query_params.get("api_token") or request.query_params.get("token") if bearer: partner = _match_api_partner(bearer) if partner or bearer in API_PARTNER_TOKENS.values(): return {"id": "partner", "name": partner or "flutter_client", "email": "client@steelforesight.ir", "status": "approved"} row = auth.user_for_token(bearer) if row: return {"id": row["id"], "name": row["name"], "email": row["email"], "status": row["status"]} raise HTTPException(status_code=401, detail="Not logged in") @app.post("/api/logout") def logout(request: Request, response: Response): auth.delete_session(request.cookies.get(SESSION_COOKIE)) response.delete_cookie(SESSION_COOKIE, path="/") return {"ok": True} SSO_SECRET = os.environ.get("SSO_SECRET", "") def _verify_sso_token(t: str): if not SSO_SECRET: return None try: raw_body, _, sig = t.partition(".") expected = hmac.new(SSO_SECRET.encode(), raw_body.encode(), hashlib.sha256).hexdigest() if not hmac.compare_digest(sig, expected): return None payload = json.loads(base64.urlsafe_b64decode(raw_body.encode() + b"==").decode()) return payload except Exception: return None @app.get("/api/sso") def sso_login(t: str = Query(...), response: Response = None): payload = _verify_sso_token(t) if not payload: return RedirectResponse(url="/", status_code=302) email = payload.get("email", "").lower() name = payload.get("name", "کاربر اندیشکده") with get_conn() as conn: row = conn.execute("SELECT id FROM users WHERE lower(email)=?", (email,)).fetchone() if not row: now = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ") conn.execute( "INSERT INTO users(name, username, password_hash, status, created_at, email) " "VALUES(?,?,?,?,?,?)", (name, email, auth.hash_password("sso_login_dummy"), "approved", now, email), ) row = conn.execute("SELECT id FROM users WHERE lower(email)=?", (email,)).fetchone() _set_session_cookie(response, auth.create_session(row["id"])) return RedirectResponse(url="/", status_code=302) data_api = APIRouter(prefix="/api/v1") @data_api.get("/prices/latest") def v1_prices_latest(): return get_prices_latest(None) @data_api.get("/live") def v1_live(): return get_live(None) @data_api.get("/latest") def v1_latest(): return get_latest(None) @data_api.get("/commodity") def v1_commodity(): return get_commodity(None) @data_api.get("/currency") def v1_currency(): return get_currency(None) @data_api.get("/gold") def v1_gold(): return get_gold(None) @data_api.get("/coin") def v1_coin(): return get_coin(None) @data_api.get("/metals/tree") def v1_metals_tree(): return get_tree(None) @data_api.get("/metals/latest") def v1_metals_latest(): return get_latest(None) @data_api.get("/metals/history") def v1_metals_history(title: str = Query(...), days: int = 0): return get_history(title, days, None) @data_api.get("/metals/search") def v1_metals_search(q: str = Query(...)): return search(q, None) @data_api.get("/steel-stocks") def v1_steel_stocks(): return get_steel_stocks(None) @data_api.get("/crypto") def v1_crypto(): return get_crypto(None) @data_api.get("/world-economy") def v1_world_economy(): return get_world_economy(None) @data_api.get("/ime/categories") def v1_ime_categories(): return ime_metals.get_ime_categories() @data_api.get("/ime/symbol/{symbol_id}") def v1_ime_symbol(symbol_id: int): return ime_metals.get_ime_symbol_detail(symbol_id) @data_api.get("/lme/diagram") def v1_lme_diagram(field: str = "WM_Cu_low"): return scrape_lme.get_entire_diagram_data(field) @data_api.get("/lme/history") def v1_lme_history(metal: str = "Copper"): return scrape_lme.get_metal_history(metal) @data_api.get("/lme") def v1_lme(): return scrape_lme.get_lme_data() app.include_router(data_api)