"""User bootstrap, JWT, and KPI helper functions. Extracted from app.py so blueprints can import without circular dependencies. All Redis access goes through ``extensions.redis_client``. """ import json import logging import os import random import threading import time import pandas as pd import requests import extensions as ext logger = logging.getLogger("apex") # Resolve apex/frontend/static/user.xlsx regardless of working directory APEX_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) _STATIC_ROOT = os.path.join(APEX_ROOT, "frontend", "static") _USER_XLSX = os.path.join(_STATIC_ROOT, "user.xlsx") _REDIS_EXPIRY = int(os.getenv("REDIS_EXPIRY", "86400")) _JWT_EXPIRY = int(os.getenv("JWT_EXPIRY", "1382400")) def get_jwt_token(email: str) -> str | None: key = f"jwt:{email}" try: jwt_data = ext.redis_client.get(key) if jwt_data: payload = json.loads(jwt_data) token = payload.get("token") if token: return token except Exception as exc: logger.warning("[JWT] Redis read failed for %s: %s", email, exc) otp = str(random.randint(100000, 999999)) try: res = requests.post( "https://misapi.rptechindia.com/api/JwtToken/CreateJWTToken", headers={"Content-Type": "application/json"}, json={"email": email, "otp": otp}, timeout=10, ).json() if res.get("Success") and "Responsedata" in res: token = res["Responsedata"] ext.redis_client.set( key, json.dumps({"token": token, "timestamp": time.time()}), ex=_JWT_EXPIRY, ) return token logger.warning("[JWT] API returned non-success for %s: %s", email, res) except Exception as exc: logger.error("[JWT] Token creation failed for %s: %s", email, exc) return None def get_user_company_data(email: str, jwt_token: str | None = None) -> dict: try: if not os.path.exists(_USER_XLSX): logger.warning("[UserData] user.xlsx not found at %s", _USER_XLSX) df = pd.DataFrame(columns=["email id"]) else: df = pd.read_excel(_USER_XLSX) df.columns = [c.strip().lower() for c in df.columns] user_row = df[df["email id"].str.lower() == (email or "").lower()] allowed_filters = [] if not user_row.empty: row = user_row.iloc[0] flag_map = { "company": "COMPANY_CODE", "branch": "PLANT", "mgr1": "MAT_GRP_NAME", "pgr": "BRANDCODE", "mgr": "MATERIALGROUP", } for col, db_col in flag_map.items(): if str(row.get(col, "")).strip().upper() == "X": allowed_filters.append(db_col) else: allowed_filters = ["COMPANY_CODE", "PLANT", "MAT_GRP_NAME", "BRANDCODE", "MATERIALGROUP"] headers = {"Content-Type": "application/json"} if jwt_token: headers["Authorization"] = f"Bearer {jwt_token}" res = requests.post( "https://misapi.rptechindia.com/api/Master/UserInfo", headers=headers, json={"token": "rpt", "userid": "", "id": "", "querytype": "2", "search": email, "active": "", "menuid": ""}, ).json() if not res.get("Success") or not res.get("Responsedata"): return {"DESIGNATIONNAME": [""]} user_id = res["Responsedata"][0]["USERID"] dn_raw = res["Responsedata"][0].get("DESIGNATIONNAME") dn = [dn_raw] if isinstance(dn_raw, str) else (dn_raw if isinstance(dn_raw, list) else []) comp_res = requests.post( "https://misapi.rptechindia.com/api/TargetAchevement/Company", headers=headers, json={"tokan": "rpt", "querytype": "13", "id1": "1,0", "id2": "", "id3": user_id, "search": "", "userid": ""}, ).json() if not comp_res.get("Success"): return {"COMPANY_CODE": ["XXXX"], "DESIGNATIONNAME": dn} mgr1_map = { "P01": "Test and Measurement", "P02": "Fire Safety and Security", "P03": "LAN Infrastructure", "P04": "Business Productivity", "P05": "Other", "Z01": "COMPONENTS", "Z02": "ENTERPRISE", "Z03": "CAREPACK", "Z04": "MOBILE", "Z05": "NETWORKING", "Z06": "ONLINE", "Z07": "OTHER", "Z08": "PCD", "Z09": "PERIPHERALS", "Z10": "RMA", "Z11": "SOFTWARE", "Z12": "EMBEDDED", "Z13": "POWER", "Z14": "MEMORY / STORAGE", "Z15": "LIFESTYLE", "Z16": "APPLE", "Z17": "CONSUMER ELECTRONICS", } column_map = { "COMPANY_CODE": "company", "PLANT": "branchcode", "MAT_GRP_NAME": "mgr1code", "BRANDCODE": "pgrcode", "MATERIALGROUP": "mgrcode", } result: dict = {} for db_col, api_key in column_map.items(): if db_col not in allowed_filters: continue values = set() for d in comp_res.get("Responsedata", []): val = d.get(api_key) if not val: continue if api_key == "mgr1code": val = mgr1_map.get(val, val) values.add(val) if values: result[db_col] = list(values) result["DESIGNATIONNAME"] = dn return result except Exception as exc: logger.error("[UserData] get_user_company_data failed for %s: %s", email, exc) return {} def get_user_where_clause(email: str) -> str: try: cached = ext.redis_client.get(email) if not cached: return "" user_data = json.loads(cached) conditions = [] for col, vals in user_data.items(): if not vals or col.upper() == "DESIGNATIONNAME": continue if col.upper() == "PLANT" and "2000" in vals: continue safe_vals = [ "'" + str(v).replace("'", "''") + "'" for v in vals if v is not None and str(v).strip() ] if safe_vals: conditions.append(f"{col} IN ({', '.join(safe_vals)})") if conditions: return " AND " + " AND ".join(conditions) except Exception as exc: logger.error("[UserData] get_user_where_clause failed for %s: %s", email, exc) return "" def _refresh_authenticated_user_cache(email: str) -> None: from cells import fetch_cells_data # local import — cells module lives at apex root try: jwt_token = get_jwt_token(email) user_data = get_user_company_data(email, jwt_token=jwt_token) if user_data: ext.redis_client.set(email, json.dumps(user_data), ex=_REDIS_EXPIRY) except Exception as exc: logger.error("[Bootstrap] User data refresh failed for %s: %s", email, exc) try: external_where = get_user_where_clause(email) kpi_data = fetch_cells_data(email, external_where=external_where) ext.redis_client.set(f"kpi:{email}", json.dumps(kpi_data), ex=_REDIS_EXPIRY) except Exception as exc: logger.error("[Bootstrap] KPI data refresh failed for %s: %s", email, exc) ext.redis_client.set(f"kpi:{email}", json.dumps({}), ex=_REDIS_EXPIRY) def start_authenticated_user_bootstrap(email: str) -> None: if not email: return threading.Thread( target=_refresh_authenticated_user_cache, args=(email,), daemon=True, ).start()