359 lines
12 KiB
Python
359 lines
12 KiB
Python
"""Deterministic mapping support for custom data sources."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.target_schema_registry import TargetSchema, get_target_schema
|
|
|
|
SECRET_KEY_PATTERN = re.compile(
|
|
r"(token|secret|password|passwd|authorization|api[_-]?key|client[_-]?secret)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
class MappingError(ValueError):
|
|
"""Raised when a mapping definition cannot be executed."""
|
|
|
|
|
|
def stable_payload_hash(payload: Any) -> str:
|
|
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode()
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def redact_for_llm(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
redacted = {}
|
|
for key, item in value.items():
|
|
if SECRET_KEY_PATTERN.search(str(key)):
|
|
redacted[key] = "[REDACTED]"
|
|
else:
|
|
redacted[key] = redact_for_llm(item)
|
|
return redacted
|
|
if isinstance(value, list):
|
|
return [redact_for_llm(item) for item in value[:20]]
|
|
return value
|
|
|
|
|
|
def extract_path(payload: Any, path: str | None) -> Any:
|
|
if not path or path == "$":
|
|
return payload
|
|
|
|
normalized = path.strip()
|
|
if normalized.startswith("$."):
|
|
normalized = normalized[2:]
|
|
elif normalized.startswith("$"):
|
|
normalized = normalized[1:]
|
|
normalized = normalized.strip(".")
|
|
if not normalized:
|
|
return payload
|
|
|
|
current = payload
|
|
for raw_segment in normalized.split("."):
|
|
segment = raw_segment.strip()
|
|
if not segment:
|
|
continue
|
|
|
|
list_all = segment.endswith("[*]")
|
|
if list_all:
|
|
segment = segment[:-3]
|
|
|
|
index = None
|
|
match = re.fullmatch(r"(.+)\[(\d+)\]", segment)
|
|
if match:
|
|
segment = match.group(1)
|
|
index = int(match.group(2))
|
|
|
|
if segment:
|
|
if isinstance(current, dict):
|
|
current = current.get(segment)
|
|
else:
|
|
return None
|
|
|
|
if list_all:
|
|
return current if isinstance(current, list) else []
|
|
|
|
if index is not None:
|
|
if not isinstance(current, list) or index >= len(current):
|
|
return None
|
|
current = current[index]
|
|
|
|
return current
|
|
|
|
|
|
def _convert_value(value: Any, target_type: str | None) -> Any:
|
|
if value is None or target_type in (None, "", "any"):
|
|
return value
|
|
|
|
if target_type == "string":
|
|
return str(value)
|
|
if target_type == "integer":
|
|
return int(value)
|
|
if target_type == "float":
|
|
return float(value)
|
|
if target_type == "boolean":
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, str):
|
|
return value.strip().lower() in {"1", "true", "yes", "y", "on"}
|
|
return bool(value)
|
|
if target_type == "datetime":
|
|
if isinstance(value, datetime):
|
|
return value
|
|
if isinstance(value, (int, float)):
|
|
return datetime.fromtimestamp(value)
|
|
if isinstance(value, str):
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
return value
|
|
if target_type == "object":
|
|
if isinstance(value, dict):
|
|
return value
|
|
raise ValueError("expected object")
|
|
if target_type == "array":
|
|
if isinstance(value, list):
|
|
return value
|
|
raise ValueError("expected array")
|
|
|
|
return value
|
|
|
|
|
|
def _apply_enum(value: Any, enum_map: Any) -> Any:
|
|
if not isinstance(enum_map, dict):
|
|
return value
|
|
key = str(value)
|
|
return enum_map.get(key, enum_map.get(value, value))
|
|
|
|
|
|
def _map_one(item: Any, field_mapping: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
|
|
output: dict[str, Any] = {}
|
|
errors: list[str] = []
|
|
|
|
for field_name, rule in field_mapping.items():
|
|
if isinstance(rule, str):
|
|
rule = {"path": rule}
|
|
if not isinstance(rule, dict):
|
|
errors.append(f"{field_name}: mapping rule must be an object or path string")
|
|
continue
|
|
|
|
value = extract_path(item, rule.get("path"))
|
|
if value is None and "default" in rule:
|
|
value = rule.get("default")
|
|
value = _apply_enum(value, rule.get("enum"))
|
|
|
|
try:
|
|
value = _convert_value(value, rule.get("type"))
|
|
except (TypeError, ValueError) as exc:
|
|
errors.append(f"{field_name}: failed to convert value {value!r}: {exc}")
|
|
continue
|
|
|
|
if value is not None or rule.get("include_null", False):
|
|
output[field_name] = value
|
|
|
|
return output, errors
|
|
|
|
|
|
def execute_mapping(
|
|
payload: Any,
|
|
mapping_json: dict[str, Any],
|
|
target_schema: str | TargetSchema,
|
|
*,
|
|
limit: int | None = None,
|
|
) -> dict[str, Any]:
|
|
schema = get_target_schema(target_schema) if isinstance(target_schema, str) else target_schema
|
|
source = mapping_json.get("source") or {}
|
|
fields = mapping_json.get("fields")
|
|
if not isinstance(fields, dict) or not fields:
|
|
raise MappingError("mapping_json.fields must be a non-empty object")
|
|
|
|
items_path = source.get("items_path") or mapping_json.get("items_path") or "$"
|
|
items = extract_path(payload, items_path)
|
|
if isinstance(items, dict):
|
|
items = [items]
|
|
elif not isinstance(items, list):
|
|
items = []
|
|
|
|
if limit is not None:
|
|
items = items[:limit]
|
|
|
|
mapped_records: list[dict[str, Any]] = []
|
|
errors: list[dict[str, Any]] = []
|
|
for index, item in enumerate(items):
|
|
mapped, mapping_errors = _map_one(item, fields)
|
|
validated, validation_errors = schema.validate_record(mapped)
|
|
all_errors = mapping_errors + validation_errors
|
|
if all_errors:
|
|
errors.append({"index": index, "errors": all_errors, "record": mapped})
|
|
continue
|
|
if validated is not None:
|
|
mapped_records.append(validated)
|
|
|
|
return {
|
|
"target_schema": schema.key,
|
|
"total_items": len(items),
|
|
"mapped_count": len(mapped_records),
|
|
"failed_count": len(errors),
|
|
"records": mapped_records,
|
|
"errors": errors,
|
|
}
|
|
|
|
|
|
def build_heuristic_mapping(sample_payload: Any, target_schema_key: str) -> dict[str, Any]:
|
|
schema = get_target_schema(target_schema_key)
|
|
items_path = "$"
|
|
sample_item = sample_payload
|
|
if isinstance(sample_payload, dict):
|
|
for key in ("data", "items", "results", "features", "vessels"):
|
|
candidate = sample_payload.get(key)
|
|
if isinstance(candidate, list) and candidate:
|
|
items_path = f"$.{key}[*]"
|
|
sample_item = candidate[0]
|
|
break
|
|
elif isinstance(sample_payload, list) and sample_payload:
|
|
items_path = "$"
|
|
sample_item = sample_payload[0]
|
|
|
|
available = _flatten_keys(sample_item if isinstance(sample_item, dict) else {})
|
|
fields: dict[str, Any] = {}
|
|
for field in schema.fields:
|
|
candidate = _best_field_match(field.name, available)
|
|
if candidate:
|
|
fields[field.name] = {"path": f"$.{candidate}", "type": field.type}
|
|
elif field.name == "data" and target_schema_key == "generic_records":
|
|
fields[field.name] = {"path": "$", "type": "object"}
|
|
elif not field.required:
|
|
fields[field.name] = {"path": f"$.{field.name}", "type": field.type, "default": None}
|
|
|
|
return {
|
|
"source": {"items_path": items_path},
|
|
"fields": fields,
|
|
"meta": {
|
|
"generated_by": "heuristic",
|
|
"requires_review": True,
|
|
},
|
|
}
|
|
|
|
|
|
def _flatten_keys(payload: dict[str, Any], prefix: str = "") -> list[str]:
|
|
keys: list[str] = []
|
|
for key, value in payload.items():
|
|
dotted = f"{prefix}.{key}" if prefix else str(key)
|
|
keys.append(dotted)
|
|
if isinstance(value, dict):
|
|
keys.extend(_flatten_keys(value, dotted))
|
|
return keys
|
|
|
|
|
|
def _best_field_match(field_name: str, candidates: list[str]) -> str | None:
|
|
aliases = {
|
|
"lat": ("lat", "latitude", "y"),
|
|
"lon": ("lon", "lng", "longitude", "x"),
|
|
"mmsi": ("mmsi",),
|
|
"sog": ("sog", "speed", "speedOverGround"),
|
|
"cog": ("cog", "course", "courseOverGround"),
|
|
"received_at": ("received_at", "timestamp", "time", "updated_at"),
|
|
"observed_at": ("observed_at", "timestamp", "time", "updated_at"),
|
|
"source_id": ("id", "source_id", "uuid"),
|
|
}.get(field_name, (field_name,))
|
|
|
|
lowered = {candidate.lower(): candidate for candidate in candidates}
|
|
for alias in aliases:
|
|
if alias.lower() in lowered:
|
|
return lowered[alias.lower()]
|
|
for candidate in candidates:
|
|
tail = candidate.split(".")[-1].lower()
|
|
if tail in {alias.lower() for alias in aliases}:
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def _parse_datetime(value: Any) -> datetime | None:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value
|
|
if isinstance(value, str):
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
return None
|
|
|
|
|
|
async def persist_mapped_records(
|
|
db: AsyncSession,
|
|
*,
|
|
datasource_name: str,
|
|
datasource_config_id: int,
|
|
target_schema: str,
|
|
records: list[dict[str, Any]],
|
|
mapping_version: int,
|
|
) -> int:
|
|
"""Persist validated mapped records to the destination for a target schema."""
|
|
if target_schema == "vessel_ais":
|
|
from app.models.vessel import VesselPosition
|
|
|
|
for record in records:
|
|
db.add(
|
|
VesselPosition(
|
|
mmsi=record["mmsi"],
|
|
lat=record["lat"],
|
|
lon=record["lon"],
|
|
sog=record.get("sog"),
|
|
cog=record.get("cog"),
|
|
heading=record.get("heading"),
|
|
received_at=_parse_datetime(record.get("received_at")) or datetime.now(UTC),
|
|
)
|
|
)
|
|
await db.commit()
|
|
return len(records)
|
|
|
|
from app.models.collected_data import CollectedData
|
|
|
|
collected_at = datetime.now(UTC)
|
|
for index, record in enumerate(records):
|
|
if target_schema == "geo_points":
|
|
source_id = record.get("source_id") or f"{datasource_config_id}:{index}"
|
|
name = record.get("name")
|
|
metadata = {
|
|
"latitude": record.get("lat"),
|
|
"longitude": record.get("lon"),
|
|
"type": record.get("type"),
|
|
"mapping_version": mapping_version,
|
|
"target_schema": target_schema,
|
|
**(record.get("metadata") or {}),
|
|
}
|
|
reference_date = _parse_datetime(record.get("observed_at"))
|
|
else:
|
|
source_id = record.get("source_id") or f"{datasource_config_id}:{index}"
|
|
name = None
|
|
metadata = {
|
|
"data": record.get("data") or {},
|
|
"mapping_version": mapping_version,
|
|
"target_schema": target_schema,
|
|
}
|
|
reference_date = _parse_datetime(record.get("observed_at"))
|
|
|
|
db.add(
|
|
CollectedData(
|
|
source=datasource_name,
|
|
source_id=str(source_id),
|
|
entity_key=f"{datasource_name}:{source_id}",
|
|
data_type=target_schema,
|
|
name=name,
|
|
title=name,
|
|
extra_data=metadata,
|
|
collected_at=collected_at,
|
|
reference_date=reference_date,
|
|
is_valid=1,
|
|
is_current=True,
|
|
change_type="created",
|
|
change_summary={},
|
|
)
|
|
)
|
|
|
|
await db.commit()
|
|
return len(records)
|