81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from typing import Any
|
|
|
|
NEXT_FLIGHT_RE = re.compile(r"self\.__next_f\.push\(\[1,\"(.*?)\"\]\)", re.DOTALL)
|
|
|
|
|
|
def extract_next_flight_strings(html: str) -> list[str]:
|
|
"""Extract decoded Next.js Flight chunks from mobile.de HTML."""
|
|
chunks: list[str] = []
|
|
for match in NEXT_FLIGHT_RE.finditer(html):
|
|
raw = match.group(1)
|
|
try:
|
|
chunks.append(json.loads(f'"{raw}"'))
|
|
except json.JSONDecodeError:
|
|
# Резервный вариант: сохраняем работоспособность парсера,
|
|
# даже если один из фрагментов имеет нестандартное экранирование.
|
|
chunks.append(raw.encode("utf-8", errors="ignore").decode("unicode_escape", errors="ignore"))
|
|
return chunks
|
|
|
|
|
|
def extract_json_object_after(text: str, marker: str) -> dict[str, Any] | None:
|
|
"""Return JSON object that starts immediately after a marker in a decoded Flight chunk."""
|
|
marker_index = text.find(marker)
|
|
if marker_index < 0:
|
|
return None
|
|
start = text.find("{", marker_index + len(marker))
|
|
if start < 0:
|
|
return None
|
|
|
|
depth = 0
|
|
in_string = False
|
|
escaped = False
|
|
for index in range(start, len(text)):
|
|
char = text[index]
|
|
if in_string:
|
|
if escaped:
|
|
escaped = False
|
|
elif char == "\\":
|
|
escaped = True
|
|
elif char == '"':
|
|
in_string = False
|
|
continue
|
|
if char == '"':
|
|
in_string = True
|
|
elif char == "{":
|
|
depth += 1
|
|
elif char == "}":
|
|
depth -= 1
|
|
if depth == 0:
|
|
candidate = text[start : index + 1]
|
|
try:
|
|
return json.loads(candidate)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
return None
|
|
|
|
|
|
def extract_search_results(html: str) -> dict[str, Any]:
|
|
"""Extract searchResults from mobile.de SRP HTML."""
|
|
for chunk in extract_next_flight_strings(html):
|
|
if '"eventScope":"page-srp"' not in chunk or '"searchResults"' not in chunk:
|
|
continue
|
|
results = extract_json_object_after(chunk, '"searchResults":')
|
|
if isinstance(results, dict):
|
|
return results
|
|
return {}
|
|
|
|
|
|
def extract_detail_listing(html: str) -> dict[str, Any]:
|
|
"""Extract listing object from mobile.de VIP/detail HTML."""
|
|
for chunk in extract_next_flight_strings(html):
|
|
if '"eventScope":"page-vip"' not in chunk or '"listing"' not in chunk:
|
|
continue
|
|
listing = extract_json_object_after(chunk, '"listing":')
|
|
if isinstance(listing, dict):
|
|
return listing
|
|
return {}
|