| """ |
| Storage layer for the Objective O4 evaluation study. |
| |
| Mirrors the resilient pattern already used in Assignment 6's phishing_app: |
| 1. Always write to a local CSV first (fast, always works, no external |
| dependency). |
| 2. Optionally ALSO mirror the write to a Google Sheet, via a service |
| account (not a personal OAuth login) -- only activates if |
| GOOGLE_CREDENTIALS_JSON and GOOGLE_SHEET_ID environment variables are |
| set. Silently skipped otherwise, so the whole study runs fully |
| locally during development, and the student can switch on live Google |
| Sheets sync later just by setting those two env vars once ethics |
| approval is granted and real data collection begins. |
| |
| Participant linking: pretest and posttest responses are matched using a |
| short, random, anonymous code (no personal information), generated at the |
| end of pretest and re-entered by the participant to unlock posttest. See |
| study/consent_ethics/ for the participant-facing explanation of this. |
| """ |
| import csv |
| import json |
| import os |
| import random |
| import string |
| from pathlib import Path |
| from datetime import datetime, timezone |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| RESPONSES_DIR = ROOT / "study" / "responses" |
| RESPONSES_DIR.mkdir(parents=True, exist_ok=True) |
|
|
| PRETEST_CSV = RESPONSES_DIR / "pretest_responses.csv" |
| POSTTEST_CSV = RESPONSES_DIR / "posttest_responses.csv" |
|
|
| PRETEST_FIELDNAMES = [ |
| "code", "timestamp_utc", |
| "age_group", "programme", "year_of_study", |
| "prior_training", "prior_phishing_experience", "self_knowledge_rating", |
| "pretest_emails_json", |
| ] |
|
|
| POSTTEST_FIELDNAMES = [ |
| "code", "timestamp_utc", |
| "posttest_emails_json", |
| "sus_1", "sus_2", "sus_3", "sus_4", "sus_5", |
| "sus_6", "sus_7", "sus_8", "sus_9", "sus_10", |
| "tlx_mental_demand", "tlx_physical_demand", "tlx_temporal_demand", |
| "tlx_performance", "tlx_effort", "tlx_frustration", |
| ] |
|
|
| CODE_CHARSET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789" |
|
|
|
|
| def _ensure_csv(path: Path, fieldnames: list): |
| if not path.exists(): |
| with open(path, "w", newline="", encoding="utf-8") as f: |
| csv.DictWriter(f, fieldnames=fieldnames).writeheader() |
|
|
|
|
| def _existing_codes(path: Path, fieldnames: list) -> set: |
| _ensure_csv(path, fieldnames) |
| with open(path, encoding="utf-8") as f: |
| reader = csv.DictReader(f) |
| return {row["code"] for row in reader} |
|
|
|
|
| def generate_unique_code() -> str: |
| existing = _existing_codes(PRETEST_CSV, PRETEST_FIELDNAMES) |
| while True: |
| code = "".join(random.choices(CODE_CHARSET, k=8)) |
| if code not in existing: |
| return code |
|
|
|
|
| def code_status(code: str) -> dict: |
| """Returns whether a code exists in pretest and whether posttest is already complete.""" |
| pretest_codes = _existing_codes(PRETEST_CSV, PRETEST_FIELDNAMES) |
| posttest_codes = _existing_codes(POSTTEST_CSV, POSTTEST_FIELDNAMES) |
| exists = code in pretest_codes |
| already_completed = code in posttest_codes |
| return {"exists": exists, "already_completed": already_completed} |
|
|
|
|
| |
| |
| |
| _gs_client = None |
| _gs_spreadsheet = None |
|
|
|
|
| def _get_gs_spreadsheet(): |
| global _gs_client, _gs_spreadsheet |
| if _gs_spreadsheet is not None: |
| return _gs_spreadsheet |
|
|
| creds_json = os.environ.get("GOOGLE_CREDENTIALS_JSON") |
| sheet_id = os.environ.get("GOOGLE_SHEET_ID") |
| if not creds_json or not sheet_id: |
| return None |
|
|
| try: |
| import gspread |
| from google.oauth2.service_account import Credentials |
|
|
| scopes = [ |
| "https://www.googleapis.com/auth/spreadsheets", |
| "https://www.googleapis.com/auth/drive", |
| ] |
| creds = Credentials.from_service_account_info(json.loads(creds_json), scopes=scopes) |
| client = gspread.authorize(creds) |
| _gs_client = client |
| _gs_spreadsheet = client.open_by_key(sheet_id) |
| return _gs_spreadsheet |
| except Exception as e: |
| print(f"[Google Sheets] Connection failed, continuing with local CSV only: {e}") |
| return None |
|
|
|
|
| def _mirror_to_sheet(tab_name: str, fieldnames: list, row: dict): |
| spreadsheet = _get_gs_spreadsheet() |
| if spreadsheet is None: |
| return |
| try: |
| try: |
| ws = spreadsheet.worksheet(tab_name) |
| except Exception: |
| ws = spreadsheet.add_worksheet(title=tab_name, rows=1000, cols=len(fieldnames)) |
| ws.append_row(fieldnames) |
| ws.append_row([str(row.get(col, "")) for col in fieldnames], value_input_option="USER_ENTERED") |
| except Exception as e: |
| print(f"[Google Sheets] Mirror write to '{tab_name}' failed: {e}") |
|
|
|
|
| |
| |
| |
| def save_pretest(demographics: dict, pretest_emails: list) -> str: |
| """Stores a pretest submission and returns the newly generated participant code.""" |
| code = generate_unique_code() |
| row = { |
| "code": code, |
| "timestamp_utc": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), |
| "age_group": demographics.get("age_group", ""), |
| "programme": demographics.get("programme", ""), |
| "year_of_study": demographics.get("year_of_study", ""), |
| "prior_training": demographics.get("prior_training", ""), |
| "prior_phishing_experience": demographics.get("prior_phishing_experience", ""), |
| "self_knowledge_rating": demographics.get("self_knowledge_rating", ""), |
| "pretest_emails_json": json.dumps(pretest_emails), |
| } |
| _ensure_csv(PRETEST_CSV, PRETEST_FIELDNAMES) |
| with open(PRETEST_CSV, "a", newline="", encoding="utf-8") as f: |
| csv.DictWriter(f, fieldnames=PRETEST_FIELDNAMES, extrasaction="ignore").writerow(row) |
| _mirror_to_sheet("PretestResponses", PRETEST_FIELDNAMES, row) |
| return code |
|
|
|
|
| def save_posttest(code: str, posttest_emails: list, sus_responses: list, tlx_responses: dict) -> bool: |
| """Stores a posttest submission. Returns False if the code is invalid or already used.""" |
| status = code_status(code) |
| if not status["exists"] or status["already_completed"]: |
| return False |
|
|
| row = { |
| "code": code, |
| "timestamp_utc": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), |
| "posttest_emails_json": json.dumps(posttest_emails), |
| } |
| for i in range(10): |
| row[f"sus_{i + 1}"] = sus_responses[i] if i < len(sus_responses) else "" |
| for key in ["mental_demand", "physical_demand", "temporal_demand", "performance", "effort", "frustration"]: |
| row[f"tlx_{key}"] = tlx_responses.get(key, "") |
|
|
| _ensure_csv(POSTTEST_CSV, POSTTEST_FIELDNAMES) |
| with open(POSTTEST_CSV, "a", newline="", encoding="utf-8") as f: |
| csv.DictWriter(f, fieldnames=POSTTEST_FIELDNAMES, extrasaction="ignore").writerow(row) |
| _mirror_to_sheet("PosttestResponses", POSTTEST_FIELDNAMES, row) |
| return True |
|
|
