sag_align

def translate_id(uuid_v4: str) -> str

from octx.sag_align import translate_id

v7 = translate_id("550e8400-e29b-41d4-a716-446655440000")
# → "550e8400-e29b-71d4-a716-446655440000"

def revert_id(uuid_v7: str) -> str

from octx.sag_align import revert_id

v4 = revert_id("550e8400-e29b-71d4-a716-446655440000")
# → "550e8400-e29b-41d4-a716-446655440000"

def ensure_octx_ids(
    records: Iterable[dict[str, Any]],
    *,
    id_fields: Collection[str] = {"id", "document_id", "chunk_id", "event_id", "entity_id", "parent_id"},
) -> Iterator[dict[str, Any]]

from octx.sag_align import ensure_octx_ids

records = [{"id": "550e8400-e29b-41d4-a716-446655440000", "title": "doc"}]
result = list(ensure_octx_ids(records))
# result[0]["id"] → "550e8400-e29b-71d4-a716-446655440000"

def default_sag_field_profile() -> SagFieldProfile

from octx.sag_align import default_sag_field_profile

profile = default_sag_field_profile()
# SagFieldProfile(id=36, name=512, entity_type=100, ...)

def validate_cardinality_for_sag(
    chunk_events: Iterable[dict[str, Any]],
    *,
    events: Collection[str] | None = None,
) -> list[SagCardinalityIssue]

from octx.sag_align import validate_cardinality_for_sag

issues = validate_cardinality_for_sag(chunk_events)
for issue in issues:
    print(issue.event_id, issue.chunk_ids)

def validate_field_lengths_for_sag(
    chunks: Iterable[dict[str, Any]],
    events: Iterable[dict[str, Any]],
    entities: Iterable[dict[str, Any]],
    chunk_events: Iterable[dict[str, Any]],
    event_entities: Iterable[dict[str, Any]],
    *,
    profile: SagFieldProfile | None = None,
) -> list[SagFieldIssue]

from octx.sag_align import validate_field_lengths_for_sag

issues = validate_field_lengths_for_sag(chunks, events, entities, ce, ee)

def find_entity_dedup_matches(
    entities: Iterable[dict[str, Any]],
    *,
    normalize: bool = True,
) -> list[SagDedupMatch]

from octx.sag_align import find_entity_dedup_matches

matches = find_entity_dedup_matches(entities)

def classify_ids(
    records: Iterable[dict[str, Any]],
    *,
    id_fields: Collection[str] = {"id", "document_id", "chunk_id", "event_id", "entity_id", "parent_id"},
) -> dict[str, int]

from octx.sag_align import classify_ids

stats = classify_ids(all_records)
# {"octx": 5, "v4": 2, "other": 0}

def validate_sag_structured(
    chunks: Iterable[dict[str, Any]],
    events: Iterable[dict[str, Any]],
    entities: Iterable[dict[str, Any]],
    chunk_events: Iterable[dict[str, Any]],
    event_entities: Iterable[dict[str, Any]],
    *,
    document_ids: Collection[str] | None = None,
) -> list[SagCoverageIssue]

from octx.sag_align import validate_sag_structured

issues = validate_sag_structured(chunks, events, entities, ce, ee, document_ids=["doc-1"])
for issue in issues:
    print(issue.rule, issue.record_id)

def validate_sag_constraints(
    chunks: Iterable[dict[str, Any]],
    events: Iterable[dict[str, Any]],
    entities: Iterable[dict[str, Any]],
    chunk_events: Iterable[dict[str, Any]],
    event_entities: Iterable[dict[str, Any]],
) -> list[SagConstraintIssue]

from octx.sag_align import validate_sag_constraints

issues = validate_sag_constraints(chunks, events, entities, ce, ee)
for issue in issues:
    print(issue.code, issue.record_id)

def write_structured_to_workspace(
    workspace: str | Path,
    *,
    chunks: Iterable[dict[str, Any]] = (),
    events: Iterable[dict[str, Any]] = (),
    entities: Iterable[dict[str, Any]] = (),
    chunk_events: Iterable[dict[str, Any]] = (),
    event_entities: Iterable[dict[str, Any]] = (),
    profile: SagFieldProfile | None = None,
) -> SagAlignmentReport

from octx.sag_align import write_structured_to_workspace

report = write_structured_to_workspace(
    "my-workspace",
    chunks=[...],
    events=[...],
    entities=[...],
    chunk_events=[...],
    event_entities=[...],
)

@dataclass(frozen=True)
class SagFieldProfile:
    id: int = 36            # UUID string length
    name: int = 512         # entity.name
    entity_type: int = 100  # entity.type
    title: int = 512        # event.title, document.title
    summary: int = 4096     # event.summary
    category: int = 128     # event.category
    description: int = 4096 # entity.description, relation description
    text: int = 0           # chunk.text / event.content (0 = unlimited)

@dataclass(frozen=True)
class SagCardinalityIssue:
    event_id: str
    chunk_ids: tuple[str, ...]  # All Chunk IDs linked to this Event
    primary_hint: str | None = None  # Suggested primary Chunk

@dataclass(frozen=True)
class SagFieldIssue:
    path: str               # JSONL file path (e.g. "data/entities.jsonl")
    line: int               # 1-indexed line number
    record_id: str | None
    field: str              # Field name exceeding the limit
    value_length: int
    limit: int

@dataclass(frozen=True)
class SagDedupMatch:
    entity_id_a: str
    entity_id_b: str
    name: str               # Matching name value
    same_package: bool = True  # Duplicate within the same package

@dataclass(frozen=True)
class SagCoverageIssue:
    rule: str               # Rule identifier (see validate_sag_structured)
    record_id: str | None   # ID of the offending record
    description: str        # Human-readable explanation

@dataclass(frozen=True)
class SagConstraintIssue:
    code: str               # Constraint code (e.g. chunk_text_empty)
    record_id: str | None   # ID of the offending record
    description: str        # Human-readable explanation

@dataclass
class SagAlignmentReport:
    cardinality_issues: list[SagCardinalityIssue]
    field_issues: list[SagFieldIssue]
    dedup_within: list[SagDedupMatch]
    coverage_issues: list[SagCoverageIssue]
    constraint_issues: list[SagConstraintIssue]
    id_translations: int        # Number of UUIDv4→v7 translations
    unchanged_ids: int          # Number of IDs left unchanged

    # Properties
    has_cardinality_conflicts: bool  # cardinality_issues non-empty?
    has_field_issues: bool           # field_issues non-empty?
    has_dedup_within: bool           # dedup_within non-empty?
    has_coverage_issues: bool        # coverage_issues non-empty?
    has_constraint_issues: bool      # constraint_issues non-empty?

    # Methods
    def merge(self, other: SagAlignmentReport) -> SagAlignmentReport
        # Merge two reports (e.g. from different packages)