211 lines
7.5 KiB
Python
211 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
||
"""queue-extract.py -- independently extract the item census from WORK-QUEUE.md.
|
||
|
||
LANE: Extract - data extraction and tables.
|
||
|
||
Item = a `## Qnn` section, or (Q103) a block-initial record that opens an id with
|
||
no heading of its own, or (Q103) a later block-initial record for an id already
|
||
an item. Each item owns the lines up to the next opening.
|
||
|
||
Output: TSV id kind line end_line lines bytes role role_detail status status_line title
|
||
"""
|
||
import re
|
||
import sys
|
||
|
||
QUEUE = "/root/troupe-standup/WORK-QUEUE.md"
|
||
|
||
HEADING = re.compile(r"^##\s+(Q\d+)\b(.*)$")
|
||
STATUS = re.compile(r"^\s*\*{0,2}Status:\s*(.*)$")
|
||
FENCE = re.compile(r"^(```|~~~)")
|
||
BULLET = re.compile(r"^[ \t>|]*(?:[-*+] ?|[-*+]$)")
|
||
OPEN_SEP = re.compile(r"\s*(?:—|–|-|:|\(|\bopened\b|\bob\b|\bOPEN\b|\bOPENED\b)")
|
||
ID_TOKEN = re.compile(r"\bQ\d+\b")
|
||
|
||
|
||
def split_lines(text):
|
||
lines = text.split("\n")
|
||
trailing = text.endswith("\n")
|
||
if trailing:
|
||
lines = lines[:-1]
|
||
return lines, trailing
|
||
|
||
|
||
def line_bytes(line, is_last, trailing):
|
||
if is_last and not trailing:
|
||
return len(line.encode("utf-8"))
|
||
return len(line.encode("utf-8")) + 1
|
||
|
||
|
||
def status_value(raw):
|
||
raw = raw.strip()
|
||
while raw.startswith("**"):
|
||
raw = raw[2:].strip()
|
||
if "**" in raw:
|
||
raw = raw.split("**", 1)[0].strip()
|
||
return raw
|
||
|
||
|
||
def fence_mask(lines):
|
||
out = set()
|
||
in_fence = False
|
||
for i, ln in enumerate(lines, 1):
|
||
if FENCE.match(ln.strip()):
|
||
in_fence = not in_fence
|
||
out.add(i)
|
||
elif in_fence:
|
||
out.add(i)
|
||
return out
|
||
|
||
|
||
def record_form(line, qid):
|
||
s = line.strip()
|
||
if HEADING.match(s) and HEADING.match(s).group(1) == qid:
|
||
return "HEADING"
|
||
body = s
|
||
m = re.match(r"^[ \t>|]*(?:[-*+] ?|[-*+]$)", body)
|
||
if m:
|
||
body = body[m.end():]
|
||
if qid not in ID_TOKEN.findall(body):
|
||
return None
|
||
if body.startswith("Status:") or body.startswith("**Status:"):
|
||
return "STATUS-LEAD"
|
||
if body.startswith("**"):
|
||
inner = body[2:].lstrip("*").lstrip()
|
||
mm = ID_TOKEN.match(inner)
|
||
if mm and mm.group(0) == qid and OPEN_SEP.match(inner[mm.end():]):
|
||
return "BOLD-LEAD"
|
||
return None
|
||
|
||
|
||
def build_items(lines, fenced):
|
||
headings = []
|
||
for i, ln in enumerate(lines, 1):
|
||
if i in fenced:
|
||
continue
|
||
m = HEADING.match(ln)
|
||
if m:
|
||
headings.append((i, m.group(1), m.group(2).strip()))
|
||
heading_ids = {h[1] for h in headings}
|
||
|
||
# first record-form line per id, and reference counts
|
||
# mention scan: first record-form line per id, and reference counts
|
||
first_record = {}
|
||
for i, ln in enumerate(lines, 1):
|
||
if i in fenced:
|
||
continue
|
||
for m in ID_TOKEN.finditer(ln):
|
||
qid = m.group(0)
|
||
if qid not in first_record:
|
||
frm = record_form(ln, qid)
|
||
if frm:
|
||
first_record[qid] = (i, frm)
|
||
# Only ids WITHOUT a heading own an item: restrict the boundary set to those,
|
||
# otherwise a heading id's first prose mention (e.g. "…appended at the end of
|
||
# this file (Q89, Q90)…") carves its section short.
|
||
first_record = {qid: v for qid, v in first_record.items() if qid not in heading_ids}
|
||
|
||
# later block-initial records: a continuation block for an already-open id
|
||
later = []
|
||
for i, ln in enumerate(lines, 1):
|
||
if i in fenced or not ln.startswith("**"):
|
||
continue
|
||
prev = lines[i - 2].strip() if i >= 2 else ""
|
||
if prev != "" and prev != "---" and not prev.startswith("<!--"):
|
||
continue
|
||
for qid in ID_TOKEN.findall(ln):
|
||
frm = record_form(ln, qid)
|
||
# A first record opens an item itself (handled by the first_record
|
||
# loop below); every later block-initial record is a continuation.
|
||
if frm == "BOLD-LEAD" and (qid, i) not in first_record_keys:
|
||
later.append((i, qid, frm))
|
||
break
|
||
|
||
boundaries = sorted({h[0] for h in headings}
|
||
| {r[0] for r in first_record.values()}
|
||
| {r[0] for r in later})
|
||
|
||
def end_of(line_no):
|
||
nxt = [b for b in boundaries if b > line_no]
|
||
return (nxt[0] - 1) if nxt else len(lines)
|
||
|
||
items = []
|
||
seen = {}
|
||
for line_no, qid, rest in headings:
|
||
role = "opening" if qid not in seen else "continuation"
|
||
detail = ""
|
||
if role == "continuation":
|
||
words = rest.lstrip("—- ").split()
|
||
detail = words[0].strip("*(") if words else ""
|
||
seen.setdefault(qid, line_no)
|
||
items.append({"id": qid, "line": line_no, "kind": "heading", "role": role,
|
||
"role_detail": detail, "title": rest})
|
||
first_record_keys = {(qid, line) for qid, (line, frm) in first_record.items()}
|
||
for qid, (line_no, frm) in first_record.items():
|
||
if qid in heading_ids:
|
||
continue
|
||
raw = lines[line_no - 1]
|
||
seen.setdefault(qid, line_no)
|
||
items.append({"id": qid, "line": line_no, "kind": "record", "role": "record",
|
||
"role_detail": frm, "title": raw.strip()})
|
||
for line_no, qid, frm in later:
|
||
raw = lines[line_no - 1]
|
||
if (qid, line_no) in first_record_keys:
|
||
continue
|
||
role = "continuation" if qid in seen else "record"
|
||
seen.setdefault(qid, line_no)
|
||
items.append({"id": qid, "line": line_no, "kind": "record", "role": role,
|
||
"role_detail": frm, "title": raw.strip()})
|
||
items.sort(key=lambda it: it["line"])
|
||
return items, end_of
|
||
|
||
|
||
def extract():
|
||
with open(QUEUE, encoding="utf-8") as fh:
|
||
text = fh.read()
|
||
lines, trailing = split_lines(text)
|
||
fenced = fence_mask(lines)
|
||
items, end_of = build_items(lines, fenced)
|
||
|
||
sizes = [line_bytes(ln, i == len(lines) - 1, trailing) for i, ln in enumerate(lines)]
|
||
for it in items:
|
||
end = end_of(it["line"])
|
||
status = None
|
||
status_line = None
|
||
for off, ln in enumerate(lines[it["line"]:end], start=it["line"] + 1):
|
||
m = STATUS.match(ln)
|
||
if m:
|
||
status = status_value(m.group(1))
|
||
status_line = off
|
||
break
|
||
it["end_line"] = end
|
||
it["lines"] = end - it["line"] + 1
|
||
it["bytes"] = sum(sizes[it["line"] - 1:end])
|
||
it["status"] = status
|
||
it["status_line"] = status_line
|
||
items.sort(key=lambda it: (it["line"], it["id"]))
|
||
return items, len(text.encode("utf-8")), len(lines)
|
||
|
||
|
||
def main():
|
||
items, total_bytes, total_lines = extract()
|
||
out = sys.stdout
|
||
out.write("\t".join(["id", "kind", "line", "end_line", "lines", "bytes", "role",
|
||
"role_detail", "status", "status_line", "title"]) + "\n")
|
||
for it in items:
|
||
title = it["title"].replace("\t", " ").replace("\n", " ")
|
||
row = [it["id"], it["kind"], str(it["line"]), str(it["end_line"]),
|
||
str(it["lines"]), str(it["bytes"]), it["role"], it["role_detail"] or "-",
|
||
it["status"] or "-", str(it["status_line"]), title]
|
||
out.write("\t".join(row) + "\n")
|
||
out.write("---\n")
|
||
out.write("items=%d bytes=%d lines=%d\n" % (len(items), total_bytes, total_lines))
|
||
closed = sum(1 for it in items if it["status"])
|
||
out.write("status_present=%d\n" % closed)
|
||
open_items = [it["id"] for it in items if not it["status"]]
|
||
out.write("open_ids=%s\n" % ",".join(open_items))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|