73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Iterable
|
|
|
|
|
|
def save_to_json(data: Any, filename: str | Path) -> None:
|
|
path = Path(filename)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def first_non_empty(values: Iterable[Any]) -> Any | None:
|
|
for value in values:
|
|
if value not in (None, "", [], {}, ()):
|
|
return value
|
|
return None
|
|
|
|
|
|
# Регулярные выражения для VIN, lot и price.
|
|
VIN_RE = re.compile(r"\b([A-HJ-NPR-Z0-9]{17})\b", re.IGNORECASE)
|
|
LOT_RE = re.compile(r"\b(\d{7,10})\b")
|
|
PRICE_RE = re.compile(r"\$\s?([\d,]+(?:\.\d{1,2})?)")
|
|
|
|
|
|
def deep_find_key(obj, target_keys: set[str], max_depth: int = 64, _depth: int = 0) -> list:
|
|
# Рекурсивно ищет значения по набору ключей в произвольном JSON-дереве.
|
|
found = []
|
|
if _depth >= max_depth:
|
|
return found
|
|
if isinstance(obj, dict):
|
|
for key, value in obj.items():
|
|
if key.lower() in target_keys:
|
|
found.append(value)
|
|
found.extend(deep_find_key(value, target_keys, max_depth=max_depth, _depth=_depth + 1))
|
|
elif isinstance(obj, list):
|
|
for item in obj:
|
|
found.extend(deep_find_key(item, target_keys, max_depth=max_depth, _depth=_depth + 1))
|
|
return found
|
|
|
|
|
|
def deep_find_all_keys(
|
|
payloads: list,
|
|
field_map: dict[str, set[str]],
|
|
max_depth: int = 64,
|
|
) -> dict[str, list]:
|
|
"""Извлекает все нужные поля за один проход по JSON."""
|
|
# Готовим обратную карту: нормализованный ключ -> имя поля.
|
|
reverse: dict[str, str] = {}
|
|
for field_name, keys in field_map.items():
|
|
for k in keys:
|
|
reverse[k.lower()] = field_name
|
|
|
|
result: dict[str, list] = {f: [] for f in field_map}
|
|
|
|
def _recurse(obj: Any, depth: int) -> None:
|
|
if depth >= max_depth:
|
|
return
|
|
if isinstance(obj, dict):
|
|
for k, v in obj.items():
|
|
field = reverse.get(k.lower())
|
|
if field is not None:
|
|
result[field].append(v)
|
|
_recurse(v, depth + 1)
|
|
elif isinstance(obj, list):
|
|
for item in obj:
|
|
_recurse(item, depth + 1)
|
|
|
|
for payload in payloads:
|
|
_recurse(payload, 0)
|
|
|
|
return result
|