scripts/seed.py

"""Seed reference data per docs/02-data-inventory.md Section 7.

Idempotent. Inserts L1+L2 categories, one activity_template per
(L1 category x procurement_type x kind), intra-category PROC->INSTALL
dependency templates, one sample cross-category rule, and (only when
SEED_TEST_DATA=true) sample contractors + projects tagged metadata.test_data.

Run:
    python -m scripts.seed
    SEED_TEST_DATA=true python -m scripts.seed
"""

import json
import os

from app import db
from scripts import migrate

# (code, name) for L1 categories — the BOQ taxonomy backbone.
L1_CATEGORIES = [
    ("FLOORING", "Flooring"),
    ("JOINERY", "Joinery"),
    ("HVAC", "HVAC"),
    ("ELECTRICAL", "Electrical"),
    ("PLUMBING", "Plumbing"),
    ("CIVIL", "Civil"),
    ("DEMOLITION", "Demolition"),
    ("PAINTING", "Painting"),
    ("WALL_CLADDING", "Wall Cladding"),
]

# (l1_code, l2_code, l2_name)
L2_CATEGORIES = [
    ("FLOORING", "TILE_FLOORING", "Tile Flooring"),
    ("FLOORING", "VINYL_FLOORING", "Vinyl Flooring"),
    ("JOINERY", "KITCHEN_CABINETRY", "Kitchen Cabinetry"),
    ("JOINERY", "WARDROBES", "Wardrobes"),
    ("HVAC", "DUCTING", "Ducting"),
    ("ELECTRICAL", "WIRING", "Wiring"),
    ("PLUMBING", "PIPING", "Piping"),
]

# Defaults per kind (Stage 2 stub values; refined with real calendars in Stage 3).
DEFAULT_LEAD_TIME_DAYS = 7
DEFAULT_EFFORT_DAYS = 2

DEFAULT_CALENDAR_ID = 1
DEFAULT_TIMEZONE = "Asia/Dubai"  # see docs/02-data-inventory.md (UAE addresses)

# UAE public holidays 2026-2027. Islamic dates are lunar-approximate.
# metadata.source = 'hardcoded' so the team audits/replaces before production.
HOLIDAYS = [
    ("2026-01-01", "New Year's Day"),
    ("2026-03-20", "Eid al-Fitr (approx)"),
    ("2026-03-21", "Eid al-Fitr holiday (approx)"),
    ("2026-05-27", "Arafat Day (approx)"),
    ("2026-05-28", "Eid al-Adha (approx)"),
    ("2026-06-17", "Hijri New Year (approx)"),
    ("2026-08-26", "Prophet Muhammad's Birthday (approx)"),
    ("2026-12-01", "Commemoration Day"),
    ("2026-12-02", "National Day"),
    ("2026-12-03", "National Day (second day)"),
    ("2027-01-01", "New Year's Day"),
    ("2027-03-10", "Eid al-Fitr (approx)"),
    ("2027-03-11", "Eid al-Fitr holiday (approx)"),
    ("2027-05-16", "Arafat Day (approx)"),
    ("2027-05-17", "Eid al-Adha (approx)"),
    ("2027-06-06", "Hijri New Year (approx)"),
    ("2027-08-15", "Prophet Muhammad's Birthday (approx)"),
    ("2027-12-01", "Commemoration Day"),
    ("2027-12-02", "National Day"),
    ("2027-12-03", "National Day (second day)"),
]


def get_or_create_l1(conn, code: str, name: str) -> int:
    row = conn.execute(
        "SELECT id FROM categories WHERE parent_id IS NULL AND code = ?", (code,)
    ).fetchone()
    if row:
        return int(row["id"])
    cur = conn.execute(
        "INSERT INTO categories (parent_id, level, code, name) VALUES (NULL, 1, ?, ?)",
        (code, name),
    )
    return int(cur.lastrowid)


def seed_categories(conn) -> dict[str, int]:
    ids: dict[str, int] = {}
    for code, name in L1_CATEGORIES:
        ids[code] = get_or_create_l1(conn, code, name)
    for l1_code, l2_code, l2_name in L2_CATEGORIES:
        conn.execute(
            "INSERT OR IGNORE INTO categories (parent_id, level, code, name) VALUES (?, 2, ?, ?)",
            (ids[l1_code], l2_code, l2_name),
        )
    return ids


def _ins_activity_template(conn, category_id, procurement_type, kind, name_template,
                           effort, lead, clock, responsibility):
    conn.execute(
        """INSERT OR IGNORE INTO activity_templates
             (category_id, procurement_type, kind, name_template,
              default_effort_days, default_lead_time_days, clock, responsibility_default)
           VALUES (?,?,?,?,?,?,?,?)""",
        (category_id, procurement_type, kind, name_template, effort, lead, clock, responsibility),
    )


def seed_activity_templates(conn, cat_ids: dict[str, int]) -> None:
    for cid in cat_ids.values():
        # SUPPLY -> PROCUREMENT
        _ins_activity_template(conn, cid, "SUPPLY", "PROCUREMENT", "Procure {name}",
                               None, DEFAULT_LEAD_TIME_DAYS, "WALL", "CONTRACTOR")
        # APPLY -> INSTALLATION
        _ins_activity_template(conn, cid, "APPLY", "INSTALLATION", "Install {name}",
                               DEFAULT_EFFORT_DAYS, None, "WORKING", "CONTRACTOR")
        # SUPPLY_AND_APPLY -> PROCUREMENT + INSTALLATION
        _ins_activity_template(conn, cid, "SUPPLY_AND_APPLY", "PROCUREMENT", "Procure {name}",
                               None, DEFAULT_LEAD_TIME_DAYS, "WALL", "CONTRACTOR")
        _ins_activity_template(conn, cid, "SUPPLY_AND_APPLY", "INSTALLATION", "Install {name}",
                               DEFAULT_EFFORT_DAYS, None, "WORKING", "CONTRACTOR")


def seed_dependency_templates(conn, cat_ids: dict[str, int]) -> None:
    # Intra-category: PROCUREMENT -> INSTALLATION, FS, lag 0.
    for cid in cat_ids.values():
        conn.execute(
            """INSERT OR IGNORE INTO dependency_templates
                 (predecessor_category_id, successor_category_id,
                  predecessor_kind, successor_kind, dep_type, lag_days)
               VALUES (?,?,?,?,'FS',0)""",
            (cid, cid, "PROCUREMENT", "INSTALLATION"),
        )
    # Cross-category sample: Demolition (INSTALLATION) -> Flooring (PROCUREMENT).
    # NOTE: cross-category rules are NOT applied by the Rules Engine in Stage 2
    # (see docs/03-phase1-notes.md); seeded for Stage 5's ScopeWiringService.
    conn.execute(
        """INSERT OR IGNORE INTO dependency_templates
             (predecessor_category_id, successor_category_id,
              predecessor_kind, successor_kind, dep_type, lag_days)
           VALUES (?,?,?,?,'FS',0)""",
        (cat_ids["DEMOLITION"], cat_ids["FLOORING"], "INSTALLATION", "PROCUREMENT"),
    )


def seed_default_calendar(conn) -> None:
    """Default calendar id=1: Mon-Fri 08:00-17:00, Asia/Dubai, 8h/day + holidays."""
    conn.execute(
        """INSERT OR IGNORE INTO calendars (id, name, timezone, parent_id, hours_per_day)
           VALUES (?, 'Default — UAE Construction', ?, NULL, 8.00)""",
        (DEFAULT_CALENDAR_ID, DEFAULT_TIMEZONE),
    )
    for weekday in (1, 2, 3, 4, 5):  # Mon-Fri
        conn.execute(
            """INSERT OR IGNORE INTO calendar_working_hours
                 (calendar_id, weekday, start_minute, end_minute)
               VALUES (?, ?, 480, 1020)""",
            (DEFAULT_CALENDAR_ID, weekday),
        )
    meta = json.dumps({"source": "hardcoded"})
    for date, reason in HOLIDAYS:
        conn.execute(
            """INSERT OR IGNORE INTO calendar_exceptions
                 (calendar_id, date, start_minute, end_minute, kind, reason, metadata)
               VALUES (?, ?, NULL, NULL, 'HOLIDAY', ?, ?)""",
            (DEFAULT_CALENDAR_ID, date, reason, meta),
        )


def seed_test_data(conn) -> None:
    meta = json.dumps({"test_data": True})
    for name in ("Test Contractor A", "Test Contractor B"):
        conn.execute(
            "INSERT INTO contractors (name, active, metadata) VALUES (?, 1, ?)", (name, meta)
        )
    conn.execute(
        "INSERT INTO projects (name, planned_start, status, metadata) VALUES (?,?,?,?)",
        ("Sample Villa Renovation", "2026-07-01", "DRAFT", meta),
    )


def run() -> dict:
    migrate.up()  # ensure schema exists
    with db.with_tx() as conn:
        cat_ids = seed_categories(conn)
        seed_activity_templates(conn, cat_ids)
        seed_dependency_templates(conn, cat_ids)
        seed_default_calendar(conn)
        if os.environ.get("SEED_TEST_DATA", "false").lower() == "true":
            seed_test_data(conn)
        counts = {
            "categories": conn.execute("SELECT COUNT(*) c FROM categories").fetchone()["c"],
            "activity_templates": conn.execute("SELECT COUNT(*) c FROM activity_templates").fetchone()["c"],
            "dependency_templates": conn.execute("SELECT COUNT(*) c FROM dependency_templates").fetchone()["c"],
            "calendars": conn.execute("SELECT COUNT(*) c FROM calendars").fetchone()["c"],
            "calendar_exceptions": conn.execute("SELECT COUNT(*) c FROM calendar_exceptions").fetchone()["c"],
        }
    return counts


if __name__ == "__main__":
    print("seeded:", run())