grimacing
3ea36ef3ab
Squash merge of phase/03-l2-modules-and-core-scripts; 4 L2s + 5 core scripts; verify_phase03.sh green.
118 lines
3.4 KiB
Python
Executable File
118 lines
3.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""l3b_agent_stub.py — D-008 / D-026 / D-021
|
|
|
|
Parses a GitHub/Gitea Issue body by keywords and emits a contract.yaml that
|
|
selects an L2 stack. This is the agentic (L3B) entry surface: deterministic
|
|
keyword matching, no external AI APIs.
|
|
|
|
D-008 keyword map (priority order — first match wins):
|
|
gas, price, ingest, data-lake -> l2-commodity-price-feed
|
|
invoice, billing -> l2-invoice-service
|
|
analytics, historical, query -> l2-energy-analytics-api
|
|
regulatory, compliance, reporting, trading
|
|
-> l2-regulatory-reporting
|
|
(no match) -> l2-invoice-service (fallback)
|
|
|
|
Output contract.yaml (D-021 schema):
|
|
stack: <mapped L2 name>
|
|
inputs:
|
|
environment: dev
|
|
owner: citizen-developer
|
|
source: l3b-agent-stub
|
|
public-ingress: false
|
|
|
|
Input:
|
|
argv[1] = issue body text (or stdin if argv[1] absent/empty)
|
|
-o <path> = write the contract to a file (default: stdout)
|
|
Exit:
|
|
0 on success, 1 on empty input
|
|
"""
|
|
import sys
|
|
|
|
|
|
# Ordered keyword groups -> L2 stack mapping (D-008). First match wins.
|
|
KEYWORD_MAP = [
|
|
(("gas", "price", "ingest", "data-lake"), "l2-commodity-price-feed"),
|
|
(("invoice", "billing"), "l2-invoice-service"),
|
|
(("analytics", "historical", "query"), "l2-energy-analytics-api"),
|
|
(("regulatory", "compliance", "reporting", "trading"), "l2-regulatory-reporting"),
|
|
]
|
|
|
|
FALLBACK_STACK = "l2-invoice-service"
|
|
|
|
|
|
def map_issue_to_stack(text: str) -> str:
|
|
lowered = text.lower()
|
|
for keywords, stack in KEYWORD_MAP:
|
|
for kw in keywords:
|
|
if kw in lowered:
|
|
return stack
|
|
return FALLBACK_STACK
|
|
|
|
|
|
def render_contract(stack: str) -> str:
|
|
# Fixed-schema YAML (D-021). Emitted as text (no yaml dependency needed).
|
|
return (
|
|
f"stack: {stack}\n"
|
|
"inputs:\n"
|
|
" environment: dev\n"
|
|
" owner: citizen-developer\n"
|
|
" source: l3b-agent-stub\n"
|
|
"public-ingress: false\n"
|
|
)
|
|
|
|
|
|
def read_issue_body(args: list) -> str:
|
|
"""Read issue body from args[0] (already-stripped argv, no script name)
|
|
or stdin. Empty -> error."""
|
|
if len(args) >= 1 and args[0].strip():
|
|
return args[0]
|
|
# Fall back to stdin if argv body is absent or empty.
|
|
if not sys.stdin.isatty():
|
|
data = sys.stdin.read()
|
|
if data.strip():
|
|
return data
|
|
return ""
|
|
|
|
|
|
def parse_output_flag(argv: list):
|
|
"""Extract -o <path> from argv (returns (rest, output_path))."""
|
|
output_path = None
|
|
rest = []
|
|
i = 1
|
|
while i < len(argv):
|
|
arg = argv[i]
|
|
if arg == "-o":
|
|
if i + 1 < len(argv):
|
|
output_path = argv[i + 1]
|
|
i += 2
|
|
continue
|
|
else:
|
|
print("l3b_agent_stub: -o requires a path argument", file=sys.stderr)
|
|
sys.exit(1)
|
|
rest.append(arg)
|
|
i += 1
|
|
return rest, output_path
|
|
|
|
|
|
def main() -> int:
|
|
rest, output_path = parse_output_flag(sys.argv)
|
|
body = read_issue_body(rest)
|
|
if not body.strip():
|
|
print("l3b_agent_stub: empty issue body (no argv[1] and no stdin)", file=sys.stderr)
|
|
return 1
|
|
|
|
stack = map_issue_to_stack(body)
|
|
contract = render_contract(stack)
|
|
|
|
if output_path:
|
|
with open(output_path, "w", encoding="utf-8") as fh:
|
|
fh.write(contract)
|
|
else:
|
|
sys.stdout.write(contract)
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |