ansible_tasks.py

#!/usr/bin/env python3
"""
Extracts Ansible tasks from input.
"""

import re

EMPTY_RE = re.compile(r"^$")
START_SEGMENT_RE = re.compile(r"^(?P<type>TASK|CHECK|PLAY|RUNNING HANDLER) \[(?P<name>.*)\] \*\*\*\**$")
START_BLOCK_RE = re.compile(r"^(?P<result>ok|changed|skipping|failed|fatal): \[([A-Za-z0-9. >-]+)\].* => {$")
END_BLOCK_RE = re.compile(r"^}$")
LINE_RE = re.compile(r"^(?P<result>ok|changed|skipping): \[([A-Za-z0-9. >-]+)\]$")

class Segment():
    def __init__(self):
        self.offset = 0
        self.length = 0
        self.metadata = {}

def rule_one_of(*seq):
    def state(offset, line, end):
        for rule in seq:
            n = rule(offset, line, end)
            if n is not None:
                return n
        return state
    return state

def rule_end(handler):
    def state(offset, line, end):
        if end:
            return handler(offset, line, end)
        return None
    return state

def rule_pattern(regexp, handler):
    def state(offset, line, end):
        match = regexp.match(line)
        if match is not None:
            return handler(match, offset, line, end)
        return None
    return state

def state_init():
    def handle_segment(match, offset, line, end):
        seg = Segment()
        seg.offset = offset
        seg.metadata["type"] = "ansible_task"
        seg.metadata["name"] = match.group("type") + " " + match.group("name")
        return state_segment(seg)
    return rule_one_of(
        rule_pattern(START_SEGMENT_RE, handle_segment),
    )

def state_segment(seg):
    def handle_empty(match, offset, line, end):
        return state_end(seg)
    def handle_block(match, offset, line, end):
        if match.group("result") in ["failed", "fatal"]:
            seg.metadata["status"] = "failure"
        elif seg.metadata.get("status") in [None, "skipped"] and match.group("result") in ["ok", "changed"]:
            seg.metadata["status"] = "success"
        elif seg.metadata.get("status") in [None] and match.group("result") in ["skipping"]:
            seg.metadata["status"] = "skipped"
        return state_block(seg)
    def handle_line(match, offset, line, end):
        if seg.metadata.get("status") in [None, "skipped"] and match.group("result") in ["ok", "changed"]:
            seg.metadata["status"] = "success"
        elif seg.metadata.get("status") in [None] and match.group("result") in ["skipping"]:
            seg.metadata["status"] = "skipped"
        return None
    return rule_one_of(
        rule_pattern(EMPTY_RE, handle_empty),
        rule_pattern(START_BLOCK_RE, handle_block),
        rule_pattern(LINE_RE, handle_line),
    )

def state_block(seg):
    def handle_end(offset, line, end):
        return state_end(seg)(offset, line, end)
    def handle_block_end(match, offset, line, end):
        return state_segment(seg)
    return rule_one_of(
        rule_end(handle_end),
        rule_pattern(END_BLOCK_RE, handle_block_end),
    )

def state_end(seg):
    def state(offset, line, end):
        import json
        print(json.dumps({
            "offset": seg.offset,
            "length": offset - seg.offset,
            "metadata": seg.metadata,
        }))
        return state_init()(offset, line, end)
    return state

def main():
    import sys
    offset = 0
    state = state_init()
    for line in sys.stdin.buffer:
        line_len = len(line)
        line = line.rstrip(b"\n")
        state = state(offset, line.decode("utf-8"), False)
        offset += line_len
    state = state(offset, "", True)
main()