190 lines
8.0 KiB
Python
190 lines
8.0 KiB
Python
"""Human-readable query helpers over the governance portfolio."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Iterable, Sequence
|
|
|
|
from .governance_models import EcosystemGovernancePortfolio, GovernanceDomain, GovernanceStatus, PlatformGovernanceCard
|
|
from .models import as_plain_data, merge_unique
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class PortfolioQuestion:
|
|
question_id: str
|
|
question: str
|
|
answer: str
|
|
evidence: tuple[str, ...]
|
|
next_action: str
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return as_plain_data(self)
|
|
|
|
|
|
def card_or_none(portfolio: EcosystemGovernancePortfolio, platform_id: str) -> PlatformGovernanceCard | None:
|
|
return portfolio.card_for(platform_id)
|
|
|
|
|
|
def summarize_card(card: PlatformGovernanceCard) -> str:
|
|
blockers = ", ".join(check.title for check in card.blockers[:3]) or "sem blocker principal"
|
|
action = card.next_actions[0] if card.next_actions else "manter regressao"
|
|
return (
|
|
f"{card.platform_id} esta em status {card.status_label}, score {card.governance_score}, "
|
|
f"maturidade {card.maturity.value}. Blockers: {blockers}. Proxima acao: {action}."
|
|
)
|
|
|
|
|
|
def strongest_platforms(portfolio: EcosystemGovernancePortfolio, limit: int = 5) -> tuple[PlatformGovernanceCard, ...]:
|
|
return tuple(sorted(portfolio.cards, key=lambda item: (-item.governance_score, item.platform_id))[:limit])
|
|
|
|
|
|
def weakest_platforms(portfolio: EcosystemGovernancePortfolio, limit: int = 5) -> tuple[PlatformGovernanceCard, ...]:
|
|
return tuple(sorted(portfolio.cards, key=lambda item: (item.governance_score, item.platform_id))[:limit])
|
|
|
|
|
|
def platforms_by_domain_gap(portfolio: EcosystemGovernancePortfolio, domain: GovernanceDomain) -> tuple[PlatformGovernanceCard, ...]:
|
|
cards: list[PlatformGovernanceCard] = []
|
|
for card in portfolio.cards:
|
|
if any(check.domain == domain and check.status in {GovernanceStatus.ATTENTION, GovernanceStatus.FAIL, GovernanceStatus.BLOCKED} for check in card.checks):
|
|
cards.append(card)
|
|
cards.sort(key=lambda item: (item.governance_score, item.platform_id))
|
|
return tuple(cards)
|
|
|
|
|
|
def blockers_for_domain(portfolio: EcosystemGovernancePortfolio, domain: GovernanceDomain) -> tuple[str, ...]:
|
|
items: list[str] = []
|
|
for card in portfolio.cards:
|
|
for check in card.blockers:
|
|
if check.domain == domain:
|
|
items.append(f"{card.platform_id}: {check.title} - {check.next_action}")
|
|
return merge_unique(items)
|
|
|
|
|
|
def build_operational_questions(portfolio: EcosystemGovernancePortfolio) -> tuple[PortfolioQuestion, ...]:
|
|
questions: list[PortfolioQuestion] = []
|
|
questions.append(
|
|
PortfolioQuestion(
|
|
question_id="estado-geral-governanca",
|
|
question="Qual e o estado geral de governanca humana do ecossistema?",
|
|
answer=(
|
|
f"O score medio de governanca e {portfolio.average_governance_score}. "
|
|
f"Plataformas bloqueadas: {len(portfolio.blocked_platforms)}. "
|
|
f"Plataformas controladas: {len(portfolio.controlled_platforms)}."
|
|
),
|
|
evidence=portfolio.executive_summary,
|
|
next_action="atuar primeiro nos blockers de dominio com maior impacto humano",
|
|
)
|
|
)
|
|
weak = weakest_platforms(portfolio)
|
|
questions.append(
|
|
PortfolioQuestion(
|
|
question_id="plataformas-mais-fracas",
|
|
question="Quais plataformas mais precisam de continuidade?",
|
|
answer="As plataformas com menor score sao: " + ", ".join(f"{card.platform_id} ({card.governance_score})" for card in weak),
|
|
evidence=tuple(summarize_card(card) for card in weak),
|
|
next_action="executar as OS vinculadas aos checks dessas plataformas",
|
|
)
|
|
)
|
|
strong = strongest_platforms(portfolio)
|
|
questions.append(
|
|
PortfolioQuestion(
|
|
question_id="plataformas-mais-fortes",
|
|
question="Quais plataformas estao mais maduras para leitura humana?",
|
|
answer="As plataformas mais fortes sao: " + ", ".join(f"{card.platform_id} ({card.governance_score})" for card in strong),
|
|
evidence=tuple(summarize_card(card) for card in strong),
|
|
next_action="usar essas plataformas como referencia de padrao e regressao",
|
|
)
|
|
)
|
|
for domain in (
|
|
GovernanceDomain.DOCS,
|
|
GovernanceDomain.INTEGRATIONS,
|
|
GovernanceDomain.IDENTITY,
|
|
GovernanceDomain.BUSINESS,
|
|
GovernanceDomain.MCP,
|
|
GovernanceDomain.CLOUD,
|
|
GovernanceDomain.OBSERVABILITY,
|
|
):
|
|
blockers = blockers_for_domain(portfolio, domain)
|
|
impacted = platforms_by_domain_gap(portfolio, domain)
|
|
answer = (
|
|
f"Dominio {domain.value} tem {len(blockers)} blockers e "
|
|
f"{len(impacted)} plataformas com gap/atencao."
|
|
)
|
|
if blockers:
|
|
answer += " Principais: " + " | ".join(blockers[:3])
|
|
questions.append(
|
|
PortfolioQuestion(
|
|
question_id=f"dominio-{domain.value}",
|
|
question=f"O que bloqueia ou exige atencao no dominio {domain.value}?",
|
|
answer=answer,
|
|
evidence=blockers[:8] or tuple(summarize_card(card) for card in impacted[:5]),
|
|
next_action=(
|
|
f"priorizar checks do dominio {domain.value} e validar owner "
|
|
"institucional antes da proxima promocao"
|
|
),
|
|
)
|
|
)
|
|
questions.append(
|
|
PortfolioQuestion(
|
|
question_id="ordens-saida-justificadas",
|
|
question="As ordens de saida estao justificadas por checks reais?",
|
|
answer=(
|
|
f"Ha {len(portfolio.order_candidates)} candidatas de OS derivadas de checks de governanca. "
|
|
"Cada candidata guarda source_check_ids e validacoes."
|
|
),
|
|
evidence=tuple(f"{candidate.candidate_id}: {', '.join(candidate.source_check_ids)}" for candidate in portfolio.order_candidates[:12]),
|
|
next_action="manter ativas apenas ordens ligadas a pendencias reais ou continuidade impossivel nesta rodada",
|
|
)
|
|
)
|
|
return tuple(questions)
|
|
|
|
|
|
def questions_markdown(questions: Sequence[PortfolioQuestion]) -> str:
|
|
lines = ["# Perguntas operacionais sobre governanca", ""]
|
|
for question in questions:
|
|
lines.append(f"## {question.question}")
|
|
lines.append("")
|
|
lines.append(question.answer)
|
|
lines.append("")
|
|
lines.append(f"Proxima acao: {question.next_action}")
|
|
if question.evidence:
|
|
lines.append("")
|
|
lines.append("Evidencias:")
|
|
for item in question.evidence[:10]:
|
|
lines.append(f"- {item}")
|
|
lines.append("")
|
|
return "\n".join(lines).strip() + "\n"
|
|
|
|
|
|
def questions_rows(questions: Sequence[PortfolioQuestion]) -> list[list[str]]:
|
|
rows = [["question_id", "question", "answer", "next_action", "evidence_count"]]
|
|
for question in questions:
|
|
rows.append([question.question_id, question.question, question.answer, question.next_action, str(len(question.evidence))])
|
|
return rows
|
|
|
|
|
|
def query_by_keyword(questions: Sequence[PortfolioQuestion], keyword: str) -> tuple[PortfolioQuestion, ...]:
|
|
lowered = keyword.lower()
|
|
return tuple(
|
|
question
|
|
for question in questions
|
|
if lowered in question.question.lower() or lowered in question.answer.lower() or any(lowered in evidence.lower() for evidence in question.evidence)
|
|
)
|
|
|
|
|
|
def unresolved_question_ids(questions: Sequence[PortfolioQuestion]) -> tuple[str, ...]:
|
|
ids: list[str] = []
|
|
for question in questions:
|
|
text = f"{question.answer} {question.next_action}".lower()
|
|
if "blocker" in text or "gap" in text or "priorizar" in text:
|
|
ids.append(question.question_id)
|
|
return tuple(ids)
|
|
|
|
|
|
def compact_question_payload(questions: Sequence[PortfolioQuestion]) -> dict[str, object]:
|
|
return {
|
|
"count": len(questions),
|
|
"unresolved": unresolved_question_ids(questions),
|
|
"questions": [question.to_dict() for question in questions],
|
|
}
|