125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Iterable
|
|
from urllib.parse import urlparse
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .models import DataSource, ExternalObservation
|
|
|
|
|
|
SOURCE_DEFAULTS = {
|
|
"rf4db": ("RF4DB", "https://rf4db.com", 70),
|
|
"rf4stat-fishing": ("RF4-STAT fishing", "https://rf4-stat.ru/fishing/", 65),
|
|
"rf4stat-post": ("RF4-STAT posts", "https://rf4-stat.ru/posts/", 60),
|
|
"rf4map": ("RF4MAP", "https://rf4map.ru", 55),
|
|
"rf4posts-spot": ("RF4 Posts spots", "https://rf4-posts.com", 50),
|
|
}
|
|
SOURCE_HOSTS = {
|
|
"rf4db": {"rf4db.com", "download.rf4db.com"},
|
|
"rf4stat-fishing": {"rf4-stat.ru"},
|
|
"rf4stat-post": {"rf4-stat.ru"},
|
|
"rf4map": {"rf4map.ru"},
|
|
"rf4posts-spot": {"rf4-posts.com"},
|
|
}
|
|
|
|
|
|
class CommunityImportError(ValueError):
|
|
pass
|
|
|
|
|
|
def stage_observations(
|
|
session: Session, records: Iterable[dict[str, Any]], *, fetched_at: datetime | None = None,
|
|
) -> tuple[int, int]:
|
|
fetched_at = fetched_at or datetime.now(timezone.utc)
|
|
created = updated = 0
|
|
for raw in records:
|
|
payload = _json_payload(raw)
|
|
source_system = _required(payload, "source_system", 50)
|
|
if source_system not in SOURCE_DEFAULTS:
|
|
raise CommunityImportError(f"unsupported source_system: {source_system}")
|
|
external_id = _required(payload, "source_external_id", 200)
|
|
source = session.get(DataSource, source_system)
|
|
if source is None:
|
|
name, base_url, confidence = SOURCE_DEFAULTS[source_system]
|
|
source = DataSource(key=source_system, name=name, base_url=base_url, default_confidence=confidence, enabled=False)
|
|
session.add(source)
|
|
session.flush()
|
|
observation = session.scalar(select(ExternalObservation).where(
|
|
ExternalObservation.source_system == source_system,
|
|
ExternalObservation.source_external_id == external_id,
|
|
))
|
|
source_url = _required(payload, "source_url", 2000)
|
|
if urlparse(source_url).scheme != "https" or urlparse(source_url).hostname not in SOURCE_HOSTS[source_system]:
|
|
raise CommunityImportError("source_url does not match source_system")
|
|
values = {
|
|
"source_url": source_url,
|
|
"fish_name": _required(payload, "fish", 200),
|
|
"fish_external_id": _optional(payload, "fish_external_id", 200),
|
|
"waterbody_name": _required(payload, "waterbody", 200),
|
|
"waterbody_external_id": _optional(payload, "waterbody_external_id", 200),
|
|
"x": _integer(payload.get("x"), maximum=10_000),
|
|
"y": _integer(payload.get("y"), maximum=10_000),
|
|
"weight_g": _integer(payload.get("weight_g"), minimum=1, maximum=3_000_000),
|
|
"published_at": _datetime(payload.get("published_at")),
|
|
"last_seen_at": fetched_at, "payload": payload,
|
|
}
|
|
if observation is None:
|
|
observation = ExternalObservation(
|
|
source_system=source_system, source_external_id=external_id,
|
|
first_seen_at=fetched_at, status="staged", **values,
|
|
)
|
|
session.add(observation)
|
|
created += 1
|
|
else:
|
|
for key, value in values.items():
|
|
setattr(observation, key, value)
|
|
updated += 1
|
|
session.commit()
|
|
return created, updated
|
|
|
|
|
|
def _json_payload(raw: dict[str, Any]) -> dict[str, Any]:
|
|
if not isinstance(raw, dict):
|
|
raise CommunityImportError("each observation must be an object")
|
|
return json.loads(json.dumps(raw, default=str))
|
|
|
|
|
|
def _required(payload: dict[str, Any], key: str, limit: int) -> str:
|
|
value = str(payload.get(key) or "").strip()
|
|
if not value or len(value) > limit:
|
|
raise CommunityImportError(f"invalid {key}")
|
|
return value
|
|
|
|
|
|
def _optional(payload: dict[str, Any], key: str, limit: int) -> str | None:
|
|
value = str(payload.get(key) or "").strip()
|
|
if len(value) > limit:
|
|
raise CommunityImportError(f"invalid {key}")
|
|
return value or None
|
|
|
|
|
|
def _integer(value: Any, *, minimum: int = -10_000, maximum: int) -> int | None:
|
|
if value is None:
|
|
return None
|
|
try:
|
|
parsed = int(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise CommunityImportError("invalid integer field") from exc
|
|
if not minimum <= parsed <= maximum:
|
|
raise CommunityImportError("integer field outside allowed range")
|
|
return parsed
|
|
|
|
|
|
def _datetime(value: Any) -> datetime | None:
|
|
if value in {None, ""}:
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(str(value))
|
|
except ValueError as exc:
|
|
raise CommunityImportError("invalid published_at") from exc
|
|
return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
|