from __future__ import annotations import json import uuid from dataclasses import dataclass from datetime import datetime, timezone from types import SimpleNamespace from typing import Any, Iterable from aiogram.types import CallbackQuery, InlineQuery, Message, TelegramObject, User from fastapi import Request from sqlalchemy.ext.asyncio import AsyncSession from database.audit import ( create_audit_reset_marker_db, delete_old_audit_events_db, ensure_audit_table, fetch_existing_audit_request_ids_db, fetch_latest_audit_reset_db, fetch_audit_events_db, fetch_audit_events_db_window, fetch_audit_rows_db, fetch_successful_payment_rows_db, ) from database.models import AuditEvent from logger import logger from .rules import ( AUDIT_STEP_LABELS, DEFAULT_FUNNEL_STEPS, _funnel_step_counts, _is_ignored_analytics_event, _normalize_path_to_step, _normalize_path_to_steps, ) try: from core.cache_config import ( AUDIT_REDIS_BUFFER_ENABLED, AUDIT_REDIS_DRAIN_BATCH, AUDIT_REDIS_FLUSH_KEY, AUDIT_REDIS_IDENTITY_PREFIX, AUDIT_REDIS_USER_PREFIX, AUDIT_REDIS_USER_TTL_SEC, ) except ImportError: AUDIT_REDIS_BUFFER_ENABLED = False AUDIT_REDIS_FLUSH_KEY = "audit:flush" AUDIT_REDIS_USER_PREFIX = "audit:user:tg:" AUDIT_REDIS_IDENTITY_PREFIX = "audit:user:identity:" AUDIT_REDIS_USER_TTL_SEC = 25 * 3600 AUDIT_REDIS_DRAIN_BATCH = 1000 _MAX_TEXT_LEN = 160 _AUDIT_REDIS_PROCESSING_KEY = f"{AUDIT_REDIS_FLUSH_KEY}:processing" _AUDIT_REDIS_DRAIN_LOCK_KEY = f"{AUDIT_REDIS_FLUSH_KEY}:drain_lock" _AUDIT_REDIS_DRAIN_LOCK_TTL_SEC = 15 * 60 @dataclass class AuditContext: request_id: str channel: str path_or_handler: str actor_identity_id: str | None = None actor_tg_id: int | None = None def new_request_id() -> str: return uuid.uuid4().hex def _naive_utc(dt: datetime) -> datetime: """Приводит datetime к naive UTC для запросов к колонкам DateTime (без timezone).""" if dt.tzinfo is not None: dt = dt.astimezone(timezone.utc) return dt.replace(tzinfo=None) def _trim(value: Any, limit: int = _MAX_TEXT_LEN) -> str | None: if value is None: return None text = str(value).strip() if not text: return None if len(text) <= limit: return text return text[: limit - 3] + "..." def _jsonable(value: Any) -> Any: if value is None or isinstance(value, (str, int, float, bool)): return value if isinstance(value, datetime): return value.isoformat() if isinstance(value, dict): return {str(k): _jsonable(v) for k, v in value.items()} if isinstance(value, (list, tuple, set)): return [_jsonable(item) for item in value] return _trim(value, 500) def _serialize(payload: dict[str, Any]) -> str: return json.dumps(_jsonable(payload), ensure_ascii=False, sort_keys=True) async def get_audit_db_reset_at(session: AsyncSession) -> datetime | None: reset_at = await fetch_latest_audit_reset_db(session, source="db") if reset_at is None: return None if reset_at.tzinfo is None: return reset_at.replace(tzinfo=timezone.utc) return reset_at.astimezone(timezone.utc) async def set_audit_db_reset_at(session: AsyncSession, at: datetime | None = None) -> datetime: created = at.astimezone(timezone.utc).replace(tzinfo=None) if at is not None and at.tzinfo else (at or datetime.utcnow()) await create_audit_reset_marker_db(session, source="db", created_at=created) await session.commit() return created.replace(tzinfo=timezone.utc) async def clear_audit_redis_buffers() -> int: from core.redis_cache import cache_delete_pattern patterns = ( AUDIT_REDIS_FLUSH_KEY, _AUDIT_REDIS_PROCESSING_KEY, _AUDIT_REDIS_DRAIN_LOCK_KEY, ) deleted = 0 for pattern in patterns: deleted += await cache_delete_pattern(pattern) return deleted def _message_text(event: TelegramObject) -> str | None: if isinstance(event, Message): return _trim(event.text or event.caption) if isinstance(event, CallbackQuery): return _trim(event.data) if isinstance(event, InlineQuery): return _trim(event.query) return None def _event_user(event: TelegramObject) -> User | None: if hasattr(event, "from_user") and isinstance(event.from_user, User): return event.from_user return None def describe_telegram_event(event: TelegramObject) -> str: if isinstance(event, Message): return f"message:{_message_text(event) or '-'}" if isinstance(event, CallbackQuery): return f"callback:{_message_text(event) or '-'}" if isinstance(event, InlineQuery): return f"inline:{_message_text(event) or '-'}" return type(event).__name__ def ensure_api_context(request: Request) -> AuditContext: context = getattr(request.state, "audit_context", None) if isinstance(context, AuditContext): return context path_or_handler = request.url.path if request.url.query: path_or_handler = f"{path_or_handler}?{request.url.query}" context = AuditContext( request_id=new_request_id(), channel="api", path_or_handler=path_or_handler, ) request.state.audit_context = context request.state.audit_request_id = context.request_id return context def get_api_context(request: Request | None) -> AuditContext | None: if request is None: return None context = getattr(request.state, "audit_context", None) if isinstance(context, AuditContext): return context return None def set_api_actor( request: Request, *, identity_id: str | None = None, tg_id: int | None = None, ) -> AuditContext: context = ensure_api_context(request) if identity_id is not None: context.actor_identity_id = identity_id if tg_id is not None: context.actor_tg_id = tg_id return context def ensure_telegram_context( data: dict[str, Any] | None, event: TelegramObject, ) -> AuditContext: if data is not None: existing = data.get("audit_context") if isinstance(existing, AuditContext): return existing user = _event_user(event) context = AuditContext( request_id=new_request_id(), channel="telegram", path_or_handler=describe_telegram_event(event), actor_tg_id=user.id if user else None, ) if data is not None: data["audit_context"] = context data["audit_request_id"] = context.request_id return context def set_telegram_actor( audit_context: AuditContext | dict[str, Any] | None, *, identity_id: str | None = None, tg_id: int | None = None, ) -> AuditContext | None: context: AuditContext | None if isinstance(audit_context, AuditContext): context = audit_context elif isinstance(audit_context, dict): context = audit_context.get("audit_context") else: context = None if not isinstance(context, AuditContext): return None if identity_id is not None: context.actor_identity_id = identity_id if tg_id is not None: context.actor_tg_id = tg_id return context def get_telegram_context(audit_context: AuditContext | dict[str, Any] | None) -> AuditContext | None: if isinstance(audit_context, AuditContext): return audit_context if isinstance(audit_context, dict): context = audit_context.get("audit_context") if isinstance(context, AuditContext): return context return None def log_api_access( request: Request, *, status_code: int, duration_ms: int, result: str, reason: str | None = None, ) -> None: context = ensure_api_context(request) client_ip = request.client.host if request.client else "-" logger.debug( f"[AUDIT_ACCESS] {_serialize({ 'request_id': context.request_id, 'channel': 'api', 'method': request.method, 'path': context.path_or_handler, 'status_code': status_code, 'duration_ms': duration_ms, 'result': result, 'reason': reason, 'client_ip': client_ip, 'actor_identity_id': context.actor_identity_id, 'actor_tg_id': context.actor_tg_id, })}" ) def log_telegram_access( event: TelegramObject, *, audit_context: AuditContext | None, result: str, reason: str | None = None, ) -> None: context = audit_context or AuditContext( request_id=new_request_id(), channel="telegram", path_or_handler=describe_telegram_event(event), ) user = _event_user(event) logger.debug( f"[AUDIT_ACCESS] {_serialize({ 'request_id': context.request_id, 'channel': 'telegram', 'path_or_handler': context.path_or_handler, 'event_type': type(event).__name__, 'message': _message_text(event), 'result': result, 'reason': reason, 'actor_identity_id': context.actor_identity_id, 'actor_tg_id': context.actor_tg_id or (user.id if user else None), 'username': getattr(user, 'username', None) if user else None, })}" ) async def record_audit_event( session: AsyncSession, *, event_type: str, channel: str, path_or_handler: str, actor_identity_id: str | None = None, actor_tg_id: int | None = None, entity_type: str | None = None, entity_id: str | int | None = None, result: str = "success", reason: str | None = None, metadata: dict[str, Any] | None = None, request_id: str | None = None, ) -> AuditEvent: await ensure_audit_table(session) event = AuditEvent( event_type=event_type, channel=channel, actor_identity_id=actor_identity_id, actor_tg_id=actor_tg_id, path_or_handler=_trim(path_or_handler, 255) or channel, entity_type=_trim(entity_type, 64), entity_id=_trim(entity_id, 255), result=_trim(result, 32) or "success", reason=_trim(reason, 1000), metadata_=_jsonable(metadata) if metadata else None, request_id=_trim(request_id, 64), ) session.add(event) await session.flush() logger.debug( f"[AUDIT_EVENT] {_serialize({ 'id': event.id, 'request_id': event.request_id, 'channel': event.channel, 'event_type': event.event_type, 'actor_identity_id': event.actor_identity_id, 'actor_tg_id': event.actor_tg_id, 'path_or_handler': event.path_or_handler, 'entity_type': event.entity_type, 'entity_id': event.entity_id, 'result': event.result, 'reason': event.reason, 'metadata': event.metadata_, })}" ) return event async def safe_record_audit_event(session: AsyncSession, **kwargs: Any) -> AuditEvent | None: try: return await record_audit_event(session, **kwargs) except Exception as exc: logger.warning(f"[Audit] Не удалось записать событие {kwargs.get('event_type')}: {exc}") return None def _telegram_access_payload( audit_context: AuditContext | None, event: TelegramObject, *, result: str = "success", reason: str | None = None, ) -> dict[str, Any]: ctx = get_telegram_context(audit_context) user = _event_user(event) path = describe_telegram_event(event) if ctx is None: ctx = AuditContext( request_id=new_request_id(), channel="telegram", path_or_handler=path, actor_tg_id=user.id if user else None, ) return { "request_id": ctx.request_id, "path_or_handler": path, "actor_identity_id": ctx.actor_identity_id, "actor_tg_id": ctx.actor_tg_id or (user.id if user else None), "result": result, "reason": reason, } async def record_telegram_access_event( session: AsyncSession, audit_context: AuditContext | None, event: TelegramObject, *, result: str = "success", reason: str | None = None, ) -> AuditEvent | None: payload = _telegram_access_payload(audit_context, event, result=result, reason=reason) return await safe_record_audit_event( session, event_type="telegram_access", channel="telegram", path_or_handler=payload["path_or_handler"], actor_identity_id=payload["actor_identity_id"], actor_tg_id=payload["actor_tg_id"], result=payload["result"], reason=payload["reason"], request_id=payload["request_id"], ) def _audit_record_for_redis( *, event_type: str = "telegram_access", channel: str = "telegram", path_or_handler: str, actor_identity_id: str | None = None, actor_tg_id: int | None = None, entity_type: str | None = None, entity_id: str | int | None = None, result: str = "success", reason: str | None = None, request_id: str | None = None, metadata_: dict | None = None, ) -> dict[str, Any]: effective_request_id = _trim(request_id, 64) if request_id else new_request_id() return { "event_type": event_type, "channel": channel, "path_or_handler": _trim(path_or_handler, 255) or channel, "actor_identity_id": actor_identity_id, "actor_tg_id": actor_tg_id, "entity_type": _trim(entity_type, 64) if entity_type else None, "entity_id": _trim(str(entity_id), 255) if entity_id is not None else None, "result": _trim(result, 32) or "success", "reason": _trim(reason, 1000) if reason else None, "request_id": effective_request_id, "metadata_": _jsonable(metadata_) if metadata_ else None, "created_at": datetime.now(timezone.utc).isoformat(), } async def _push_audit_record_to_redis(record: dict[str, Any]) -> None: from core.redis_cache import cache_expire, cache_rpush n = await cache_rpush(AUDIT_REDIS_FLUSH_KEY, record) if n == 0: raise RuntimeError("Redis unavailable (cache_rpush returned 0)") actor_tg_id = record.get("actor_tg_id") actor_identity_id = record.get("actor_identity_id") if actor_tg_id is not None: user_key = f"{AUDIT_REDIS_USER_PREFIX}{actor_tg_id}" await cache_rpush(user_key, record) await cache_expire(user_key, AUDIT_REDIS_USER_TTL_SEC) if actor_identity_id: identity_key = f"{AUDIT_REDIS_IDENTITY_PREFIX}{actor_identity_id}" await cache_rpush(identity_key, record) await cache_expire(identity_key, AUDIT_REDIS_USER_TTL_SEC) async def record_audit_event_to_redis( *, request_id: str | None = None, path_or_handler: str = "", actor_identity_id: str | None = None, actor_tg_id: int | None = None, result: str = "success", reason: str | None = None, ) -> None: record = _audit_record_for_redis( path_or_handler=path_or_handler, actor_identity_id=actor_identity_id, actor_tg_id=actor_tg_id, result=result, reason=reason, request_id=request_id, ) await _push_audit_record_to_redis(record) async def record_api_access_event_to_redis( *, request_id: str | None = None, path_or_handler: str = "", actor_identity_id: str | None = None, actor_tg_id: int | None = None, result: str = "success", reason: str | None = None, ) -> None: record = _audit_record_for_redis( event_type="api_access", channel="api", path_or_handler=path_or_handler, actor_identity_id=actor_identity_id, actor_tg_id=actor_tg_id, result=result, reason=reason, request_id=request_id, ) await _push_audit_record_to_redis(record) async def record_api_access_event_background( session_factory: Any, request: Request, *, result: str = "success", reason: str | None = None, status_code: int = 200, ) -> None: context = ensure_api_context(request) path_or_handler = f"{request.method} {request.url.path}" if request.url.query: path_or_handler = f"{path_or_handler}?{request.url.query}" path_or_handler = _trim(path_or_handler, 255) or "api" if AUDIT_REDIS_BUFFER_ENABLED: try: await record_api_access_event_to_redis( request_id=context.request_id, path_or_handler=path_or_handler, actor_identity_id=context.actor_identity_id, actor_tg_id=context.actor_tg_id, result=result, reason=reason, ) except Exception as exc: logger.warning("[Audit] Запись api_access в Redis-буфер не удалась: %s", exc) return try: async with session_factory() as session: await ensure_audit_table(session) await record_audit_event( session, event_type="api_access", channel="api", path_or_handler=path_or_handler, actor_identity_id=context.actor_identity_id, actor_tg_id=context.actor_tg_id, result=result, reason=reason, request_id=_trim(context.request_id, 64), ) await session.commit() except Exception as exc: logger.warning( "[Audit] Фоновая запись api_access не удалась: %s", exc, extra={"path_or_handler": path_or_handler[:80] if path_or_handler else None}, ) async def record_telegram_access_event_background( session_factory: Any, *, request_id: str | None, path_or_handler: str, actor_identity_id: str | None = None, actor_tg_id: int | None = None, result: str = "success", reason: str | None = None, ) -> None: if AUDIT_REDIS_BUFFER_ENABLED: try: await record_audit_event_to_redis( request_id=request_id, path_or_handler=path_or_handler, actor_identity_id=actor_identity_id, actor_tg_id=actor_tg_id, result=result, reason=reason, ) return except Exception as exc: logger.warning(f"[Audit] Запись в Redis-буфер не удалась, пишем в БД: {exc}") try: async with session_factory() as session: await ensure_audit_table(session) await record_audit_event( session, event_type="telegram_access", channel="telegram", path_or_handler=_trim(path_or_handler, 255) or "telegram", actor_identity_id=actor_identity_id, actor_tg_id=actor_tg_id, result=result, reason=reason, request_id=_trim(request_id, 64), ) await session.commit() except Exception as exc: logger.warning( "[Audit] Фоновая запись telegram_access не удалась: %s", exc, extra={"path_or_handler": path_or_handler[:80] if path_or_handler else None}, ) async def safe_record_api_event( session: AsyncSession, request: Request, *, event_type: str, entity_type: str | None = None, entity_id: str | int | None = None, result: str = "success", reason: str | None = None, metadata: dict[str, Any] | None = None, actor_identity_id: str | None = None, actor_tg_id: int | None = None, path_or_handler: str | None = None, ) -> AuditEvent | None: context = ensure_api_context(request) return await safe_record_audit_event( session, event_type=event_type, channel="api", path_or_handler=path_or_handler or context.path_or_handler, actor_identity_id=actor_identity_id if actor_identity_id is not None else context.actor_identity_id, actor_tg_id=actor_tg_id if actor_tg_id is not None else context.actor_tg_id, entity_type=entity_type, entity_id=entity_id, result=result, reason=reason, metadata=metadata, request_id=context.request_id, ) async def safe_record_telegram_event( session: AsyncSession, audit_context: AuditContext | dict[str, Any] | None, *, event_type: str, entity_type: str | None = None, entity_id: str | int | None = None, result: str = "success", reason: str | None = None, metadata: dict[str, Any] | None = None, actor_identity_id: str | None = None, actor_tg_id: int | None = None, path_or_handler: str | None = None, ) -> AuditEvent | None: context = get_telegram_context(audit_context) return await safe_record_audit_event( session, event_type=event_type, channel="telegram", path_or_handler=path_or_handler or (context.path_or_handler if context else "telegram"), actor_identity_id=actor_identity_id if actor_identity_id is not None else (context.actor_identity_id if context else None), actor_tg_id=actor_tg_id if actor_tg_id is not None else (context.actor_tg_id if context else None), entity_type=entity_type, entity_id=entity_id, result=result, reason=reason, metadata=metadata, request_id=context.request_id if context else None, ) def _redis_record_to_event_like(rec: dict[str, Any]) -> SimpleNamespace: created = rec.get("created_at") if isinstance(created, str): try: created = datetime.fromisoformat(created.replace("Z", "+00:00")) except Exception: created = datetime.now(timezone.utc) elif created is None: created = datetime.now(timezone.utc) return SimpleNamespace( id=None, event_type=rec.get("event_type", "telegram_access"), channel=rec.get("channel", "telegram"), path_or_handler=rec.get("path_or_handler") or "", actor_identity_id=rec.get("actor_identity_id"), actor_tg_id=rec.get("actor_tg_id"), entity_type=rec.get("entity_type"), entity_id=rec.get("entity_id"), result=rec.get("result", "success"), reason=rec.get("reason"), metadata_=rec.get("metadata_"), request_id=rec.get("request_id"), created_at=created, ) async def _list_audit_events_from_redis( tg_id: int | None, identity_id: str | None, channel: str | None, event_types: list[str] | None, max_events: int = 3000, ) -> list[SimpleNamespace]: if not AUDIT_REDIS_BUFFER_ENABLED: return [] from core.redis_cache import cache_lrange out: list[SimpleNamespace] = [] seen: set[tuple[str, str]] = set() keys_to_read = [] if tg_id is not None: keys_to_read.append(f"{AUDIT_REDIS_USER_PREFIX}{tg_id}") if identity_id: keys_to_read.append(f"{AUDIT_REDIS_IDENTITY_PREFIX}{identity_id}") for key in keys_to_read: raw = await cache_lrange(key, -max_events, -1) for rec in reversed(raw): if not isinstance(rec, dict): continue created = rec.get("created_at") rid = rec.get("request_id") or "" if (created, rid) in seen: continue if channel and rec.get("channel") != channel: continue if event_types and rec.get("event_type") not in event_types: continue seen.add((str(created), rid)) out.append(_redis_record_to_event_like(rec)) out.sort(key=lambda e: e.created_at, reverse=True) return out[:max_events] async def list_audit_events_from_redis_buffer(max_events: int = 5000) -> list[SimpleNamespace]: if not AUDIT_REDIS_BUFFER_ENABLED: return [] from core.redis_cache import cache_lrange raw = await cache_lrange(AUDIT_REDIS_FLUSH_KEY, -max_events, -1) out = [] for rec in raw: if not isinstance(rec, dict): continue out.append(_redis_record_to_event_like(rec)) out.sort(key=lambda e: e.created_at) return out def _aggregate_audit_rows( rows: list[tuple[Any, Any, Any, Any]], ) -> tuple[dict[str, dict[str, Any]], list[dict[str, Any]], set[tuple[str, int | str]]]: by_step: dict[str, dict[str, Any]] = {} all_actors: set[tuple[str, int | str]] = set() for row in rows: path, res, tg_id, identity_id = row[0], row[1], row[2], row[3] if _is_ignored_analytics_event(path or ""): continue actor = _audit_actor_key(identity_id, tg_id) if actor is not None: all_actors.add(actor) for step in _normalize_path_to_steps(path or ""): if step not in by_step: by_step[step] = {"total": 0, "success": 0, "fail": 0, "actors": set()} by_step[step]["total"] += 1 if res == "success": by_step[step]["success"] += 1 else: by_step[step]["fail"] += 1 if actor is not None: by_step[step]["actors"].add(actor) by_path_list = [] for step, data in sorted(by_step.items(), key=lambda x: -x[1]["total"]): total = data["total"] fail = data["fail"] unique = len(data["actors"]) fail_rate = round(100.0 * fail / total, 1) if total else 0 by_path_list.append( { "step": step, "label": AUDIT_STEP_LABELS.get(step, step), "total": total, "success": data["success"], "fail": fail, "unique_users": unique, "fail_rate_pct": fail_rate, } ) return by_step, by_path_list, all_actors def _audit_actor_key(identity_id: Any, tg_id: Any) -> tuple[str, str | int] | None: if tg_id not in (None, 0, "0", ""): try: return ("tg", int(tg_id)) except (TypeError, ValueError): return ("tg", str(tg_id)) if identity_id not in (None, ""): return ("identity", str(identity_id)) return None def _event_like_dedupe_key(event: AuditEvent | SimpleNamespace) -> tuple[Any, ...]: request_id = getattr(event, "request_id", None) event_type = getattr(event, "event_type", None) path = getattr(event, "path_or_handler", None) result = getattr(event, "result", None) if request_id: return ("request", request_id, event_type, path, result) return ( "raw", event_type, getattr(event, "channel", None), path, getattr(event, "actor_tg_id", None), getattr(event, "actor_identity_id", None), result, getattr(event, "reason", None), str(getattr(event, "created_at", None)), ) def _dedupe_event_like(events: list[AuditEvent | SimpleNamespace]) -> list[AuditEvent | SimpleNamespace]: seen: set[tuple[Any, ...]] = set() unique: list[AuditEvent | SimpleNamespace] = [] for event in events: key = _event_like_dedupe_key(event) if key in seen: continue seen.add(key) unique.append(event) return unique async def list_audit_events( session: AsyncSession, *, identity_id: str | None = None, tg_id: int | None = None, channel: str | None = None, event_type: str | None = None, event_types: Iterable[str] | None = None, limit: int = 100, offset: int = 0, ) -> list[AuditEvent | SimpleNamespace]: event_types_list = sorted(event_types) if event_types else None if not AUDIT_REDIS_BUFFER_ENABLED: return await fetch_audit_events_db( session, identity_id=identity_id, tg_id=tg_id, channel=channel, event_type=event_type, event_types=event_types_list, limit=limit, offset=offset, ) redis_events = await _list_audit_events_from_redis( tg_id, identity_id, channel, event_types_list, max_events=3000 ) need = offset + limit + len(redis_events) db_events = await fetch_audit_events_db_window( session, identity_id=identity_id, tg_id=tg_id, channel=channel, event_type=event_type, event_types=event_types_list, limit=min(5000, need), ) merged = _dedupe_event_like(redis_events + db_events) merged.sort(key=lambda e: (e.created_at, getattr(e, "id", 0)), reverse=True) return merged[offset : offset + limit] async def delete_old_audit_events( session: AsyncSession, *, older_than_days: int = 90, ) -> int: return await delete_old_audit_events_db(session, older_than_days=older_than_days) async def get_audit_stats( session: AsyncSession, *, date_from: datetime, date_to: datetime, max_events: int = 100_000, ) -> dict[str, Any]: d_from = _naive_utc(date_from) d_to = _naive_utc(date_to) rows = await fetch_audit_rows_db(session, date_from=d_from, date_to=d_to, limit=max_events) rows.extend(await fetch_successful_payment_rows_db(session, date_from=d_from, date_to=d_to, limit=max_events)) by_step, by_path_list, all_actors = _aggregate_audit_rows(rows) raw_total_events = sum(1 for row in rows if not _is_ignored_analytics_event(row[0] or "")) analytics_total_events = sum(row["total"] for row in by_path_list) return { "summary": { "date_from": date_from.isoformat(), "date_to": date_to.isoformat(), "total_events": raw_total_events, "raw_total_events": raw_total_events, "analytics_total_events": analytics_total_events, "unique_users": len(all_actors), }, "by_path": by_path_list, } async def get_audit_stats_from_redis(max_events: int = 5000) -> dict[str, Any] | None: if not AUDIT_REDIS_BUFFER_ENABLED: return None events = await list_audit_events_from_redis_buffer(max_events=max_events) filtered_events = events rows = [(e.path_or_handler, e.result, e.actor_tg_id, e.actor_identity_id) for e in filtered_events] by_step, by_path_list, all_actors = _aggregate_audit_rows(rows) raw_total_events = sum(1 for row in rows if not _is_ignored_analytics_event(row[0] or "")) analytics_total_events = sum(row["total"] for row in by_path_list) return { "summary": { "source": "redis", "total_events": raw_total_events, "raw_total_events": raw_total_events, "analytics_total_events": analytics_total_events, "unique_users": len(all_actors), }, "by_path": by_path_list, } async def get_audit_stats_from_redis_since( *, date_from: datetime | None = None, max_events: int = 5000, ) -> dict[str, Any] | None: if not AUDIT_REDIS_BUFFER_ENABLED: return None events = await list_audit_events_from_redis_buffer(max_events=max_events) filtered_events = events if date_from is not None: threshold = date_from.astimezone(timezone.utc) if date_from.tzinfo else date_from.replace(tzinfo=timezone.utc) filtered_events = [e for e in events if getattr(e, "created_at", None) and e.created_at >= threshold] rows = [(e.path_or_handler, e.result, e.actor_tg_id, e.actor_identity_id) for e in filtered_events] by_step, by_path_list, all_actors = _aggregate_audit_rows(rows) raw_total_events = sum(1 for row in rows if not _is_ignored_analytics_event(row[0] or "")) analytics_total_events = sum(row["total"] for row in by_path_list) return { "summary": { "source": "redis", "total_events": raw_total_events, "raw_total_events": raw_total_events, "analytics_total_events": analytics_total_events, "unique_users": len(all_actors), }, "by_path": by_path_list, } async def get_audit_funnel( session: AsyncSession, *, date_from: datetime, date_to: datetime, steps_ordered: tuple[str, ...] | None = None, max_events: int = 50_000, ) -> list[dict[str, Any]]: steps = steps_ordered or DEFAULT_FUNNEL_STEPS d_from = _naive_utc(date_from) d_to = _naive_utc(date_to) rows = await fetch_audit_rows_db(session, date_from=d_from, date_to=d_to, limit=max_events) rows.extend(await fetch_successful_payment_rows_db(session, date_from=d_from, date_to=d_to, limit=max_events)) return _funnel_from_rows(rows, steps) def _funnel_from_rows( rows: list[tuple[Any, ...]], steps_ordered: tuple[str, ...] | None = None, ) -> list[dict[str, Any]]: steps = steps_ordered or DEFAULT_FUNNEL_STEPS step_actors: dict[str, set[tuple[str, str | int]]] = {step: set() for step in steps} for row in rows: if len(row) >= 4: path, result, tg_id, identity_id = row[0], row[1], row[2], row[3] else: path, tg_id, identity_id = row[0], row[1], row[2] result = "success" if _is_ignored_analytics_event(path or ""): continue key = _audit_actor_key(identity_id, tg_id) if key is None: continue for step in _normalize_path_to_steps(path or ""): if not _funnel_step_counts(path or "", str(result or ""), step): continue if step in step_actors: step_actors[step].add(key) funnel_list = [] prev_actors: set[tuple[str, str | int]] | None = None for step in steps: actors = step_actors.get(step, set()) count = len(actors) conversion = None if prev_actors is not None and prev_actors: overlap = len(prev_actors & actors) conversion = round(100.0 * overlap / len(prev_actors), 1) funnel_list.append( { "step": step, "label": AUDIT_STEP_LABELS.get(step, step), "count": count, "conversion_from_prev_pct": conversion, } ) prev_actors = actors return funnel_list async def get_audit_funnel_from_redis( max_events: int = 5000, steps_ordered: tuple[str, ...] | None = None, ) -> list[dict[str, Any]] | None: if not AUDIT_REDIS_BUFFER_ENABLED: return None events = await list_audit_events_from_redis_buffer(max_events=max_events) rows = [(e.path_or_handler, e.result, e.actor_tg_id, e.actor_identity_id) for e in events] return _funnel_from_rows(rows, steps_ordered) async def get_audit_funnel_from_redis_since( *, date_from: datetime | None = None, max_events: int = 5000, steps_ordered: tuple[str, ...] | None = None, ) -> list[dict[str, Any]] | None: if not AUDIT_REDIS_BUFFER_ENABLED: return None events = await list_audit_events_from_redis_buffer(max_events=max_events) filtered_events = events if date_from is not None: threshold = date_from.astimezone(timezone.utc) if date_from.tzinfo else date_from.replace(tzinfo=timezone.utc) filtered_events = [e for e in events if getattr(e, "created_at", None) and e.created_at >= threshold] rows = [(e.path_or_handler, e.result, e.actor_tg_id, e.actor_identity_id) for e in filtered_events] return _funnel_from_rows(rows, steps_ordered) async def drain_audit_redis_to_db(session_factory: Any) -> int: from core.redis_cache import cache_delete, cache_lmove_batch, cache_lpop_batch, cache_lrange, cache_setnx if not await cache_setnx(_AUDIT_REDIS_DRAIN_LOCK_KEY, 1, _AUDIT_REDIS_DRAIN_LOCK_TTL_SEC): logger.info("[Audit] drain_audit_redis_to_db пропущен: уже выполняется другой drain") return 0 total = 0 try: while True: raw_batch = await cache_lrange(_AUDIT_REDIS_PROCESSING_KEY, 0, AUDIT_REDIS_DRAIN_BATCH - 1) if not raw_batch: raw_batch = await cache_lmove_batch( AUDIT_REDIS_FLUSH_KEY, _AUDIT_REDIS_PROCESSING_KEY, AUDIT_REDIS_DRAIN_BATCH, ) if not raw_batch: break batch = [rec for rec in raw_batch if isinstance(rec, dict)] if not batch: logger.warning("[Audit] drain_audit_redis_to_db: отброшен пустой/битый батч (%s элементов)", len(raw_batch)) await cache_lpop_batch(_AUDIT_REDIS_PROCESSING_KEY, len(raw_batch)) continue try: async with session_factory() as session: await ensure_audit_table(session) existing_request_ids = await fetch_existing_audit_request_ids_db( session, [str(rec.get("request_id")) for rec in batch if rec.get("request_id")], ) inserted_count = 0 seen_batch_request_ids: set[str] = set() for rec in batch: request_id = rec.get("request_id") if request_id and (request_id in existing_request_ids or request_id in seen_batch_request_ids): continue if request_id: seen_batch_request_ids.add(request_id) created = rec.get("created_at") if isinstance(created, str): try: created = datetime.fromisoformat(created.replace("Z", "+00:00")) except Exception: created = datetime.now(timezone.utc) elif created is None: created = datetime.now(timezone.utc) event = AuditEvent( event_type=rec.get("event_type", "telegram_access"), channel=rec.get("channel", "telegram"), path_or_handler=rec.get("path_or_handler") or "telegram", actor_identity_id=rec.get("actor_identity_id"), actor_tg_id=rec.get("actor_tg_id"), entity_type=rec.get("entity_type"), entity_id=rec.get("entity_id"), result=rec.get("result", "success"), reason=rec.get("reason"), metadata_=rec.get("metadata_"), request_id=request_id, created_at=created, ) session.add(event) inserted_count += 1 await session.commit() await cache_lpop_batch(_AUDIT_REDIS_PROCESSING_KEY, len(raw_batch)) total += inserted_count except Exception as exc: logger.warning("[Audit] drain_audit_redis_to_db батч не записан, останется в Redis processing: %s", exc) break return total finally: await cache_delete(_AUDIT_REDIS_DRAIN_LOCK_KEY)