Spaces:
Sleeping
Sleeping
| """Strict, release-only loader for live linguistic knowledge.""" | |
| from __future__ import annotations | |
| import csv | |
| from pathlib import Path | |
| import pandas as pd | |
| from src.consent_lineage import ( | |
| CONSENT_LINEAGE_COLUMNS, | |
| CROWDSOURCING_RECORD_COLUMNS, | |
| DATASET_GOVERNANCE_COLUMNS, | |
| DATASET_PROVENANCE_COLUMNS, | |
| consent_lineage_is_verified, | |
| ) | |
| _KNOWLEDGE_COLUMNS = ( | |
| "Utterance", "Clarification", "Dialect", "Tone_Category", | |
| "Linguistic_Context", "Pragmatic_Analysis", | |
| ) | |
| _REQUIRED_COLUMNS = set( | |
| (*_KNOWLEDGE_COLUMNS[:3], *CROWDSOURCING_RECORD_COLUMNS, | |
| *CONSENT_LINEAGE_COLUMNS, *DATASET_GOVERNANCE_COLUMNS) | |
| ) | |
| def _read_governed_csv_strict(path: Path) -> pd.DataFrame: | |
| with path.open("r", encoding="utf-8-sig", newline="") as source: | |
| reader = csv.reader(source, strict=True) | |
| try: | |
| header = next(reader) | |
| except StopIteration as exc: | |
| raise ValueError("empty CSV") from exc | |
| if any(not str(column).strip() for column in header): | |
| raise ValueError("blank CSV header") | |
| if len(header) != len(set(header)): | |
| raise ValueError("duplicate CSV headers") | |
| missing = sorted(_REQUIRED_COLUMNS - set(header)) | |
| if missing: | |
| raise ValueError("missing governance columns: " + ", ".join(missing)) | |
| for row_number, row in enumerate(reader, start=2): | |
| if len(row) != len(header): | |
| raise ValueError(f"row {row_number} has a width mismatch") | |
| frame = pd.read_csv(path, dtype=str, encoding="utf-8-sig", keep_default_na=False) | |
| if list(frame.columns) != header: | |
| raise ValueError("CSV schema changed while reading") | |
| return frame | |
| def load_verified_release_knowledge(dataset_dir) -> pd.DataFrame: | |
| root = Path(dataset_dir).expanduser().resolve() | |
| released = [] | |
| for path in sorted(root.glob("*/data/*.csv")): | |
| try: | |
| frame = _read_governed_csv_strict(path) | |
| except Exception as exc: | |
| print( | |
| "🔒 Skipped non-governed knowledge file " | |
| f"({type(exc).__name__}); path and details redacted." | |
| ) | |
| continue | |
| mask = frame.apply(lambda row: consent_lineage_is_verified(row.to_dict()), axis=1) | |
| mask &= frame["Release_Eligible"].str.strip().str.casefold().eq("true") | |
| mask &= frame["Ethics_Review_Status"].str.strip().str.casefold().eq("approved_verified_consent") | |
| mask &= frame["License_Compatibility_Status"].str.strip().str.casefold().eq("exact_match") | |
| mask &= frame[list(DATASET_PROVENANCE_COLUMNS)].apply( | |
| lambda row: all(str(value).strip() for value in row), axis=1 | |
| ) | |
| eligible = frame.loc[mask].copy() | |
| if not eligible.empty: | |
| released.append(eligible) | |
| if not released: | |
| return pd.DataFrame(columns=list(_KNOWLEDGE_COLUMNS)) | |
| return pd.concat(released, ignore_index=True) | |