37 lines
2 KiB
Python
37 lines
2 KiB
Python
"""Read registered channel context. The persona selects; this module never guesses intent."""
|
|
import json
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
def normalize_channel_id(channel, rows):
|
|
if channel is None:
|
|
return None, 'NO_CHANNEL_PROVIDED'
|
|
requested = str(channel).strip()
|
|
matches = [item['id'] for item in rows if requested == item['id'] or requested in item.get('machine_aliases', [])]
|
|
if len(matches) != 1:
|
|
raise ValueError('UNKNOWN_OR_AMBIGUOUS_PRIVATE_CHANNEL')
|
|
return matches[0], 'EXACT' if requested == matches[0] else 'MACHINE_ALIAS_NORMALIZED'
|
|
|
|
def load_channels(channel=None, root=ROOT):
|
|
data = json.loads((root / 'routing/persona-channel-context-map.json').read_text())
|
|
if data['map_id'] != 'ZY-PERSONA-CHANNEL-CONTEXT-001':
|
|
raise ValueError('CHANNEL_CONTEXT_MAP_INVALID')
|
|
rows = data['channels']
|
|
if len({c['id'] for c in rows}) != len(rows):
|
|
raise ValueError('CHANNEL_IDS_AMBIGUOUS')
|
|
registry = json.loads((root / 'routing/fifth-domain-number-registry.json').read_text())
|
|
numbers = {entry['id']: entry for entry in registry['registrations']}
|
|
for item in rows:
|
|
profile = json.loads((root / item['profile']).read_text())
|
|
if item['id'] not in numbers or profile['channel_id'] != item['id']:
|
|
raise ValueError('CHANNEL_NOT_REGISTERED')
|
|
registered = numbers[item['id']]
|
|
if profile['world_path'] != registered['world_path']:
|
|
raise ValueError('CHANNEL_PATH_MISMATCH')
|
|
item['world_path'] = profile['world_path']
|
|
normalized, normalization_state = normalize_channel_id(channel, rows)
|
|
return {**data, 'requested_channel': channel, 'normalized_channel': normalized, 'channel_normalization_state': normalization_state,
|
|
'selected_channel': next((c for c in rows if c['id'] == normalized), None),
|
|
'selection_state': 'PERSONA_SELECTED' if normalized else 'PERSONA_INTERPRETATION_REQUIRED',
|
|
'authority_granted': False}
|