ddmitry/clickstream-ch-kafka-superset-demo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(generator): сохранена фактура визита при восстановлении

Browse Source 
- Зачем:
  - визит после восстановления не должен менять браузер и источники перехода внутри одного click_id.
- Что:
  - добавлен base_click_id в state v3 для восстановления донора фактуры.
  - исправлено восстановление timestamp offset без потери микросекунд.
  - расширены тесты и стыковая проверка browser/source и device/os/geo.
- Проверка:
  - uv run --with-requirements generator/requirements.txt pytest generator/tests -q.
  - bash -n scripts/check_generated_analytics.sh.
  - git diff --cached --check.
This commit is contained in:
ddadmin
2026-07-04 21:46:23 +03:00
parent 0cbfe9b32e
commit e2d06841de
13 changed files with 464 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
generator/src/clickstream_generator/generation.py
+28 -4
View File
@@ -142,13 +142,27 @@ class EventGenerator:
user_profile: dict[str, dict] | None = None,
) -> dict[str, list[dict]]:
"""Генерирует один визит с сохранением связей."""
batch, _base_click_id = self.generate_visit_batch(
batch_size,
planned_start_at=planned_start_at,
user_profile=user_profile,
)
return batch
def generate_visit_batch(
self,
batch_size: int,
planned_start_at: datetime | None = None,
user_profile: dict[str, dict] | None = None,
) -> tuple[dict[str, list[dict]], str | None]:
"""Генерирует визит и возвращает click_id донора фактуры."""
if not self.dictionary.browser_events:
return {
"browser_events": [],
"location_events": [],
"device_events": [],
"geo_events": [],
}
}, None
batch = {
"browser_events": [],
@@ -158,7 +172,7 @@ class EventGenerator:
}
if batch_size <= 0:
return batch
return batch, None
max_visit_events = min(batch_size, self.config.max_session_events)
min_visit_events = min(2, max_visit_events)
@@ -181,8 +195,18 @@ class EventGenerator:
base_browser_events = self.dictionary.browser_by_click_id[base_click_id][:len(visit_path)]
else:
base_browser = self.rng.choice(self.dictionary.browser_events)
# Запасная ветка тоже восстановима: state хранит донора, а индекс идёт по кругу.
base_click_id = base_browser["click_id"]
base_browser_events = [base_browser for _ in range(len(visit_path))]
source_events = self.dictionary.browser_by_click_id[base_click_id]
if not all(
event["event_id"] in self.dictionary.location_by_event_id
for event in source_events
):
raise ValueError(f"Fixture base_click_id has incomplete locations: {base_click_id}")
base_browser_events = [
source_events[event_index % len(source_events)]
for event_index in range(len(visit_path))
]
base_device = (
user_profile["device"]
@@ -229,4 +253,4 @@ class EventGenerator:
planned_timestamp += timedelta(seconds=self._visit_pause_seconds())
return batch
return batch, base_click_id
generator/src/clickstream_generator/runtime.py
+30 -12
View File
@@ -35,6 +35,7 @@ class ActiveVisit:
batch: dict[str, list[dict]]
timestamps: list[datetime]
base_click_id: str
user: UserProfile | None = None
next_index: int = 0
@@ -130,7 +131,12 @@ def _datetime_to_state(value: datetime | None) -> str | None:
def _timestamp_to_state_offset(started_at: datetime, timestamp: datetime) -> int:
return int((timestamp - started_at).total_seconds() * 1_000_000)
delta = timestamp - started_at
return (
delta.days * 86_400_000_000
+ delta.seconds * 1_000_000
+ delta.microseconds
)
def _format_event_timestamp(timestamp: datetime) -> str:
@@ -217,6 +223,7 @@ class TickStreamGenerator:
return {
"user_domain_id": visit.user.user_domain_id if visit.user else None,
"click_id": browser_events[0]["click_id"],
"base_click_id": visit.base_click_id,
"next_index": visit.next_index,
"started_at": started_at.isoformat(),
"offsets_us": [
@@ -235,7 +242,7 @@ class TickStreamGenerator:
resume_model_at: datetime | None = None,
restarted_at: datetime | None = None,
) -> None:
"""Восстанавливает популяцию и активные визиты из state v2."""
"""Восстанавливает популяцию и активные визиты из state."""
users = [
self._user_from_state(item)
for item in state.population
@@ -302,6 +309,7 @@ class TickStreamGenerator:
]
batch = self._compact_visit_batch(
click_id=item["click_id"],
base_click_id=item["base_click_id"],
user=user,
timestamps=timestamps,
page_url_paths=item["page_url_paths"],
@@ -309,6 +317,7 @@ class TickStreamGenerator:
return ActiveVisit(
batch=batch,
timestamps=timestamps,
base_click_id=item["base_click_id"],
user=user,
next_index=item["next_index"],
)
@@ -316,24 +325,26 @@ class TickStreamGenerator:
def _compact_visit_batch(
self,
click_id: str,
base_click_id: str,
user: UserProfile,
timestamps: list[datetime],
page_url_paths: list[str],
) -> dict[str, list[dict]]:
batch = _empty_batch()
browser_templates = self.generator.dictionary.browser_by_click_id.get(
user.seed_click_id,
self.generator.dictionary.browser_events,
)
if base_click_id not in self.generator.dictionary.browser_by_click_id:
raise ValueError(f"Unknown fixture base_click_id: {base_click_id}")
browser_templates = self.generator.dictionary.browser_by_click_id[base_click_id]
if not browser_templates:
raise ValueError(f"Fixture base_click_id has no browser events: {base_click_id}")
for event_index, (timestamp, page_url_path) in enumerate(
zip(timestamps, page_url_paths)
):
browser_template = browser_templates[event_index % len(browser_templates)]
location_template = self.generator.dictionary.location_by_event_id.get(
browser_template["event_id"],
self.generator.dictionary.location_events[0],
)
source_event_id = browser_template["event_id"]
if source_event_id not in self.generator.dictionary.location_by_event_id:
raise ValueError(f"Unknown fixture location event_id: {source_event_id}")
location_template = self.generator.dictionary.location_by_event_id[source_event_id]
event_id = _stable_event_id(click_id, event_index)
batch["browser_events"].append(
{
@@ -428,7 +439,7 @@ class TickStreamGenerator:
self._pending_visit_births = 0.0
return
visit_batch = self.generator.generate_batch(
visit_batch, base_click_id = self.generator.generate_visit_batch(
self.generator.config.max_session_events,
planned_start_at=tick_time,
user_profile={"device": user.device, "geo": user.geo},
@@ -439,11 +450,18 @@ class TickStreamGenerator:
]
if not timestamps:
break
if base_click_id is None:
raise ValueError("Generated active visit is missing fixture base_click_id")
click_id = visit_batch["browser_events"][0]["click_id"]
self.population.start_visit(user, click_id)
self.active_visits.append(
ActiveVisit(batch=visit_batch, timestamps=timestamps, user=user)
ActiveVisit(
batch=visit_batch,
timestamps=timestamps,
base_click_id=base_click_id,
user=user,
)
)
self._pending_visit_births -= 1.0
generator/src/clickstream_generator/service.py
+1 -1
View File
@@ -205,7 +205,7 @@ class GeneratorService:
state,
wall_now_utc: datetime,
) -> datetime:
"""Считает модельную точку live-восстановления по state v2."""
"""Считает модельную точку live-восстановления по state."""
wall_now_utc = self._as_aware_utc(wall_now_utc)
wall_saved_at = self._as_aware_utc(state.wall_timestamp)
idle_seconds = max(0.0, (wall_now_utc - wall_saved_at).total_seconds())
generator/src/clickstream_generator/state.py
+7 -3
View File
@@ -10,7 +10,7 @@ from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
logger = logging.getLogger("generator")
STATE_VERSION = "2.0"
STATE_VERSION = "3.0"
def _nested_list_to_tuple(obj):
@@ -74,7 +74,7 @@ def _validate_resume_fields(data: dict) -> None:
raise ValueError("gen_seed must be an integer or null")
def _validate_v2_payload(data: dict) -> None:
def _validate_v3_payload(data: dict) -> None:
_validate_resume_fields(data)
population = data.get("population")
active_visits = data.get("active_visits")
@@ -123,6 +123,7 @@ def _validate_v2_payload(data: dict) -> None:
(
"user_domain_id",
"click_id",
"base_click_id",
"next_index",
"started_at",
"offsets_us",
@@ -132,6 +133,7 @@ def _validate_v2_payload(data: dict) -> None:
)
user_domain_id = visit["user_domain_id"]
click_id = visit["click_id"]
base_click_id = visit["base_click_id"]
offsets = visit["offsets_us"]
page_url_paths = visit["page_url_paths"]
next_index = visit["next_index"]
@@ -139,6 +141,8 @@ def _validate_v2_payload(data: dict) -> None:
raise ValueError(f"active_visits[{index}].user_domain_id is unknown")
if not isinstance(click_id, str) or not click_id:
raise ValueError(f"active_visits[{index}].click_id must be a string")
if not isinstance(base_click_id, str) or not base_click_id:
raise ValueError(f"active_visits[{index}].base_click_id must be a string")
if not isinstance(offsets, list) or not offsets:
raise ValueError(f"active_visits[{index}].offsets_us must be a non-empty list")
if not all(isinstance(offset, int) and offset >= 0 for offset in offsets):
@@ -235,7 +239,7 @@ class GeneratorState:
version,
)
raise ValueError(f"unsupported state version: {version}")
_validate_v2_payload(data)
_validate_v3_payload(data)
model_timestamp = _parse_aware_utc(
data["model_timestamp"],
"model_timestamp",