From 53d700126abc918d4e72c396ce79ecae7f3ff5aa Mon Sep 17 00:00:00 2001 From: IK Date: Thu, 3 Sep 2026 18:41:32 +0700 Subject: [PATCH] feat: stage external source observations --- README.md | 11 +- .../versions/0008_external_observations.py | 60 +++++++++ apps/api/app/cli.py | 24 +++- apps/api/app/community_importer.py | 120 ++++++++++++++++++ apps/api/app/models.py | 31 +++++ apps/api/tests/test_community_importer.py | 95 ++++++++++++++ docs/ROADMAP.md | 3 +- docs/community-source-pilot.md | 17 ++- 8 files changed, 352 insertions(+), 9 deletions(-) create mode 100644 apps/api/alembic/versions/0008_external_observations.py create mode 100644 apps/api/app/community_importer.py create mode 100644 apps/api/tests/test_community_importer.py diff --git a/README.md b/README.md index 82527b5..1f11dee 100644 --- a/README.md +++ b/README.md @@ -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`; diff --git a/apps/api/alembic/versions/0008_external_observations.py b/apps/api/alembic/versions/0008_external_observations.py new file mode 100644 index 0000000..10adc39 --- /dev/null +++ b/apps/api/alembic/versions/0008_external_observations.py @@ -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") diff --git a/apps/api/app/cli.py b/apps/api/app/cli.py index d8255a1..56a27ca 100644 --- a/apps/api/app/cli.py +++ b/apps/api/app/cli.py @@ -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: - 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}") + 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 diff --git a/apps/api/app/community_importer.py b/apps/api/app/community_importer.py new file mode 100644 index 0000000..1fc1856 --- /dev/null +++ b/apps/api/app/community_importer.py @@ -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) diff --git a/apps/api/app/models.py b/apps/api/app/models.py index 4281069..78a6e08 100644 --- a/apps/api/app/models.py +++ b/apps/api/app/models.py @@ -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() diff --git a/apps/api/tests/test_community_importer.py b/apps/api/tests/test_community_importer.py new file mode 100644 index 0000000..4e85269 --- /dev/null +++ b/apps/api/tests/test_community_importer.py @@ -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() diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 2c2ee0f..4f1173d 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -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 — пилот diff --git a/docs/community-source-pilot.md b/docs/community-source-pilot.md index 249b3fd..829e1fe 100644 --- a/docs/community-source-pilot.md +++ b/docs/community-source-pilot.md @@ -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`.