release: bump version to 0.43.0
This commit is contained in:
linkong
358
backend/app/services/datasource_mapping.py
Normal file
358
backend/app/services/datasource_mapping.py
Normal file
@@ -0,0 +1,358 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user