141 lines
6.1 KiB
Python
141 lines
6.1 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
from collections import Counter
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from .models import CatchReport, ModerationStatus
|
|
from .schemas import ActivityOut
|
|
|
|
|
|
def activity_rows(
|
|
session: Session,
|
|
*,
|
|
hours: int,
|
|
waterbody: str | None = None,
|
|
fish: str | None = None,
|
|
method: str | None = None,
|
|
now: datetime | None = None,
|
|
) -> list[ActivityOut]:
|
|
now = _aware(now or datetime.now(timezone.utc))
|
|
query = (
|
|
select(CatchReport)
|
|
.options(
|
|
joinedload(CatchReport.fish), joinedload(CatchReport.waterbody),
|
|
joinedload(CatchReport.spot), joinedload(CatchReport.bait),
|
|
)
|
|
.where(
|
|
CatchReport.moderation_status == ModerationStatus.approved,
|
|
CatchReport.deleted_at.is_(None),
|
|
CatchReport.spot_id.is_not(None),
|
|
CatchReport.reported_at >= now - timedelta(hours=hours),
|
|
)
|
|
)
|
|
if waterbody:
|
|
query = query.where(CatchReport.waterbody.has(slug=waterbody))
|
|
if fish:
|
|
query = query.where(CatchReport.fish.has(slug=fish))
|
|
if method:
|
|
query = query.where(CatchReport.fishing_method == method)
|
|
reports = list(session.scalars(query))
|
|
|
|
groups: dict[tuple[object, object], list[CatchReport]] = {}
|
|
for report in reports:
|
|
groups.setdefault((report.spot_id, report.fish_id), []).append(report)
|
|
|
|
result: list[ActivityOut] = []
|
|
for items in groups.values():
|
|
first = items[0]
|
|
players = {r.player_name.strip().casefold() for r in items if r.player_name and r.player_name.strip()}
|
|
freshness = [math.exp(-max(0.0, (now - _aware(r.reported_at)).total_seconds()) / 3600 / 18) for r in items]
|
|
weighted = sum(value * r.source_confidence / 100 for value, r in zip(freshness, items))
|
|
trophies = sum(bool(r.fish.trophy_weight_g and r.weight_g >= r.fish.trophy_weight_g) for r in items)
|
|
activity = round(55 * min(1, weighted / 12) + 25 * min(1, len(players) / 6) + 20 * min(1, trophies / 3))
|
|
average_confidence = sum(r.source_confidence for r in items) / len(items)
|
|
confidence = round(45 * min(1, len(items) / 10) + 35 * min(1, len(players) / 5) + 20 * average_confidence / 100)
|
|
# Cap confidence: 1 player → max 50%, 2 players → max 65%
|
|
if len(players) == 1:
|
|
confidence = min(confidence, 50)
|
|
elif len(players) == 2:
|
|
confidence = min(confidence, 65)
|
|
coordinate_precisions = {_coordinate_precision(item) for item in items}
|
|
coordinate_precision = max(coordinate_precisions, key=_precision_rank)
|
|
coordinate_sources = sorted({_source_system(item) for item in items})
|
|
latest = max(_aware(r.reported_at) for r in items)
|
|
baits = Counter(r.bait.name for r in items if r.bait)
|
|
freshness_text = _freshness_text(now - latest)
|
|
result.append(ActivityOut(
|
|
spot_id=first.spot.id, waterbody_slug=first.waterbody.slug,
|
|
waterbody=first.waterbody.name_ru, fish_slug=first.fish.slug,
|
|
fish=first.fish.name_ru, x=first.spot.x, y=first.spot.y,
|
|
best_bait=baits.most_common(1)[0][0] if baits else None,
|
|
catches=len(items), unique_players=len(players),
|
|
average_weight_g=round(sum(r.weight_g for r in items) / len(items)),
|
|
max_weight_g=max(r.weight_g for r in items), last_confirmed_at=latest,
|
|
activity_score=activity, confidence_score=confidence,
|
|
explanation=_explanation(len(items), len(players), freshness_text, activity, confidence),
|
|
sources=coordinate_sources,
|
|
coordinate_precision=coordinate_precision,
|
|
coordinate_sources=coordinate_sources,
|
|
))
|
|
return sorted(result, key=lambda row: (row.activity_score, row.last_confirmed_at), reverse=True)
|
|
|
|
|
|
def _source_system(report: CatchReport) -> str:
|
|
provenance = (report.raw_payload or {}).get("provenance", {})
|
|
if isinstance(provenance, dict) and provenance.get("source_system"):
|
|
return str(provenance["source_system"])
|
|
if report.source_type.value == "official_record":
|
|
return "rf4-official"
|
|
if report.source_type.value == "user":
|
|
return "players"
|
|
return "manual-import"
|
|
|
|
|
|
def _coordinate_precision(report: CatchReport) -> str:
|
|
provenance = (report.raw_payload or {}).get("provenance", {})
|
|
value = provenance.get("coordinate_precision") if isinstance(provenance, dict) else None
|
|
return value if value in {"exact", "approximate", "area", "missing"} else "exact"
|
|
|
|
|
|
def _precision_rank(value: str) -> int:
|
|
return {"exact": 0, "approximate": 1, "area": 2, "missing": 3}[value]
|
|
|
|
|
|
def _aware(value: datetime) -> datetime:
|
|
return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
def _freshness_text(delta: timedelta) -> str:
|
|
minutes = max(0, round(delta.total_seconds() / 60))
|
|
if minutes < 60:
|
|
return f"{minutes} мин. назад"
|
|
return f"{minutes // 60} ч. назад"
|
|
|
|
|
|
def _explanation(catches: int, players: int, freshness: str, activity: int, confidence: int) -> str:
|
|
activity_label = "Высокая" if activity >= 60 else "Средняя" if activity >= 40 else "Низкая"
|
|
confidence_label = "высокая" if confidence >= 70 else "средняя" if confidence >= 40 else "низкая"
|
|
summary = (
|
|
f"{activity_label} активность: {_count(catches, 'свежий улов', 'свежих улова', 'свежих уловов')} "
|
|
f"от {_count(players, 'игрока', 'игроков', 'игроков')}. "
|
|
f"Последнее подтверждение {freshness}. Уверенность {confidence_label}."
|
|
)
|
|
return summary + (" Данных мало: нужно хотя бы 3 наблюдения." if catches < 3 else "")
|
|
|
|
|
|
def _count(value: int, one: str, few: str, many: str) -> str:
|
|
remainder = value % 100
|
|
if 11 <= remainder <= 14:
|
|
word = many
|
|
elif value % 10 == 1:
|
|
word = one
|
|
elif 2 <= value % 10 <= 4:
|
|
word = few
|
|
else:
|
|
word = many
|
|
return f"{value} {word}"
|