170 lines
4.7 KiB
Python
170 lines
4.7 KiB
Python
import json
|
|
import re
|
|
from html.parser import HTMLParser
|
|
from typing import Any, Mapping, Optional
|
|
|
|
|
|
_MARKDOWN_CODE_FENCE_RE = re.compile(r"^\s*```[a-zA-Z0-9_-]*\s*$")
|
|
_MARKDOWN_HEADING_RE = re.compile(r"^\s{0,3}#{1,6}\s*")
|
|
_MARKDOWN_TRAILING_HEADING_RE = re.compile(r"\s+#{1,6}\s*$")
|
|
_MARKDOWN_BULLET_RE = re.compile(r"^\s*[-*+]\s+")
|
|
_MARKDOWN_BLOCKQUOTE_RE = re.compile(r"^\s*>\s?")
|
|
_MARKDOWN_TABLE_SEPARATOR_RE = re.compile(r"^\s*\|?\s*:?-{3,}:?\s*(\|\s*:?-{3,}:?\s*)+\|?\s*$")
|
|
|
|
|
|
class ContractHTMLTextExtractor(HTMLParser):
|
|
_BLOCK_TAGS = {
|
|
"address",
|
|
"article",
|
|
"aside",
|
|
"blockquote",
|
|
"br",
|
|
"dd",
|
|
"div",
|
|
"dl",
|
|
"dt",
|
|
"fieldset",
|
|
"figcaption",
|
|
"figure",
|
|
"footer",
|
|
"form",
|
|
"h1",
|
|
"h2",
|
|
"h3",
|
|
"h4",
|
|
"h5",
|
|
"h6",
|
|
"header",
|
|
"hr",
|
|
"li",
|
|
"main",
|
|
"nav",
|
|
"ol",
|
|
"p",
|
|
"pre",
|
|
"section",
|
|
"table",
|
|
"tbody",
|
|
"td",
|
|
"tfoot",
|
|
"th",
|
|
"thead",
|
|
"tr",
|
|
"ul",
|
|
}
|
|
_SKIP_TAGS = {"script", "style"}
|
|
|
|
def __init__(self):
|
|
super().__init__(convert_charrefs=True)
|
|
self._parts: list[str] = []
|
|
self._skip_depth = 0
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]):
|
|
normalized = tag.lower()
|
|
if normalized in self._SKIP_TAGS:
|
|
self._skip_depth += 1
|
|
return
|
|
if normalized in self._BLOCK_TAGS:
|
|
self._append_break()
|
|
|
|
def handle_endtag(self, tag: str):
|
|
normalized = tag.lower()
|
|
if normalized in self._SKIP_TAGS and self._skip_depth > 0:
|
|
self._skip_depth -= 1
|
|
return
|
|
if normalized in self._BLOCK_TAGS:
|
|
self._append_break()
|
|
|
|
def handle_data(self, data: str):
|
|
if self._skip_depth or not data:
|
|
return
|
|
self._parts.append(data)
|
|
|
|
def get_text(self) -> str:
|
|
text = "".join(self._parts)
|
|
text = re.sub(r"[ \t\r\f\v]+", " ", text)
|
|
text = re.sub(r" *\n *", "\n", text)
|
|
text = re.sub(r"\n{3,}", "\n\n", text)
|
|
return text.strip()
|
|
|
|
def _append_break(self):
|
|
if self._parts and not str(self._parts[-1]).endswith("\n"):
|
|
self._parts.append("\n")
|
|
|
|
|
|
def normalize_contract_content(value: Any) -> str:
|
|
if isinstance(value, Mapping):
|
|
return _extract_mapping_contract_content(value)
|
|
|
|
text = str(value or "").strip()
|
|
if not text:
|
|
return ""
|
|
|
|
parsed = parse_contract_json_text(text)
|
|
if isinstance(parsed, Mapping):
|
|
return _extract_mapping_contract_content(parsed) or text
|
|
|
|
if looks_like_html(text):
|
|
stripped = html_to_text(text)
|
|
return markdown_to_text(stripped or text)
|
|
|
|
return markdown_to_text(text)
|
|
|
|
|
|
def parse_contract_json_text(value: str) -> Optional[Mapping[str, Any]]:
|
|
cleaned = str(value or "").strip()
|
|
if not cleaned:
|
|
return None
|
|
if cleaned.startswith("```"):
|
|
cleaned = cleaned.removeprefix("```json").removeprefix("```").strip()
|
|
if cleaned.endswith("```"):
|
|
cleaned = cleaned[:-3].strip()
|
|
if not cleaned.startswith("{"):
|
|
return None
|
|
try:
|
|
parsed = json.loads(cleaned)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
return parsed if isinstance(parsed, Mapping) else None
|
|
|
|
|
|
def html_to_text(value: str) -> str:
|
|
parser = ContractHTMLTextExtractor()
|
|
parser.feed(str(value or ""))
|
|
parser.close()
|
|
return parser.get_text()
|
|
|
|
|
|
def markdown_to_text(value: str) -> str:
|
|
lines: list[str] = []
|
|
for raw_line in str(value or "").splitlines():
|
|
line = raw_line.strip()
|
|
if not line:
|
|
lines.append("")
|
|
continue
|
|
if _MARKDOWN_CODE_FENCE_RE.match(line) or _MARKDOWN_TABLE_SEPARATOR_RE.match(line):
|
|
continue
|
|
line = _MARKDOWN_BLOCKQUOTE_RE.sub("", line)
|
|
line = _MARKDOWN_HEADING_RE.sub("", line)
|
|
line = _MARKDOWN_TRAILING_HEADING_RE.sub("", line)
|
|
line = _MARKDOWN_BULLET_RE.sub("", line)
|
|
line = re.sub(r"(\*\*|__)(.*?)\1", r"\2", line)
|
|
line = re.sub(r"`([^`]+)`", r"\1", line)
|
|
line = line.replace("\\#", "#").strip()
|
|
lines.append(line)
|
|
text = "\n".join(lines)
|
|
text = re.sub(r"\n{3,}", "\n\n", text)
|
|
return text.strip()
|
|
|
|
|
|
def looks_like_html(value: str) -> bool:
|
|
return bool(re.search(r"</?[a-zA-Z][^>]*>", str(value or "")))
|
|
|
|
|
|
def _extract_mapping_contract_content(payload: Mapping[str, Any]) -> str:
|
|
for key in ("content", "text", "body", "markdown", "document", "html"):
|
|
value = payload.get(key)
|
|
if isinstance(value, str) and value.strip():
|
|
return normalize_contract_content(value)
|
|
return ""
|