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 .community_review import publish_observation from .models import DataSource, ExternalEntityAlias, ExternalObservation, Fish, ModerationStatus, Waterbody 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 touched: list[ExternalObservation] = [] 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=True) 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: # Preserve the published snapshot, but withdraw it from activity until # a moderator confirms the changed source record. changed = any(getattr(observation, key) != values[key] for key in ( "source_url", "fish_name", "fish_external_id", "waterbody_name", "waterbody_external_id", "x", "y", "weight_g", )) or observation.payload != payload if observation.status != "rejected" and changed and observation.catch_report is not None: observation.catch_report.moderation_status = ModerationStatus.pending observation.status = "staged" observation.fish = None observation.waterbody = None observation.review_note = "Source record changed; manual mapping and publication required" for key, value in values.items(): setattr(observation, key, value) updated += 1 touched.append(observation) session.commit() for observation in touched: _auto_publish(session, observation) return created, updated def _auto_publish(session: Session, observation: ExternalObservation) -> bool: """Publish only complete observations covered by previously reviewed aliases.""" if ( observation.catch_report_id is not None or observation.status not in {"staged", "mapped", "ready"} or not observation.source.enabled or observation.x is None or observation.y is None or observation.weight_g is None ): return False # Fish: prefer external alias, fall back to exact name match fish = None if observation.fish_external_id is not None: fish_alias = session.scalar(select(ExternalEntityAlias).where( ExternalEntityAlias.source_system == observation.source_system, ExternalEntityAlias.entity_type == "fish", ExternalEntityAlias.external_id == observation.fish_external_id, )) if fish_alias and fish_alias.fish: fish = fish_alias.fish if fish is None: # Fallback: exact name match fish = session.scalar( select(Fish).where(Fish.name_ru == observation.fish_name) ) if fish is None: return False # Waterbody: prefer external alias, fall back to exact name match waterbody = None if observation.waterbody_external_id is not None: waterbody_alias = session.scalar(select(ExternalEntityAlias).where( ExternalEntityAlias.source_system == observation.source_system, ExternalEntityAlias.entity_type == "waterbody", ExternalEntityAlias.external_id == observation.waterbody_external_id, )) if waterbody_alias and waterbody_alias.waterbody: waterbody = waterbody_alias.waterbody if waterbody is None: # Fallback: exact name match waterbody = session.scalar( select(Waterbody).where(Waterbody.name_ru == observation.waterbody_name) ) if waterbody is None: return False observation.fish = fish observation.waterbody = waterbody observation.status = "ready" # A07: Describe actual matching method used fish_method = "external_id" if observation.fish_external_id else "name" wb_method = "external_id" if observation.waterbody_external_id else "name" observation.review_note = f"Auto-matched: fish via {fish_method}, waterbody via {wb_method}" publish_observation(session, observation) return True 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)