diegoecab/oci-deal-accelerator
1
1
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

CLI improvements: --json output, filters, arch-search, ecal-score, deck fixes

Browse Source 
6 CLI improvements from MCP integration testing + deck bug fixes:

- kb_cli.py: --json on search, new arch-search and ecal-score subcommands,
  _load_yaml handles multi-document YAML (fixes health command on 32 files)
- findings_cli.py: --json on search/list/stats, filter flags on search
  (--product, --severity, --tag, --client, --status, --category)
- validate-architecture.py: --profile optional, --output defaults to stdout
- common-objections.yaml + ecal-artefacts-catalog.yaml: merge front matter
- oci_deck_gen.py: fix slide 9 cost_notes char iteration, fix slide 12
  migration field mapping (duration/milestones), dynamic phase spacing
- proposal-spec.yaml: replace real name with fictional Carlos Mendoza

All 14 skill options validated with generated deliverables.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
2026-04-12 23:02:40 -03:00
parent ca93a0aa4e
commit b48d155c1b
27 changed files with 2857 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
tools/findings_cli.py
View File

@@ -28,6 +28,7 @@ Usage:
import argparse
import difflib
import json
import os
import sys
from collections import Counter
@@ -378,6 +379,10 @@ def cmd_list(args: argparse.Namespace) -> None:
category=getattr(args, "category", None),
)
if getattr(args, "json_output", False):
print(json.dumps(results, indent=2, default=str))
return
if not results:
print("No findings match the given filters.")
return
@@ -401,6 +406,21 @@ def cmd_search(args: argparse.Namespace) -> None:
"""Full-text search."""
results = search(args.query)
# Apply optional filters (mirrors list subcommand)
results = filter_findings(
results,
severity=getattr(args, "severity", None),
product=getattr(args, "product", None),
tag=getattr(args, "tag", None),
client=getattr(args, "client", None),
status=getattr(args, "status", None),
category=getattr(args, "category", None),
)
if getattr(args, "json_output", False):
print(json.dumps(results, indent=2, default=str))
return
if not results:
print(f"No findings matching '{args.query}'.")
return
@@ -600,6 +620,10 @@ def cmd_stats(args: argparse.Namespace) -> None:
"""Display summary statistics."""
s = stats()
if getattr(args, "json_output", False):
print(json.dumps(s, indent=2))
return
print(f"Total findings: {s['total']}\n")
print("By Severity:")
@@ -640,10 +664,18 @@ def build_parser() -> argparse.ArgumentParser:
p_list.add_argument("--client", help="Filter by client (partial match).")
p_list.add_argument("--status", help="Filter by status.")
p_list.add_argument("--category", help="Filter by category.")
p_list.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON.")
# --- search ---
p_search = subparsers.add_parser("search", help="Full-text search.")
p_search.add_argument("query", help="Search term.")
p_search.add_argument("--severity", help="Filter by severity.")
p_search.add_argument("--product", help="Filter by product (partial match).")
p_search.add_argument("--tag", help="Filter by tag.")
p_search.add_argument("--client", help="Filter by client (partial match).")
p_search.add_argument("--status", help="Filter by status.")
p_search.add_argument("--category", help="Filter by category.")
p_search.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON.")
# --- add ---
p_add = subparsers.add_parser("add", help="Add a new finding.")
@@ -685,7 +717,8 @@ def build_parser() -> argparse.ArgumentParser:
subparsers.add_parser("aer", help="After-Engagement Review (interactive).")
# --- stats ---
subparsers.add_parser("stats", help="Show summary statistics.")
p_stats = subparsers.add_parser("stats", help="Show summary statistics.")
p_stats.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON.")
return parser

228
tools/kb_cli.py
View File

@@ -22,6 +22,7 @@ Usage:
"""
import argparse
import json
import os
import re
import sys
@@ -41,6 +42,8 @@ GOVERNANCE_PATH = os.path.join(PROJECT_ROOT, "config", "kb-governance.yaml")
OWNERS_PATH = os.path.join(PROJECT_ROOT, "config", "kb-owners.yaml")
TRACKER_PATH = os.path.join(PROJECT_ROOT, "kb", "field-findings", "tracker.yaml")
MATRIX_PATH = os.path.join(PROJECT_ROOT, "kb", "compatibility", "adb-feature-matrix.yaml")
ARCH_CATALOG_PATH = os.path.join(KB_ROOT, "architecture-center", "catalog.yaml")
ECAL_CATALOG_PATH = os.path.join(KB_ROOT, "patterns", "ecal-artefacts-catalog.yaml")
_RESET = "\033[0m"
_GREEN = "\033[32m"
@@ -69,7 +72,13 @@ def _load_yaml(path: str) -> Optional[Dict[str, Any]]:
if not os.path.isfile(resolved):
return None
with open(resolved, "r", encoding="utf-8") as fh:
return yaml.safe_load(fh)
docs = list(yaml.safe_load_all(fh))
# Merge multi-document YAML (front matter + body) into single dict
merged: Dict[str, Any] = {}
for doc in docs:
if isinstance(doc, dict):
merged.update(doc)
return merged or None
def _parse_date(d: Any) -> Optional[date]:
@@ -401,6 +410,14 @@ def cmd_search(args: argparse.Namespace) -> None:
matches.append((i, line.strip()[:80]))
results.append((rel, matches))
if getattr(args, "json_output", False):
json_results = [
{"file": fp, "matches": [{"line": ln, "text": txt} for ln, txt in ms]}
for fp, ms in results
]
print(json.dumps(json_results, indent=2))
return
if not results:
print(f"No results for '{args.query}'.")
return
@@ -438,6 +455,197 @@ def cmd_owners(args: argparse.Namespace) -> None:
print(f"{area:<30} {name_display:<25} {team:<25} {cadence}")
# =============================================================================
# Architecture Center search command
# =============================================================================
def cmd_arch_search(args: argparse.Namespace) -> None:
"""Search Architecture Center catalog by services and/or tags."""
data = _load_yaml(ARCH_CATALOG_PATH)
if not data or "entries" not in data:
print("Error: Architecture catalog not found or empty.", file=sys.stderr)
sys.exit(1)
results = data["entries"]
# Filter by services (OR match, case-insensitive)
if args.services:
svc_set = {s.lower() for s in args.services}
results = [
e for e in results
if svc_set & {s.lower() for s in e.get("services", [])}
]
# Filter by tags (OR match, case-insensitive)
if args.tags:
tag_set = {t.lower() for t in args.tags}
results = [
e for e in results
if tag_set & {t.lower() for t in e.get("tags", [])}
]
# Optional text search across title + summary
if args.query:
q = args.query.lower()
results = [
e for e in results
if q in e.get("title", "").lower() or q in e.get("summary", "").lower()
]
if getattr(args, "json_output", False):
print(json.dumps(results, indent=2, default=str))
return
if not results:
print("No matching architecture references found.")
return
print(f"Found {len(results)} architecture reference(s):\n")
for e in results:
svcs = ", ".join(e.get("services", []))
tags = ", ".join(e.get("tags", []))
print(f" {_c(e.get('title', ''), _CYAN)}")
print(f" Services: {svcs}")
print(f" Tags: {tags}")
print(f" URL: {e.get('url', 'N/A')}")
if e.get("terraform"):
print(f" Terraform: {e['terraform']}")
print()
# =============================================================================
# ECAL score command
# =============================================================================
def _load_ecal_artefacts() -> List[Dict[str, Any]]:
"""Load and flatten all ECAL artefacts from the catalog."""
data = _load_yaml(ECAL_CATALOG_PATH)
if not data:
return []
artefacts = []
for phase in ["define", "design", "deliver"]:
phase_data = data.get(phase, {})
if not isinstance(phase_data, dict):
continue
for step, items in phase_data.items():
if not isinstance(items, list):
continue
for item in items:
if isinstance(item, dict) and "id" in item:
item["_phase"] = phase
item["_step"] = step
artefacts.append(item)
return artefacts
def cmd_ecal_score(args: argparse.Namespace) -> None:
"""Evaluate engagement readiness against ECAL 3.1 artefacts."""
artefacts = _load_ecal_artefacts()
if not artefacts:
print("Error: Could not load ECAL artefacts catalog.", file=sys.stderr)
sys.exit(1)
summary_lower = args.summary.lower()
summary_words = set(summary_lower.split())
stopwords = {
"the", "a", "an", "and", "or", "for", "to", "in", "of", "is",
"with", "on", "at", "by", "as", "it", "be", "this", "that", "from",
}
# Filter by phase if specified
if args.phase:
artefacts = [a for a in artefacts if a.get("_phase") == args.phase]
scored = []
for art in artefacts:
name_lower = art.get("name", "").lower()
what_lower = art.get("what", "").lower()
id_lower = art.get("id", "").lower()
# Check if artefact name or ID appears in summary
name_match = name_lower in summary_lower
id_match = id_lower in summary_lower
# Keyword overlap with the "what" field
what_words = set(what_lower.split()) - stopwords
meaningful_overlap = summary_words & what_words
if name_match or id_match:
relevance = "mentioned"
elif len(meaningful_overlap) >= 3:
relevance = "likely_relevant"
elif len(meaningful_overlap) >= 1:
relevance = "possibly_relevant"
else:
relevance = "none"
scored.append({
"id": art.get("id"),
"name": art.get("name"),
"phase": art.get("_phase"),
"step": art.get("_step"),
"skill_support": art.get("skill_support"),
"customer_facing": art.get("customer_facing"),
"relevance": relevance,
})
# Compute summary stats
mentioned = [s for s in scored if s["relevance"] == "mentioned"]
likely = [s for s in scored if s["relevance"] == "likely_relevant"]
possibly = [s for s in scored if s["relevance"] == "possibly_relevant"]
not_covered = [s for s in scored if s["relevance"] == "none"]
relevant = [s for s in scored if s["relevance"] != "none"]
auto_gen = [s for s in relevant if s["skill_support"] == "auto_generate"]
assist = [s for s in relevant if s["skill_support"] == "assist"]
if getattr(args, "json_output", False):
output = {
"summary": {
"total_artefacts": len(scored),
"mentioned": len(mentioned),
"likely_relevant": len(likely),
"possibly_relevant": len(possibly),
"not_covered": len(not_covered),
"skill_auto_generate": len(auto_gen),
"skill_assist": len(assist),
},
"artefacts": scored,
}
print(json.dumps(output, indent=2, default=str))
return
# Text output
print(f"\n{_c('ECAL 3.1 READINESS ASSESSMENT', _BOLD)}")
print("=" * 50)
print(f"\nArtefacts assessed: {len(scored)}")
print(f" Mentioned in summary: {_c(str(len(mentioned)), _GREEN)}")
print(f" Likely relevant: {_c(str(len(likely)), _YELLOW)}")
print(f" Possibly relevant: {str(len(possibly))}")
print(f" Not covered: {_c(str(len(not_covered)), _RED)}")
print(f"\nSkill can auto-generate: {len(auto_gen)} relevant artefact(s)")
print(f"Skill can assist: {len(assist)} relevant artefact(s)")
if mentioned or likely:
print(f"\n{'ID':<10} {'Phase':<8} {'Name':<35} {'Support':<15} Relevance")
print("-" * 85)
for s in mentioned + likely:
rel_color = _GREEN if s["relevance"] == "mentioned" else _YELLOW
print(
f"{s['id']:<10} {s['phase']:<8} "
f"{str(s['name']):<35} {str(s['skill_support']):<15} "
f"{_c(s['relevance'], rel_color)}"
)
if not_covered:
print(f"\nNot covered ({len(not_covered)} artefact(s)) — consider adding to engagement:")
for s in not_covered[:10]:
cf = " [customer-facing]" if s.get("customer_facing") else ""
print(f" {s['id']} {s['name']}{cf}")
if len(not_covered) > 10:
print(f" ... and {len(not_covered) - 10} more (use --json for full list)")
# =============================================================================
# CLI entry point
# =============================================================================
@@ -469,10 +677,24 @@ def build_parser() -> argparse.ArgumentParser:
# --- search ---
p_search = subparsers.add_parser("search", help="Search across all KB files.")
p_search.add_argument("query", help="Search term.")
p_search.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON.")
# --- owners ---
subparsers.add_parser("owners", help="Show domain owners.")
# --- arch-search ---
p_arch = subparsers.add_parser("arch-search", help="Search Architecture Center catalog.")
p_arch.add_argument("query", nargs="?", default=None, help="Optional text search term.")
p_arch.add_argument("--services", nargs="+", help="Filter by service names (e.g., adb-s exacs).")
p_arch.add_argument("--tags", nargs="+", help="Filter by tags (e.g., ha-dr multicloud).")
p_arch.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON.")
# --- ecal-score ---
p_ecal = subparsers.add_parser("ecal-score", help="ECAL engagement readiness assessment.")
p_ecal.add_argument("--summary", required=True, help="Text describing the engagement.")
p_ecal.add_argument("--phase", choices=["define", "design", "deliver"], help="Limit to a specific ECAL phase.")
p_ecal.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON.")
return parser
@@ -495,6 +717,10 @@ def main() -> None:
cmd_search(args)
elif args.command == "owners":
cmd_owners(args)
elif args.command == "arch-search":
cmd_arch_search(args)
elif args.command == "ecal-score":
cmd_ecal_score(args)
else:
parser.print_help()
sys.exit(1)

151
tools/oci_deck_gen.py
View File

@@ -633,16 +633,20 @@ class OCIDeckGenerator(OraclePresBase):
col_x += Inches(3.2)
def add_environment_catalogue_slide(self, environments: list,
cost_notes: list = None):
cost_notes=None):
"""Environment Catalogue slide — Prod/Pre-Prod/Dev-Test/DR per workload.
environments: list of {"environment": str, "tier": str,
"databases": str, "ocpus": str, "isolation": str}
cost_notes: optional list of cost optimization notes
environments: list of {"name"|"environment": str, "sizing": str,
"isolation": str, "cost_pct": int}
cost_notes: optional list of strings or single string
"""
slide = self._add_blank_slide()
self._add_title_bar(slide, "Environment Catalogue", margin=self.MARGIN)
# Normalize cost_notes to a list
if isinstance(cost_notes, str):
cost_notes = [s.strip() for s in cost_notes.split(".") if s.strip()]
rows = len(environments) + 1
table = self._add_table(
slide, rows, 5,
@@ -650,28 +654,65 @@ class OCIDeckGenerator(OraclePresBase):
Inches(12), Inches(0.4 * rows),
)
table.columns[0].width = Inches(2.2)
table.columns[1].width = Inches(1.8)
table.columns[2].width = Inches(2.5)
table.columns[3].width = Inches(1.5)
table.columns[4].width = Inches(4.0)
# Detect schema: new (name/sizing/isolation/cost_pct) vs legacy (environment/tier/databases/ocpus/isolation)
sample = environments[0] if environments else {}
use_new_schema = "sizing" in sample or "cost_pct" in sample
headers = ["Environment", "Tier", "Databases", "OCPUs", "Isolation"]
for j, h in enumerate(headers):
self._style_table_cell(
table.cell(0, j), h, font_size=11, bold=True,
color=Colors.WHITE, bg_color=Colors.TEAL,
alignment=PP_ALIGN.CENTER,
if use_new_schema:
table = self._add_table(
slide, rows, 4,
self.MARGIN, Inches(1.2),
Inches(12), Inches(0.4 * rows),
)
table.columns[0].width = Inches(2.5)
table.columns[1].width = Inches(5.0)
table.columns[2].width = Inches(2.5)
table.columns[3].width = Inches(2.0)
for i, env in enumerate(environments):
row_idx = i + 1
bg = Colors.TABLE_ALT_ROW if row_idx % 2 == 0 else None
self._style_table_cell(table.cell(row_idx, 0), env.get("environment", ""), font_size=10, bold=True, bg_color=bg)
self._style_table_cell(table.cell(row_idx, 1), env.get("tier", ""), font_size=10, bg_color=bg, alignment=PP_ALIGN.CENTER)
self._style_table_cell(table.cell(row_idx, 2), env.get("databases", ""), font_size=10, bg_color=bg)
self._style_table_cell(table.cell(row_idx, 3), env.get("ocpus", ""), font_size=10, bg_color=bg, alignment=PP_ALIGN.CENTER)
self._style_table_cell(table.cell(row_idx, 4), env.get("isolation", ""), font_size=10, bg_color=bg)
headers = ["Environment", "Sizing", "Isolation", "Cost %"]
for j, h in enumerate(headers):
self._style_table_cell(
table.cell(0, j), h, font_size=11, bold=True,
color=Colors.WHITE, bg_color=Colors.TEAL,
alignment=PP_ALIGN.CENTER,
)
for i, env in enumerate(environments):
row_idx = i + 1
bg = Colors.TABLE_ALT_ROW if row_idx % 2 == 0 else None
self._style_table_cell(table.cell(row_idx, 0), env.get("name", env.get("environment", "")), font_size=10, bold=True, bg_color=bg)
self._style_table_cell(table.cell(row_idx, 1), env.get("sizing", ""), font_size=10, bg_color=bg)
self._style_table_cell(table.cell(row_idx, 2), env.get("isolation", ""), font_size=10, bg_color=bg, alignment=PP_ALIGN.CENTER)
cost_pct = env.get("cost_pct", "")
self._style_table_cell(table.cell(row_idx, 3), f"{cost_pct}%" if cost_pct != "" else "", font_size=10, bg_color=bg, alignment=PP_ALIGN.CENTER)
else:
table = self._add_table(
slide, rows, 5,
self.MARGIN, Inches(1.2),
Inches(12), Inches(0.4 * rows),
)
table.columns[0].width = Inches(2.2)
table.columns[1].width = Inches(1.8)
table.columns[2].width = Inches(2.5)
table.columns[3].width = Inches(1.5)
table.columns[4].width = Inches(4.0)
headers = ["Environment", "Tier", "Databases", "OCPUs", "Isolation"]
for j, h in enumerate(headers):
self._style_table_cell(
table.cell(0, j), h, font_size=11, bold=True,
color=Colors.WHITE, bg_color=Colors.TEAL,
alignment=PP_ALIGN.CENTER,
)
for i, env in enumerate(environments):
row_idx = i + 1
bg = Colors.TABLE_ALT_ROW if row_idx % 2 == 0 else None
self._style_table_cell(table.cell(row_idx, 0), env.get("environment", ""), font_size=10, bold=True, bg_color=bg)
self._style_table_cell(table.cell(row_idx, 1), env.get("tier", ""), font_size=10, bg_color=bg, alignment=PP_ALIGN.CENTER)
self._style_table_cell(table.cell(row_idx, 2), env.get("databases", ""), font_size=10, bg_color=bg)
self._style_table_cell(table.cell(row_idx, 3), env.get("ocpus", ""), font_size=10, bg_color=bg, alignment=PP_ALIGN.CENTER)
self._style_table_cell(table.cell(row_idx, 4), env.get("isolation", ""), font_size=10, bg_color=bg)
if cost_notes:
notes_y = Inches(1.2) + Inches(0.4 * rows) + Inches(0.3)
@@ -841,71 +882,81 @@ class OCIDeckGenerator(OraclePresBase):
def add_migration_slide(self, phases: list, tools: list = None,
downtime: str = ""):
"""Slide 9: Migration Approach.
"""Migration Approach slide.
phases: list of {"name": str, "weeks": str, "tasks": [...]}
phases: list of {"name": str, "duration"|"weeks": str,
"milestones"|"tasks": [...]}
"""
slide = self._add_blank_slide()
self._add_title_bar(slide, "Migration Approach", margin=self.MARGIN)
y = Inches(1.3)
phase_colors = [Colors.TEAL, Colors.ORACLE_RED, Colors.BURNT_ORANGE, Colors.FOREST]
phase_colors = [Colors.TEAL, Colors.BURNT_ORANGE, Colors.FOREST, Colors.PURPLE]
n = len(phases)
# Scale spacing to fit all phases: leave room for tools/downtime at bottom
avail = 5.0 # inches available for phases (from y=1.3 to ~6.3)
step = min(1.1, avail / max(n, 1))
y = Inches(1.3)
bar_left = Inches(2.8)
bar_width = Inches(9.2)
# Phase timeline bars
bar_left = Inches(2.5)
bar_width = Inches(9.5)
for i, phase in enumerate(phases):
color = phase_colors[i % len(phase_colors)]
duration = phase.get("duration", phase.get("weeks", ""))
milestones = phase.get("milestones", phase.get("tasks", []))
# Phase label
# Phase label (left of bar)
self._add_textbox(
slide, self.MARGIN, y,
Inches(2), Inches(0.45),
text=phase["name"], font_size=13, bold=True,
Inches(2.3), Inches(0.4),
text=phase["name"], font_size=11, bold=True,
)
# Phase bar
# Colored phase bar with duration
bar = self._add_rect(
slide, bar_left, y + Inches(0.05),
bar_width * 0.9, Inches(0.4),
slide, bar_left, y + Inches(0.02),
bar_width * 0.9, Inches(0.36),
fill_color=color,
)
bar.text_frame.paragraphs[0].text = phase.get("weeks", "")
bar.text_frame.paragraphs[0].font.size = Pt(12)
bar.text_frame.paragraphs[0].text = duration
bar.text_frame.paragraphs[0].font.size = Pt(11)
bar.text_frame.paragraphs[0].font.color.rgb = Colors.WHITE
bar.text_frame.paragraphs[0].font.name = self.FONT
bar.text_frame.paragraphs[0].alignment = PP_ALIGN.CENTER
# Tasks below bar
if phase.get("tasks"):
tasks_text = " • ".join(phase["tasks"][:4])
# Milestones/tasks below bar (compact)
if milestones:
tasks_text = " • ".join(str(m) for m in milestones[:4])
self._add_textbox(
slide, bar_left, y + Inches(0.48),
bar_width, Inches(0.38),
text=tasks_text, font_size=11, italic=True,
slide, bar_left, y + Inches(0.40),
bar_width, Inches(0.30),
text=tasks_text, font_size=9, italic=True,
color=Colors.SECONDARY_TEXT,
)
y += Inches(1.1)
y += Inches(step)
# Tools
if tools:
y += Inches(0.2)
y = max(y, Inches(1.3 + avail)) + Inches(0.15)
self._add_textbox(
slide, self.MARGIN, y,
Inches(12), Inches(0.35),
text=f"Migration Tools: {', '.join(tools)}",
font_size=11, bold=True, color=Colors.TEAL,
text=f"Migration Tools: {', '.join(tools[:4])}",
font_size=10, bold=True, color=Colors.TEAL,
)
y += Inches(0.35)
# Downtime approach
if downtime:
y += Inches(0.5)
if not tools:
y = max(y, Inches(1.3 + avail)) + Inches(0.15)
self._add_textbox(
slide, self.MARGIN, y,
Inches(12), Inches(0.35),
text=f"Downtime Approach: {downtime}",
font_size=11, color=Colors.PRIMARY_TEXT,
font_size=10, color=Colors.PRIMARY_TEXT,
)
def add_operational_raci_slide(self, raci_items: list,