Spaces:
Running
Running
| """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 | |
| """ | |
| class BatchIngestionDay: | |
| day: date | |
| videos: int | |
| known_seconds: float | None = None | |
| known_videos: int | None = None | |
| unknown_videos: int | None = None | |
| def duration_hours(self) -> float | None: | |
| return self.known_seconds / 3600 if self.known_seconds is not None and self.known_videos else None | |
| 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 | |
| def duration_hours(self) -> float | None: | |
| return self.known_seconds / 3600 if self.known_seconds is not None and self.known_videos else None | |
| 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 | |
| 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) | |
| 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) | |