265 lines
9.6 KiB
Python
265 lines
9.6 KiB
Python
"""Addresses, anchors and their resolution.
|
|
|
|
Non-normative; see spec/06-anchors.md.
|
|
"""
|
|
|
|
import re
|
|
from fractions import Fraction
|
|
|
|
from . import hashing, jcs
|
|
from .model import CONTAINERS_WITH_CONTENT, LEAVES, RUN_KINDS, Doc, anchor_class
|
|
from .text import norm
|
|
|
|
ADDRESS_RE = re.compile(
|
|
r"^(?:(?P<store>[a-z0-9][a-z0-9._-]*)/)?"
|
|
r"(?P<key>[a-z0-9][a-z0-9._-]*)"
|
|
r"#(?P<node>[a-z0-9][a-z0-9._-]*)"
|
|
r"(?:@(?P<pin>[0-9a-f]{6}))?"
|
|
r"(?:\[(?P<start>0|[1-9][0-9]*):(?P<end>0|[1-9][0-9]*)\])?$"
|
|
)
|
|
QUOTE_CAP = 200
|
|
CONTEXT_CAP = 32
|
|
|
|
|
|
def parse_address(s):
|
|
m = ADDRESS_RE.match(s)
|
|
if not m:
|
|
return None
|
|
d = m.groupdict()
|
|
out = {"store": d["store"], "key": d["key"], "node": d["node"], "pin": d["pin"], "span": None}
|
|
if d["start"] is not None:
|
|
a, b = int(d["start"]), int(d["end"])
|
|
if a > b:
|
|
return None
|
|
out["span"] = [a, b]
|
|
return out
|
|
|
|
|
|
def format_address(store, key, node, pin=None, span=None):
|
|
s = (store + "/" if store else "") + key + "#" + node
|
|
if pin:
|
|
s += "@" + pin[:6]
|
|
if span is not None:
|
|
s += "[%d:%d]" % tuple(span)
|
|
return s
|
|
|
|
|
|
def cell_context(doc, cell_id):
|
|
"""Table label and the text of the row and column header paths of a cell."""
|
|
cell = doc.nodes[cell_id]
|
|
row = doc.nodes[cell["parent"]]
|
|
table_id = doc.ancestors(cell_id)[-3]
|
|
table = doc.nodes[table_id]
|
|
cols = {c["id"]: c for c in table.get("attrs", {}).get("columns", [])}
|
|
col = cols.get(cell.get("attrs", {}).get("column"), {"header_path": []})
|
|
|
|
def txt(nid):
|
|
return norm(hashing.plain(doc, nid))
|
|
|
|
return {
|
|
"table_label": table.get("label"),
|
|
"row_path": [txt(x) for x in row.get("attrs", {}).get("header_path", [])],
|
|
"column_path": [txt(x) for x in col["header_path"]],
|
|
}
|
|
|
|
|
|
def make_anchor(doc, values, node, rev, span=None, store=None):
|
|
n = doc.nodes[node]
|
|
a = {"format": "scienting-anchor/1"}
|
|
if store:
|
|
a["store"] = store
|
|
a.update({"doc": doc.key, "node": node, "kind": n["kind"], "own_hash": values[node]["own_hash"],
|
|
"node_hash": values[node]["node_hash"]})
|
|
if n["kind"] in LEAVES:
|
|
text = hashing.plain(doc, node)
|
|
if span is not None:
|
|
a["span"] = list(span)
|
|
exact = text[span[0]:span[1]]
|
|
pre, suf = text[max(0, span[0] - CONTEXT_CAP):span[0]], text[span[1]:span[1] + CONTEXT_CAP]
|
|
else:
|
|
exact, pre, suf = text, "", ""
|
|
q = {}
|
|
if len(exact) <= QUOTE_CAP:
|
|
q["exact"] = exact
|
|
else:
|
|
q["exact_hash"] = hashing.PLAIN(norm(exact))
|
|
q["length"] = len(exact)
|
|
if pre:
|
|
q["prefix"] = pre
|
|
if suf:
|
|
q["suffix"] = suf
|
|
a["quote"] = q
|
|
if n["kind"] == "cell":
|
|
a["context"] = cell_context(doc, node)
|
|
elif anchor_class(n["kind"]) == "container-without-content":
|
|
a["leaves"] = [[x, values[x]["match_hash"]] for x in doc.leaves(node)]
|
|
a["rev"] = rev
|
|
return a
|
|
|
|
|
|
def _quote_hits(text, q):
|
|
"""Start positions where the quote matches `text`."""
|
|
if "exact" in q:
|
|
ex = q["exact"]
|
|
hits = [i for i in range(len(text) - len(ex) + 1) if text.startswith(ex, i)]
|
|
length = len(ex)
|
|
else:
|
|
length = q["length"]
|
|
hits = [i for i in range(len(text) - length + 1)
|
|
if hashing.PLAIN(norm(text[i:i + length])) == q["exact_hash"]]
|
|
if len(hits) > 1 and ("prefix" in q or "suffix" in q):
|
|
pre, suf = q.get("prefix", ""), q.get("suffix", "")
|
|
hits = [i for i in hits if text[:i].endswith(pre) and text[i + length:].startswith(suf)]
|
|
return hits, length
|
|
|
|
|
|
def _drift(state, doc, anchor, nid, threshold):
|
|
leaves = set(doc.leaves(nid))
|
|
touched = set()
|
|
count = 0
|
|
for c in state.corrections:
|
|
if c["index"] <= anchor["rev"]:
|
|
continue
|
|
inside = nid in (c.get("path") or []) or nid in (c.get("to_path") or [])
|
|
t = set(c.get("touched", []))
|
|
if inside and t:
|
|
count += 1
|
|
touched |= t
|
|
denom = len(leaves | touched)
|
|
share = Fraction(len(touched), denom) if denom else Fraction(0)
|
|
level = "none" if count == 0 else ("high" if share > threshold else "low")
|
|
return {"count": count, "share": _decimal(share), "level": level}
|
|
|
|
|
|
def _decimal(fr):
|
|
"""A share as a decimal string with four places, rounded half up."""
|
|
v = (fr * 10000 + Fraction(1, 2)).__floor__()
|
|
return "%d.%04d" % (v // 10000, v % 10000)
|
|
|
|
|
|
def _follow(state, nid):
|
|
"""Follow aliases and retirements. Returns (live id or None, chain, via)."""
|
|
live = {n["id"] for n in state.nodes}
|
|
table = {e["id"]: e for e in state.ids}
|
|
chain, via = [], None
|
|
seen = set()
|
|
while nid not in live:
|
|
if nid in seen or nid not in table:
|
|
return None, chain, via
|
|
seen.add(nid)
|
|
e = table[nid]
|
|
if e["kind"] == "retired" or via is None:
|
|
via = e["kind"]
|
|
chain.append(nid)
|
|
nid = e.get("successor")
|
|
if nid is None:
|
|
return None, chain, via
|
|
return nid, chain, via
|
|
|
|
|
|
def _relabel_reverted_own(state, doc, nid, anchor):
|
|
"""The own_hash the node had before the relabel corrections since the anchor, or None."""
|
|
rel = [c for c in state.corrections if c["index"] > anchor["rev"] and c.get("target") == nid]
|
|
if not rel or any(c["op"] != "relabel" for c in rel):
|
|
return None
|
|
n = dict(doc.nodes[nid], label=rel[0]["from"])
|
|
tmp = Doc(doc.key, [n if x["id"] == nid else x for x in [doc.nodes[i] for i in doc.order]])
|
|
return hashing.PLAIN(jcs.dumpb(hashing.own_content(tmp, nid, lambda h: hashing.PLAIN(
|
|
norm(hashing.plain(tmp, h))))))
|
|
|
|
|
|
def resolve(state, anchor, threshold=Fraction(1, 4)):
|
|
doc = state.doc()
|
|
values = hashing.compute(doc)
|
|
kind = anchor.get("kind")
|
|
original = anchor["node"]
|
|
nid, chain, via = _follow(state, original)
|
|
live_kind = doc.nodes[nid]["kind"] if nid else None
|
|
kind = kind or live_kind or next((e["node_kind"] for e in state.ids if e["id"] == original), None)
|
|
cls = anchor_class(kind) if kind else "leaf"
|
|
is_span = "span" in anchor
|
|
result = {"node": nid}
|
|
retired = via == "retired"
|
|
|
|
def out(res, **kw):
|
|
r = {"result": res, "node": kw.pop("node", nid)}
|
|
if r["node"]:
|
|
r["node_hash"] = values[r["node"]]["node_hash"]
|
|
if chain:
|
|
r["via"] = chain
|
|
r.update(kw)
|
|
return r
|
|
|
|
# The cell rule: a retired cell is found again by its table label and header paths.
|
|
def cell_rule():
|
|
ctx = anchor.get("context")
|
|
q = anchor.get("quote")
|
|
if kind != "cell" or not ctx or not q:
|
|
return None
|
|
found = []
|
|
for x in doc.order:
|
|
if doc.nodes[x]["kind"] != "cell":
|
|
continue
|
|
if cell_context(doc, x) != ctx:
|
|
continue
|
|
hits, ln = _quote_hits(hashing.plain(doc, x), q)
|
|
if len(hits) == 1:
|
|
found.append((x, hits[0], ln))
|
|
if len(found) == 1:
|
|
x, i, ln = found[0]
|
|
kw = {"node": x}
|
|
if is_span:
|
|
kw["span"] = [i, i + ln]
|
|
return out("reattached", **kw)
|
|
return None
|
|
|
|
if nid is None:
|
|
return cell_rule() or out("stale", node=None, reason="retired without a successor" if chain
|
|
else "unknown node")
|
|
|
|
if cls == "leaf":
|
|
text = hashing.plain(doc, nid)
|
|
if not retired and values[nid]["node_hash"] == anchor["node_hash"]:
|
|
return out("current", **({"span": anchor["span"]} if is_span else {}))
|
|
q = anchor.get("quote")
|
|
if retired:
|
|
if q:
|
|
hits, ln = _quote_hits(text, q)
|
|
if (is_span and len(hits) == 1) or (not is_span and hits):
|
|
return out("moved", **({"span": [hits[0], hits[0] + ln]} if is_span else {}))
|
|
return cell_rule() or out("stale", reason="the successor does not contain the quote")
|
|
if q:
|
|
hits, ln = _quote_hits(text, q)
|
|
if len(hits) == 1:
|
|
return out("reattached", **({"span": [hits[0], hits[0] + ln]} if is_span else {}))
|
|
own = _relabel_reverted_own(state, doc, nid, anchor)
|
|
if own is not None and own == anchor["own_hash"]:
|
|
return out("reattached", reason="relabel")
|
|
return out("stale", reason="the node changed and the quote does not occur exactly once")
|
|
|
|
if cls == "container-with-content":
|
|
if values[nid]["own_hash"] == anchor["own_hash"]:
|
|
if retired:
|
|
return out("moved")
|
|
if values[nid]["node_hash"] == anchor["node_hash"]:
|
|
return out("current")
|
|
return out("changed-below", drift=_drift(state, doc, anchor, nid, threshold))
|
|
if not retired:
|
|
own = _relabel_reverted_own(state, doc, nid, anchor)
|
|
if own is not None and own == anchor["own_hash"]:
|
|
return out("reattached", reason="relabel")
|
|
return out("stale", reason="the node's own content changed")
|
|
|
|
# Container without content.
|
|
if retired:
|
|
orig = anchor.get("leaves", [])
|
|
now = set(doc.leaves(nid))
|
|
now_match = {values[x]["match_hash"] for x in now}
|
|
held = sum(1 for i, m in orig if i in now or m in now_match)
|
|
if orig and 2 * held >= len(orig):
|
|
return out("moved")
|
|
return out("stale", reason="the successor holds fewer than half of the original leaves")
|
|
if values[nid]["node_hash"] == anchor["node_hash"]:
|
|
return out("current")
|
|
return out("changed-below", drift=_drift(state, doc, anchor, nid, threshold))
|