#!/usr/bin/env python3 """Load seed/brentwood-raw.json (the review app export) into the volleyboard schema. Standalone + idempotent: re-running deletes the "brentwood" match (DB-level cascade wipes sides/nodes/members/courts/balls/touches) and reloads cleanly; players are get-or-create by username and never deleted. Usage (dev DB): DATABASE_URL="postgresql+psycopg2://volleyboard:volleyboard@127.0.0.1:5433/volleyboard" \ backend/.venv/bin/python tools/load-seed.py Reads DATABASE_URL from env. Imports the backend's models + constants (one authority for schema and vocab — this script re-types nothing). Transform decisions (also reported in the M1 build notes): - pages.staging_url/live_url are ALL null in the export; the review app renders link buttons from project domain + page slug. We compose https://{staging_domain}{slug} / https://{live_domain}{slug} (preferring a raw per-page URL if one ever exists) so the board's link column works. - features/issues have no title in the source; title = first non-empty line of the markdown (images stripped, "#" markers stripped, collapsed whitespace, 80 chars max) — the body keeps the full text. - issues.image_path is IGNORED (screenshots were not exported; attachments are M4). notes/comments/issue_comments are DEFERRED to M4 — not loaded. - created_at preserved verbatim wherever the export has it. M6 additions (plan §9 "perspective-sensitive seeds ship with M6"): - EVERY ball gets a synthesized serve touch (author where known, erin — the client rep — for authorless pages/features), stamped RECENTLY (deterministic spread over the last ~48h): aging counts from the latest touch, so without this the months-old Brentwood created_at stamps would wobble the whole board on first load. - ~4 Tingang sub-rallies (2 issues + 2 pages): the fleshout flow — erin serve -> rian pass adi -> adi pass zeina -> tingang-internal -> back to adi -> adi spike rian — acting_node stamped per ACTOR's member node (exactly what services.touches.acting_node_id_for would record), plus Tingang-scoped internal comments. - One Rob back-office rally on an issues ball (erin -> rob -> erin) + a Rob comment scoped to Back office. - 2 pages + 1 issues ball deliberately last-touched 5-7 days ago: the CURATED aging-wobble demo (plan §9 — demonstrated, not epidemic). - Adi's autopass rule: from_subteam -> rian, active (§4.4's worked example; rules cascade away with the match delete, so the reload stays idempotent). Comments are polymorphic (parent_id has NO FK), so the reload deletes the match's ball comments explicitly before dropping the match — DB cascade cannot reach them. """ from __future__ import annotations import json import os import re import sys from datetime import datetime, timedelta, timezone from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[1] BACKEND_DIR = PROJECT_ROOT / "backend" SEED_FILE = PROJECT_ROOT / "seed" / "brentwood-raw.json" # Dev runs from the repo (app package under backend/); in the container the # backend IS the project root (/app) — point at whichever exists. sys.path.insert(0, str(BACKEND_DIR if BACKEND_DIR.is_dir() else PROJECT_ROOT)) from sqlalchemy import create_engine, delete, select # noqa: E402 from sqlalchemy.orm import Session # noqa: E402 from app import constants # noqa: E402 from app.models import ( # noqa: E402 AutopassRule, Ball, Comment, Court, Match, Player, Side, TeamMember, TeamNode, Touch, ) MATCH_SLUG = "brentwood" MATCH_NAME = "Brentwood" # Seed email -> volleyboard username (the four real people in the export). EMAIL_TO_USERNAME = { "erin.coulson@brentwood.ca": "erin", "info@adipramono.com": "adi", "rian@rian.ca": "rian", "tracy@mcallistermarketing.com": "tracy", } # username, display_name, email, stable avatar_color (tailwind 500-series # hexes; assigned once, never reshuffled). rob/zeina/rahman/stevan are the # fleshout-only players — no seed content, they exist for perspectives (M6). ROSTER = [ ("rian", "Rian", "rian@rian.ca", "#3b82f6"), # blue-500 ("adi", "Adi", "info@adipramono.com", "#8b5cf6"), # violet-500 ("erin", "Erin", "erin.coulson@brentwood.ca", "#10b981"), # emerald-500 ("tracy", "Tracy", "tracy@mcallistermarketing.com", "#f59e0b"), # amber-500 ("rob", "Rob", None, "#ef4444"), # red-500 ("zeina", "Zeina", None, "#ec4899"), # pink-500 ("rahman", "Rahman", None, "#14b8a6"), # teal-500 ("stevan", "Stevan", None, "#f97316"), # orange-500 ] MD_IMAGE_RE = re.compile(r"!\[[^\]]*\]\([^)]*\)") # the review preview idiom def parse_ts(value: str | None) -> datetime | None: if not value: return None return datetime.fromisoformat(value) def _created(row: dict) -> dict: """created_at kwarg only when the export has one (preserve verbatim); otherwise omit so the column's server default (now()) applies.""" ts = parse_ts(row.get("created_at")) return {"created_at": ts} if ts else {} def snippet_title(markdown: str | None, limit: int = 80) -> str: """First non-empty line of a markdown blob, de-imaged, de-headinged, whitespace-collapsed, truncated. Body keeps the full text.""" text = MD_IMAGE_RE.sub("", markdown or "") line = "" for raw in text.splitlines(): candidate = raw.strip().lstrip("#").strip() if candidate: line = candidate break if not line: return "Untitled" line = re.sub(r"\s+", " ", line) if len(line) > limit: line = line[: limit - 3].rstrip() + "..." return line def compose_url(raw: str | None, domain: str | None, slug: str | None) -> str | None: """Per-page URL when the export has one; else domain + slug.""" if raw: return raw if domain and slug: return f"https://{domain}{slug}" return None def get_or_create_players(db: Session) -> dict[str, Player]: players: dict[str, Player] = {} for username, display_name, email, color in ROSTER: player = db.execute( select(Player).where(Player.username == username) ).scalar_one_or_none() if player is None: player = Player(username=username) db.add(player) player.display_name = display_name player.email = email player.avatar_color = color players[username] = player db.flush() return players def emails_to_player_ids(emails: list[str] | None, players: dict[str, Player]) -> list: ids = [] for email in emails or []: username = EMAIL_TO_USERNAME.get((email or "").lower()) if username and username in players: ids.append(players[username].id) return ids def author_id_for(email: str | None, players: dict[str, Player]): username = EMAIL_TO_USERNAME.get((email or "").lower()) return players[username].id if username else None def build_team_tree(db: Session, match: Match, players: dict[str, Player]) -> dict: """Plan §4.1 worked example: SIDE Bowden Works: Front line (boundary rian): rian, adi -> Tingang (boundary adi): zeina, rahman, stevan SIDE Brentwood: Front line (boundary erin): erin, tracy -> Back office (boundary erin): rob A player belongs to exactly ONE node per match; a sub-node's boundary player is NOT a member of it (near-side excludes the boundary, §4.2). Returns the four nodes by key so the M6 rally builder can stamp acting_node_id per actor. """ bowden = Side(match_id=match.id, name="Bowden Works", position=1) brentwood = Side(match_id=match.id, name="Brentwood", position=2) db.add_all([bowden, brentwood]) db.flush() bowden_front = TeamNode( side_id=bowden.id, name="Front line", boundary_player_id=players["rian"].id ) brentwood_front = TeamNode( side_id=brentwood.id, name="Front line", boundary_player_id=players["erin"].id ) db.add_all([bowden_front, brentwood_front]) db.flush() tingang = TeamNode( side_id=bowden.id, parent_node_id=bowden_front.id, name="Tingang", boundary_player_id=players["adi"].id, ) back_office = TeamNode( side_id=brentwood.id, parent_node_id=brentwood_front.id, name="Back office", boundary_player_id=players["erin"].id, ) db.add_all([tingang, back_office]) db.flush() membership = [ (bowden_front, ["rian", "adi"]), (tingang, ["zeina", "rahman", "stevan"]), (brentwood_front, ["erin", "tracy"]), (back_office, ["rob"]), ] for node, usernames in membership: for username in usernames: db.add( TeamMember( match_id=match.id, node_id=node.id, player_id=players[username].id ) ) db.flush() return { "bowden_front": bowden_front, "tingang": tingang, "brentwood_front": brentwood_front, "back_office": back_office, } # --------------------------------------------------------------------- M6 def _member_node_of(nodes: dict, username: str): """The actor's MEMBER node — mirrors services.touches.acting_node_id_for (a boundary player's touches carry their member node, not the node they lead: Adi acts from the front line even when passing into Tingang).""" by_user = { "rian": "bowden_front", "adi": "bowden_front", "zeina": "tingang", "rahman": "tingang", "stevan": "tingang", "erin": "brentwood_front", "tracy": "brentwood_front", "rob": "back_office", } key = by_user.get(username) return nodes[key].id if key else None def build_m6_seed( db: Session, match: Match, players: dict[str, Player], nodes: dict, pages_balls: list[Ball], features_balls: list[Ball], issues_balls: list[Ball], counts: dict[str, int], ) -> None: """Rallies, scoped comments, the curated aging demo, and Adi's autopass rule (module docstring). Deterministic: ball selection is by status + position order, timestamps are index-derived offsets from now — a rerun rebuilds the same shape anchored to the new now.""" now = datetime.now(timezone.utc) touch_count = 0 comment_count = 0 def add_touch(ball, seq, kind, frm, to, at, via_rule_id=None): nonlocal touch_count db.add( Touch( ball_id=ball.id, seq=seq, kind=kind, from_player=players[frm].id if frm else None, to_players=[players[u].id for u in to], acting_node_id=_member_node_of(nodes, frm) if frm else None, via_rule_id=via_rule_id, created_at=at, ) ) touch_count += 1 def add_comment(ball, author, node_key, body, at): nonlocal comment_count db.add( Comment( parent_type="ball", parent_id=ball.id, author_id=players[author].id, acting_node_id=nodes[node_key].id, body=body, created_at=at, ) ) comment_count += 1 # ---- ball selection (stable: status + position order) ---- open_issues = sorted( (b for b in issues_balls if b.status == "Open"), key=lambda b: b.position ) open_pages = sorted( (b for b in pages_balls if b.status == "open"), key=lambda b: b.position ) tingang_balls = open_issues[:2] + open_pages[:2] tingang_ids = {b.id for b in tingang_balls} # Rob's ball: the export has almost no Open issues (the two that exist # go to Tingang above), so take the first still-live issue — a back- # office consult on an Addressing/Address Later ball reads naturally. rob_ball = next( ( b for b in sorted(issues_balls, key=lambda b: b.position) if b.id not in tingang_ids and b.status != "Resolved" ), None, ) special_ids = tingang_ids | ({rob_ball.id} if rob_ball else set()) # Curated aging demo: the LAST balls by position (bottom of the board, # so the wobble is discoverable, not in the visitor's face), 5-7 days # stale. Never a ball that just got a fresh rally. aged: list[Ball] = [] for candidate in sorted(pages_balls, key=lambda b: b.position, reverse=True): if candidate.id not in special_ids and candidate.status != "completed": aged.append(candidate) if len(aged) == 2: break for candidate in sorted(issues_balls, key=lambda b: b.position, reverse=True): if candidate.id not in special_ids and candidate.status not in ("Resolved",): aged.append(candidate) break aged_ids = {b.id for b in aged} # ---- baseline: every OTHER ball opens with one recent serve ---- # Author where the export knew one; erin (the client rep, whose site # these pages are) otherwise. Spread over ~2-46h ago, index-derived. all_balls = pages_balls + features_balls + issues_balls id_to_username = {p.id: u for u, p in players.items()} for i, ball in enumerate(all_balls): if ball.id in special_ids or ball.id in aged_ids: continue frm = id_to_username.get(ball.author_id, "erin") to = [id_to_username[p] for p in (ball.action_to or []) if p in id_to_username] at = now - timedelta(minutes=120 + (i * 97) % (44 * 60)) add_touch(ball, 1, "serve", frm, to, at) # Aged balls: one serve, 5-7 days ago — old enough to wobble (every # court config ships aging_days=3). for j, ball in enumerate(aged): frm = id_to_username.get(ball.author_id, "erin") at = now - timedelta(days=5, hours=10 + j * 17) add_touch(ball, 1, "serve", frm, [], at) # ---- Tingang sub-rallies: the fleshout flow, minutes apart ---- # erin serve -> rian pass adi -> adi pass zeina -> zeina pass rahman -> # rahman back to adi (over Tingang's net = its boundary) -> adi spike # rian. Rian's projected timeline shows the zeina/rahman stretch as ONE # collapsed entry; Adi and the Tingang three see every touch. inner_two = [("zeina", "rahman"), ("rahman", "zeina")] # vary the pair for k, ball in enumerate(tingang_balls): start = now - timedelta(hours=3 + k * 9) step = timedelta(minutes=13) a, b = inner_two[k % 2] add_touch(ball, 1, "serve", "erin", ["rian"], start) add_touch(ball, 2, "pass", "rian", ["adi"], start + step) add_touch(ball, 3, "pass", "adi", [a], start + 2 * step) add_touch(ball, 4, "pass", a, [b], start + 3 * step) add_touch(ball, 5, "pass", b, ["adi"], start + 4 * step) add_touch(ball, 6, "spike", "adi", ["rian"], start + 5 * step) # Final state consistent with the spike: ball back in rian's hands, # rally memory holding adi (who had it before the spike). ball.action_to = [players["rian"].id] ball.prev_action_to = [players["adi"].id] if ball.status == "Open": # issues vocab only; pages keep theirs ball.status = "Ready for Review" if len(tingang_balls) >= 1: add_comment( tingang_balls[0], "zeina", "tingang", "Internal: refactoring the hero block before this goes back over " "the net - the image grid swap lands first. Don't surface to the " "client yet.", now - timedelta(hours=3) + timedelta(minutes=40), ) if len(tingang_balls) >= 2: add_comment( tingang_balls[1], "rahman", "tingang", "Internal: root cause is the cached srcset. Zeina is rebuilding " "the block; I'll retest on staging before Adi reviews.", now - timedelta(hours=12) + timedelta(minutes=45), ) # ---- Rob's back-office rally: erin -> rob -> erin ---- if rob_ball is not None: start = now - timedelta(hours=20) step = timedelta(minutes=25) add_touch(rob_ball, 1, "serve", "erin", ["rob"], start) add_touch(rob_ball, 2, "pass", "rob", ["erin"], start + step) add_comment( rob_ball, "rob", "back_office", "Checked the ledger export - the totals mismatch is on our side, " "not the site. Sending Erin the corrected sheet.", start + step - timedelta(minutes=5), ) rob_ball.action_to = [players["erin"].id] # ---- Adi's autopass: from_subteam -> rian (§4.4 worked example) ---- db.add( AutopassRule( match_id=match.id, owner_player_id=players["adi"].id, trigger="from_subteam", target_player_id=players["rian"].id, active=True, ) ) counts["touches"] = touch_count counts["comments"] = comment_count counts["autopass_rules"] = 1 counts["tingang_rallies"] = len(tingang_balls) counts["aged_balls"] = len(aged) def load(db: Session, seed: dict) -> dict[str, int]: players = get_or_create_players(db) existing = db.execute( select(Match).where(Match.slug == MATCH_SLUG) ).scalar_one_or_none() if existing is not None: # Comments have a POLYMORPHIC parent (no FK) — the match delete's # cascade cannot reach them, so clear them by ball id first or # every rerun would strand orphans. ball_ids = ( db.execute( select(Ball.id) .join(Court, Ball.court_id == Court.id) .where(Court.match_id == existing.id) ) .scalars() .all() ) if ball_ids: db.execute( delete(Comment).where( Comment.parent_type == "ball", Comment.parent_id.in_(ball_ids) ) ) db.delete(existing) # DB-level ON DELETE CASCADE clears the rest db.flush() match = Match(slug=MATCH_SLUG, name=MATCH_NAME) db.add(match) db.flush() nodes = build_team_tree(db, match, players) courts: dict[str, Court] = {} for position, key in enumerate(["pages", "features", "issues", "warmup"]): cfg = constants.COURT_CONFIGS[key] court = Court( match_id=match.id, config_key=key, name=cfg["name"], position=position, config=cfg, # snapshot of the authority for the frontend ) db.add(court) courts[key] = court db.flush() project = seed.get("project") or {} staging_domain = project.get("staging_domain") live_domain = project.get("live_domain") counts = {"players": len(players), "courts": len(courts)} pages_balls: list[Ball] = [] for page in seed["pages"]: ball = Ball( court_id=courts["pages"].id, title=page["page_name"], status=page["status"], review_state=page["signoff"], tags=page.get("tags") or [], url_staging=compose_url(page.get("staging_url"), staging_domain, page.get("slug")), url_live=compose_url(page.get("live_url"), live_domain, page.get("slug")), position=page.get("position") or 0, **_created(page), ) db.add(ball) pages_balls.append(ball) counts["balls_pages"] = len(seed["pages"]) features_balls: list[Ball] = [] for feature in seed["features"]: ball = Ball( court_id=courts["features"].id, title=snippet_title(feature.get("description")), body=feature.get("description"), status=feature["status"], review_state=feature["signoff"], position=feature.get("position") or 0, author_id=author_id_for(feature.get("author_email"), players), **_created(feature), ) db.add(ball) features_balls.append(ball) counts["balls_features"] = len(seed["features"]) issues_balls: list[Ball] = [] for index, issue in enumerate(seed["issues"]): ball = Ball( court_id=courts["issues"].id, title=snippet_title(issue.get("content")), body=issue.get("content"), status=issue["status"], category=issue.get("category"), action_to=emails_to_player_ids(issue.get("action_to"), players), prev_action_to=emails_to_player_ids(issue.get("prev_action_to"), players), author_id=author_id_for(issue.get("author_email"), players), position=index, # export order = creation order **_created(issue), ) db.add(ball) issues_balls.append(ball) counts["balls_issues"] = len(seed["issues"]) counts["balls_total"] = ( counts["balls_pages"] + counts["balls_features"] + counts["balls_issues"] ) # Warm-up court (M7, plan §2 config 4): the pre-game — access requests, # credentials, the requirements trickle. Hand-authored (the review app # had no pre-game concept, so there is nothing to export); mostly # inbound serves from the client side, a couple already put away — # the "wrangling the project into existence" feel from idea.md. warmup_specs = [ ("WordPress admin access for the team", "Request", "erin", ["rian"], "Resolved", -9), ("Brand fonts + logo pack", "Request", "rian", ["erin"], "Resolved", -8), ("Staging DNS: dev.brentwood.ca", "Request", "rian", ["erin"], "Resolved", -8), ("Which programs pages carry over as-is?", "Question", "rian", ["erin", "tracy"], "Open", -2), ("Google Analytics + Search Console access", "Request", "rian", ["erin"], "Open", -1), ("Accessibility target is WCAG 2.1 AA", "Requirement", "erin", ["rian", "adi"], "FYI Only", -3), ] now = datetime.now(timezone.utc) warmup_balls: list[Ball] = [] for index, (title, category, author, to, status, days_ago) in enumerate( warmup_specs ): at = now + timedelta(days=days_ago, hours=index) resolved = status == "Resolved" ball = Ball( court_id=courts["warmup"].id, title=title, status=status, category=category, action_to=[] if resolved else [players[u].id for u in to], prev_action_to=[players[u].id for u in to] if resolved else [], author_id=players[author].id, resolved_at=at + timedelta(hours=6) if resolved else None, position=index, created_at=at, ) db.add(ball) warmup_balls.append(ball) db.flush() for ball, (title, category, author, to, status, days_ago) in zip( warmup_balls, warmup_specs ): db.add( Touch( ball_id=ball.id, seq=1, kind="serve", from_player=players[author].id, to_players=[players[u].id for u in to], acting_node_id=_member_node_of(nodes, author), created_at=ball.created_at, ) ) if ball.resolved_at is not None: db.add( Touch( ball_id=ball.id, seq=2, kind="resolve", from_player=players[to[0]].id, to_players=[], acting_node_id=_member_node_of(nodes, to[0]), created_at=ball.resolved_at, ) ) counts["balls_warmup"] = len(warmup_balls) counts["balls_total"] += len(warmup_balls) db.flush() # ball ids for the rally builder below build_m6_seed( db, match, players, nodes, pages_balls, features_balls, issues_balls, counts ) return counts def main() -> int: database_url = os.environ.get("DATABASE_URL", "") if not database_url: print("ERROR: DATABASE_URL not set", file=sys.stderr) return 2 if not SEED_FILE.is_file(): print(f"ERROR: seed file missing: {SEED_FILE}", file=sys.stderr) return 2 with open(SEED_FILE) as fh: seed = json.load(fh) engine = create_engine(database_url, future=True) with Session(engine) as db: counts = load(db, seed) db.commit() print(f"Seed loaded into match {MATCH_SLUG!r}:") for key, value in counts.items(): print(f" {key}: {value}") return 0 if __name__ == "__main__": raise SystemExit(main())