"""Example 1 — Ingest: create an empty store and import a source into it.

A **source import** is a *grow op*: it creates canonical sample rows in a store
from a WebDataset tar you upload. This is the raw→canonical boundary — every
sample gets a server-assigned canonical id, and the keys in the tar become the
store's keys.

    DATAVO_SERVER=https://dev.datavo.io python examples/01_ingest_source.py

See examples/README.md for auth and local-stack setup.
"""

from __future__ import annotations

from _common import build_source_tar, cleanup, make_client, step, unique

from datavo_sdk import SourceImportCreateRequest

# A source shard is just samples keyed by a caller-chosen id, each carrying one
# or more keys. Here: raw audio (bytes) + a transcript (text).
SAMPLES = [
    {"__key__": "0001", "audio.wav": b"RIFF....fake-audio-one", "transcript.txt": "one"},
    {"__key__": "0002", "audio.wav": b"RIFF....fake-audio-two", "transcript.txt": "two"},
    {"__key__": "0003", "audio.wav": b"RIFF....fake-audio-three", "transcript.txt": "three"},
]


def main() -> None:
    client = make_client()
    store = unique("example_ingest")
    try:
        step(f"create an empty store {store!r}")
        client.create_store(
            name=store,
            display_name="Ingest example",
            metadata={"example": "01_ingest_source"},
        )

        step("open a source import against the store")
        imp = client.create_source_import(
            sample_store=store,
            # REPLAY: operation_id is yours. Re-running this script with the same one
            # returns the existing import instead of opening a second.
            operation_id="demo-batch-001",
            default_tags={"language": "en"},  # applied to every sample it introduces
        )

        step("upload the source shard as part 0 (reserve → upload → complete, in one call)")
        tar = build_source_tar(SAMPLES, root_key="audio.wav")
        client.upload_source_import_part(imp.source_import_id, 0, tar)

        step("commit, then wait for the worker to materialize canonical samples")
        client.commit_source_import(imp.source_import_id)
        done = client.wait_for_source_import(imp.source_import_id, timeout_seconds=180)

        detail = client.get_store(store)
        print(f"\n  import state={done.state!r}, sample_count={done.sample_count}")
        print(f"  store {store!r} now has {detail.sample_count} samples; keys={[k.key for k in detail.keys]}")
    finally:
        cleanup(client, store)


if __name__ == "__main__":
    main()