89 lines
3.0 KiB
Python
89 lines
3.0 KiB
Python
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
START_MARKER = "/*WellFargoProprietary%"
|
|
END_MARKER = "%WellFargoProprietary*/"
|
|
|
|
|
|
class WellsFargoPayloadExtractor:
|
|
"""Extract wrapped WellFargo payloads from HAR entries and parse them as JSON."""
|
|
|
|
def __init__(self) -> None:
|
|
self.objects: list[Any] = []
|
|
self.errors: list[str] = []
|
|
|
|
def _normalize_text(self, raw_text: Any) -> str | None:
|
|
"""Return a normalized text payload or None when input is not a string."""
|
|
if not isinstance(raw_text, str):
|
|
return None
|
|
|
|
text = raw_text.strip()
|
|
|
|
# Some HAR exports store content.text as an escaped JSON string literal.
|
|
if len(text) >= 2 and text[0] == '"' and text[-1] == '"':
|
|
try:
|
|
decoded = json.loads(text)
|
|
if isinstance(decoded, str):
|
|
text = decoded
|
|
except json.JSONDecodeError:
|
|
self.errors.append("Could not decode an escaped text wrapper")
|
|
|
|
return text
|
|
|
|
def extract_from_data(self, har_data: dict[str, Any]) -> list[Any]:
|
|
"""Parse and store objects from log.entries[*].response.content.text."""
|
|
self.objects = []
|
|
self.errors = []
|
|
|
|
entries = har_data.get("log", {}).get("entries", [])
|
|
if not isinstance(entries, list):
|
|
self.errors.append("Expected log.entries to be a list")
|
|
return self.objects
|
|
|
|
for index, entry in enumerate(entries):
|
|
text_value = (
|
|
entry.get("response", {})
|
|
.get("content", {})
|
|
.get("text")
|
|
if isinstance(entry, dict)
|
|
else None
|
|
)
|
|
text = self._normalize_text(text_value)
|
|
if not text or not text.startswith(START_MARKER):
|
|
continue
|
|
|
|
end_idx = text.find(END_MARKER, len(START_MARKER))
|
|
if end_idx == -1:
|
|
self.errors.append(f"Entry {index}: missing end marker")
|
|
continue
|
|
|
|
payload = text[len(START_MARKER) : end_idx]
|
|
if not payload:
|
|
self.errors.append(f"Entry {index}: payload between markers is empty")
|
|
continue
|
|
|
|
try:
|
|
self.objects.append(json.loads(payload))
|
|
except json.JSONDecodeError as ex:
|
|
self.errors.append(f"Entry {index}: invalid payload JSON ({ex.msg})")
|
|
|
|
return self.objects
|
|
|
|
def extract_from_file(self, input_path: str | Path) -> list[Any]:
|
|
"""Load HAR JSON from a file path and extract payload objects."""
|
|
path = Path(input_path)
|
|
with path.open("r", encoding="utf-8") as f:
|
|
har_data = json.load(f)
|
|
return self.extract_from_data(har_data)
|
|
|
|
def get_objects(self) -> list[Any]:
|
|
"""Return the latest parsed objects."""
|
|
return self.objects
|
|
|
|
|
|
def extract_wf_objects(har_data: dict[str, Any]) -> list[Any]:
|
|
"""Backward-compatible functional API."""
|
|
return WellsFargoPayloadExtractor().extract_from_data(har_data)
|
|
|