import os from collections import defaultdict from importlib import import_module from dotenv import load_dotenv from pathlib import Path env_path = Path(__file__).resolve().parent / ".env" load_dotenv(dotenv_path=env_path) DB_HOST = os.getenv("DB_HOST", "127.0.0.1") DB_PORT = int(os.getenv("DB_PORT", "3306")) DB_NAME = os.getenv("DB_NAME", "qaclide") DB_USER = os.getenv("DB_USER", "") DB_PASSWORD = os.getenv("DB_PASSWORD", "") DB_TABLE = os.getenv("DROPDOWN_TABLE", "").strip() # Explicit field SQL queries mapping DROPDOWN_QUERIES = { "Company": "SELECT DISTINCT company_name FROM clide_company_master WHERE company_name IS NOT NULL AND soft_delete=0 AND company_name != '' ORDER BY company_name", "Service": "SELECT DISTINCT stream_name FROM clide_stream WHERE stream_name IS NOT NULL AND stream_name != '' ORDER BY stream_name", "PlantProjectClient": "SELECT DISTINCT sub_stream_name FROM clide_sub_stream WHERE sub_stream_name IS NOT NULL AND sub_stream_name != '' ORDER BY sub_stream_name", "Department": "SELECT DISTINCT department FROM clide_company_department WHERE department IS NOT NULL AND department != '' ORDER BY department", "Contractor": "SELECT DISTINCT contractor FROM clide_company_contractor WHERE contractor IS NOT NULL AND contractor != '' ORDER BY contractor", "Zone": "SELECT DISTINCT sub_stream_zone_name FROM clide_sub_stream_zone WHERE sub_stream_zone_name IS NOT NULL AND sub_stream_zone_name != '' ORDER BY sub_stream_zone_name", "Location": "SELECT DISTINCT sub_stream_zone_location_name FROM clide_sub_stream_zone_location WHERE sub_stream_zone_location_name IS NOT NULL AND sub_stream_zone_location_name != '' ORDER BY sub_stream_zone_location_name", "Activity": "SELECT DISTINCT activity FROM clide_sub_stream_hira_activity WHERE activity IS NOT NULL AND activity != '' ORDER BY activity", "SubActivity": "SELECT DISTINCT name FROM sub_activity WHERE name IS NOT NULL AND name != '' ORDER BY name", "Hazard": "SELECT DISTINCT name FROM hazard WHERE name IS NOT NULL AND name != '' ORDER BY name", "SubHazard": "SELECT DISTINCT name FROM sub_hazard WHERE name IS NOT NULL AND name != '' ORDER BY name", "SourceRules": "SELECT DISTINCT source FROM clide_company_sourcemaster WHERE source IS NOT NULL AND source != '' ORDER BY source", "ControlMeasureViolations": "SELECT DISTINCT name FROM existing_control_measure WHERE name IS NOT NULL AND name != '' ORDER BY name", "ResolveRights": "SELECT DISTINCT em.display_name FROM clide_module_oiac_closing_rights_category_user cru JOIN clide_module_oiac_users ou ON cru.oiac_closing_rights_user_id = ou.oiac_closing_rights_user_id JOIN clide_user cu ON ou.user_id = cu.user_id JOIN employee_master em ON cu.employee_master_id = em.id WHERE em.display_name IS NOT NULL AND em.display_name != '' ORDER BY em.display_name", } # Static dropdowns explicitly specified by user STATIC_DROPDOWNS = { "UAUCType": ["UA", "UC"], "RiskLevel": ["Extreme", "High", "Medium", "Low"], } def _load_pymysql(): try: pymysql = import_module("pymysql") dict_cursor = import_module("pymysql.cursors").DictCursor return pymysql, dict_cursor except ModuleNotFoundError as exc: raise RuntimeError( "PyMySQL is required to read dropdown values from the QA MySQL database. " "Install the backend requirements before starting the app." ) from exc def get_connection(): pymysql, dict_cursor = _load_pymysql() return pymysql.connect( host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, database=DB_NAME, cursorclass=dict_cursor, charset="utf8mb4", autocommit=False, ) def init_db(): try: conn = get_connection() try: with conn.cursor() as cur: cur.execute("SELECT 1") print(f"[database] Successfully connected to database '{DB_NAME}' at {DB_HOST}:{DB_PORT}.") finally: conn.close() except Exception as err: print(f"[database ERROR] Could not connect to QA database '{DB_NAME}' ({err}).") def get_all_dropdowns() -> dict: result: dict[str, list[str]] = {} # 1. Apply static dropdowns for field, options in STATIC_DROPDOWNS.items(): result[field] = options # 2. Query dynamic fields from MySQL try: conn = get_connection() try: with conn.cursor() as cur: for field, query in DROPDOWN_QUERIES.items(): try: cur.execute(query) rows = cur.fetchall() values = [list(r.values())[0] for r in rows if r and list(r.values())[0]] result[field] = values except Exception as q_err: print(f"[database ERROR] Query for field '{field}' failed: {q_err}") result[field] = [] finally: conn.close() except Exception as err: print(f"[database ERROR] Failed connecting to DB: {err}") return result def resolve_related_fields(data: dict) -> dict: location = data.get("Location") zone = data.get("Zone") plant = data.get("PlantProjectClient") sub_hazard = data.get("SubHazard") sub_activity = data.get("SubActivity") try: conn = get_connection() try: with conn.cursor() as cur: # 1. Resolve Location -> Zone, Plant, Service if location and (not zone or not plant or not data.get("Service")): query = """ SELECT stream_id, sub_stream_id, sub_stream_zone_id FROM clide_sub_stream_zone_location WHERE sub_stream_zone_location_name = %s LIMIT 1 """ cur.execute(query, (location,)) row = cur.fetchone() if row: stream_id = row.get("stream_id") sub_stream_id = row.get("sub_stream_id") zone_id = row.get("sub_stream_zone_id") if zone_id and not zone: cur.execute("SELECT sub_stream_zone_name FROM clide_sub_stream_zone WHERE sub_stream_zone_id = %s LIMIT 1", (zone_id,)) z_row = cur.fetchone() if z_row: data["Zone"] = list(z_row.values())[0] zone = data["Zone"] if sub_stream_id and not plant: cur.execute("SELECT sub_stream_name FROM clide_sub_stream WHERE sub_stream_id = %s LIMIT 1", (sub_stream_id,)) p_row = cur.fetchone() if p_row: data["PlantProjectClient"] = list(p_row.values())[0] plant = data["PlantProjectClient"] if stream_id and not data.get("Service"): cur.execute("SELECT stream_name FROM clide_stream WHERE stream_id = %s LIMIT 1", (stream_id,)) s_row = cur.fetchone() if s_row: data["Service"] = list(s_row.values())[0] # 2. Resolve Zone -> Plant, Service if zone and (not plant or not data.get("Service")): query = """ SELECT stream_id, sub_stream_id FROM clide_sub_stream_zone WHERE sub_stream_zone_name = %s LIMIT 1 """ cur.execute(query, (zone,)) row = cur.fetchone() if row: stream_id = row.get("stream_id") sub_stream_id = row.get("sub_stream_id") if sub_stream_id and not plant: cur.execute("SELECT sub_stream_name FROM clide_sub_stream WHERE sub_stream_id = %s LIMIT 1", (sub_stream_id,)) p_row = cur.fetchone() if p_row: data["PlantProjectClient"] = list(p_row.values())[0] plant = data["PlantProjectClient"] if stream_id and not data.get("Service"): cur.execute("SELECT stream_name FROM clide_stream WHERE stream_id = %s LIMIT 1", (stream_id,)) s_row = cur.fetchone() if s_row: data["Service"] = list(s_row.values())[0] # 3. Resolve Plant -> Service, Region, State, City, ProjectCategory, Company if plant: query = """ SELECT stream_id, company_id, region, sub_stream_state, sub_stream_city, project_category_id FROM clide_sub_stream WHERE sub_stream_name = %s LIMIT 1 """ cur.execute(query, (plant,)) row = cur.fetchone() if row: stream_id = row.get("stream_id") company_id = row.get("company_id") region = row.get("region") state = row.get("sub_stream_state") city = row.get("sub_stream_city") pc_id = row.get("project_category_id") if stream_id and not data.get("Service"): cur.execute("SELECT stream_name FROM clide_stream WHERE stream_id = %s LIMIT 1", (stream_id,)) s_row = cur.fetchone() if s_row: data["Service"] = list(s_row.values())[0] if company_id and not data.get("Company"): cur.execute("SELECT company_name FROM clide_company_master WHERE company_id = %s LIMIT 1", (company_id,)) c_row = cur.fetchone() if c_row: data["Company"] = list(c_row.values())[0] data["Region"] = region or "" data["State"] = state or "" data["City"] = city or "" if pc_id: cur.execute("SELECT project_category_name FROM clide_project_category WHERE project_category_id = %s LIMIT 1", (pc_id,)) pc_row = cur.fetchone() if pc_row: data["ProjectCategory"] = list(pc_row.values())[0] else: data["ProjectCategory"] = "" else: data["ProjectCategory"] = "" # 4. Resolve SubHazard -> Hazard if sub_hazard and not data.get("Hazard"): query = """ SELECT hazard_id FROM sub_hazard WHERE name = %s LIMIT 1 """ cur.execute(query, (sub_hazard,)) row = cur.fetchone() if row: h_id = row.get("hazard_id") if h_id: cur.execute("SELECT name FROM hazard WHERE id = %s LIMIT 1", (h_id,)) h_row = cur.fetchone() if h_row: data["Hazard"] = list(h_row.values())[0] # 5. Resolve SubActivity -> Activity if sub_activity and not data.get("Activity"): query = """ SELECT activity_id FROM sub_activity WHERE name = %s LIMIT 1 """ cur.execute(query, (sub_activity,)) row = cur.fetchone() if row: act_id = row.get("activity_id") if act_id: cur.execute("SELECT activity FROM clide_sub_stream_hira_activity WHERE activity_master_id = %s LIMIT 1", (act_id,)) act_row = cur.fetchone() if act_row: data["Activity"] = list(act_row.values())[0] # 6. Resolve Service -> Company if Company is empty if data.get("Service") and not data.get("Company"): cur.execute("SELECT stream_id FROM clide_stream WHERE stream_name = %s LIMIT 1", (data.get("Service"),)) s_row = cur.fetchone() if s_row: stream_id = s_row.get("stream_id") if stream_id: cur.execute(""" SELECT c.company_name FROM clide_company_stream cs JOIN clide_company_master c ON cs.company_id = c.company_id WHERE cs.stream_id = %s AND cs.soft_delete = 0 LIMIT 1 """, (stream_id,)) c_row = cur.fetchone() if c_row: data["Company"] = c_row.get("company_name") finally: conn.close() except Exception as err: print(f"[database ERROR] resolve_related_fields failed: {err}") return data