"""Public-safe batch ingestion facts from active PostgreSQL membership only. Video creation is database ingestion, not camera collection. These facts describe currently active rows and cannot reconstruct a historical deletion audit. """ from __future__ import annotations from collections import Counter from dataclasses import dataclass from datetime import date, datetime, timezone import hashlib import math import re from uuid import UUID from zoneinfo import ZoneInfo import psycopg from annotation_statistics import validate_public_labels from live_inventory import ( _boundary_path, _city_for_coordinates, _load_boundaries, _MISSING_COORDINATES, _OUTSIDE_BOUNDARIES, _valid_float, ) _BEIJING = ZoneInfo('Asia/Shanghai') _BATCH_SQL = """ SELECT b.batch_id, array_agg(v.collect_lon), array_agg(v.collect_lat), array_agg(v.created_at), array_agg(v.updated_at), bool_or(v.sha256 = b.representative_video_sha256) AS has_representative, b.processing_info, b.validation_passed, b.review_status, b.labels, array_agg(v.duration_seconds) FROM aerial.videos v LEFT JOIN aerial.annotation_batch_videos bv ON bv.video_sha256 = v.sha256 LEFT JOIN aerial.annotation_batches b ON b.batch_id = bv.batch_id WHERE v.deleted_at IS NULL GROUP BY b.batch_id """ @dataclass(frozen=True) class BatchIngestionDay: day: date videos: int known_seconds: float | None = None known_videos: int | None = None unknown_videos: int | None = None @property def duration_hours(self) -> float | None: return self.known_seconds / 3600 if self.known_seconds is not None and self.known_videos else None @dataclass(frozen=True) class BatchUpdateRecord: code: str city: str videos: int first_ingested_at: datetime | None last_ingested_at: datetime | None source_updated_at: datetime | None annotation_completed_at: datetime | None annotations_available: bool ingestion_days: tuple[BatchIngestionDay, ...] known_seconds: float | None = None known_videos: int | None = None unknown_videos: int | None = None @property def duration_hours(self) -> float | None: return self.known_seconds / 3600 if self.known_seconds is not None and self.known_videos else None @dataclass(frozen=True) class BatchUpdateCatalog: records: tuple[BatchUpdateRecord, ...] unlinked_videos: int = 0 unlinked_known_seconds: float | None = None unlinked_known_videos: int | None = None unlinked_unknown_videos: int | None = None @property def unlinked_duration_hours(self) -> float | None: return (self.unlinked_known_seconds / 3600 if self.unlinked_known_seconds is not None and self.unlinked_known_videos else None) @property def includes_durations(self) -> bool: """An empty measured catalog is available; legacy count-only catalogs are not.""" return (self.unlinked_known_seconds is not None and self.unlinked_known_videos is not None and self.unlinked_unknown_videos is not None and all(record.known_seconds is not None and record.known_videos is not None and record.unknown_videos is not None and all(day.known_seconds is not None and day.known_videos is not None and day.unknown_videos is not None for day in record.ingestion_days) for record in self.records)) def _public_code(batch_id: UUID) -> str: identifier = batch_id if isinstance(batch_id, UUID) else UUID(str(batch_id)) return 'B-' + hashlib.sha256(b'nds-public-batch-v1\0' + identifier.bytes).hexdigest()[:16] def _finished_at(value: object) -> datetime | None: if not isinstance(value, str) or not re.fullmatch( r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|[+-]\d{2}:\d{2})', value ): return None try: return datetime.fromisoformat(value).astimezone(timezone.utc) except (ValueError, OverflowError): return None def fetch_batch_updates(dsn: str) -> BatchUpdateCatalog: """Read every active membership once; private identifiers stay local. A batch exists here if it has at least one active video. The source membership constraint assigns each video to one batch. GPS votes prefer known administrative cities, breaking ties lexically, then use explicit fallbacks. Member arrays are aggregated before transfer so labels travel only once per batch. """ path = _boundary_path() boundaries = _load_boundaries(path, path.stat().st_mtime_ns) with psycopg.connect(dsn, options='-c default_transaction_read_only=on') as connection: with connection.cursor() as cursor: cursor.execute(_BATCH_SQL) rows = cursor.fetchall() groups = {} unlinked = 0 unlinked_seconds, unlinked_known, unlinked_unknown = [], 0, 0 coordinate_cities = {} for batch_id, lons, lats, created_times, updated_times, representative, processing, validated, review, labels, durations in rows: if batch_id is None: unlinked += len(created_times) for seconds in durations: duration = _valid_float(seconds) if duration is None: unlinked_unknown += 1 else: unlinked_known += 1 unlinked_seconds.append(float(duration)) continue group = dict(cities=Counter(), created=[], updated=[], days={}, seconds=[], known=0, unknown=0, videos=len(created_times), representative=bool(representative), processing=processing, validated=validated, review=review, labels=labels) groups[batch_id] = group for lon, lat, created, updated, seconds in zip(lons, lats, created_times, updated_times, durations): duration = _valid_float(seconds) if duration is None: group['unknown'] += 1 else: group['known'] += 1 group['seconds'].append(float(duration)) if (lon, lat) not in coordinate_cities: coordinate_cities[(lon, lat)] = _city_for_coordinates(lon, lat, boundaries) group['cities'][coordinate_cities[(lon, lat)]] += 1 if created is not None: group['created'].append(created) day = group['days'].setdefault(created.astimezone(_BEIJING).date(), [0, [], 0, 0]) day[0] += 1 if duration is None: day[3] += 1 else: day[1].append(float(duration)) day[2] += 1 if updated is not None: group['updated'].append(updated) records = [] for batch_id, group in groups.items(): cities = group['cities'] known = {city: count for city, count in cities.items() if city not in {_MISSING_COORDINATES, _OUTSIDE_BOUNDARIES}} city = min(known or cities, key=lambda name: (-(known or cities)[name], name)) processing = group['processing'] if isinstance(group['processing'], dict) else {} completed = processing.get('status') == 'completed' available = bool(completed and group['validated'] is True and group['review'] != 'rejected' and group['representative']) if available: try: validate_public_labels(group['labels']) except ValueError: available = False records.append(BatchUpdateRecord( _public_code(batch_id), city, group['videos'], min(group['created'], default=None), max(group['created'], default=None), max(group['updated'], default=None), _finished_at(processing.get('finished_at')) if completed else None, available, tuple(BatchIngestionDay(day, values[0], math.fsum(values[1]), values[2], values[3]) for day, values in sorted(group['days'].items())), math.fsum(group['seconds']), group['known'], group['unknown'], )) return BatchUpdateCatalog(tuple(sorted(records, key=lambda record: record.code)), unlinked, math.fsum(unlinked_seconds), unlinked_known, unlinked_unknown)