#!/usr/bin/env python3
"""Migration: normalize and backfill field_config for all sites.
Run once to:
1. Normalize field keys to lowercase
2. Backfill missing optional fields (validation, condition, calculation, step)
3. Ensure all fields have 'key', 'name', 'label', 'type', 'required'
Usage:
python scripts/migrate_field_configs.py
Dry run:
python scripts/migrate_field_configs.py --dry-run
"""
import argparse
import json
import os
import sqlite3
import sys
from datetime import datetime
# Support running from project root or scripts dir
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
DB_PATH = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "relay.db")
# Type normalization: old values -> new values
TYPE_MAP = {
"tel": "phone",
"text": "text",
"email": "email",
"phone": "phone",
"number": "number",
"select": "select",
"textarea": "textarea",
"checkbox": "checkbox",
"radio": "radio",
"file": "file",
}
# Default type for unrecognized types
DEFAULT_TYPE = "text"
def normalize_field(field):
"""Normalize a single field definition, adding missing defaults."""
normalized = {}
# Key / name - normalize to lowercase, ensure both exist
key = (field.get("key") or field.get("name") or "").lower().strip()
normalized["key"] = key
normalized["name"] = key
# Label
normalized["label"] = field.get("label") or key.replace("_", " ").title()
# Type
raw_type = field.get("type", DEFAULT_TYPE).lower()
normalized["type"] = TYPE_MAP.get(raw_type, DEFAULT_TYPE)
# Required
normalized["required"] = bool(field.get("required", False))
# Optional fields - backfill with defaults
normalized["placeholder"] = field.get("placeholder", "")
normalized["default"] = field.get("default", "")
# Options (for select/radio)
if normalized["type"] in ("select", "radio"):
options = field.get("options", [])
if not options:
# Backfill with generic options if missing
options = ["Option 1", "Option 2", "Option 3"]
normalized["options"] = options
else:
normalized["options"] = []
# Validation
validation = field.get("validation", {})
if not validation:
# Backfill with type-appropriate validation
if normalized["type"] == "email":
validation = {"type": "email"}
elif normalized["type"] == "phone":
validation = {"type": "pattern", "pattern": "^[+]?\\d{7,15}$"}
elif normalized["type"] == "number":
validation = {"type": "numeric"}
normalized["validation"] = validation
# Pattern / error message (legacy support)
if field.get("pattern"):
normalized["pattern"] = field["pattern"]
if field.get("errorMessage"):
normalized["errorMessage"] = field["errorMessage"]
# Numeric range
if field.get("min") is not None:
normalized["min"] = field["min"]
if field.get("max") is not None:
normalized["max"] = field["max"]
if field.get("minLength") or field.get("min_length"):
normalized["minLength"] = field.get("minLength") or field.get("min_length")
if field.get("maxLength") or field.get("max_length"):
normalized["maxLength"] = field.get("maxLength") or field.get("max_length")
# Condition - empty by default
normalized["condition"] = field.get("condition", None)
# Calculation - empty by default
normalized["calculation"] = field.get("calculation", None)
# Step (multi-step)
if field.get("step") is not None:
normalized["step"] = field["step"]
return normalized
def migrate_field_config(fc_json):
"""Normalize a field_config JSON string. Returns (original, normalized) or (original, None) if no changes."""
if not fc_json:
return fc_json, None
fields = json.loads(fc_json)
if not fields:
return fc_json, None
normalized = [normalize_field(f) for f in fields]
new_json = json.dumps(normalized)
# Return None if no changes
if new_json == fc_json:
return fc_json, None
return fc_json, normalized
def main():
parser = argparse.ArgumentParser(description="Migrate field_configs to normalized format")
parser.add_argument("--dry-run", action="store_true", help="Show changes without applying")
args = parser.parse_args()
if not os.path.exists(DB_PATH):
print(f"ERROR: Database not found at {DB_PATH}")
sys.exit(1)
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Get all sites with field_config
sites = cursor.execute(
"SELECT id, token, name, field_config FROM sites WHERE field_config IS NOT NULL"
).fetchall()
changes = []
skipped = []
for site in sites:
original, normalized = migrate_field_config(site["field_config"])
if normalized:
changes.append({
"id": site["id"],
"token": site["token"],
"name": site["name"],
"original": json.loads(original),
"normalized": normalized,
})
else:
skipped.append(site["name"])
# Report
print(f"Migration: field_config normalization")
print(f"Timestamp: {datetime.now().isoformat()}")
print(f"{'[DRY RUN] ' if args.dry_run else ''}")
print(f"\nSites needing update: {len(changes)}")
print(f"Already normalized: {len(skipped)}")
for change in changes:
print(f"\n--- {change['name']} ({change['token']}) ---")
print(f"Fields: {len(change['original'])} -> {len(change['normalized'])}")
for i, (old, new) in enumerate(zip(change["original"], change["normalized"])):
print(f" [{i}] {old.get('key', old.get('name'))} -> {new['key']}")
# Apply if not dry run
if not args.dry_run and changes:
for change in changes:
cursor.execute(
"UPDATE sites SET field_config = ? WHERE id = ?",
(json.dumps(change["normalized"]), change["id"]),
)
conn.commit()
print(f"\n✓ Applied {len(changes)} updates to database.")
elif args.dry_run:
print("\n[Dry run - no changes applied. Remove --dry-run to apply.]")
conn.close()
if __name__ == "__main__":
main()