Explorer
/proc/287/root/tmp/aa053-fix-policy.py
← Zurück ↓ Download
from pathlib import Path
import yaml
src=Path('/opt/struktur/social-media-radar/architecture/data-ownership.yaml')
doc=yaml.safe_load(src.read_text())
tech=doc['field_groups']['signals_technical']
tech['examples']=[x for x in tech['examples'] if x not in {'external_id','observed_at','published_at','content_hash'}]
mixed=doc['field_groups']['mixed']
mixed['examples']=[x for x in mixed['examples'] if x != 'processing_state']
items=list(doc['field_groups'].items())
new=[]
inserted=False
for name,group in items:
    new.append((name,group))
    if name=='signal_sources':
        new.append(('signals_enrichment',{
            'table':'signals',
            'owner':'signal_intake',
            'examples':['source_platform','external_id','observed_at','published_at','content_hash','processing_state','ad_start_at','last_seen_at','active_status'],
            'known_writers':['sma_core/signal_enrichment.py'],
            'status':'ASSIGNED',
        }))
        inserted=True
if not inserted: raise SystemExit('signal_sources group missing')
doc['field_groups']=dict(new)
Path('/tmp/aa053-data-ownership-v5.yaml').write_text(yaml.safe_dump(doc,sort_keys=False,allow_unicode=True),encoding='utf-8')