fix(etl): rebuild results from artifacts when logs are unavailable in CI

Author: melvinbarbaux

src/etl.py

@@ -19,6 +19,8 @@
 EXIT_CODE_RE = re.compile(r"\bEXIT_CODE=(-?\d+)\b")
 SEED_RE = re.compile(r"\bseed=(\d+)\b")
 EXCEPTION_RE = re.compile(r"([A-Za-z_]\w*(?:Error|Exception)):\s*(.+)")
+RUN_ID_REGIME_RE = re.compile(r"^(R\d+)_")
+TOKEN_REGIME_RE = re.compile(r"(?:^|[_-])(R\d+)(?:[_-]|$)", re.IGNORECASE)
 
 
 def to_float(value):
@@ -724,6 +726,160 @@ def extract_run_data_from_bench_log(log_path):
     return results
 
 
+def get_nested(mapping, path, default=None):
+    cursor = mapping
+    for key in path:
+        if not isinstance(cursor, dict):
+            return default
+        cursor = cursor.get(key)
+        if cursor is None:
+            return default
+    return cursor
+
+
+def normalize_status(value):
+    if value is None:
+        return None
+    status = str(value).strip().upper()
+    if not status:
+        return None
+    if status in {'SUCCESS', 'SUCCEEDED', 'OK'}:
+        return 'OK'
+    if status in {'FAIL', 'FAILED', 'ERROR'}:
+        return 'FAIL'
+    return status
+
+
+def infer_target_regime(run_id, run_payload):
+    if run_id:
+        direct = RUN_ID_REGIME_RE.match(run_id)
+        if direct:
+            return direct.group(1).upper()
+
+    for candidate in (
+        get_nested(run_payload, ('config', 'target_regime')),
+        get_nested(run_payload, ('run', 'target_regime')),
+        get_nested(run_payload, ('run', 'name')),
+    ):
+        if not candidate:
+            continue
+        match = TOKEN_REGIME_RE.search(str(candidate))
+        if match:
+            return match.group(1).upper()
+
+    return None
+
+
+def extract_run_data_from_artifact_dir(artifact_dir):
+    artifact_dir = Path(artifact_dir)
+    if not artifact_dir.is_dir():
+        return None
+
+    run_json_path = artifact_dir / 'run.json'
+    if not run_json_path.exists():
+        return None
+
+    run_id = artifact_dir.name
+    try:
+        with open(run_json_path, 'r', encoding='utf-8', errors='ignore') as f:
+            payload = json.load(f)
+    except Exception as e:
+        print(f"Error reading artifact run json {run_json_path}: {e}")
+        return None
+
+    if not isinstance(payload, dict):
+        return None
+
+    config_payload = get_nested(payload, ('config',), default={})
+    if not isinstance(config_payload, dict):
+        config_payload = {}
+
+    method_id = (
+        get_nested(payload, ('artifacts', 'method', 'id'))
+        or get_nested(payload, ('method', 'id'))
+        or get_nested(config_payload, ('method', 'id'))
+    )
+    paradigm = (
+        get_nested(payload, ('artifacts', 'method', 'kind'))
+        or get_nested(payload, ('run', 'benchmark_mode'))
+    )
+    modality = (
+        get_nested(payload, ('artifacts', 'dataset', 'info', 'modality'))
+        or get_nested(config_payload, ('dataset', 'modality'))
+    )
+    dataset_id = (
+        get_nested(payload, ('artifacts', 'dataset', 'id'))
+        or get_nested(config_payload, ('dataset', 'id'))
+    )
+
+    status = normalize_status(get_nested(payload, ('run', 'status')))
+    exit_code = (
+        to_int(get_nested(payload, ('run', 'exit_code')))
+        if get_nested(payload, ('run', 'exit_code')) is not None
+        else to_int(get_nested(payload, ('run', 'error_code')))
+    )
+    if exit_code is None and status == 'OK':
+        exit_code = 0
+
+    error_value = payload.get('error')
+    if isinstance(error_value, dict):
+        error_value = json.dumps(error_value, ensure_ascii=False)
+    elif error_value is not None and not isinstance(error_value, str):
+        error_value = str(error_value)
+    if status == 'FAIL' and not error_value:
+        if exit_code is not None:
+            error_value = f"status=FAIL exit_code={exit_code}"
+        else:
+            error_value = "status=FAIL"
+
+    results = {
+        'run_id': run_id,
+        'raw_data_urls': {
+            'run': f"data/artifacts/{run_id}/run.json",
+        },
+        'history': [],
+        'method_id': method_id,
+        'paradigm': paradigm,
+        'modality': modality,
+        'dataset_id': dataset_id,
+        'target_regime': infer_target_regime(run_id, payload),
+        'status': status,
+        'exit_code': exit_code,
+        'error': error_value,
+        'seed': to_int(get_nested(payload, ('run', 'seed'))),
+        'duration_s': to_float(get_nested(payload, ('run', 'duration_s'))),
+        'test_accuracy': to_float(get_nested(payload, ('metrics', 'test', 'accuracy'))),
+        'test_macro_f1': to_float(get_nested(payload, ('metrics', 'test', 'macro_f1'))),
+        'val.accuracy': to_float(get_nested(payload, ('metrics', 'val', 'accuracy'))),
+        'val.macro_f1': to_float(get_nested(payload, ('metrics', 'val', 'macro_f1'))),
+    }
+
+    config_path = artifact_dir / 'config.yaml'
+    if config_path.exists():
+        results['raw_data_urls']['config'] = f"data/artifacts/{run_id}/config.yaml"
+
+    log_path = artifact_dir / 'run.log'
+    if log_path.exists():
+        results['raw_data_urls']['log'] = f"data/artifacts/{run_id}/run.log"
+
+    method_params = (
+        get_nested(config_payload, ('method', 'params'))
+        if isinstance(config_payload, dict)
+        else None
+    )
+    if isinstance(method_params, dict):
+        for key, value in flatten_dict(method_params, parent_key='params').items():
+            results[key] = value
+
+    sampling_stats = get_nested(payload, ('artifacts', 'sampling', 'stats'))
+    if isinstance(sampling_stats, dict):
+        for key, value in flatten_dict(sampling_stats, parent_key='sampling.stats').items():
+            results[key] = value
+
+    ensure_fallback_artifacts(results, artifact_dir, run_id, source_row=None)
+    return results
+
+
 def main():
     output_dir = Path('dashboard/public/data')
     output_dir.mkdir(parents=True, exist_ok=True)
@@ -755,6 +911,20 @@ def main():
                     all_runs.append(run_data)
                     seen_run_ids.add(run_id)
 
+    if not all_runs:
+        artifacts_root = output_dir / 'artifacts'
+        if artifacts_root.exists():
+            print('No logs found; rebuilding results from artifacts.')
+            for artifact_dir in sorted(artifacts_root.iterdir()):
+                run_data = extract_run_data_from_artifact_dir(artifact_dir)
+                if not run_data:
+                    continue
+                run_id = run_data.get('run_id')
+                if not run_id or run_id in seen_run_ids:
+                    continue
+                all_runs.append(run_data)
+                seen_run_ids.add(run_id)
+
     output_file = output_dir / 'results.json'
     with open(output_file, 'w', encoding='utf-8') as f:
         json.dump(all_runs, f, indent=2)