import re from fastapi import FastAPI, Response import json from fastapi.middleware.cors import CORSMiddleware import os from sklearn.metrics.pairwise import cosine_similarity from schemas import PredictionRequest from utils import get_ollama_embedding, parse_combined_text_to_dict app = FastAPI( title="preparedness-api", version="0.0.1", description="General API for the preparedness campaign.", contact={ "email": "contact@jamestwose.com", }, # lifespan=lifespan, ) # Whitelist: only these variables will be exposed to the client form renderer. # Update this list when you want to add/remove fields shown in the UI. WHITELIST = { "risks_cntry_most_exposed_to_firstly", "risks_pers_most_exposed_to_firstly", "risks_pers_most_exposed_to_number_of_mentioned_risks", "pot_info_sources_to_learn_about_disaster_risks_firstly", "statements_disaster_risks_readseenheard_info_in_last_12m", "statements_disaster_risks_feel_well_informed", "statements_disaster_risks_trust_information_by_pub_auth_on_risks_where_you_live", "statements_disaster_risks_easy_to_find_information_by_pub_auth_on_risks_where_you_live", "statements_disaster_risks_know_where_to_find_info_when_travelling_to_oth_eu_cntry", "disaster_measures_in_hh_emergency_supply_drinksfood", "disaster_measures_in_hh_emergency_supply_water_cookinghygiene", "disaster_measures_in_hh_flashlightcandles", "disaster_measures_in_hh_batterypowered_radio", "disaster_measures_in_hh_emergency_pharmacy", "disaster_measures_in_hh_copies_imp_documentsstored_safely", "disaster_measures_in_hh_emergency_grabbag", "disaster_measures_in_hh_signed_up_for_alerts", "disaster_measures_in_hh_participated_in_trainingexercise", "disaster_measures_in_hh_informed_about_official_response_plan", "disaster_measures_in_hh_agreed_with_friendsfamily_to_contact", "disaster_measures_in_hh_discussed_common_prot_measures_in_neighbourhood", "disaster_measures_in_hh_invested_in_prot_measures_in_home", "how_many_days_meet_water_needs_if_water_services_disrupted", "how_many_days_power_essent_appliances_if_elec_interrupted", "how_many_days_cook_mealsheat_if_gas_disrupted", "how_many_days_provide_food_if_transportation_disrupted", "how_many_days_continued_treatment_if_medication_supply_disrupted", "personal_disaster_preparedness_better_able_to_cope_by_prep", "personal_disaster_preparedness_feel_well_prepared", "personal_disaster_preparedness_no_timefin_resources_to_prep", "personal_disaster_preparedness_easy_to_find_info_on_how_to_prep", "personal_disaster_preparedness_need_more_info_to_prep", "personal_disaster_preparedness_know_how_emerg_services_will_alert", "personal_disaster_preparedness_know_what_to_do_in_event_of_disaster", "personal_disaster_preparedness_employerschool_encourages_trainingprep", "personal_disaster_preparedness_emerg_services_encourage_trainingprep", "relying_on_in_first_days_of_disaster_familyfriends", "relying_on_in_first_days_of_disaster_people_in_neighbourhood", "relying_on_in_first_days_of_disaster_assocsnonprofit_orgs", "relying_on_in_first_days_of_disaster_emerg_services", "relying_on_in_first_days_of_disaster_local_authgvmt_services", "relying_on_in_first_days_of_disaster_workemployerschooledu_institution", "relying_on_in_first_days_of_disaster_private_sector_entities", "trust_in_emerg_services_to_handle_disastersemerg_situations_properly", "engaging_in_voluntary_work_for_emerg_responder_orgs", "occupation_of_respondent", "age_recoded_6_categories", "size_of_community", "direction_things_are_going_life_personally", "political_discussion_local_matters", "political_discussion_national_matters", "internet_use_total", "gender", "age_education", "standard_of_living_last_5yrs_in_light_of_crises", "personal_living_conditions_in_one_years_time", "standard_of_living_next_5yrs", # region and education levels are included but will often be null in examples "region_spain", "region_austria", "region_belgium", "education_level_bachelor_or_equivalent", } def format_column_name(col: str) -> str: """Format a column name to match the style used in the WHITELIST.""" col = re.sub(r"[.\s]+", "_", col) col = re.sub(r"[^0-9a-zA-Z_]", "", col) col = col.lower() col = re.sub(r"__+", "_", col) # Replace double (or more) underscores with single col = col.strip("_") # Remove leading/trailing underscores return col origins = ["*"] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/") async def root(): template_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), "preparedness/templates", "user_survey.html", ) try: with open(template_path, "r", encoding="utf-8") as f: html_content = f.read() return Response(content=html_content, media_type="text/html") except Exception as e: return Response( content=f"Error loading template: {e}", media_type="text/plain", status_code=500, ) @app.post("/predict") async def predict(request: PredictionRequest): example_best_user = "risks_cntry_most_exposed_to_firstly: Terrorist attacks | risks_pers_most_exposed_to_firstly: Extreme weather events (violent storms, droughts, heatwaves, cold waves, etc.) | risks_pers_most_exposed_to_number_of_mentioned_risks: 2 mentions | pot_info_sources_to_learn_about_disaster_risks_firstly: National media | statements_disaster_risks_readseenheard_info_in_last_12m: Tend to agree | statements_disaster_risks_feel_well_informed: Tend to agree | statements_disaster_risks_trust_information_by_pub_auth_on_risks_where_you_live: Totally agree | statements_disaster_risks_easy_to_find_information_by_pub_auth_on_risks_where_you_live: Tend to agree | statements_disaster_risks_know_where_to_find_info_when_travelling_to_oth_eu_cntry: Totally agree | disaster_measures_in_hh_emergency_supply_drinksfood: Keep an emergency supply stock/pack of drinks, food | disaster_measures_in_hh_emergency_supply_water_cookinghygiene: Keep an emergency supply of water for cooking and hygiene | disaster_measures_in_hh_flashlightcandles: Have flashlight or candles accessible | disaster_measures_in_hh_batterypowered_radio: Have a battery-powered radio accessible | disaster_measures_in_hh_emergency_pharmacy: Keep a home pharmacy for emergencies | disaster_measures_in_hh_copies_imp_documentsstored_safely: Have made sure you have copies of your most important documents or have stored them safely | disaster_measures_in_hh_emergency_grabbag: Have prepared a grab-bag, in case you need to evacuate rapidly in an emergency | disaster_measures_in_hh_signed_up_for_alerts: Have signed up for alerts and warnings from emergency services or authorities | disaster_measures_in_hh_participated_in_trainingexercise: Have participated in a training or exercise, to learn how to react in an emergency | disaster_measures_in_hh_informed_about_official_response_plan: Got informed on the response plan your city, region or country has for a disaster or emergency (e.g. (...) | disaster_measures_in_hh_agreed_with_friendsfamily_to_contact: Agreed with family, friends on how to contact each other in case of an emergency | disaster_measures_in_hh_discussed_common_prot_measures_in_neighbourhood: Discussed common protective measures in your neighbourhood | disaster_measures_in_hh_invested_in_prot_measures_in_home: Have invested in protective measures in your home (e.g. flood-proofed the electricity installation, cleared (...) | how_many_days_meet_water_needs_if_water_services_disrupted: More than 7 days | how_many_days_power_essent_appliances_if_elec_interrupted: More than 7 days | how_many_days_cook_mealsheat_if_gas_disrupted: More than 7 days | how_many_days_provide_food_if_transportation_disrupted: More than 7 days | how_many_days_continued_treatment_if_medication_supply_disrupted: More than 7 days | personal_disaster_preparedness_better_able_to_cope_by_prep: Totally agree | personal_disaster_preparedness_feel_well_prepared: Tend to agree | personal_disaster_preparedness_no_timefin_resources_to_prep: Tend to disagree | personal_disaster_preparedness_easy_to_find_info_on_how_to_prep: Tend to agree | personal_disaster_preparedness_need_more_info_to_prep: Totally agree | personal_disaster_preparedness_know_how_emerg_services_will_alert: Tend to agree | personal_disaster_preparedness_know_what_to_do_in_event_of_disaster: Tend to agree | personal_disaster_preparedness_employerschool_encourages_trainingprep: Totally disagree | personal_disaster_preparedness_emerg_services_encourage_trainingprep: Tend to agree | relying_on_in_first_days_of_disaster_familyfriends: A great deal | relying_on_in_first_days_of_disaster_people_in_neighbourhood: A great deal | relying_on_in_first_days_of_disaster_assocsnonprofit_orgs: A great deal | relying_on_in_first_days_of_disaster_emerg_services: A great deal | relying_on_in_first_days_of_disaster_local_authgvmt_services: A great deal | relying_on_in_first_days_of_disaster_workemployerschooledu_institution: Not a lot | relying_on_in_first_days_of_disaster_private_sector_entities: Not a lot | trust_in_emerg_services_to_handle_disastersemerg_situations_properly: Tend to trust | engaging_in_voluntary_work_for_emerg_responder_orgs: No, you have never engaged in voluntary work and do not plan to do so | occupation_of_respondent: Skilled manual worker | age_recoded_6_categories: 35-44 | size_of_community: Towns/suburbs | direction_things_are_going_life_personally: Things are going in the right direction | political_discussion_local_matters: Never | political_discussion_national_matters: Never | internet_use_total: Everyday/almost everyday (at least once 1 in d62_1 to d62_4) | gender: Woman | age_education: 23.0 | standard_of_living_last_5yrs_in_light_of_crises: Your standard of living has not changed | personal_living_conditions_in_one_years_time: Worse | standard_of_living_next_5yrs: Your standard of living will not change | region_austria: nan | region_belgium: nan | region_bulgaria: nan | region_croatia: nan | region_cyprus: nan | region_czechia: nan | region_denmark: nan | region_germany: nan | region_estonia: nan | region_finland: nan | region_france: nan | region_greece: nan | region_hungary: nan | region_ireland: nan | region_italy: nan | region_latvia: nan | region_lithuania: nan | region_luxembourg: nan | region_malta: nan | region_netherlands: nan | region_poland: nan | region_portugal: nan | region_romania: nan | region_slovenia: nan | region_slovakia: nan | region_spain: ES61 - Andalucia | region_sweden: nan | education_level_preprimary_education_incl_no_education: Not mentioned | education_level_primary_education: Not mentioned | education_level_lower_secondary_education: Not mentioned | education_level_upper_secondary_education: Not mentioned | education_level_postsecondary_non_tertiary_incl_prevocationalvocational: Not mentioned | education_level_shortcycle_tertiary: Not mentioned | education_level_bachelor_or_equivalent: Bachelor or equivalent | education_level_master_or_equivalent: Not mentioned | education_level_doctoral_or_equivalent: Not mentioned" # parse combined example and request text into dicts for field-wise comparison example_dict = parse_combined_text_to_dict(example_best_user) # Load metadata and build an inverted mapping from internal_label -> canonical variable id meta_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), "preparedness", "data", "za8841_meta.json", ) try: with open(meta_path, "r", encoding="utf-8") as f: meta = json.load(f) var_labels = meta.get("column_names_to_labels", {}) col_names = meta.get("column_names", []) # Build mapping from internal_label -> canonical id using the var_labels and column list internal_to_id = {} for col in col_names: human_label = var_labels.get(col) if isinstance(human_label, str): internal = format_column_name(human_label) internal_to_id[internal] = col except Exception: internal_to_id = {} request_parsed_internal = parse_combined_text_to_dict(request.text) # # Map internal_label keys back to canonical variable ids when possible # request_dict = {} # for ik, val in request_parsed_internal.items(): # # prefer exact match by internal label # varid = internal_to_id.get(ik) # if not varid: # # try case-insensitive normalized match # for internal_label, cid in internal_to_id.items(): # if internal_label.strip().lower() == ik.strip().lower(): # varid = cid # break # # fall back to the original internal key if we couldn't map # request_dict[varid or ik] = val # compute simple field diffs: keys present in either dict; value equal, different or missing keys = sorted(set(list(example_dict.keys()) + list(request_parsed_internal.keys()))) field_diffs = {} for k in keys: if "region" in k: continue a = example_dict.get(k) b = request_parsed_internal.get(k) if a == b: # field_diffs[k] = {"status": "same", "example": a, "request": b} continue else: field_diffs[k] = {"status": "different", "example": a, "request": b} example_best_user_embedding = get_ollama_embedding(example_best_user) prompt_embedding = get_ollama_embedding(request.text) similarity = float( cosine_similarity([example_best_user_embedding], [prompt_embedding])[0][0] ) return { "similarity": similarity, "example_parsed": example_dict, # "request_parsed": request_dict, "request_parsed_internal_keys": request_parsed_internal, "field_diffs": field_diffs, } @app.get("/value_labels") async def value_labels(): """Return the variable-level value labels extracted from the za8841 metadata JSON. This endpoint returns the `variable_value_labels` section as a JSON object so the client can use canonical, human-friendly labels for variables where available. """ meta_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), "preparedness", "data", "za8841_meta.json", ) try: with open(meta_path, "r", encoding="utf-8") as f: meta = json.load(f) variable_value_labels = meta.get("variable_value_labels", {}) # filter to whitelist filtered = {k: v for k, v in variable_value_labels.items() if k in WHITELIST} return filtered except Exception as e: return Response( content=json.dumps({"error": str(e)}), media_type="application/json", status_code=500, ) @app.get("/variable_map") async def variable_map(): """Return a mapping from human-readable variable label -> variable id. The source JSON contains a mapping of variable id -> human label (e.g. "qc5_3": "STATEMENTS ..."). This endpoint inverts that mapping so the client can look up the variable id by visible label. """ meta_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), "preparedness", "data", "za8841_meta.json", ) try: with open(meta_path, "r", encoding="utf-8") as f: meta = json.load(f) var_labels = meta.get("column_names_to_labels", {}) # filter to whitelist filtered = {k: v for k, v in var_labels.items() if k in WHITELIST} # invert: map human label -> variable id inv = {v: k for k, v in filtered.items() if isinstance(v, str)} return inv except Exception as e: return Response( content=json.dumps({"error": str(e)}), media_type="application/json", status_code=500, ) @app.get("/variables") async def variables(): """Return an ordered list of variable metadata suitable for client-side form rendering. The response format is a list of objects: [{"id": "qc5_3", "label": "STATEMENTS ...", "values": {"1.0": "Totally agree", ...}}, ...] This lets the client render each variable using canonical labels and codes. """ meta_path = os.path.join( os.path.dirname(os.path.dirname(__file__)), "preparedness", "data", "za8841_meta.json", ) try: with open(meta_path, "r", encoding="utf-8") as f: meta = json.load(f) col_names = meta.get("column_names", []) # human-friendly labels live under `column_names_to_labels` in the metadata var_labels = meta.get("column_names_to_labels", {}) value_labels = meta.get("variable_value_labels", {}) vars_out = [] for col in col_names: if format_column_name(var_labels.get(col)) not in WHITELIST: print(format_column_name(var_labels.get(col))) continue item = { "id": col, "internal_label": format_column_name(var_labels.get(col)), "label": var_labels.get(col) or col, "values": value_labels.get(col) or {}, } vars_out.append(item) return vars_out except Exception as e: return Response( content=json.dumps({"error": str(e)}), media_type="application/json", status_code=500, )