feat: stage external source observations

2026-09-03 18:41:32 +07:00
parent 16d64606f8
commit 53d700126a
8 changed files with 352 additions and 9 deletions
@@ -24,6 +24,15 @@ python -m rf4_research.community_cli rf4stat-posts --limit 25

 Команды печатают нормализованный JSON в stdout и ничего не записывают в базу. Для регулярного получения необходимо соблюдать согласованные лимиты; для RF4-STAT — не менее пяти секунд между запросами разных страниц.

+Проверенный JSON можно идемпотентно загрузить в изолированный staging, не влияющий на публичную статистику:
+
+```bash
+python -m rf4_research.community_cli rf4db --limit 25 \
+  | docker compose exec -T api python -m app.cli stage-community-json --input -
+```
+
+Staging проверяет происхождение URL и диапазоны значений. Источники по умолчанию выключены; автоматического преобразования в одобренные уловы нет.
+
 ## Запуск через Docker

 Требуются Docker Engine и Docker Compose. Это основной и рекомендуемый сценарий:
@@ -59,7 +68,7 @@ docker compose up --build
 ## Что реализовано

 - FastAPI и SQLAlchemy 2;
- PostgreSQL 17 и миграции Alembic до `0007`;
+- PostgreSQL 17 и миграции Alembic до `0008`;
 - идемпотентный seed с двумя точками и свежими демо-уловами;
 - `GET /api/v1/activity` с фильтрами периода, водоёма, рыбы, способа и сортировки;
 - `GET /api/v1/spots/{id}` и `/catches`;
@@ -0,0 +1,60 @@
+"""Staging for authorized external observations."""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "0008"
+down_revision = "0007"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "data_source",
+        sa.Column("key", sa.String(50), primary_key=True),
+        sa.Column("name", sa.String(100), nullable=False),
+        sa.Column("base_url", sa.Text(), nullable=False),
+        sa.Column("default_confidence", sa.Integer(), nullable=False),
+        sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.false()),
+    )
+    op.create_table(
+        "external_observation",
+        sa.Column("id", sa.Uuid(), primary_key=True),
+        sa.Column("source_system", sa.String(50), sa.ForeignKey("data_source.key"), nullable=False),
+        sa.Column("source_external_id", sa.String(200), nullable=False),
+        sa.Column("source_url", sa.Text(), nullable=False),
+        sa.Column("fish_name", sa.String(200), nullable=False),
+        sa.Column("fish_external_id", sa.String(200)),
+        sa.Column("waterbody_name", sa.String(200), nullable=False),
+        sa.Column("waterbody_external_id", sa.String(200)),
+        sa.Column("x", sa.Integer()),
+        sa.Column("y", sa.Integer()),
+        sa.Column("weight_g", sa.Integer()),
+        sa.Column("published_at", sa.DateTime(timezone=True)),
+        sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("status", sa.String(30), nullable=False, server_default="staged"),
+        sa.Column("payload", sa.JSON(), nullable=False),
+        sa.UniqueConstraint("source_system", "source_external_id"),
+    )
+    op.create_index("ix_external_observation_source_system", "external_observation", ["source_system"])
+    op.create_index("ix_external_observation_last_seen_at", "external_observation", ["last_seen_at"])
+    op.create_index("ix_external_observation_status", "external_observation", ["status"])
+    sources = sa.table(
+        "data_source",
+        sa.column("key", sa.String), sa.column("name", sa.String), sa.column("base_url", sa.Text),
+        sa.column("default_confidence", sa.Integer), sa.column("enabled", sa.Boolean),
+    )
+    op.bulk_insert(sources, [
+        {"key": "rf4db", "name": "RF4DB", "base_url": "https://rf4db.com", "default_confidence": 70, "enabled": False},
+        {"key": "rf4stat-fishing", "name": "RF4-STAT fishing", "base_url": "https://rf4-stat.ru/fishing/", "default_confidence": 65, "enabled": False},
+        {"key": "rf4stat-post", "name": "RF4-STAT posts", "base_url": "https://rf4-stat.ru/posts/", "default_confidence": 60, "enabled": False},
+    ])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_external_observation_status", table_name="external_observation")
+    op.drop_index("ix_external_observation_last_seen_at", table_name="external_observation")
+    op.drop_index("ix_external_observation_source_system", table_name="external_observation")
+    op.drop_table("external_observation")
+    op.drop_table("data_source")
@@ -1,9 +1,12 @@
 from __future__ import annotations

 import argparse
+import json
+import sys

 from .database import SessionLocal
 from .importer import import_records
+from .community_importer import stage_observations


 def main() -> int:
@@ -13,10 +16,27 @@ def main() -> int:
    command.add_argument("--url", default="https://rf4game.de/records/region/RU/")
    command.add_argument("--region", default="RU")
    command.add_argument("--category", default="records")
+    community = sub.add_parser("stage-community-json")
+    community.add_argument("--input", default="-", help="JSON array path or - for stdin")
+    community.add_argument("--limit", type=int, default=500)
    args = parser.parse_args()
    with SessionLocal() as session:
+        if args.command == "import-records":
            run = import_records(session, url=args.url, region=args.region, category=args.category)
            print(f"import {run.status.value}: seen={run.rows_seen} created={run.rows_created} updated={run.rows_updated}")
+        else:
+            if not 1 <= args.limit <= 5000:
+                parser.error("--limit must be between 1 and 5000")
+            stream = sys.stdin if args.input == "-" else open(args.input, encoding="utf-8")
+            try:
+                payload = json.load(stream)
+            finally:
+                if stream is not sys.stdin:
+                    stream.close()
+            if not isinstance(payload, list):
+                parser.error("input must be a JSON array")
+            created, updated = stage_observations(session, payload[:args.limit])
+            print(f"staged: created={created} updated={updated}")
    return 0


@@ -0,0 +1,120 @@
+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),
+}
+SOURCE_HOSTS = {
+    "rf4db": {"rf4db.com", "download.rf4db.com"},
+    "rf4stat-fishing": {"rf4-stat.ru"},
+    "rf4stat-post": {"rf4-stat.ru"},
+}
+
+
+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)
@@ -136,3 +136,34 @@ class SubmissionAttempt(Base):
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    client_hash: Mapped[str] = mapped_column(String(64), index=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
+
+
+class DataSource(Base):
+    __tablename__ = "data_source"
+    key: Mapped[str] = mapped_column(String(50), primary_key=True)
+    name: Mapped[str] = mapped_column(String(100))
+    base_url: Mapped[str] = mapped_column(Text)
+    default_confidence: Mapped[int]
+    enabled: Mapped[bool] = mapped_column(default=False)
+
+
+class ExternalObservation(Base):
+    __tablename__ = "external_observation"
+    __table_args__ = (UniqueConstraint("source_system", "source_external_id"),)
+    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
+    source_system: Mapped[str] = mapped_column(ForeignKey("data_source.key"), index=True)
+    source_external_id: Mapped[str] = mapped_column(String(200))
+    source_url: Mapped[str] = mapped_column(Text)
+    fish_name: Mapped[str] = mapped_column(String(200))
+    fish_external_id: Mapped[str | None] = mapped_column(String(200))
+    waterbody_name: Mapped[str] = mapped_column(String(200))
+    waterbody_external_id: Mapped[str | None] = mapped_column(String(200))
+    x: Mapped[int | None]
+    y: Mapped[int | None]
+    weight_g: Mapped[int | None]
+    published_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    first_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
+    last_seen_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
+    status: Mapped[str] = mapped_column(String(30), default="staged", index=True)
+    payload: Mapped[dict] = mapped_column(JSON)
+    source: Mapped[DataSource] = relationship()
@@ -0,0 +1,95 @@
+import json
+from dataclasses import asdict
+from datetime import datetime, timezone
+from pathlib import Path
+
+import pytest
+from sqlalchemy import create_engine, func, select
+from sqlalchemy.orm import Session
+
+from app.community_importer import CommunityImportError, stage_observations
+from app.database import Base
+from app.models import DataSource, ExternalObservation
+from rf4_research.community_sources import parse_rf4db_catches
+
+
+FIXTURE = Path(__file__).parents[3] / "tests" / "fixtures" / "rf4db_catches_sample.html"
+
+
+def record(source: str = "rf4db", external_id: str = "catch-1") -> dict[str, object]:
+    return {
+        "source_system": source,
+        "source_external_id": external_id,
+        "source_url": f"https://rf4db.com/ru/catches/{external_id}" if source == "rf4db" else f"https://rf4-stat.ru/fishing/{external_id}",
+        "fish": "Щука",
+        "fish_external_id": "pike",
+        "waterbody": "Тестовое озеро",
+        "waterbody_external_id": "test-lake",
+        "x": 71,
+        "y": 92,
+        "weight_g": None,
+        "published_at": None,
+        "bait": "Приманка",
+    }
+
+
+@pytest.fixture
+def db() -> Session:
+    engine = create_engine("sqlite://")
+    Base.metadata.create_all(engine)
+    with Session(engine) as session:
+        yield session
+    engine.dispose()
+
+
+def test_staging_is_idempotent_and_preserves_first_seen(db: Session) -> None:
+    first = datetime(2026, 9, 3, 10, tzinfo=timezone.utc)
+    second = datetime(2026, 9, 3, 11, tzinfo=timezone.utc)
+
+    assert stage_observations(db, [record()], fetched_at=first) == (1, 0)
+    changed = record() | {"x": 73, "weight_g": 5_000}
+    assert stage_observations(db, [changed], fetched_at=second) == (0, 1)
+
+    item = db.scalar(select(ExternalObservation))
+    assert item is not None
+    assert (item.x, item.weight_g, item.status) == (73, 5_000, "staged")
+    assert item.first_seen_at.replace(tzinfo=timezone.utc) == first
+    assert item.last_seen_at.replace(tzinfo=timezone.utc) == second
+    assert db.scalar(select(func.count()).select_from(ExternalObservation)) == 1
+    source = db.get(DataSource, "rf4db")
+    assert source is not None and source.enabled is False
+
+
+def test_external_ids_are_isolated_by_source(db: Session) -> None:
+    created, updated = stage_observations(db, [record("rf4db"), record("rf4stat-fishing")])
+
+    assert (created, updated) == (2, 0)
+
+
+def test_parser_json_can_be_staged_without_losing_provenance(db: Session) -> None:
+    parsed = parse_rf4db_catches(FIXTURE.read_text(encoding="utf-8"))
+    payload = json.loads(json.dumps([asdict(item) for item in parsed], default=str))
+
+    assert stage_observations(db, payload) == (1, 0)
+    item = db.scalar(select(ExternalObservation))
+    assert item is not None
+    assert (item.source_system, item.fish_external_id, item.x, item.y) == ("rf4db", "pike", 71, 92)
+
+
+def test_invalid_source_rolls_back_caller_transaction(db: Session) -> None:
+    with pytest.raises(CommunityImportError, match="unsupported source_system"):
+        stage_observations(db, [record("unknown")])
+    db.rollback()
+
+    assert db.scalar(select(func.count()).select_from(ExternalObservation)) == 0
+
+
+@pytest.mark.parametrize("change", [
+    {"source_url": "https://attacker.example/catch-1"},
+    {"x": 10_001},
+    {"weight_g": 3_000_001},
+])
+def test_invalid_provenance_and_ranges_are_rejected(db: Session, change: dict[str, object]) -> None:
+    with pytest.raises(CommunityImportError):
+        stage_observations(db, [record() | change])
+    db.rollback()
@@ -74,7 +74,8 @@
 - [x] Получено подтверждение владельца проекта о разрешениях RF4DB и RF4-STAT; добавлены пилотные HTML-парсеры и отчёт `docs/community-source-pilot.md`.
 - [x] Добавлены общий nullable-контракт, парсер detail-страницы RF4DB и ограниченный read-only CLI для RF4DB/RF4-STAT.
 - [ ] Зафиксировать сами подтверждения разрешений и согласованные лимиты/атрибуцию в репозитории или закрытой операционной документации.
- [ ] Добавить staging-модель внешних наблюдений и идемпотентный импорт RF4DB/RF4-STAT без автоматического влияния на индекс.
+- [x] Добавить staging-модель внешних наблюдений и идемпотентный импорт RF4DB/RF4-STAT без автоматического влияния на индекс (миграция `0008`, сквозной контрактный тест).
+- [ ] Добавить административную очередь сопоставления staging-записей с каноническими рыбами/водоёмами и явную публикацию в `catch_report`.
 - [ ] Согласовать один добровольный канал сообщества и правила происхождения, модерации и удаления сообщений.

 ## Этап 5 — пилот
@@ -63,8 +63,15 @@ Detail-страница дополнительно содержит ветер,
 ## Перед продуктивным импортом

 1. Сохранить подтверждение разрешения и согласованные лимиты запросов.
-2. Добавить миграцию `data_source`/`source_system` и nullable-модель внешнего наблюдения либо обоснованно сделать вес nullable.
-3. Ввести алиасы рыб и водоёмов по внешним slug, а не только по названию.
-4. Назначить разные начальные уровни доверия для RF4DB и двух каналов RF4-STAT.
-5. Добавить идемпотентную staging-загрузку и ручную модерацию перед влиянием на индекс.
-6. Не хранить и не проксировать изображения без отдельного условия разрешения; на первом этапе достаточно исходной ссылки.
+2. Ввести алиасы рыб и водоёмов по внешним slug, а не только по названию.
+3. Согласовать начальные уровни доверия для RF4DB и двух каналов RF4-STAT (в staging записаны консервативные значения 70/65/60, но источники выключены).
+4. Добавить ручную модерацию и явное преобразование staging-записи перед влиянием на индекс.
+5. Не хранить и не проксировать изображения без отдельного условия разрешения; на первом этапе достаточно исходной ссылки.
+
+## Staging
+
+Миграция `0008` добавляет справочник `data_source` и таблицу `external_observation`. Уникальность `(source_system, source_external_id)` делает повторную загрузку идемпотентной; `first_seen_at` сохраняется, `last_seen_at` и payload обновляются. Вес, координаты и время могут отсутствовать.
+
+Загрузка принимает только известные источники, HTTPS-ссылки соответствующего домена и значения в допустимых диапазонах. Все записи получают статус `staged`; таблица не связана с расчётом активности. Тест доказывает путь `HTML → ExternalCatch → JSON → external_observation` без потери provenance.
+
+Контрольная загрузка в PostgreSQL создала 139 staging-записей: 23 `rf4db` с координатами, 100 `rf4stat-fishing` с весом и 16 `rf4stat-post` с весом. Немедленный повтор дал `created=0, updated=139`, подтвердив идемпотентность на реальной БД. Все три источника остались `enabled=false`.