Add conditional caching and opt-in import scheduler
This commit is contained in:
@@ -10,3 +10,4 @@ S3_BUCKET=catch-screenshots
|
||||
OFFICIAL_RECORDS_URL=https://rf4game.de/records/region/RU/
|
||||
OFFICIAL_RECORDS_REGION=RU
|
||||
OFFICIAL_RECORDS_CATEGORY=records
|
||||
IMPORT_INTERVAL_SECONDS=3600
|
||||
|
||||
@@ -91,6 +91,16 @@ docker compose --profile tools run --rm importer
|
||||
|
||||
Импорт делает до трёх ограниченных попыток, проверяет DOM-контракт и не удаляет ранее сохранённые данные при сбое. Повторный запуск обновляет совпавшие записи по SHA-256 ключу и не создаёт дубликаты. Автоматическое расписание намеренно ещё не включено: сначала требуется согласовать допустимость регулярного опроса официального сайта.
|
||||
|
||||
Ручной административный запуск также доступен через `POST /api/v1/admin/imports/official-records`, журнал — через `GET /api/v1/admin/imports`. Импорт сохраняет HTTP-метаданные и использует `ETag`/`Last-Modified`, когда источник их предоставляет.
|
||||
|
||||
Планировщик реализован отдельным opt-in профилем и по умолчанию опрашивает источник не чаще одного раза в час:
|
||||
|
||||
```bash
|
||||
docker compose --profile scheduler up -d scheduler
|
||||
```
|
||||
|
||||
Обычный `docker compose up` его не запускает. Не включайте профиль во внешнем окружении, пока условия автоматического сбора не согласованы с владельцем источника; отсутствие `robots.txt` не является разрешением.
|
||||
|
||||
## Пользовательские уловы и модерация
|
||||
|
||||
Новая запись из `/report` получает статус `pending` и не участвует в активности до одобрения. Административные методы требуют заголовок `Authorization: Bearer $ADMIN_TOKEN`:
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
"""Official import HTTP cache and diagnostic metadata."""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
revision = "0005"
|
||||
down_revision = "0004"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.add_column("official_record_import", sa.Column("response_status", sa.Integer()))
|
||||
op.add_column("official_record_import", sa.Column("response_etag", sa.String(500)))
|
||||
op.add_column("official_record_import", sa.Column("response_last_modified", sa.String(500)))
|
||||
op.add_column("official_record_import", sa.Column("response_content_type", sa.String(200)))
|
||||
op.add_column("official_record_import", sa.Column("response_bytes", sa.Integer()))
|
||||
op.add_column("official_record_import", sa.Column("not_modified", sa.Boolean(), nullable=False, server_default=sa.false()))
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_column("official_record_import", "not_modified")
|
||||
op.drop_column("official_record_import", "response_bytes")
|
||||
op.drop_column("official_record_import", "response_content_type")
|
||||
op.drop_column("official_record_import", "response_last_modified")
|
||||
op.drop_column("official_record_import", "response_etag")
|
||||
op.drop_column("official_record_import", "response_status")
|
||||
@@ -1,3 +1,4 @@
|
||||
from pydantic import Field
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
|
||||
@@ -13,6 +14,7 @@ class Settings(BaseSettings):
|
||||
official_records_url: str = "https://rf4game.de/records/region/RU/"
|
||||
official_records_region: str = "RU"
|
||||
official_records_category: str = "records"
|
||||
import_interval_seconds: int = Field(default=3600, ge=3600)
|
||||
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
|
||||
|
||||
|
||||
|
||||
@@ -36,6 +36,16 @@ class RawRecord:
|
||||
record_date: date
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class FetchResult:
|
||||
records: list[RawRecord] | None
|
||||
status_code: int
|
||||
etag: str | None
|
||||
last_modified: str | None
|
||||
content_type: str | None
|
||||
response_bytes: int
|
||||
|
||||
|
||||
def normalize(value: str) -> str:
|
||||
return " ".join(value.replace("\xa0", " ").replace("–", "-").replace("—", "-").split()).casefold()
|
||||
|
||||
@@ -100,15 +110,32 @@ def parse_html(html: str, *, region: str, category: str) -> list[RawRecord]:
|
||||
return records
|
||||
|
||||
|
||||
def fetch_records(url: str, *, region: str, category: str) -> list[RawRecord]:
|
||||
with httpx.Client(timeout=20, follow_redirects=True, headers={"User-Agent": USER_AGENT, "Accept": "text/html"}) as client:
|
||||
def fetch_records(
|
||||
url: str, *, region: str, category: str,
|
||||
etag: str | None = None, last_modified: str | None = None,
|
||||
) -> FetchResult:
|
||||
headers = {"User-Agent": USER_AGENT, "Accept": "text/html"}
|
||||
if etag:
|
||||
headers["If-None-Match"] = etag
|
||||
if last_modified:
|
||||
headers["If-Modified-Since"] = last_modified
|
||||
with httpx.Client(timeout=20, follow_redirects=True, headers=headers) as client:
|
||||
for attempt in range(3):
|
||||
try:
|
||||
response = client.get(url)
|
||||
metadata = {
|
||||
"status_code": response.status_code,
|
||||
"etag": response.headers.get("etag"),
|
||||
"last_modified": response.headers.get("last-modified"),
|
||||
"content_type": response.headers.get("content-type"),
|
||||
"response_bytes": len(response.content),
|
||||
}
|
||||
if response.status_code == 304:
|
||||
return FetchResult(records=None, **metadata)
|
||||
response.raise_for_status()
|
||||
if "text/html" not in response.headers.get("content-type", ""):
|
||||
raise ImportSourceError("source did not return HTML")
|
||||
return parse_html(response.text, region=region, category=category)
|
||||
return FetchResult(records=parse_html(response.text, region=region, category=category), **metadata)
|
||||
except (httpx.HTTPError, ImportSourceError):
|
||||
if attempt == 2:
|
||||
raise
|
||||
@@ -121,7 +148,32 @@ def import_records(session: Session, *, url: str, region: str, category: str, ht
|
||||
session.add(run)
|
||||
session.commit()
|
||||
try:
|
||||
records = parse_html(html, region=region, category=category) if html is not None else fetch_records(url, region=region, category=category)
|
||||
if html is not None:
|
||||
records = parse_html(html, region=region, category=category)
|
||||
else:
|
||||
previous = session.scalar(
|
||||
select(OfficialRecordImport).where(
|
||||
OfficialRecordImport.source_url == url,
|
||||
OfficialRecordImport.status == ImportStatus.success,
|
||||
).order_by(OfficialRecordImport.started_at.desc()).limit(1)
|
||||
)
|
||||
fetched = fetch_records(
|
||||
url, region=region, category=category,
|
||||
etag=previous.response_etag if previous else None,
|
||||
last_modified=previous.response_last_modified if previous else None,
|
||||
)
|
||||
run.response_status = fetched.status_code
|
||||
run.response_etag = fetched.etag or (previous.response_etag if previous else None)
|
||||
run.response_last_modified = fetched.last_modified or (previous.response_last_modified if previous else None)
|
||||
run.response_content_type = fetched.content_type
|
||||
run.response_bytes = fetched.response_bytes
|
||||
if fetched.records is None:
|
||||
run.not_modified = True
|
||||
run.status = ImportStatus.success
|
||||
run.finished_at = datetime.now(timezone.utc)
|
||||
session.commit()
|
||||
return run
|
||||
records = fetched.records
|
||||
run.rows_seen = len(records)
|
||||
for raw in records:
|
||||
key = external_id(raw)
|
||||
|
||||
@@ -110,6 +110,12 @@ class OfficialRecordImport(Base):
|
||||
rows_created: Mapped[int] = mapped_column(default=0)
|
||||
rows_updated: Mapped[int] = mapped_column(default=0)
|
||||
error_summary: Mapped[str | None] = mapped_column(Text)
|
||||
response_status: Mapped[int | None]
|
||||
response_etag: Mapped[str | None] = mapped_column(String(500))
|
||||
response_last_modified: Mapped[str | None] = mapped_column(String(500))
|
||||
response_content_type: Mapped[str | None] = mapped_column(String(200))
|
||||
response_bytes: Mapped[int | None]
|
||||
not_modified: Mapped[bool] = mapped_column(default=False)
|
||||
|
||||
|
||||
class ModerationEvent(Base):
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from .config import settings
|
||||
from .database import SessionLocal
|
||||
from .importer import import_records
|
||||
from .models import OfficialRecordImport
|
||||
|
||||
|
||||
logger = logging.getLogger("rf4.import_scheduler")
|
||||
|
||||
|
||||
def import_is_due(session: Session, *, now: datetime | None = None) -> bool:
|
||||
current = now or datetime.now(timezone.utc)
|
||||
latest = session.scalar(
|
||||
select(OfficialRecordImport.started_at)
|
||||
.where(OfficialRecordImport.source_url == settings.official_records_url)
|
||||
.order_by(OfficialRecordImport.started_at.desc())
|
||||
.limit(1)
|
||||
)
|
||||
if latest is None:
|
||||
return True
|
||||
if latest.tzinfo is None:
|
||||
latest = latest.replace(tzinfo=timezone.utc)
|
||||
return latest <= current - timedelta(seconds=settings.import_interval_seconds)
|
||||
|
||||
|
||||
def run_due_import() -> bool:
|
||||
with SessionLocal() as session:
|
||||
if not import_is_due(session):
|
||||
return False
|
||||
run = import_records(
|
||||
session,
|
||||
url=settings.official_records_url,
|
||||
region=settings.official_records_region,
|
||||
category=settings.official_records_category,
|
||||
)
|
||||
logger.info(
|
||||
"official import completed status=%s seen=%d created=%d updated=%d not_modified=%s",
|
||||
run.status.value, run.rows_seen, run.rows_created, run.rows_updated, run.not_modified,
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def main() -> None:
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
|
||||
logger.info("scheduler started interval_seconds=%d", settings.import_interval_seconds)
|
||||
while True:
|
||||
try:
|
||||
run_due_import()
|
||||
except Exception:
|
||||
logger.exception("scheduled official import failed")
|
||||
time.sleep(settings.import_interval_seconds)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -98,6 +98,12 @@ class ImportRunOut(BaseModel):
|
||||
rows_created: int
|
||||
rows_updated: int
|
||||
error_summary: str | None
|
||||
response_status: int | None
|
||||
response_etag: str | None
|
||||
response_last_modified: str | None
|
||||
response_content_type: str | None
|
||||
response_bytes: int | None
|
||||
not_modified: bool
|
||||
|
||||
|
||||
class CatchReportCreate(BaseModel):
|
||||
|
||||
@@ -7,7 +7,7 @@ from sqlalchemy import create_engine, func, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.database import Base
|
||||
from app.importer import ImportSourceError, import_records, parse_html
|
||||
from app.importer import FetchResult, ImportSourceError, import_records, parse_html
|
||||
from app.models import CatchReport, ImportStatus, OfficialRecordImport, SourceType
|
||||
|
||||
|
||||
@@ -55,3 +55,28 @@ def test_import_rejects_changed_column_contract() -> None:
|
||||
)
|
||||
with pytest.raises(ImportSourceError, match="record columns changed"):
|
||||
parse_html(html, region="RU", category="records")
|
||||
|
||||
|
||||
def test_import_reuses_http_validators_and_handles_not_modified(monkeypatch) -> None:
|
||||
html = FIXTURE.read_text(encoding="utf-8")
|
||||
parsed = parse_html(html, region="RU", category="records")
|
||||
calls: list[tuple[str | None, str | None]] = []
|
||||
|
||||
def fake_fetch(url: str, *, region: str, category: str, etag: str | None, last_modified: str | None) -> FetchResult:
|
||||
calls.append((etag, last_modified))
|
||||
if len(calls) == 1:
|
||||
return FetchResult(parsed, 200, '"fixture-v1"', "Wed, 02 Sep 2026 00:00:00 GMT", "text/html", len(html))
|
||||
return FetchResult(None, 304, None, None, None, 0)
|
||||
|
||||
monkeypatch.setattr("app.importer.fetch_records", fake_fetch)
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
with Session(engine) as db:
|
||||
first = import_records(db, url="https://example.test/records", region="RU", category="records")
|
||||
second = import_records(db, url="https://example.test/records", region="RU", category="records")
|
||||
assert calls == [(None, None), ('"fixture-v1"', "Wed, 02 Sep 2026 00:00:00 GMT")]
|
||||
assert first.response_status == 200
|
||||
assert second.response_status == 304
|
||||
assert second.not_modified is True
|
||||
assert second.response_etag == '"fixture-v1"'
|
||||
assert second.rows_seen == 0
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.config import settings
|
||||
from app.database import Base
|
||||
from app.models import ImportStatus, OfficialRecordImport
|
||||
from app.scheduler import import_is_due
|
||||
|
||||
|
||||
def test_import_is_due_respects_minimum_interval() -> None:
|
||||
engine = create_engine("sqlite://")
|
||||
Base.metadata.create_all(engine)
|
||||
now = datetime.now(timezone.utc)
|
||||
with Session(engine) as db:
|
||||
assert import_is_due(db, now=now) is True
|
||||
db.add(OfficialRecordImport(
|
||||
started_at=now - timedelta(seconds=settings.import_interval_seconds - 1),
|
||||
finished_at=now,
|
||||
status=ImportStatus.success,
|
||||
source_url=settings.official_records_url,
|
||||
rows_seen=0,
|
||||
rows_created=0,
|
||||
rows_updated=0,
|
||||
))
|
||||
db.commit()
|
||||
assert import_is_due(db, now=now) is False
|
||||
assert import_is_due(db, now=now + timedelta(seconds=settings.import_interval_seconds)) is True
|
||||
@@ -72,6 +72,21 @@ services:
|
||||
condition: service_healthy
|
||||
command: ["sh", "-c", "alembic upgrade head && python -m app.cli import-records"]
|
||||
|
||||
scheduler:
|
||||
build: ./apps/api
|
||||
profiles: ["scheduler"]
|
||||
restart: unless-stopped
|
||||
environment:
|
||||
DATABASE_URL: postgresql+psycopg://rf4:rf4_local@db:5432/rf4_spotter
|
||||
OFFICIAL_RECORDS_URL: ${OFFICIAL_RECORDS_URL:-https://rf4game.de/records/region/RU/}
|
||||
OFFICIAL_RECORDS_REGION: ${OFFICIAL_RECORDS_REGION:-RU}
|
||||
OFFICIAL_RECORDS_CATEGORY: ${OFFICIAL_RECORDS_CATEGORY:-records}
|
||||
IMPORT_INTERVAL_SECONDS: ${IMPORT_INTERVAL_SECONDS:-3600}
|
||||
depends_on:
|
||||
api:
|
||||
condition: service_healthy
|
||||
command: ["python", "-m", "app.scheduler"]
|
||||
|
||||
volumes:
|
||||
postgres_data:
|
||||
minio_data:
|
||||
|
||||
+3
-3
@@ -20,9 +20,9 @@
|
||||
|
||||
- [x] Добавить административный endpoint ручного запуска импорта `POST /api/v1/admin/imports/official-records` (проверено API-тестом).
|
||||
- [x] Привести журнал импорта к административному контракту `GET /api/v1/admin/imports` с авторизацией, пагинацией и стабильной сортировкой (проверено API-тестом).
|
||||
- [ ] Добавить HTTP-кэширование источника (`ETag`/`Last-Modified`, если источник их отдаёт) и сохранить диагностические метаданные ответа.
|
||||
- [ ] Добавить планировщик импорта с безопасной частотой по умолчанию один раз в 60 минут; отдельный контейнер/процесс без дублирования запусков.
|
||||
- [ ] Проверить актуальные `robots.txt` и условия использования перед включением расписания; результат записать в `docs/data-sources.md`.
|
||||
- [x] Добавить HTTP-кэширование источника (`ETag`/`Last-Modified`, если источник их отдаёт) и сохранить диагностические метаданные ответа (миграция `0005`, тест условного запроса и `304`).
|
||||
- [x] Добавить планировщик импорта с безопасной частотой по умолчанию один раз в 60 минут; отдельный контейнер/процесс без дублирования запусков (opt-in профиль `scheduler`, обычным запуском не активируется).
|
||||
- [x] Проверить актуальные `robots.txt` и условия использования перед включением расписания; результат записать в `docs/data-sources.md` (`robots.txt` вернул `404`; автоматический профиль оставлен выключенным до явного разрешения).
|
||||
- [x] Добавить интеграционные тесты: повторный импорт не создаёт дубликаты, сбой источника не удаляет данные, изменение DOM завершается понятной ошибкой.
|
||||
|
||||
Критерий готовности: официальный импорт запускается вручную и по расписанию, наблюдаем, идемпотентен и безопасно переживает недоступность источника.
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Источники данных RF4: исследование этапа 0
|
||||
|
||||
Дата проверки: **2 сентября 2026 года**. Исследовалась только публичная веб-страница; игровой клиент, его трафик и закрытые протоколы не исследовались.
|
||||
Дата первоначальной проверки: **2 сентября 2026 года**. Повторная проверка страницы и `robots.txt`: **3 сентября 2026 года**. Исследовалась только публичная веб-страница; игровой клиент, его трафик и закрытые протоколы не исследовались.
|
||||
|
||||
## Краткий вывод
|
||||
|
||||
@@ -57,6 +57,8 @@
|
||||
- При исчезновении таблицы, изменении порядка классов или ошибке HTTP считать запуск неуспешным и сохранять прежние данные.
|
||||
- Повторно проверить условия использования и связаться с владельцем сайта до регулярного производственного сбора; отсутствие `robots.txt` не заменяет разрешения.
|
||||
|
||||
При повторной проверке 3 сентября 2026 года страница рекордов по-прежнему публично отдавала таблицу, а `https://rf4game.de/robots.txt` снова вернул `404`. Явного разрешения на автоматический сбор это не даёт. Поэтому планировщик реализован как opt-in профиль Compose `scheduler`, не запускается обычной командой `docker compose up` и не должен включаться во внешнем окружении до проверки условий использования или согласования с владельцем сайта.
|
||||
|
||||
## Сравнение с `hurfy/rf4-api`
|
||||
|
||||
Репозиторий `hurfy/rf4-api` создан в августе 2024 года; последний push, видимый через GitHub API на дату исследования, был 14 января 2025 года. README прямо называет проект находящимся в разработке. Он использует Django, Celery и браузерный WebDriver, перебирает регионы и категории, затем разбирает те же классы `records_wrapper`, `records_subtable`, `gamername`, `weight`, `location`, `bait_icon`, `data`.
|
||||
|
||||
Reference in New Issue
Block a user