Spaces:
Running
Running
| """Shared dataset-column contract for authoritative consent lineage.""" | |
| from __future__ import annotations | |
| import json | |
| import re | |
| import threading | |
| from typing import Any, Mapping | |
| VERIFIED_CONSENT_STATUS = "verified_v2" | |
| VERIFIED_CONSENT_PROVENANCE = "collection_v2_server_receipt" | |
| LEGACY_CONSENT_STATUS = "legacy_unverified" | |
| LEGACY_NO_RECEIPT_PROVENANCE = "legacy_no_receipt" | |
| # Shared by the collection withdrawal propagator and Trust dataset writer. | |
| IEDID_DATASET_LOCK = threading.Lock() | |
| CONSENT_LINEAGE_COLUMNS = ( | |
| "Consent_ID", | |
| "Consent_Status", | |
| "Consent_Provenance", | |
| "Consent_Confirmed", | |
| "Consent_Version", | |
| "Consent_Recorded_At", | |
| "Privacy_Notice_Version", | |
| "Dataset_License", | |
| "Permitted_Modalities", | |
| "Consent_Surface", | |
| "Consent_Locale", | |
| "Consent_Text_SHA256", | |
| "Consent_Record_SHA256", | |
| "Consent_Schema_Version", | |
| "Research_Purposes", | |
| "Retention_Policy_Version", | |
| "Withdrawal_Policy_Version", | |
| "Eligibility_Confirmed", | |
| "Public_Release_Confirmed", | |
| "Commercial_AI_Reuse_Confirmed", | |
| "Privacy_Risks_Acknowledged", | |
| "Operational_Processing_Acknowledged", | |
| "Voluntary_Participation_Confirmed", | |
| "Contribution_Rights_Confirmed", | |
| "Raw_Audio_Consent", | |
| "Data_Controller_Name", | |
| "Data_Controller_Contact", | |
| "Data_Recipient_Categories", | |
| ) | |
| DATASET_GOVERNANCE_COLUMNS = ( | |
| "Repository_License", | |
| "License_Compatibility_Status", | |
| "Release_Eligible", | |
| "Ethics_Review_Status", | |
| ) | |
| DATASET_PROVENANCE_COLUMNS = ( | |
| "Source_App_ID", | |
| "Source_App_Version", | |
| "Source_Game_ID", | |
| "Collected_At", | |
| ) | |
| CROWDSOURCING_RECORD_COLUMNS = ( | |
| "Participant_ID", | |
| "Submission_ID", | |
| "Task_ID", | |
| *DATASET_PROVENANCE_COLUMNS, | |
| ) | |
| WITHDRAWAL_AUDIT_COLUMNS = ( | |
| "Withdrawal_Request_ID", | |
| "Withdrawal_Requested_At", | |
| "Deletion_Propagation_Status", | |
| ) | |
| _CONSENT_ID_RE = re.compile(r"^cns_[0-9a-f]{32}$") | |
| _SHA256_RE = re.compile(r"^[0-9a-f]{64}$") | |
| def canonical_json_list(values: list[str]) -> str: | |
| return json.dumps( | |
| sorted(set(values)), | |
| ensure_ascii=False, | |
| separators=(",", ":"), | |
| ) | |
| def legacy_consent_lineage( | |
| provenance: str = LEGACY_NO_RECEIPT_PROVENANCE, | |
| ) -> dict[str, str]: | |
| row = {column: "" for column in CONSENT_LINEAGE_COLUMNS} | |
| row["Consent_Status"] = LEGACY_CONSENT_STATUS | |
| row["Consent_Provenance"] = provenance | |
| return row | |
| def _cell(metadata: Mapping[str, Any], key: str) -> str: | |
| value = metadata.get(key, "") | |
| if value is None: | |
| return "" | |
| return str(value).strip() | |
| def _is_true(value: Any) -> bool: | |
| return str(value).strip().casefold() in {"true", "1", "yes"} | |
| def consent_lineage_is_verified(metadata: Mapping[str, Any]) -> bool: | |
| if _cell(metadata, "Consent_Status") != VERIFIED_CONSENT_STATUS: | |
| return False | |
| if _cell(metadata, "Consent_Provenance") != VERIFIED_CONSENT_PROVENANCE: | |
| return False | |
| if not _CONSENT_ID_RE.fullmatch(_cell(metadata, "Consent_ID")): | |
| return False | |
| if not _is_true(metadata.get("Consent_Confirmed")): | |
| return False | |
| if not _is_true(metadata.get("Eligibility_Confirmed")): | |
| return False | |
| for field in ( | |
| "Public_Release_Confirmed", | |
| "Commercial_AI_Reuse_Confirmed", | |
| "Privacy_Risks_Acknowledged", | |
| "Operational_Processing_Acknowledged", | |
| "Voluntary_Participation_Confirmed", | |
| "Contribution_Rights_Confirmed", | |
| ): | |
| if not _is_true(metadata.get(field)): | |
| return False | |
| if _cell(metadata, "Raw_Audio_Consent").casefold() not in {"true", "false"}: | |
| return False | |
| if not _SHA256_RE.fullmatch(_cell(metadata, "Consent_Text_SHA256")): | |
| return False | |
| if not _SHA256_RE.fullmatch(_cell(metadata, "Consent_Record_SHA256")): | |
| return False | |
| required = ( | |
| "Consent_Version", | |
| "Consent_Recorded_At", | |
| "Privacy_Notice_Version", | |
| "Dataset_License", | |
| "Permitted_Modalities", | |
| "Consent_Surface", | |
| "Consent_Locale", | |
| "Consent_Schema_Version", | |
| "Research_Purposes", | |
| "Retention_Policy_Version", | |
| "Withdrawal_Policy_Version", | |
| "Data_Controller_Name", | |
| "Data_Controller_Contact", | |
| "Data_Recipient_Categories", | |
| ) | |
| if any(not _cell(metadata, field) for field in required): | |
| return False | |
| try: | |
| modalities = json.loads(_cell(metadata, "Permitted_Modalities")) | |
| purposes = json.loads(_cell(metadata, "Research_Purposes")) | |
| recipients = json.loads(_cell(metadata, "Data_Recipient_Categories")) | |
| except (TypeError, ValueError, json.JSONDecodeError): | |
| return False | |
| return all(isinstance(values, list) and values for values in (modalities, purposes, recipients)) | |
| def raw_audio_release_is_allowed(metadata: Mapping[str, Any]) -> bool: | |
| return consent_lineage_is_verified(metadata) and _is_true( | |
| metadata.get("Raw_Audio_Consent") | |
| ) | |
| def repository_license_status( | |
| dataset_license: str, | |
| repository_license: str, | |
| ) -> str: | |
| row_license = str(dataset_license or "").strip().casefold() | |
| repo_license = str(repository_license or "").strip().casefold() | |
| if not repo_license: | |
| return "repository_license_unconfigured" | |
| if row_license == repo_license: | |
| return "exact_match" | |
| return "mismatch_review_required" | |