#!/usr/bin/env python3 """ Project Registry Module (ADR-144)
Central project registry stored in org.db (Tier 2) that indexes all CODITECT projects across submodules, customer tenants, and contributor workspaces.
Task ID: FPA.F.1.7 Created: 2026-02-03 ADR: ADR-144 Multi-Project Registry Architecture
Usage: python3 scripts/project_registry.py register --id PILOT --name "CODITECT PILOT" --scope platform python3 scripts/project_registry.py list python3 scripts/project_registry.py show PILOT python3 scripts/project_registry.py switch PILOT python3 scripts/project_registry.py scan --root ~/PROJECTS """
import json import os import re import sqlite3 import subprocess import sys from datetime import datetime, timezone from pathlib import Path from typing import Dict, List, Optional, Tuple
Add parent to path for imports
sys.path.insert(0, str(Path(file).parent.parent))
try: from scripts.core.paths import ( get_org_db_path, get_projects_db_path, discover_projects_dir, get_machine_uuid, ORG_DB, PROJECTS_DIR, ) except ImportError: # Fallback for standalone execution def get_org_db_path() -> Path: home = Path.home() candidates = [ home / "PROJECTS" / ".coditect-data" / "context-storage" / "org.db", home / ".coditect-data" / "context-storage" / "org.db", ] for c in candidates: if c.exists(): return c return candidates[0]
def discover_projects_dir() -> Path:
return Path.home() / "PROJECTS"
def get_machine_uuid() -> Optional[str]:
return None
ORG_DB = get_org_db_path()
PROJECTS_DIR = discover_projects_dir()
=============================================================================
Database Schema (ADR-144)
=============================================================================
SCHEMA_SQL = """ -- ============================================================ -- Table: projects (ADR-144) -- Purpose: Central registry of all CODITECT projects -- Location: org.db (Tier 2 - irreplaceable) -- ============================================================ CREATE TABLE IF NOT EXISTS projects ( project_id TEXT PRIMARY KEY, slug TEXT UNIQUE NOT NULL, name TEXT NOT NULL, description TEXT, scope TEXT NOT NULL CHECK (scope IN ('platform', 'org', 'customer', 'project')), tenant_id TEXT, owner TEXT NOT NULL, plan_location TEXT NOT NULL, tracks_config TEXT, status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'paused', 'completed', 'archived')), created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), metadata TEXT, cloud_synced_at TEXT, cloud_project_uuid TEXT );
-- ============================================================ -- Table: project_tracks (ADR-144) -- Purpose: Map projects to their assigned tracks -- Location: org.db (Tier 2 - irreplaceable) -- ============================================================ CREATE TABLE IF NOT EXISTS project_tracks ( project_id TEXT NOT NULL REFERENCES projects(project_id) ON DELETE CASCADE, track_letter TEXT NOT NULL, track_name TEXT NOT NULL, track_file TEXT NOT NULL, tier INTEGER NOT NULL CHECK (tier IN (1, 2, 3)), progress_pct REAL DEFAULT 0.0, status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'paused', 'completed', 'deferred')), PRIMARY KEY (project_id, track_letter) );
-- ============================================================ -- Indexes for common query patterns -- ============================================================ CREATE INDEX IF NOT EXISTS idx_projects_scope ON projects(scope); CREATE INDEX IF NOT EXISTS idx_projects_status ON projects(status); CREATE INDEX IF NOT EXISTS idx_projects_tenant ON projects(tenant_id); CREATE INDEX IF NOT EXISTS idx_projects_slug ON projects(slug); CREATE INDEX IF NOT EXISTS idx_project_tracks_project ON project_tracks(project_id); CREATE INDEX IF NOT EXISTS idx_project_tracks_letter ON project_tracks(track_letter);
-- ============================================================ -- View: active_projects -- Purpose: Convenience view for common filtering -- ============================================================ CREATE VIEW IF NOT EXISTS active_projects AS SELECT p.project_id, p.slug, p.name, p.scope, p.owner, p.plan_location, p.status, COUNT(pt.track_letter) AS track_count, AVG(pt.progress_pct) AS avg_progress FROM projects p LEFT JOIN project_tracks pt ON p.project_id = pt.project_id WHERE p.status = 'active' GROUP BY p.project_id; """
=============================================================================
Database Initialization
=============================================================================
def init_registry_tables() -> bool: """ Initialize project registry tables in org.db.
Returns:
True if successful, False otherwise
"""
db_path = get_org_db_path()
# Ensure directory exists
db_path.parent.mkdir(parents=True, exist_ok=True)
try:
conn = sqlite3.connect(str(db_path))
conn.executescript(SCHEMA_SQL)
conn.commit()
conn.close()
return True
except sqlite3.Error as e:
print(f"Error initializing registry tables: {e}")
return False
def check_registry_exists() -> bool: """Check if the projects table exists in org.db.""" db_path = get_org_db_path() if not db_path.exists(): return False
try:
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
cursor.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='projects'"
)
exists = cursor.fetchone() is not None
conn.close()
return exists
except sqlite3.Error:
return False
=============================================================================
Registry Operations
=============================================================================
def register_project( project_id: str, name: str, scope: str, owner: str, plan_location: str, slug: Optional[str] = None, description: Optional[str] = None, tenant_id: Optional[str] = None, tracks: Optional[str] = None, metadata: Optional[Dict] = None, ) -> bool: """ Register a project in the registry.
Args:
project_id: Unique project identifier (e.g., 'PILOT', 'CUST-avivatec-fpa')
name: Human-readable project name
scope: Isolation scope ('platform', 'org', 'customer', 'project')
owner: Owner identifier (e.g., 'core-team', 'hal.casteel')
plan_location: Path to project plan (relative to PROJECTS_DIR or absolute)
slug: URL-safe identifier (auto-generated from project_id if not provided)
description: Optional project description
tenant_id: Tenant UUID for customer projects
tracks: Comma-separated track letters (e.g., 'A,B,C,D,E,F')
metadata: Additional project metadata
Returns:
True if registration successful, False otherwise
"""
# Ensure registry exists
if not check_registry_exists():
if not init_registry_tables():
return False
# Generate slug if not provided
if slug is None:
slug = project_id.lower().replace('_', '-')
db_path = get_org_db_path()
try:
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
# Check if already registered
cursor.execute("SELECT project_id, name FROM projects WHERE project_id = ?", (project_id,))
existing = cursor.fetchone()
if existing:
print(f"Project already registered: {existing['project_id']} ({existing['name']})")
return True
# Prepare tracks config
tracks_config = None
if tracks:
track_letters = [t.strip().upper() for t in tracks.split(',')]
tracks_config = json.dumps({
'letters': track_letters,
'tier': 1 if scope == 'platform' else 2
})
# Insert project
cursor.execute("""
INSERT INTO projects (
project_id, slug, name, description, scope,
tenant_id, owner, plan_location, tracks_config,
status, metadata
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?)
""", (
project_id,
slug,
name,
description,
scope,
tenant_id,
owner,
plan_location,
tracks_config,
json.dumps(metadata) if metadata else None,
))
conn.commit()
print("=" * 60)
print("PROJECT REGISTERED (ADR-144)")
print("=" * 60)
print(f"\n Project ID: {project_id}")
print(f" Name: {name}")
print(f" Scope: {scope}")
print(f" Owner: {owner}")
print(f" Plan Location: {plan_location}")
if tracks:
print(f" Tracks: {tracks}")
print(f"\n Stored in: {db_path}")
print("=" * 60)
# Register tracks if provided
if tracks:
register_project_tracks(project_id, tracks, conn)
conn.close()
return True
except sqlite3.Error as e:
print(f"Database error: {e}")
return False
def register_project_tracks( project_id: str, tracks: str, conn: Optional[sqlite3.Connection] = None, ) -> bool: """ Register tracks for a project.
Args:
project_id: Project ID
tracks: Comma-separated track letters
conn: Optional existing database connection
Returns:
True if successful
"""
close_conn = False
if conn is None:
conn = sqlite3.connect(str(get_org_db_path()))
close_conn = True
try:
cursor = conn.cursor()
track_letters = [t.strip().upper() for t in tracks.split(',')]
# Get project scope to determine tier
cursor.execute("SELECT scope, plan_location FROM projects WHERE project_id = ?", (project_id,))
project = cursor.fetchone()
if not project:
print(f"Project not found: {project_id}")
return False
scope = project[0]
plan_dir = project[1]
# Determine tier based on scope and track letter
for letter in track_letters:
# Tier 1: A-N (Technical) for platform/org
# Tier 2: O-AA (PCF Business) or customer
if letter in 'ABCDEFGHIJKLMN':
tier = 1
else:
tier = 2
# Generate track name from letter
track_names = {
'A': 'Backend API', 'B': 'Frontend', 'C': 'Infrastructure',
'D': 'Security', 'E': 'Testing/QA', 'F': 'Documentation',
'G': 'AI/ML Pipeline', 'H': 'Data/ELT', 'I': 'Integration',
'J': 'Intelligence', 'K': 'Workflow', 'L': 'Extended Testing',
'M': 'Extended Security', 'N': 'GTM/Launch',
'O': 'Vision & Strategy', 'P': 'Products & Services',
'Q': 'Marketing & Sales', 'R': 'Physical Delivery',
'S': 'Service Delivery', 'T': 'Customer Service',
'U': 'Human Capital', 'V': 'Information Technology',
'W': 'Financial Resources', 'X': 'Asset Management',
'Y': 'Risk & Compliance', 'Z': 'External Relationships',
'AA': 'Business Capabilities',
}
track_name = track_names.get(letter, f'Track {letter}')
# Generate track file path
track_file = f"{plan_dir}/TRACK-{project_id.split('-')[0]}-{letter}-{track_name.lower().replace(' ', '-').replace('/', '-')}.md"
cursor.execute("""
INSERT OR REPLACE INTO project_tracks (
project_id, track_letter, track_name, track_file, tier, progress_pct, status
) VALUES (?, ?, ?, ?, ?, 0.0, 'active')
""", (project_id, letter, track_name, track_file, tier))
conn.commit()
print(f" Registered {len(track_letters)} tracks for {project_id}")
return True
except sqlite3.Error as e:
print(f"Error registering tracks: {e}")
return False
finally:
if close_conn:
conn.close()
def list_projects( scope: Optional[str] = None, status: str = 'active', owner: Optional[str] = None, ) -> List[Dict]: """ List registered projects with optional filtering.
Args:
scope: Filter by scope ('platform', 'org', 'customer', 'project')
status: Filter by status (default: 'active')
owner: Filter by owner
Returns:
List of project dictionaries
"""
if not check_registry_exists():
print("Project registry not initialized. Run: python3 scripts/project_registry.py init")
return []
db_path = get_org_db_path()
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
try:
# Build query with filters
query = """
SELECT
p.project_id, p.slug, p.name, p.scope, p.owner,
p.plan_location, p.status, p.tenant_id, p.created_at,
COUNT(pt.track_letter) as track_count,
COALESCE(AVG(pt.progress_pct), 0) as avg_progress
FROM projects p
LEFT JOIN project_tracks pt ON p.project_id = pt.project_id
WHERE 1=1
"""
params = []
if scope:
query += " AND p.scope = ?"
params.append(scope)
if status:
query += " AND p.status = ?"
params.append(status)
if owner:
query += " AND p.owner = ?"
params.append(owner)
query += " GROUP BY p.project_id ORDER BY p.name"
cursor.execute(query, params)
projects = [dict(row) for row in cursor.fetchall()]
print("=" * 70)
print("REGISTERED PROJECTS (ADR-144)")
print("=" * 70)
if not projects:
print("\nNo projects found matching criteria.")
print("\nRegister a project:")
print(" python3 scripts/project_registry.py register --id MY-PROJECT --name 'My Project' --scope project --owner me --plan-location ./")
print("=" * 70)
return []
print(f"\n{len(projects)} project(s):\n")
for p in projects:
status_icon = {'active': '🟢', 'paused': '🟡', 'completed': '✅', 'archived': '📦'}.get(p['status'], '❓')
scope_icon = {'platform': '🏗️', 'org': '🏢', 'customer': '👤', 'project': '📁'}.get(p['scope'], '❓')
progress = p['avg_progress']
progress_bar = f"[{'█' * int(progress / 10)}{'░' * (10 - int(progress / 10))}] {progress:.0f}%"
print(f" {status_icon} {scope_icon} {p['project_id']}")
print(f" Name: {p['name']}")
print(f" Scope: {p['scope']} | Owner: {p['owner']}")
print(f" Tracks: {p['track_count']} tracks | Progress: {progress_bar}")
print(f" Location: {p['plan_location']}")
print()
print("=" * 70)
return projects
except sqlite3.Error as e:
print(f"Database error: {e}")
return []
finally:
conn.close()
def show_project(project_id: str) -> Optional[Dict]: """ Show detailed information about a project.
Args:
project_id: Project ID to show
Returns:
Project dictionary or None
"""
if not check_registry_exists():
print("Project registry not initialized.")
return None
db_path = get_org_db_path()
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
try:
# Get project
cursor.execute("SELECT * FROM projects WHERE project_id = ?", (project_id,))
project = cursor.fetchone()
if not project:
print(f"Project not found: {project_id}")
return None
project_dict = dict(project)
# Get tracks
cursor.execute("""
SELECT track_letter, track_name, track_file, tier, progress_pct, status
FROM project_tracks
WHERE project_id = ?
ORDER BY track_letter
""", (project_id,))
tracks = [dict(row) for row in cursor.fetchall()]
print("=" * 70)
print(f"PROJECT DETAILS: {project_id}")
print("=" * 70)
print(f"\n Name: {project_dict['name']}")
print(f" Slug: {project_dict['slug']}")
print(f" Scope: {project_dict['scope']}")
print(f" Owner: {project_dict['owner']}")
print(f" Status: {project_dict['status']}")
print(f" Plan Location: {project_dict['plan_location']}")
if project_dict.get('tenant_id'):
print(f" Tenant ID: {project_dict['tenant_id']}")
if project_dict.get('description'):
print(f" Description: {project_dict['description']}")
print(f"\n Created: {project_dict['created_at']}")
print(f" Updated: {project_dict['updated_at']}")
if project_dict.get('cloud_synced_at'):
print(f" Cloud Synced: {project_dict['cloud_synced_at']}")
if tracks:
print(f"\n TRACKS ({len(tracks)}):")
print(" " + "-" * 60)
for t in tracks:
status_icon = {'active': '🔵', 'paused': '🟡', 'completed': '✅', 'deferred': '⏸️'}.get(t['status'], '❓')
progress = t['progress_pct']
print(f" {status_icon} [{t['track_letter']}] {t['track_name']}")
print(f" Tier: {t['tier']} | Progress: {progress:.0f}% | Status: {t['status']}")
print("\n" + "=" * 70)
project_dict['tracks'] = tracks
return project_dict
except sqlite3.Error as e:
print(f"Database error: {e}")
return None
finally:
conn.close()
def update_project( project_id: str, name: Optional[str] = None, status: Optional[str] = None, owner: Optional[str] = None, description: Optional[str] = None, metadata: Optional[Dict] = None, ) -> bool: """ Update a project's fields.
Args:
project_id: Project ID to update
name: New name
status: New status
owner: New owner
description: New description
metadata: New metadata (merged with existing)
Returns:
True if successful
"""
if not check_registry_exists():
print("Project registry not initialized.")
return False
db_path = get_org_db_path()
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
try:
# Check project exists
cursor.execute("SELECT metadata FROM projects WHERE project_id = ?", (project_id,))
existing = cursor.fetchone()
if not existing:
print(f"Project not found: {project_id}")
return False
# Build update query
updates = []
params = []
if name:
updates.append("name = ?")
params.append(name)
if status:
updates.append("status = ?")
params.append(status)
if owner:
updates.append("owner = ?")
params.append(owner)
if description:
updates.append("description = ?")
params.append(description)
if metadata:
# Merge with existing metadata
existing_meta = json.loads(existing[0]) if existing[0] else {}
existing_meta.update(metadata)
updates.append("metadata = ?")
params.append(json.dumps(existing_meta))
if not updates:
print("No updates specified.")
return False
updates.append("updated_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now')")
query = f"UPDATE projects SET {', '.join(updates)} WHERE project_id = ?"
params.append(project_id)
cursor.execute(query, params)
conn.commit()
print(f"Updated project: {project_id}")
return True
except sqlite3.Error as e:
print(f"Database error: {e}")
return False
finally:
conn.close()
def switch_project(project_id: str) -> bool: """ Switch to a project context.
Sets $CODITECT_PROJECT environment variable and prints instructions.
Args:
project_id: Project ID to switch to
Returns:
True if project exists and switch successful
"""
if not check_registry_exists():
print("Project registry not initialized.")
return False
db_path = get_org_db_path()
conn = sqlite3.connect(str(db_path))
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
try:
cursor.execute(
"SELECT project_id, name, plan_location, tracks_config FROM projects WHERE project_id = ? AND status = 'active'",
(project_id,)
)
project = cursor.fetchone()
if not project:
print(f"Project not found or not active: {project_id}")
return False
# Set environment variable
os.environ['CODITECT_PROJECT'] = project_id
print("=" * 60)
print("PROJECT SWITCHED (ADR-144)")
print("=" * 60)
print(f"\n Active Project: {project_id}")
print(f" Name: {project['name']}")
print(f" Plan Location: {project['plan_location']}")
# Parse tracks config
if project['tracks_config']:
tracks_config = json.loads(project['tracks_config'])
tracks = tracks_config.get('letters', [])
print(f" Tracks: {', '.join(tracks)}")
print(f"\n Environment: CODITECT_PROJECT={project_id}")
print(f"\n To persist, add to shell:")
print(f" export CODITECT_PROJECT={project_id}")
print("=" * 60)
return True
except sqlite3.Error as e:
print(f"Database error: {e}")
return False
finally:
conn.close()
def scan_projects(root_dir: str, dry_run: bool = False) -> List[Dict]: """ Scan a directory tree for projects to register.
Looks for:
- PROJECT-PLAN.md files
- CLAUDE.md with project metadata
- package.json / pyproject.toml / Cargo.toml
Args:
root_dir: Root directory to scan
dry_run: If True, don't actually register
Returns:
List of discovered projects
"""
root_path = Path(root_dir).expanduser().resolve()
if not root_path.exists():
print(f"Directory not found: {root_path}")
return []
print("=" * 60)
print("SCANNING FOR PROJECTS (ADR-144)")
print("=" * 60)
print(f"\n Root: {root_path}")
print(f" Mode: {'Dry run' if dry_run else 'Register'}")
discovered = []
# Walk directory tree
for plan_file in root_path.rglob('PROJECT-PLAN.md'):
project_dir = plan_file.parent
# Skip if too deep (likely not a project root)
try:
rel_depth = len(project_dir.relative_to(root_path).parts)
if rel_depth > 4:
continue
except ValueError:
continue
# Extract project info from plan file
project_info = extract_project_info(project_dir, plan_file)
if project_info:
discovered.append(project_info)
print(f"\n Found: {project_info['project_id']}")
print(f" Name: {project_info['name']}")
print(f" Path: {project_dir}")
if not dry_run:
register_project(
project_id=project_info['project_id'],
name=project_info['name'],
scope=project_info.get('scope', 'project'),
owner=project_info.get('owner', 'unknown'),
plan_location=str(project_dir.relative_to(root_path)),
description=project_info.get('description'),
)
print(f"\n Discovered: {len(discovered)} project(s)")
print("=" * 60)
return discovered
def extract_project_info(project_dir: Path, plan_file: Path) -> Optional[Dict]: """ Extract project information from a project directory.
Args:
project_dir: Project directory
plan_file: Path to PROJECT-PLAN.md
Returns:
Dictionary with project info or None
"""
info = {
'project_id': project_dir.name.upper().replace('-', '_').replace(' ', '_'),
'name': project_dir.name,
'scope': 'project',
'owner': 'unknown',
}
# Try to read CLAUDE.md for metadata
claude_md = project_dir / 'CLAUDE.md'
if claude_md.exists():
try:
with open(claude_md) as f:
content = f.read(3000)
# Extract from frontmatter
if content.startswith('---'):
try:
import yaml
end = content.find('---', 3)
if end > 0:
frontmatter = yaml.safe_load(content[3:end])
if frontmatter:
info['name'] = frontmatter.get('title', info['name'])
info['scope'] = frontmatter.get('scope', info['scope'])
info['owner'] = frontmatter.get('owner', info['owner'])
except:
pass
# Extract Codename
match = re.search(r'\*\*Codename\*\*[:\s]*([^\n]+)', content)
if match:
info['name'] = match.group(1).strip()
# Generate project_id from directory name if customer project
if 'customer' in str(project_dir).lower() or 'cust-' in str(project_dir).lower():
info['scope'] = 'customer'
info['project_id'] = f"CUST-{project_dir.name.replace('coditect-', '').replace('jv-', '')}"
except IOError:
pass
return info
=============================================================================
Seed Data
=============================================================================
def seed_pilot_project() -> bool: """ Seed the PILOT project (primary CODITECT framework project).
Returns:
True if successful
"""
success = register_project(
project_id='PILOT',
name='CODITECT PILOT Launch',
scope='platform',
owner='core-team',
plan_location='internal/project/plans/tracks/',
description='Primary CODITECT framework development project',
tracks='A,B,C,D,E,F,G,H,I,J,K,L,M,N',
)
if success:
# Update track progress (known values from CLAUDE.md)
db_path = get_org_db_path()
conn = sqlite3.connect(str(db_path))
cursor = conn.cursor()
track_progress = {
'A': (100.0, 'completed'),
'B': (84.0, 'active'),
'C': (81.0, 'active'),
'D': (100.0, 'completed'),
'E': (100.0, 'completed'),
'F': (40.0, 'active'),
'G': (0.0, 'deferred'),
'H': (56.0, 'active'),
'I': (100.0, 'completed'),
'J': (67.0, 'active'),
'K': (0.0, 'deferred'),
'L': (0.0, 'deferred'),
'M': (0.0, 'deferred'),
'N': (61.0, 'active'),
}
for letter, (progress, status) in track_progress.items():
cursor.execute("""
UPDATE project_tracks
SET progress_pct = ?, status = ?
WHERE project_id = 'PILOT' AND track_letter = ?
""", (progress, status, letter))
conn.commit()
conn.close()
return success
def seed_avivatec_project() -> bool: """ Seed the Avivatec FP&A customer project.
Returns:
True if successful
"""
return register_project(
project_id='CUST-avivatec-fpa',
name='Avivatec AI-First FP&A Platform',
scope='customer',
owner='coditect-jv',
plan_location='submodules/ventures/coditect-jv-avivatec/',
description='AI-First Financial Planning & Analysis platform for Brazil + USA markets',
tracks='A,B,C,D,E,F,G,H,I,J,K,L,M,N,O,P,Q,R,S,T,U,V,W,X,Y,Z,AA',
metadata={
'codename': 'Avivatec FP&A',
'markets': ['Brazil', 'USA'],
'compliance': ['LGPD', 'SOC 2'],
'launch_target': '2026-Q4',
}
)
=============================================================================
CLI
=============================================================================
def main(): import argparse
parser = argparse.ArgumentParser(
description="CODITECT Project Registry (ADR-144)",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
Initialize registry
python3 scripts/project_registry.py init
Register a project
python3 scripts/project_registry.py register --id MY-PROJECT --name "My Project" --scope project --owner me --plan-location ./
List projects
python3 scripts/project_registry.py list python3 scripts/project_registry.py list --scope customer
Show project details
python3 scripts/project_registry.py show PILOT
Switch to project
python3 scripts/project_registry.py switch CUST-avivatec-fpa
Scan for projects
python3 scripts/project_registry.py scan --root ~/PROJECTS --dry-run
Seed default projects
python3 scripts/project_registry.py seed """ )
subparsers = parser.add_subparsers(dest='command', help='Commands')
# init
init_parser = subparsers.add_parser('init', help='Initialize registry tables')
# register
register_parser = subparsers.add_parser('register', help='Register a project')
register_parser.add_argument('--id', required=True, help='Project ID')
register_parser.add_argument('--name', required=True, help='Project name')
register_parser.add_argument('--scope', required=True, choices=['platform', 'org', 'customer', 'project'])
register_parser.add_argument('--owner', required=True, help='Owner identifier')
register_parser.add_argument('--plan-location', required=True, help='Path to project plan')
register_parser.add_argument('--slug', help='URL-safe slug')
register_parser.add_argument('--description', help='Project description')
register_parser.add_argument('--tenant-id', help='Tenant UUID (for customer projects)')
register_parser.add_argument('--tracks', help='Comma-separated track letters')
# list
list_parser = subparsers.add_parser('list', help='List projects')
list_parser.add_argument('--scope', choices=['platform', 'org', 'customer', 'project'])
list_parser.add_argument('--status', default='active')
list_parser.add_argument('--owner', help='Filter by owner')
# show
show_parser = subparsers.add_parser('show', help='Show project details')
show_parser.add_argument('project_id', help='Project ID to show')
# update
update_parser = subparsers.add_parser('update', help='Update a project')
update_parser.add_argument('project_id', help='Project ID to update')
update_parser.add_argument('--name', help='New name')
update_parser.add_argument('--status', choices=['active', 'paused', 'completed', 'archived'])
update_parser.add_argument('--owner', help='New owner')
update_parser.add_argument('--description', help='New description')
# switch
switch_parser = subparsers.add_parser('switch', help='Switch to project context')
switch_parser.add_argument('project_id', help='Project ID to switch to')
# scan
scan_parser = subparsers.add_parser('scan', help='Scan for projects')
scan_parser.add_argument('--root', required=True, help='Root directory to scan')
scan_parser.add_argument('--dry-run', action='store_true', help='Preview only')
# seed
seed_parser = subparsers.add_parser('seed', help='Seed default projects (PILOT, Avivatec)')
args = parser.parse_args()
if args.command == 'init':
if init_registry_tables():
print("Registry tables initialized successfully.")
sys.exit(0)
else:
sys.exit(1)
elif args.command == 'register':
success = register_project(
project_id=args.id,
name=args.name,
scope=args.scope,
owner=args.owner,
plan_location=args.plan_location,
slug=args.slug,
description=args.description,
tenant_id=args.tenant_id,
tracks=args.tracks,
)
sys.exit(0 if success else 1)
elif args.command == 'list':
list_projects(scope=args.scope, status=args.status, owner=args.owner)
elif args.command == 'show':
project = show_project(args.project_id)
sys.exit(0 if project else 1)
elif args.command == 'update':
success = update_project(
project_id=args.project_id,
name=args.name,
status=args.status,
owner=args.owner,
description=args.description,
)
sys.exit(0 if success else 1)
elif args.command == 'switch':
success = switch_project(args.project_id)
sys.exit(0 if success else 1)
elif args.command == 'scan':
scan_projects(args.root, dry_run=args.dry_run)
elif args.command == 'seed':
print("Seeding default projects...")
seed_pilot_project()
seed_avivatec_project()
print("\nDone. List with: python3 scripts/project_registry.py list")
else:
parser.print_help()
if name == "main": main()