"""Seed baseline data for CalMiner in an idempotent manner. Usage examples -------------- ```powershell # Use existing environment variables (or load from setup_test.env.example) python scripts/seed_data.py --currencies --units --defaults # Dry-run to preview actions python scripts/seed_data.py --currencies --dry-run ``` """ from __future__ import annotations import argparse import logging import os from typing import Iterable, Optional import psycopg2 from psycopg2 import errors from psycopg2.extras import execute_values from scripts.setup_database import DatabaseConfig logger = logging.getLogger(__name__) CURRENCY_SEEDS = ( ("USD", "United States Dollar", "USD$", True), ("EUR", "Euro", "EUR", True), ("CLP", "Chilean Peso", "CLP$", True), ("RMB", "Chinese Yuan", "RMB", True), ("GBP", "British Pound", "GBP", True), ("CAD", "Canadian Dollar", "CAD$", True), ("AUD", "Australian Dollar", "AUD$", True), ) MEASUREMENT_UNIT_SEEDS = ( ("tonnes", "Tonnes", "t", "mass", True), ("kilograms", "Kilograms", "kg", "mass", True), ("pounds", "Pounds", "lb", "mass", True), ("liters", "Liters", "L", "volume", True), ("cubic_meters", "Cubic Meters", "m3", "volume", True), ("kilowatt_hours", "Kilowatt Hours", "kWh", "energy", True), ) THEME_SETTING_SEEDS = ( ("--color-background", "#f4f5f7", "color", "theme", "CSS variable --color-background", True), ("--color-surface", "#ffffff", "color", "theme", "CSS variable --color-surface", True), ("--color-text-primary", "#2a1f33", "color", "theme", "CSS variable --color-text-primary", True), ("--color-text-secondary", "#624769", "color", "theme", "CSS variable --color-text-secondary", True), ("--color-text-muted", "#64748b", "color", "theme", "CSS variable --color-text-muted", True), ("--color-text-subtle", "#94a3b8", "color", "theme", "CSS variable --color-text-subtle", True), ("--color-text-invert", "#ffffff", "color", "theme", "CSS variable --color-text-invert", True), ("--color-text-dark", "#0f172a", "color", "theme", "CSS variable --color-text-dark", True), ("--color-text-strong", "#111827", "color", "theme", "CSS variable --color-text-strong", True), ("--color-primary", "#5f320d", "color", "theme", "CSS variable --color-primary", True), ("--color-primary-strong", "#7e4c13", "color", "theme", "CSS variable --color-primary-strong", True), ("--color-primary-stronger", "#837c15", "color", "theme", "CSS variable --color-primary-stronger", True), ("--color-accent", "#bff838", "color", "theme", "CSS variable --color-accent", True), ("--color-border", "#e2e8f0", "color", "theme", "CSS variable --color-border", True), ("--color-border-strong", "#cbd5e1", "color", "theme", "CSS variable --color-border-strong", True), ("--color-highlight", "#eef2ff", "color", "theme", "CSS variable --color-highlight", True), ("--color-panel-shadow", "rgba(15, 23, 42, 0.08)", "color", "theme", "CSS variable --color-panel-shadow", True), ("--color-panel-shadow-deep", "rgba(15, 23, 42, 0.12)", "color", "theme", "CSS variable --color-panel-shadow-deep", True), ("--color-surface-alt", "#f8fafc", "color", "theme", "CSS variable --color-surface-alt", True), ("--color-success", "#047857", "color", "theme", "CSS variable --color-success", True), ("--color-error", "#b91c1c", "color", "theme", "CSS variable --color-error", True), ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Seed baseline CalMiner data") parser.add_argument( "--currencies", action="store_true", help="Seed currency table" ) parser.add_argument("--units", action="store_true", help="Seed unit table") parser.add_argument( "--theme", action="store_true", help="Seed theme settings" ) parser.add_argument( "--defaults", action="store_true", help="Seed default records" ) parser.add_argument( "--dry-run", action="store_true", help="Print actions without executing" ) parser.add_argument( "--verbose", "-v", action="count", default=0, help="Increase logging verbosity", ) return parser.parse_args() def _configure_logging(args: argparse.Namespace) -> None: level = logging.WARNING - (10 * min(args.verbose, 2)) logging.basicConfig( level=max(level, logging.INFO), format="%(levelname)s %(message)s" ) def main() -> None: args = parse_args() run_with_namespace(args) def run_with_namespace( args: argparse.Namespace, *, config: Optional[DatabaseConfig] = None, ) -> None: if not hasattr(args, "verbose"): args.verbose = 0 if not hasattr(args, "dry_run"): args.dry_run = False _configure_logging(args) currencies = bool(getattr(args, "currencies", False)) units = bool(getattr(args, "units", False)) theme = bool(getattr(args, "theme", False)) defaults = bool(getattr(args, "defaults", False)) dry_run = bool(getattr(args, "dry_run", False)) if not any((currencies, units, theme, defaults)): logger.info("No seeding options provided; exiting") return config = config or DatabaseConfig.from_env() with psycopg2.connect(config.application_dsn()) as conn: conn.autocommit = True with conn.cursor() as cursor: if currencies: _seed_currencies(cursor, dry_run=dry_run) if units: _seed_units(cursor, dry_run=dry_run) if theme: _seed_theme(cursor, dry_run=dry_run) if defaults: _seed_defaults(cursor, dry_run=dry_run) def _seed_currencies(cursor, *, dry_run: bool) -> None: logger.info("Seeding currency table (%d rows)", len(CURRENCY_SEEDS)) if dry_run: for code, name, symbol, active in CURRENCY_SEEDS: logger.info("Dry run: would upsert currency %s (%s)", code, name) return execute_values( cursor, """ INSERT INTO currency (code, name, symbol, is_active) VALUES %s ON CONFLICT (code) DO UPDATE SET name = EXCLUDED.name, symbol = EXCLUDED.symbol, is_active = EXCLUDED.is_active """, CURRENCY_SEEDS, ) logger.info("Currency seed complete") def _seed_units(cursor, *, dry_run: bool) -> None: total = len(MEASUREMENT_UNIT_SEEDS) logger.info("Seeding measurement_unit table (%d rows)", total) if dry_run: for code, name, symbol, unit_type, _ in MEASUREMENT_UNIT_SEEDS: logger.info( "Dry run: would upsert measurement unit %s (%s - %s)", code, name, unit_type, ) return try: execute_values( cursor, """ INSERT INTO measurement_unit (code, name, symbol, unit_type, is_active) VALUES %s ON CONFLICT (code) DO UPDATE SET name = EXCLUDED.name, symbol = EXCLUDED.symbol, unit_type = EXCLUDED.unit_type, is_active = EXCLUDED.is_active """, MEASUREMENT_UNIT_SEEDS, ) except errors.UndefinedTable: logger.warning( "measurement_unit table does not exist; skipping unit seeding." ) cursor.connection.rollback() return logger.info("Measurement unit seed complete") def _seed_theme(cursor, *, dry_run: bool) -> None: logger.info("Seeding theme settings (%d rows)", len(THEME_SETTING_SEEDS)) if dry_run: for key, value, _, _, _, _ in THEME_SETTING_SEEDS: logger.info( "Dry run: would upsert theme setting %s = %s", key, value) return try: execute_values( cursor, """ INSERT INTO application_setting (key, value, value_type, category, description, is_editable) VALUES %s ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, value_type = EXCLUDED.value_type, category = EXCLUDED.category, description = EXCLUDED.description, is_editable = EXCLUDED.is_editable """, THEME_SETTING_SEEDS, ) except errors.UndefinedTable: logger.warning( "application_setting table does not exist; skipping theme seeding." ) cursor.connection.rollback() return logger.info("Theme settings seed complete") def _seed_defaults(cursor, *, dry_run: bool) -> None: logger.info("Seeding default records") _seed_theme(cursor, dry_run=dry_run) logger.info("Default records seed complete") if __name__ == "__main__": main()