diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..43b5e0c --- /dev/null +++ b/.gitignore @@ -0,0 +1,216 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +.python-version + +# pipenv +Pipfile.lock + +# poetry +poetry.lock + +# pdm +.pdm.toml +.pdm-python +.pdm-build/ + +# PEP 582 +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +.idea/ +*.iml +*.iws +.idea_modules/ + +# VS Code +.vscode/ +*.code-workspace + +# Sublime Text +*.sublime-project +*.sublime-workspace + +# macOS +.DS_Store +.AppleDouble +.LSOverride + +# Windows +Thumbs.db +Thumbs.db:encryptable +ehthumbs.db +ehthumbs_vista.db +*.stackdump +[Dd]esktop.ini +$RECYCLE.BIN/ +*.lnk + +# Linux +*~ +.fuse_hidden* +.directory +.Trash-* +.nfs* + +# Project-specific +*.log +*.csv +*.xlsx +*.docx +*.pdf +demo.yml +test_data/ +uploads/ +downloads/ +data/ +exemple/data/ +temp/ +tmp/ + +# API keys and secrets (safety net) +*apikey* +*api_key* +*secret* +*password* +*.pem +*.key +*.crt + +# Dataverse uploader specific +DVUploaderLog_* +refresh.txt + +# Backup files +*.bak +*.swp +*.swo +*~ + +# Archive files (unless intentionally tracked) +*.zip +*.tar.gz +*.tar.bz2 +*.tar.xz +*.7z +*.rar \ No newline at end of file diff --git a/Csv_to_json - Citation.csv b/Csv_to_json - Citation.csv index e3ad0e4..78a74ef 100644 --- a/Csv_to_json - Citation.csv +++ b/Csv_to_json - Citation.csv @@ -1,2 +1,2 @@ doi,title,subtitle,alternativeTitle,otherId: otherIdAgency,author: authorName; authorAffiliation,datasetContact: datasetContactName; datasetContactAffiliation,dsDescription: dsDescriptionValue; dsDescriptionDate,subject,keyword: keywordValue; keywordVocabulary,topicClassification: topicClassValue; topicClassVocab,publication: publicationRelationType; publicationCitation; publicationIDType; publicationIDNumber,notesText,language,producer: producerName; producerAffiliation; producerAbbreviation,productionDate,productionPlace,contributor: contributorType; contributorName,grantNumber: grantNumberAgency; grantNumberValue,distributor: distributorName; distributorAffiliation; distributorAbbreviation; distributorURL,distributionDate,depositor,dateOfDeposit,timePeriodCovered: timePeriodCoveredStart; timePeriodCoveredEnd,dateOfCollection: dateOfCollectionStart; dateOfCollectionEnd,kindOfData,series: seriesName; seriesInformation,software: softwareName,relatedMaterial,relatedDatasets,otherReferences,dataSources,originOfSources,characteristicOfSources,accessToSources,citation -,Test: csv to json file,conversion,alternativeTitle,Agency,zeynep Cevik; University of Toronto,University of toronto,2025,Social Sciences,csv: json: test,test,test1,notest1,English,statscan,2025,Canada,zeynep cevik,,,2025,Zeynep Cevik,2025,,,micro,Json,,,,,,,,, +,Test: csv to json file,conversion,alternativeTitle,Agency,zeynep Cevik; University of Toronto,University of toronto,2025,Social Sciences,csv: json: test,test,test1,"L'enquête a révélé que plusieurs ménages n'avaient pas reçu le courriel initial à cause d'une défaillance technique.",English,statscan,2025,Canada,zeynep cevik,,,2025,Zeynep Cevik,2025,,,micro,Json,,,,,,,,, diff --git a/TEST_README.md b/TEST_README.md new file mode 100644 index 0000000..e782c4e --- /dev/null +++ b/TEST_README.md @@ -0,0 +1,169 @@ +# Test Documentation + +This document describes the test suite for the CSV to Dataverse JSON converter. + +## Running Tests + +### Prerequisites + +Install the required test dependencies: + +```bash +pip install --break-system-packages pytest pandas +``` + +Or install all dev dependencies: + +```bash +pip install --break-system-packages -e ".[dev]" +``` + +### Basic Test Commands + +**Run all tests:** +```bash +python -m pytest test_csv_to_dataverse_json.py +``` + +**Run with verbose output:** +```bash +python -m pytest test_csv_to_dataverse_json.py -v +``` + +**Run with coverage report:** +```bash +python -m pytest test_csv_to_dataverse_json.py --cov=csv_to_dataverse_json +``` + +**Run a specific test class:** +```bash +python -m pytest test_csv_to_dataverse_json.py::TestUTF8Encoding -v +``` + +**Run a specific test:** +```bash +python -m pytest test_csv_to_dataverse_json.py::TestIntegration::test_full_conversion_with_all_fields -v +``` + +## Test Structure + +The test suite is organized into the following test classes: + +### TestUTF8Encoding +Tests UTF-8 character encoding handling. + +- `test_utf8_french_text_in_csv` - Verifies French accented characters are preserved +- `test_utf8_special_characters` - Tests various special characters (Japanese, Chinese, Arabic, etc.) + +### TestParseCompound +Tests the `parse_compound()` function for parsing compound field values. + +- `test_parse_author_single` - Single author with name and affiliation +- `test_parse_author_multiple` - Multiple authors separated by pipe +- `test_parse_contact_with_email` - Contact with name, affiliation, and email +- `test_parse_empty_subfields` - Handling of empty subfield values + +### TestEnsureRequiredFields +Tests the `ensure_required_fields()` function for adding missing required fields. + +- `test_adds_default_author_when_missing` - Adds default author when none provided +- `test_uses_depositor_as_fallback_author` - Uses depositor field as fallback +- `test_adds_default_contact_email_when_missing` - Adds default contact email +- `test_adds_default_description_when_missing` - Adds default description + +### TestIntegration +Integration tests for the complete conversion process. + +- `test_full_conversion_with_all_fields` - Tests conversion with various field types including custom subfield headers +- `test_conversion_with_defaults` - Tests conversion with default values +- `test_multiple_rows_conversion` - Tests converting CSV with multiple rows + +### TestGeospatialBlock +Tests the `create_geospatial_block()` function. + +- `test_create_geospatial_block_with_coverage` - Creates geospatial block with geographic coverage +- `test_create_geospatial_block_returns_none_when_empty` - Returns None when no geospatial fields present + +### TestSocialscienceBlock +Tests the `create_socialscience_block()` function. + +- `test_create_socialscience_block_with_unit` - Creates social science block with unit of analysis + +## Test Features + +### Custom Subfield Headers +The converter supports CSV headers with custom subfield specifications: + +```csv +"author: authorName; authorAffiliation","keyword: keywordValue; keywordVocabulary" +``` + +This allows you to specify which subfields to use for compound fields directly in the CSV header. + +### Temporary Files +Tests use Python's `tempfile` module to create temporary CSV and JSON files, which are automatically cleaned up after each test. + +### UTF-8 Encoding +All tests verify that UTF-8 encoding is properly preserved throughout the conversion process. + +## Expected Test Results + +When all tests pass, you should see: + +``` +============================= test session starts ============================== +... +test_csv_to_dataverse_json.py::TestUTF8Encoding::test_utf8_french_text_in_csv PASSED +test_csv_to_dataverse_json.py::TestUTF8Encoding::test_utf8_special_characters PASSED +... +============================== 16 passed in X.XXs ============================== +``` + +## Troubleshooting + +**Import errors:** +Make sure `csv_to_dataverse_json.py` is in the same directory as the test file, or install the package in development mode: +```bash +pip install --break-system-packages -e . +``` + +**Missing dependencies:** +Install pytest and pandas: +```bash +pip install --break-system-packages pytest pandas +``` + +**Permission errors:** +If you get permission errors, use the `--break-system-packages` flag or create a virtual environment: +```bash +python3 -m venv venv +source venv/bin/activate +pip install pytest pandas +``` + +## Writing New Tests + +To add new tests: + +1. Create a new test class or add to an existing one +2. Test method names must start with `test_` +3. Use descriptive docstrings to explain what the test does +4. Clean up any temporary files in a `finally` block +5. Use assertions to verify expected behavior + +Example: +```python +class TestMyFeature: + """Test description""" + + def test_my_new_feature(self): + """Test that my feature works correctly""" + # Arrange + input_data = "test" + + # Act + result = my_function(input_data) + + # Assert + assert result == "expected_output" +``` diff --git a/__pycache__/csv_to_dataverse_json.cpython-312.pyc b/__pycache__/csv_to_dataverse_json.cpython-312.pyc deleted file mode 100644 index e031203..0000000 Binary files a/__pycache__/csv_to_dataverse_json.cpython-312.pyc and /dev/null differ diff --git a/csv_to_dataverse_json.py b/csv_to_dataverse_json.py index 30a4198..65ba616 100644 --- a/csv_to_dataverse_json.py +++ b/csv_to_dataverse_json.py @@ -18,30 +18,38 @@ def ensure_required_fields(dataset_json, row, defaults=None): """Ensure Dataverse-required fields exist; fill with placeholders if missing.""" try: - citation_fields = dataset_json["datasetVersion"]["metadataBlocks"]["citation"]["fields"] + citation_fields = dataset_json["datasetVersion"]["metadataBlocks"]["citation"][ + "fields" + ] except Exception: return def find_field(name): for f in citation_fields: - if f.get('typeName') == name: + if f.get("typeName") == name: return f return None # 1) Author (authorName required) - author_field = find_field('author') - if not author_field or not author_field.get('value'): + author_field = find_field("author") + if not author_field or not author_field.get("value"): # try pull from depositor or CSV 'author' raw string raw_author = None - if 'author' in row and not pd.isna(row['author']): - raw_author = str(row['author']).split(';')[0].split('|')[0].strip() - elif 'depositor' in row and not pd.isna(row['depositor']): - raw_author = str(row['depositor']).strip() + if "author" in row and not pd.isna(row["author"]): + raw_author = str(row["author"]).split(";")[0].split("|")[0].strip() + elif "depositor" in row and not pd.isna(row["depositor"]): + raw_author = str(row["depositor"]).strip() else: # fallback to provided default author, environment, or generic - raw_author = (defaults.get('author') if defaults and defaults.get('author') else - os.getenv('DATAVERSE_DEFAULT_AUTHOR') if os.getenv('DATAVERSE_DEFAULT_AUTHOR') else - 'Unknown Author') + raw_author = ( + defaults.get("author") + if defaults and defaults.get("author") + else ( + os.getenv("DATAVERSE_DEFAULT_AUTHOR") + if os.getenv("DATAVERSE_DEFAULT_AUTHOR") + else "Unknown Author" + ) + ) new_author = { "typeName": "author", @@ -49,9 +57,14 @@ def find_field(name): "typeClass": "compound", "value": [ { - "authorName": {"typeName": "authorName", "multiple": False, "typeClass": "primitive", "value": raw_author} + "authorName": { + "typeName": "authorName", + "multiple": False, + "typeClass": "primitive", + "value": raw_author, + } } - ] + ], } # replace or append if author_field: @@ -59,29 +72,51 @@ def find_field(name): citation_fields.append(new_author) # 2) Dataset contact email (datasetContact -> datasetContactEmail required) - contact_field = find_field('datasetContact') - if not contact_field or not contact_field.get('value') or not any('datasetContactEmail' in v for v in (contact_field.get('value') or [])): + contact_field = find_field("datasetContact") + if ( + not contact_field + or not contact_field.get("value") + or not any( + "datasetContactEmail" in v for v in (contact_field.get("value") or []) + ) + ): # try CSV values contact_name = None contact_email = None - if 'datasetContact' in row and not pd.isna(row['datasetContact']): - parts = [p.strip() for p in str(row['datasetContact']).split(';')] + if "datasetContact" in row and not pd.isna(row["datasetContact"]): + parts = [p.strip() for p in str(row["datasetContact"]).split(";")] if len(parts) >= 1: contact_name = parts[0] if len(parts) >= 3: contact_email = parts[2] - if not contact_name and 'depositor' in row and not pd.isna(row['depositor']): - contact_name = str(row['depositor']).strip() - if not contact_email and 'datasetContactEmail' in row and not pd.isna(row['datasetContactEmail']): - contact_email = str(row['datasetContactEmail']).strip() + if not contact_name and "depositor" in row and not pd.isna(row["depositor"]): + contact_name = str(row["depositor"]).strip() + if ( + not contact_email + and "datasetContactEmail" in row + and not pd.isna(row["datasetContactEmail"]) + ): + contact_email = str(row["datasetContactEmail"]).strip() if not contact_email: - contact_email = (defaults.get('email') if defaults and defaults.get('email') else - os.getenv('DATAVERSE_DEFAULT_EMAIL') if os.getenv('DATAVERSE_DEFAULT_EMAIL') else - 'no-reply@example.com') + contact_email = ( + defaults.get("email") + if defaults and defaults.get("email") + else ( + os.getenv("DATAVERSE_DEFAULT_EMAIL") + if os.getenv("DATAVERSE_DEFAULT_EMAIL") + else "no-reply@example.com" + ) + ) if not contact_name: - contact_name = (defaults.get('author') if defaults and defaults.get('author') else - os.getenv('DATAVERSE_DEFAULT_AUTHOR') if os.getenv('DATAVERSE_DEFAULT_AUTHOR') else - 'Dataset Contact') + contact_name = ( + defaults.get("author") + if defaults and defaults.get("author") + else ( + os.getenv("DATAVERSE_DEFAULT_AUTHOR") + if os.getenv("DATAVERSE_DEFAULT_AUTHOR") + else "Dataset Contact" + ) + ) new_contact = { "typeName": "datasetContact", @@ -89,28 +124,53 @@ def find_field(name): "typeClass": "compound", "value": [ { - "datasetContactName": {"typeName": "datasetContactName", "multiple": False, "typeClass": "primitive", "value": contact_name}, - "datasetContactAffiliation": {"typeName": "datasetContactAffiliation", "multiple": False, "typeClass": "primitive", "value": str(row.get('datasetContactAffiliation','')).strip()}, - "datasetContactEmail": {"typeName": "datasetContactEmail", "multiple": False, "typeClass": "primitive", "value": contact_email} + "datasetContactName": { + "typeName": "datasetContactName", + "multiple": False, + "typeClass": "primitive", + "value": contact_name, + }, + "datasetContactAffiliation": { + "typeName": "datasetContactAffiliation", + "multiple": False, + "typeClass": "primitive", + "value": str(row.get("datasetContactAffiliation", "")).strip(), + }, + "datasetContactEmail": { + "typeName": "datasetContactEmail", + "multiple": False, + "typeClass": "primitive", + "value": contact_email, + }, } - ] + ], } if contact_field: citation_fields.remove(contact_field) citation_fields.append(new_contact) # 3) Description (dsDescription -> dsDescriptionValue required) - desc_field = find_field('dsDescription') - if not desc_field or not desc_field.get('value') or not any('dsDescriptionValue' in v for v in (desc_field.get('value') or [])): + desc_field = find_field("dsDescription") + if ( + not desc_field + or not desc_field.get("value") + or not any("dsDescriptionValue" in v for v in (desc_field.get("value") or [])) + ): desc_text = None - if 'dsDescription' in row and not pd.isna(row['dsDescription']): - desc_text = str(row['dsDescription']).split(';')[0].strip() - elif 'citation' in row and not pd.isna(row['citation']): - desc_text = str(row['citation']).strip() + if "dsDescription" in row and not pd.isna(row["dsDescription"]): + desc_text = str(row["dsDescription"]).split(";")[0].strip() + elif "citation" in row and not pd.isna(row["citation"]): + desc_text = str(row["citation"]).strip() else: - desc_text = (defaults.get('description') if defaults and defaults.get('description') else - os.getenv('DATAVERSE_DEFAULT_DESCRIPTION') if os.getenv('DATAVERSE_DEFAULT_DESCRIPTION') else - 'No description provided.') + desc_text = ( + defaults.get("description") + if defaults and defaults.get("description") + else ( + os.getenv("DATAVERSE_DEFAULT_DESCRIPTION") + if os.getenv("DATAVERSE_DEFAULT_DESCRIPTION") + else "No description provided." + ) + ) new_desc = { "typeName": "dsDescription", @@ -118,9 +178,14 @@ def find_field(name): "typeClass": "compound", "value": [ { - "dsDescriptionValue": {"typeName": "dsDescriptionValue", "multiple": False, "typeClass": "primitive", "value": desc_text} + "dsDescriptionValue": { + "typeName": "dsDescriptionValue", + "multiple": False, + "typeClass": "primitive", + "value": desc_text, + } } - ] + ], } if desc_field: citation_fields.remove(desc_field) @@ -137,59 +202,198 @@ def csv_to_dataverse_json(csv_file_path, output_json_path, defaults=None): # Field type directory - defines structure for all citation fields directory = { - 'title': {"typeName": "title", "multiple": False, "typeClass": "primitive"}, - 'subtitle': {"typeName": "subtitle", "multiple": False, "typeClass": "primitive"}, - 'alternativeTitle': {"typeName": "alternativeTitle", "multiple": True, "typeClass": "primitive"}, - 'otherId': {"typeName": "otherId", "multiple": True, "typeClass": "compound"}, - 'author': {"typeName": "author", "multiple": True, "typeClass": "compound"}, - 'datasetContact': {"typeName": "datasetContact", "multiple": True, "typeClass": "compound"}, - 'dsDescription': {"typeName": "dsDescription", "multiple": True, "typeClass": "compound"}, - 'subject': {"typeName": "subject", "multiple": True, "typeClass": "controlledVocabulary"}, - 'keyword': {"typeName": "keyword", "multiple": True, "typeClass": "compound"}, - 'topicClassification': {"typeName": "topicClassification", "multiple": True, "typeClass": "compound"}, - 'publication': {"typeName": "publication", "multiple": True, "typeClass": "compound"}, - 'notesText': {"typeName": "notesText", "multiple": False, "typeClass": "primitive"}, - 'language': {"typeName": "language", "multiple": True, "typeClass": "controlledVocabulary"}, - 'producer': {"typeName": "producer", "multiple": True, "typeClass": "compound"}, - 'productionDate': {"typeName": "productionDate", "multiple": False, "typeClass": "primitive"}, - 'productionPlace': {"typeName": "productionPlace", "multiple": True, "typeClass": "primitive"}, - 'contributor': {"typeName": "contributor", "multiple": True, "typeClass": "compound"}, - 'grantNumber': {"typeName": "grantNumber", "multiple": True, "typeClass": "compound"}, - 'distributor': {"typeName": "distributor", "multiple": True, "typeClass": "compound"}, - 'distributionDate': {"typeName": "distributionDate", "multiple": False, "typeClass": "primitive"}, - 'depositor': {"typeName": "depositor", "multiple": False, "typeClass": "primitive"}, - 'dateOfDeposit': {"typeName": "dateOfDeposit", "multiple": False, "typeClass": "primitive"}, - 'timePeriodCovered': {"typeName": "timePeriodCovered", "multiple": True, "typeClass": "compound"}, - 'dateOfCollection': {"typeName": "dateOfCollection", "multiple": True, "typeClass": "compound"}, - 'kindOfData': {"typeName": "kindOfData", "multiple": True, "typeClass": "primitive"}, - 'series': {"typeName": "series", "multiple": True, "typeClass": "compound"}, - 'software': {"typeName": "software", "multiple": True, "typeClass": "compound"}, - 'relatedMaterial': {"typeName": "relatedMaterial", "multiple": True, "typeClass": "primitive"}, - 'relatedDatasets': {"typeName": "relatedDatasets", "multiple": True, "typeClass": "primitive"}, - 'otherReferences': {"typeName": "otherReferences", "multiple": True, "typeClass": "primitive"}, - 'dataSources': {"typeName": "dataSources", "multiple": True, "typeClass": "primitive"}, - 'originOfSources': {"typeName": "originOfSources", "multiple": False, "typeClass": "primitive"}, - 'characteristicOfSources': {"typeName": "characteristicOfSources", "multiple": False, "typeClass": "primitive"}, - 'accessToSources': {"typeName": "accessToSources", "multiple": False, "typeClass": "primitive"} + "title": {"typeName": "title", "multiple": False, "typeClass": "primitive"}, + "subtitle": { + "typeName": "subtitle", + "multiple": False, + "typeClass": "primitive", + }, + "alternativeTitle": { + "typeName": "alternativeTitle", + "multiple": True, + "typeClass": "primitive", + }, + "otherId": {"typeName": "otherId", "multiple": True, "typeClass": "compound"}, + "author": {"typeName": "author", "multiple": True, "typeClass": "compound"}, + "datasetContact": { + "typeName": "datasetContact", + "multiple": True, + "typeClass": "compound", + }, + "dsDescription": { + "typeName": "dsDescription", + "multiple": True, + "typeClass": "compound", + }, + "subject": { + "typeName": "subject", + "multiple": True, + "typeClass": "controlledVocabulary", + }, + "keyword": {"typeName": "keyword", "multiple": True, "typeClass": "compound"}, + "topicClassification": { + "typeName": "topicClassification", + "multiple": True, + "typeClass": "compound", + }, + "publication": { + "typeName": "publication", + "multiple": True, + "typeClass": "compound", + }, + "notesText": { + "typeName": "notesText", + "multiple": False, + "typeClass": "primitive", + }, + "language": { + "typeName": "language", + "multiple": True, + "typeClass": "controlledVocabulary", + }, + "producer": {"typeName": "producer", "multiple": True, "typeClass": "compound"}, + "productionDate": { + "typeName": "productionDate", + "multiple": False, + "typeClass": "primitive", + }, + "productionPlace": { + "typeName": "productionPlace", + "multiple": True, + "typeClass": "primitive", + }, + "contributor": { + "typeName": "contributor", + "multiple": True, + "typeClass": "compound", + }, + "grantNumber": { + "typeName": "grantNumber", + "multiple": True, + "typeClass": "compound", + }, + "distributor": { + "typeName": "distributor", + "multiple": True, + "typeClass": "compound", + }, + "distributionDate": { + "typeName": "distributionDate", + "multiple": False, + "typeClass": "primitive", + }, + "depositor": { + "typeName": "depositor", + "multiple": False, + "typeClass": "primitive", + }, + "dateOfDeposit": { + "typeName": "dateOfDeposit", + "multiple": False, + "typeClass": "primitive", + }, + "timePeriodCovered": { + "typeName": "timePeriodCovered", + "multiple": True, + "typeClass": "compound", + }, + "dateOfCollection": { + "typeName": "dateOfCollection", + "multiple": True, + "typeClass": "compound", + }, + "kindOfData": { + "typeName": "kindOfData", + "multiple": True, + "typeClass": "primitive", + }, + "series": {"typeName": "series", "multiple": True, "typeClass": "compound"}, + "software": {"typeName": "software", "multiple": True, "typeClass": "compound"}, + "relatedMaterial": { + "typeName": "relatedMaterial", + "multiple": True, + "typeClass": "primitive", + }, + "relatedDatasets": { + "typeName": "relatedDatasets", + "multiple": True, + "typeClass": "primitive", + }, + "otherReferences": { + "typeName": "otherReferences", + "multiple": True, + "typeClass": "primitive", + }, + "dataSources": { + "typeName": "dataSources", + "multiple": True, + "typeClass": "primitive", + }, + "originOfSources": { + "typeName": "originOfSources", + "multiple": False, + "typeClass": "primitive", + }, + "characteristicOfSources": { + "typeName": "characteristicOfSources", + "multiple": False, + "typeClass": "primitive", + }, + "accessToSources": { + "typeName": "accessToSources", + "multiple": False, + "typeClass": "primitive", + }, } # Compound field subfield mappings compound_fields = { - 'otherId': ['otherIdAgency', 'otherIdValue'], - 'author': ['authorName', 'authorAffiliation', 'authorIdentifierScheme', 'authorIdentifier'], - 'datasetContact': ['datasetContactName', 'datasetContactAffiliation', 'datasetContactEmail'], - 'dsDescription': ['dsDescriptionValue', 'dsDescriptionDate'], - 'keyword': ['keywordValue', 'keywordVocabulary', 'keywordVocabularyURI'], - 'topicClassification': ['topicClassValue', 'topicClassVocab', 'topicClassVocabURI'], - 'publication': ['publicationRelationType', 'publicationCitation', 'publicationIDType', 'publicationIDNumber', 'publicationURL'], - 'producer': ['producerName', 'producerAffiliation', 'producerAbbreviation', 'producerURL', 'producerLogoURL'], - 'contributor': ['contributorType', 'contributorName'], - 'grantNumber': ['grantNumberAgency', 'grantNumberValue'], - 'distributor': ['distributorName', 'distributorAffiliation', 'distributorAbbreviation', 'distributorURL', 'distributorLogoURL'], - 'timePeriodCovered': ['timePeriodCoveredStart', 'timePeriodCoveredEnd'], - 'dateOfCollection': ['dateOfCollectionStart', 'dateOfCollectionEnd'], - 'series': ['seriesName', 'seriesInformation'], - 'software': ['softwareName', 'softwareVersion'] + "otherId": ["otherIdAgency", "otherIdValue"], + "author": [ + "authorName", + "authorAffiliation", + "authorIdentifierScheme", + "authorIdentifier", + ], + "datasetContact": [ + "datasetContactName", + "datasetContactAffiliation", + "datasetContactEmail", + ], + "dsDescription": ["dsDescriptionValue", "dsDescriptionDate"], + "keyword": ["keywordValue", "keywordVocabulary", "keywordVocabularyURI"], + "topicClassification": [ + "topicClassValue", + "topicClassVocab", + "topicClassVocabURI", + ], + "publication": [ + "publicationRelationType", + "publicationCitation", + "publicationIDType", + "publicationIDNumber", + "publicationURL", + ], + "producer": [ + "producerName", + "producerAffiliation", + "producerAbbreviation", + "producerURL", + "producerLogoURL", + ], + "contributor": ["contributorType", "contributorName"], + "grantNumber": ["grantNumberAgency", "grantNumberValue"], + "distributor": [ + "distributorName", + "distributorAffiliation", + "distributorAbbreviation", + "distributorURL", + "distributorLogoURL", + ], + "timePeriodCovered": ["timePeriodCoveredStart", "timePeriodCoveredEnd"], + "dateOfCollection": ["dateOfCollectionStart", "dateOfCollectionEnd"], + "series": ["seriesName", "seriesInformation"], + "software": ["softwareName", "softwareVersion"], } # Get current date and time for defaults @@ -206,7 +410,7 @@ def format_date_to_year(date_value): date_str = str(date_value).strip() # Extract year using regex - year_match = re.search(r'\b(19|20)\d{2}\b', date_str) + year_match = re.search(r"\b(19|20)\d{2}\b", date_str) if year_match: return year_match.group(0) @@ -214,153 +418,373 @@ def format_date_to_year(date_value): # Read CSV file and process each row df = pd.read_csv(csv_file_path) + + # Parse column headers to handle "field: subfield1; subfield2" format + column_mapping = {} # Maps actual column name to (base_field_name, custom_subfields) + for col in df.columns: + if ': ' in col: + # Extract base field name and custom subfields + base_field, subfields_str = col.split(': ', 1) + base_field = base_field.strip() + custom_subfields = [s.strip() for s in subfields_str.split(';') if s.strip()] + column_mapping[col] = (base_field, custom_subfields) + else: + column_mapping[col] = (col.strip(), None) + + # Create a temporary compound_fields_override for this CSV + compound_fields_override = compound_fields.copy() + for col, (base_field, custom_subfields) in column_mapping.items(): + if custom_subfields and base_field in directory: + compound_fields_override[base_field] = custom_subfields + all_datasets = [] for idx, row in df.iterrows(): # Generate unique IDs if not provided - dataset_id = int(row.get('id', 0)) if row.get('id') and not pd.isna(row.get('id')) else 1000 + idx - version_id = int(row.get('versionId', 0)) if row.get('versionId') and not pd.isna(row.get('versionId')) else 2000 + idx - + dataset_id = ( + int(row.get("id", 0)) + if row.get("id") and not pd.isna(row.get("id")) + else 1000 + idx + ) + version_id = ( + int(row.get("versionId", 0)) + if row.get("versionId") and not pd.isna(row.get("versionId")) + else 2000 + idx + ) + # Generate identifier/DOI if not provided - if row.get('identifier') and not pd.isna(row.get('identifier')): - identifier = str(row.get('identifier')) + if row.get("identifier") and not pd.isna(row.get("identifier")): + identifier = str(row.get("identifier")) else: identifier = f"FK2/{uuid.uuid4().hex[:8].upper()}" # Get protocol and authority - protocol = str(row.get('protocol', 'doi')).strip() if row.get('protocol') and not pd.isna(row.get('protocol')) else 'doi' - authority = str(row.get('authority', '10.70122')).strip() if row.get('authority') and not pd.isna(row.get('authority')) else '10.70122' - + protocol = ( + str(row.get("protocol", "doi")).strip() + if row.get("protocol") and not pd.isna(row.get("protocol")) + else "doi" + ) + authority = ( + str(row.get("authority", "10.70122")).strip() + if row.get("authority") and not pd.isna(row.get("authority")) + else "10.70122" + ) + # Build complete JSON structure with all top-level fields dataset_json = { "id": dataset_id, "identifier": identifier, - "persistentUrl": f"https://doi.org/{authority}/{identifier}" if protocol == "doi" else f"hdl:{authority}/{identifier}", + "persistentUrl": ( + f"https://doi.org/{authority}/{identifier}" + if protocol == "doi" + else f"hdl:{authority}/{identifier}" + ), "protocol": protocol, "authority": authority, "separator": "/", - "publisher": str(row.get('publisher', 'Dataverse')).strip() if row.get('publisher') and not pd.isna(row.get('publisher')) else 'Dataverse', - "publicationDate": str(row.get('publicationDate', current_date)).strip() if row.get('publicationDate') and not pd.isna(row.get('publicationDate')) else current_date, - "storageIdentifier": f"s3://{authority}/{identifier}" if row.get('storageIdentifier') is None or pd.isna(row.get('storageIdentifier')) else str(row.get('storageIdentifier')), - "datasetType": str(row.get('datasetType', 'dataset')).strip() if row.get('datasetType') and not pd.isna(row.get('datasetType')) else 'dataset', + "publisher": ( + str(row.get("publisher", "Dataverse")).strip() + if row.get("publisher") and not pd.isna(row.get("publisher")) + else "Dataverse" + ), + "publicationDate": ( + str(row.get("publicationDate", current_date)).strip() + if row.get("publicationDate") + and not pd.isna(row.get("publicationDate")) + else current_date + ), + "storageIdentifier": ( + f"s3://{authority}/{identifier}" + if row.get("storageIdentifier") is None + or pd.isna(row.get("storageIdentifier")) + else str(row.get("storageIdentifier")) + ), + "datasetType": ( + str(row.get("datasetType", "dataset")).strip() + if row.get("datasetType") and not pd.isna(row.get("datasetType")) + else "dataset" + ), "datasetVersion": { "id": version_id, "datasetId": dataset_id, "datasetPersistentId": f"{protocol}:{authority}/{identifier}", - "datasetType": str(row.get('datasetType', 'dataset')).strip() if row.get('datasetType') and not pd.isna(row.get('datasetType')) else 'dataset', - "storageIdentifier": f"s3://{authority}:{uuid.uuid4().hex[:12]}-{uuid.uuid4().hex[:12]}" if row.get('storageIdentifier') is None or pd.isna(row.get('storageIdentifier')) else str(row.get('storageIdentifier')), - "versionNumber": int(row.get('versionNumber', 1)) if row.get('versionNumber') and not pd.isna(row.get('versionNumber')) else 1, - "internalVersionNumber": int(row.get('internalVersionNumber', 1)) if row.get('internalVersionNumber') and not pd.isna(row.get('internalVersionNumber')) else 1, - "versionMinorNumber": int(row.get('versionMinorNumber', 0)) if row.get('versionMinorNumber') and not pd.isna(row.get('versionMinorNumber')) else 0, - "versionState": str(row.get('versionState', 'DRAFT')).strip() if row.get('versionState') and not pd.isna(row.get('versionState')) else 'DRAFT', - "latestVersionPublishingState": str(row.get('latestVersionPublishingState', 'DRAFT')).strip() if row.get('latestVersionPublishingState') and not pd.isna(row.get('latestVersionPublishingState')) else 'DRAFT', - "UNF": str(row.get('UNF', '')).strip() if row.get('UNF') and not pd.isna(row.get('UNF')) else '', - "lastUpdateTime": str(row.get('lastUpdateTime', current_datetime)).strip() if row.get('lastUpdateTime') and not pd.isna(row.get('lastUpdateTime')) else current_datetime, - "releaseTime": str(row.get('releaseTime', '')).strip() if row.get('releaseTime') and not pd.isna(row.get('releaseTime')) else '', - "createTime": str(row.get('createTime', current_datetime)).strip() if row.get('createTime') and not pd.isna(row.get('createTime')) else current_datetime, - "publicationDate": str(row.get('publicationDate', current_date)).strip() if row.get('publicationDate') and not pd.isna(row.get('publicationDate')) else current_date, - "citationDate": str(row.get('citationDate', current_date)).strip() if row.get('citationDate') and not pd.isna(row.get('citationDate')) else current_date, - "termsOfUse": str(row.get('termsOfUse', '')).strip() if row.get('termsOfUse') and not pd.isna(row.get('termsOfUse')) else '', - "citationRequirements": str(row.get('citationRequirements', '')).strip() if row.get('citationRequirements') and not pd.isna(row.get('citationRequirements')) else '', - "conditions": str(row.get('conditions', '')).strip() if row.get('conditions') and not pd.isna(row.get('conditions')) else '', - "termsOfAccess": str(row.get('termsOfAccess', '')).strip() if row.get('termsOfAccess') and not pd.isna(row.get('termsOfAccess')) else '', + "datasetType": ( + str(row.get("datasetType", "dataset")).strip() + if row.get("datasetType") and not pd.isna(row.get("datasetType")) + else "dataset" + ), + "storageIdentifier": ( + f"s3://{authority}:{uuid.uuid4().hex[:12]}-{uuid.uuid4().hex[:12]}" + if row.get("storageIdentifier") is None + or pd.isna(row.get("storageIdentifier")) + else str(row.get("storageIdentifier")) + ), + "versionNumber": ( + int(row.get("versionNumber", 1)) + if row.get("versionNumber") + and not pd.isna(row.get("versionNumber")) + else 1 + ), + "internalVersionNumber": ( + int(row.get("internalVersionNumber", 1)) + if row.get("internalVersionNumber") + and not pd.isna(row.get("internalVersionNumber")) + else 1 + ), + "versionMinorNumber": ( + int(row.get("versionMinorNumber", 0)) + if row.get("versionMinorNumber") + and not pd.isna(row.get("versionMinorNumber")) + else 0 + ), + "versionState": ( + str(row.get("versionState", "DRAFT")).strip() + if row.get("versionState") and not pd.isna(row.get("versionState")) + else "DRAFT" + ), + "latestVersionPublishingState": ( + str(row.get("latestVersionPublishingState", "DRAFT")).strip() + if row.get("latestVersionPublishingState") + and not pd.isna(row.get("latestVersionPublishingState")) + else "DRAFT" + ), + "UNF": ( + str(row.get("UNF", "")).strip() + if row.get("UNF") and not pd.isna(row.get("UNF")) + else "" + ), + "lastUpdateTime": ( + str(row.get("lastUpdateTime", current_datetime)).strip() + if row.get("lastUpdateTime") + and not pd.isna(row.get("lastUpdateTime")) + else current_datetime + ), + "releaseTime": ( + str(row.get("releaseTime", "")).strip() + if row.get("releaseTime") and not pd.isna(row.get("releaseTime")) + else "" + ), + "createTime": ( + str(row.get("createTime", current_datetime)).strip() + if row.get("createTime") and not pd.isna(row.get("createTime")) + else current_datetime + ), + "publicationDate": ( + str(row.get("publicationDate", current_date)).strip() + if row.get("publicationDate") + and not pd.isna(row.get("publicationDate")) + else current_date + ), + "citationDate": ( + str(row.get("citationDate", current_date)).strip() + if row.get("citationDate") and not pd.isna(row.get("citationDate")) + else current_date + ), + "termsOfUse": ( + str(row.get("termsOfUse", "")).strip() + if row.get("termsOfUse") and not pd.isna(row.get("termsOfUse")) + else "" + ), + "citationRequirements": ( + str(row.get("citationRequirements", "")).strip() + if row.get("citationRequirements") + and not pd.isna(row.get("citationRequirements")) + else "" + ), + "conditions": ( + str(row.get("conditions", "")).strip() + if row.get("conditions") and not pd.isna(row.get("conditions")) + else "" + ), + "termsOfAccess": ( + str(row.get("termsOfAccess", "")).strip() + if row.get("termsOfAccess") + and not pd.isna(row.get("termsOfAccess")) + else "" + ), "license": { - "name": str(row.get('licenseName', 'CC0 1.0')).strip() if row.get('licenseName') and not pd.isna(row.get('licenseName')) else 'CC0 1.0', - "uri": str(row.get('licenseUri', 'http://creativecommons.org/publicdomain/zero/1.0')).strip() if row.get('licenseUri') and not pd.isna(row.get('licenseUri')) else 'http://creativecommons.org/publicdomain/zero/1.0', - "iconUri": str(row.get('licenseIconUri', 'https://licensebuttons.net/p/zero/1.0/88x31.png')).strip() if row.get('licenseIconUri') and not pd.isna(row.get('licenseIconUri')) else 'https://licensebuttons.net/p/zero/1.0/88x31.png', - "rightsIdentifier": str(row.get('rightsIdentifier', 'CC0-1.0')).strip() if row.get('rightsIdentifier') and not pd.isna(row.get('rightsIdentifier')) else 'CC0-1.0', - "rightsIdentifierScheme": str(row.get('rightsIdentifierScheme', 'SPDX')).strip() if row.get('rightsIdentifierScheme') and not pd.isna(row.get('rightsIdentifierScheme')) else 'SPDX', - "schemeUri": str(row.get('schemeUri', 'https://spdx.org/licenses/')).strip() if row.get('schemeUri') and not pd.isna(row.get('schemeUri')) else 'https://spdx.org/licenses/', - "languageCode": str(row.get('languageCode', 'en')).strip() if row.get('languageCode') and not pd.isna(row.get('languageCode')) else 'en' + "name": ( + str(row.get("licenseName", "CC0 1.0")).strip() + if row.get("licenseName") + and not pd.isna(row.get("licenseName")) + else "CC0 1.0" + ), + "uri": ( + str( + row.get( + "licenseUri", + "http://creativecommons.org/publicdomain/zero/1.0", + ) + ).strip() + if row.get("licenseUri") and not pd.isna(row.get("licenseUri")) + else "http://creativecommons.org/publicdomain/zero/1.0" + ), + "iconUri": ( + str( + row.get( + "licenseIconUri", + "https://licensebuttons.net/p/zero/1.0/88x31.png", + ) + ).strip() + if row.get("licenseIconUri") + and not pd.isna(row.get("licenseIconUri")) + else "https://licensebuttons.net/p/zero/1.0/88x31.png" + ), + "rightsIdentifier": ( + str(row.get("rightsIdentifier", "CC0-1.0")).strip() + if row.get("rightsIdentifier") + and not pd.isna(row.get("rightsIdentifier")) + else "CC0-1.0" + ), + "rightsIdentifierScheme": ( + str(row.get("rightsIdentifierScheme", "SPDX")).strip() + if row.get("rightsIdentifierScheme") + and not pd.isna(row.get("rightsIdentifierScheme")) + else "SPDX" + ), + "schemeUri": ( + str(row.get("schemeUri", "https://spdx.org/licenses/")).strip() + if row.get("schemeUri") and not pd.isna(row.get("schemeUri")) + else "https://spdx.org/licenses/" + ), + "languageCode": ( + str(row.get("languageCode", "en")).strip() + if row.get("languageCode") + and not pd.isna(row.get("languageCode")) + else "en" + ), }, - "fileAccessRequest": bool(row.get('fileAccessRequest', True)) if row.get('fileAccessRequest') and not pd.isna(row.get('fileAccessRequest')) else True, + "fileAccessRequest": ( + bool(row.get("fileAccessRequest", True)) + if row.get("fileAccessRequest") + and not pd.isna(row.get("fileAccessRequest")) + else True + ), "metadataBlocks": { "citation": { "displayName": "Citation Metadata", "name": "citation", - "fields": [] + "fields": [], } - } - } + }, + }, } fields = dataset_json["datasetVersion"]["metadataBlocks"]["citation"]["fields"] - # Process each metadata field - for field_name, field_config in directory.items(): - if field_name not in row or pd.isna(row[field_name]) or row[field_name] == "": + # Process each CSV column + for col in df.columns: + # Get the base field name and whether it has custom subfields + base_field, custom_subfields = column_mapping.get(col, (col, None)) + + # Check if this field is in our directory + if base_field not in directory: + continue + + # Check if row has a value for this column + if col not in row or pd.isna(row[col]) or row[col] == "": continue - value = str(row[field_name]).strip() + value = str(row[col]).strip() if not value: continue + field_config = directory[base_field] + # Build field structure field_entry = { "typeName": field_config["typeName"], "multiple": field_config["multiple"], - "typeClass": field_config["typeClass"] + "typeClass": field_config["typeClass"], } # Process based on type if field_config["typeClass"] == "primitive": # Convert date fields to year-only format - if field_name in ['productionDate', 'distributionDate', 'dateOfDeposit']: + if base_field in [ + "productionDate", + "distributionDate", + "dateOfDeposit", + ]: value = format_date_to_year(value) if field_config["multiple"]: # Multiple primitive: split by pipe - field_entry["value"] = [v.strip() for v in value.split('|') if v.strip()] + field_entry["value"] = [ + v.strip() for v in value.split("|") if v.strip() + ] else: # Single primitive field_entry["value"] = value elif field_config["typeClass"] == "controlledVocabulary": # Controlled vocabulary: split by pipe - field_entry["value"] = [v.strip() for v in value.split('|') if v.strip()] + field_entry["value"] = [ + v.strip() for v in value.split("|") if v.strip() + ] elif field_config["typeClass"] == "compound": - # Compound: parse with subfields - field_entry["value"] = parse_compound(value, field_name, compound_fields) + # Compound: parse with subfields (use override if custom subfields specified) + field_entry["value"] = parse_compound( + value, base_field, compound_fields_override + ) # Add to fields list if field_entry.get("value"): fields.append(field_entry) # Add geospatial metadata block if present - if any(col in row for col in ['geographicCoverage', 'geographicUnit', 'geographicBoundingBox']): + if any( + col in row + for col in ["geographicCoverage", "geographicUnit", "geographicBoundingBox"] + ): geo_block = create_geospatial_block(row) if geo_block: - dataset_json["datasetVersion"]["metadataBlocks"]["geospatial"] = geo_block + dataset_json["datasetVersion"]["metadataBlocks"][ + "geospatial" + ] = geo_block # Add social science metadata block if present - if any(col in row for col in ['unitOfAnalysis', 'universe', 'timeMethod', 'samplingProcedure']): + if any( + col in row + for col in ["unitOfAnalysis", "universe", "timeMethod", "samplingProcedure"] + ): social_block = create_socialscience_block(row) if social_block: - dataset_json["datasetVersion"]["metadataBlocks"]["socialscience"] = social_block + dataset_json["datasetVersion"]["metadataBlocks"][ + "socialscience" + ] = social_block # Add files array if present - if 'files' in row and row['files'] and not pd.isna(row['files']): + if "files" in row and row["files"] and not pd.isna(row["files"]): try: - files_data = json.loads(row['files']) if isinstance(row['files'], str) else row['files'] - dataset_json["datasetVersion"]["files"] = files_data if isinstance(files_data, list) else [files_data] + files_data = ( + json.loads(row["files"]) + if isinstance(row["files"], str) + else row["files"] + ) + dataset_json["datasetVersion"]["files"] = ( + files_data if isinstance(files_data, list) else [files_data] + ) except json.JSONDecodeError: print(f" ⚠ Warning: Could not parse files JSON in row {idx + 1}") # Add citation field if present - if 'citation' in row and row['citation'] and not pd.isna(row['citation']): - dataset_json["citation"] = str(row['citation']).strip() + if "citation" in row and row["citation"] and not pd.isna(row["citation"]): + dataset_json["citation"] = str(row["citation"]).strip() # Ensure required fields exist (author, datasetContact email, description) ensure_required_fields(dataset_json, row, defaults) all_datasets.append(dataset_json) - print(f"✓ Row {idx + 1}: Dataset ID={dataset_id}, Processed {len(fields)} citation fields") + print( + f"✓ Row {idx + 1}: Dataset ID={dataset_id}, Processed {len(fields)} citation fields" + ) # Write output JSON file # If single row, write as single object; if multiple rows, write as array output_data = all_datasets[0] if len(all_datasets) == 1 else all_datasets - with open(output_json_path, 'w', encoding='utf-8') as f: + with open(output_json_path, "w", encoding="utf-8") as f: json.dump(output_data, f, indent=2, ensure_ascii=False) print(f"\n✓ Successfully converted CSV to JSON: {output_json_path}") @@ -377,18 +801,18 @@ def parse_compound(value, field_name, compound_fields): return [] subfield_names = compound_fields[field_name] - entries = [e.strip() for e in value.split('|') if e.strip()] + entries = [e.strip() for e in value.split("|") if e.strip()] result = [] for entry in entries: - parts = [p.strip() for p in entry.split(';')] + parts = [p.strip() for p in entry.split(";")] entry_obj = {} for i, subfield in enumerate(subfield_names): - if i < len(parts) and parts[i] and parts[i].lower() != 'nan': + if i < len(parts) and parts[i] and parts[i].lower() != "nan": # Special handling for dsDescriptionDate - convert to year - if subfield == 'dsDescriptionDate': - year_match = re.search(r'\b(19|20)\d{2}\b', parts[i]) + if subfield == "dsDescriptionDate": + year_match = re.search(r"\b(19|20)\d{2}\b", parts[i]) if year_match: parts[i] = year_match.group(0) else: @@ -398,7 +822,7 @@ def parse_compound(value, field_name, compound_fields): "typeName": subfield, "multiple": False, "typeClass": "primitive", - "value": parts[i] + "value": parts[i], } if entry_obj: @@ -412,29 +836,52 @@ def create_geospatial_block(row): geospatial = { "displayName": "Geospatial Metadata", "name": "geospatial", - "fields": [] + "fields": [], } # Geographic Coverage - if 'geographicCoverage' in row and row['geographicCoverage'] and not pd.isna(row['geographicCoverage']): - countries = [c.strip() for c in str(row['geographicCoverage']).split('|') if c.strip()] - geospatial["fields"].append({ - "typeName": "geographicCoverage", - "multiple": True, - "typeClass": "compound", - "value": [{"country": {"typeName": "country", "multiple": False, - "typeClass": "controlledVocabulary", "value": c}} for c in countries] - }) + if ( + "geographicCoverage" in row + and row["geographicCoverage"] + and not pd.isna(row["geographicCoverage"]) + ): + countries = [ + c.strip() for c in str(row["geographicCoverage"]).split("|") if c.strip() + ] + geospatial["fields"].append( + { + "typeName": "geographicCoverage", + "multiple": True, + "typeClass": "compound", + "value": [ + { + "country": { + "typeName": "country", + "multiple": False, + "typeClass": "controlledVocabulary", + "value": c, + } + } + for c in countries + ], + } + ) # Geographic Unit - if 'geographicUnit' in row and row['geographicUnit'] and not pd.isna(row['geographicUnit']): - units = [u.strip() for u in str(row['geographicUnit']).split('|') if u.strip()] - geospatial["fields"].append({ - "typeName": "geographicUnit", - "multiple": True, - "typeClass": "primitive", - "value": units - }) + if ( + "geographicUnit" in row + and row["geographicUnit"] + and not pd.isna(row["geographicUnit"]) + ): + units = [u.strip() for u in str(row["geographicUnit"]).split("|") if u.strip()] + geospatial["fields"].append( + { + "typeName": "geographicUnit", + "multiple": True, + "typeClass": "primitive", + "value": units, + } + ) return geospatial if geospatial["fields"] else None @@ -444,31 +891,33 @@ def create_socialscience_block(row): socialscience = { "displayName": "Social Science and Humanities Metadata", "name": "socialscience", - "fields": [] + "fields": [], } # Simple fields mapping simple_fields = { - 'unitOfAnalysis': {'multiple': True, 'typeClass': 'primitive'}, - 'universe': {'multiple': True, 'typeClass': 'primitive'}, - 'timeMethod': {'multiple': False, 'typeClass': 'primitive'}, - 'frequencyOfDataCollection': {'multiple': False, 'typeClass': 'primitive'}, - 'samplingProcedure': {'multiple': False, 'typeClass': 'primitive'}, - 'collectionMode': {'multiple': True, 'typeClass': 'primitive'}, - 'dataCollectionSituation': {'multiple': False, 'typeClass': 'primitive'}, - 'weighting': {'multiple': False, 'typeClass': 'primitive'} + "unitOfAnalysis": {"multiple": True, "typeClass": "primitive"}, + "universe": {"multiple": True, "typeClass": "primitive"}, + "timeMethod": {"multiple": False, "typeClass": "primitive"}, + "frequencyOfDataCollection": {"multiple": False, "typeClass": "primitive"}, + "samplingProcedure": {"multiple": False, "typeClass": "primitive"}, + "collectionMode": {"multiple": True, "typeClass": "primitive"}, + "dataCollectionSituation": {"multiple": False, "typeClass": "primitive"}, + "weighting": {"multiple": False, "typeClass": "primitive"}, } for field_name, config in simple_fields.items(): if field_name in row and row[field_name] and not pd.isna(row[field_name]): field_entry = { "typeName": field_name, - "multiple": config['multiple'], - "typeClass": config['typeClass'] + "multiple": config["multiple"], + "typeClass": config["typeClass"], } - if config['multiple']: - field_entry["value"] = [v.strip() for v in str(row[field_name]).split('|') if v.strip()] + if config["multiple"]: + field_entry["value"] = [ + v.strip() for v in str(row[field_name]).split("|") if v.strip() + ] else: field_entry["value"] = str(row[field_name]).strip() @@ -480,21 +929,45 @@ def create_socialscience_block(row): # Main execution if __name__ == "__main__": - parser = argparse.ArgumentParser(description='Convert CSV to Dataverse JSON with optional defaults') - parser.add_argument('csv_input', nargs='?', default='Csv_to_json - Citation.csv', help='Input CSV file') - parser.add_argument('json_output', nargs='?', default='output_metadata.json', help='Output JSON file') - parser.add_argument('--default-author', dest='default_author', help='Default author name if none provided') - parser.add_argument('--default-email', dest='default_email', help='Default contact email if none provided') - parser.add_argument('--default-description', dest='default_description', help='Default description if none provided') + parser = argparse.ArgumentParser( + description="Convert CSV to Dataverse JSON with optional defaults" + ) + parser.add_argument( + "csv_input", + nargs="?", + default="Csv_to_json - Citation.csv", + help="Input CSV file", + ) + parser.add_argument( + "json_output", + nargs="?", + default="output_metadata.json", + help="Output JSON file", + ) + parser.add_argument( + "--default-author", + dest="default_author", + help="Default author name if none provided", + ) + parser.add_argument( + "--default-email", + dest="default_email", + help="Default contact email if none provided", + ) + parser.add_argument( + "--default-description", + dest="default_description", + help="Default description if none provided", + ) args = parser.parse_args() defaults = {} if args.default_author: - defaults['author'] = args.default_author + defaults["author"] = args.default_author if args.default_email: - defaults['email'] = args.default_email + defaults["email"] = args.default_email if args.default_description: - defaults['description'] = args.default_description + defaults["description"] = args.default_description csv_to_dataverse_json(args.csv_input, args.json_output, defaults=defaults) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..8482349 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,23 @@ +[project] +name = "csv-to-dataverse-json" +version = "1.0.0" +description = "Simple tool to convert CSV files into Dataverse/Borealis JSON format" +readme = "README.md" +requires-python = ">=3.12" +dependencies = [ + "pandas>=1.3.0", +] + +[project.optional-dependencies] +dev = [ + "black>=24.0.0", + "pytest>=7.0.0", + "pytest-cov>=4.0.0", +] + +[project.scripts] +csv-to-dataverse = "csv_to_dataverse_json:main" + +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" diff --git a/test_csv_to_dataverse_json.py b/test_csv_to_dataverse_json.py new file mode 100644 index 0000000..338f75d --- /dev/null +++ b/test_csv_to_dataverse_json.py @@ -0,0 +1,459 @@ +# -*- coding: utf-8 -*- +""" +Tests for CSV to Dataverse JSON Converter +""" + +import os +import json +import tempfile +import pytest +import pandas as pd +from csv_to_dataverse_json import ( + ensure_required_fields, + parse_compound, + csv_to_dataverse_json, + create_geospatial_block, + create_socialscience_block, +) + + +class TestUTF8Encoding: + """Test UTF-8 encoding handling""" + + def test_utf8_french_text_in_csv(self): + """Test that French UTF-8 characters are correctly processed""" + # Create a temporary CSV file with French text + csv_content = """title,notesText +"Test Dataset","L'enquête a révélé que plusieurs ménages n'avaient pas reçu le courriel initial à cause d'une défaillance technique." +""" + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, encoding="utf-8" + ) as f: + f.write(csv_content) + csv_path = f.name + + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False + ) as f: + json_path = f.name + + try: + # Convert CSV to JSON + csv_to_dataverse_json(csv_path, json_path) + + # Read the output JSON + with open(json_path, "r", encoding="utf-8") as f: + result = json.load(f) + + # Check that the French text is preserved + notes_field = None + for field in result["datasetVersion"]["metadataBlocks"]["citation"][ + "fields" + ]: + if field["typeName"] == "notesText": + notes_field = field + break + + assert notes_field is not None + assert ( + "L'enquête a révélé" + in notes_field["value"] + ) + assert "ménages" in notes_field["value"] + assert "défaillance" in notes_field["value"] + + finally: + # Cleanup + os.unlink(csv_path) + os.unlink(json_path) + + def test_utf8_special_characters(self): + """Test various UTF-8 special characters""" + special_chars = "Ñoño: café, naïve, 日本語, 中文, العربية" + csv_content = f"""title,notesText +"Test Dataset","{special_chars}" +""" + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, encoding="utf-8" + ) as f: + f.write(csv_content) + csv_path = f.name + + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False + ) as f: + json_path = f.name + + try: + csv_to_dataverse_json(csv_path, json_path) + + with open(json_path, "r", encoding="utf-8") as f: + result = json.load(f) + + notes_field = None + for field in result["datasetVersion"]["metadataBlocks"]["citation"][ + "fields" + ]: + if field["typeName"] == "notesText": + notes_field = field + break + + assert notes_field is not None + assert special_chars in notes_field["value"] + + finally: + os.unlink(csv_path) + os.unlink(json_path) + + +class TestParseCompound: + """Test parse_compound function""" + + def test_parse_author_single(self): + """Test parsing single author with name and affiliation""" + compound_fields = { + "author": ["authorName", "authorAffiliation"], + } + + result = parse_compound( + "John Smith; Harvard University", "author", compound_fields + ) + + assert len(result) == 1 + assert result[0]["authorName"]["value"] == "John Smith" + assert result[0]["authorAffiliation"]["value"] == "Harvard University" + + def test_parse_author_multiple(self): + """Test parsing multiple authors separated by pipe""" + compound_fields = { + "author": ["authorName", "authorAffiliation"], + } + + result = parse_compound( + "John Smith; Harvard University | Jane Doe; MIT", "author", compound_fields + ) + + assert len(result) == 2 + assert result[0]["authorName"]["value"] == "John Smith" + assert result[0]["authorAffiliation"]["value"] == "Harvard University" + assert result[1]["authorName"]["value"] == "Jane Doe" + assert result[1]["authorAffiliation"]["value"] == "MIT" + + def test_parse_contact_with_email(self): + """Test parsing dataset contact with name, affiliation, and email""" + compound_fields = { + "datasetContact": [ + "datasetContactName", + "datasetContactAffiliation", + "datasetContactEmail", + ], + } + + result = parse_compound( + "Contact Person; University; contact@email.com", + "datasetContact", + compound_fields, + ) + + assert len(result) == 1 + assert result[0]["datasetContactName"]["value"] == "Contact Person" + assert result[0]["datasetContactAffiliation"]["value"] == "University" + assert result[0]["datasetContactEmail"]["value"] == "contact@email.com" + + def test_parse_empty_subfields(self): + """Test parsing with empty subfields""" + compound_fields = { + "keyword": ["keywordValue", "keywordVocabulary"], + } + + result = parse_compound("sample; | test;", "keyword", compound_fields) + + assert len(result) == 2 + assert result[0]["keywordValue"]["value"] == "sample" + # keywordVocabulary should be omitted when empty + assert "keywordVocabulary" not in result[0] + + +class TestEnsureRequiredFields: + """Test ensure_required_fields function""" + + def test_adds_default_author_when_missing(self): + """Test that default author is added when missing""" + dataset_json = { + "datasetVersion": {"metadataBlocks": {"citation": {"fields": []}}} + } + row = pd.Series({}) + defaults = {"author": "Default Author"} + + ensure_required_fields(dataset_json, row, defaults) + + fields = dataset_json["datasetVersion"]["metadataBlocks"]["citation"]["fields"] + author_field = next((f for f in fields if f["typeName"] == "author"), None) + + assert author_field is not None + assert author_field["value"][0]["authorName"]["value"] == "Default Author" + + def test_uses_depositor_as_fallback_author(self): + """Test that depositor is used as fallback author""" + dataset_json = { + "datasetVersion": {"metadataBlocks": {"citation": {"fields": []}}} + } + row = pd.Series({"depositor": "Jane Depositor"}) + + ensure_required_fields(dataset_json, row) + + fields = dataset_json["datasetVersion"]["metadataBlocks"]["citation"]["fields"] + author_field = next((f for f in fields if f["typeName"] == "author"), None) + + assert author_field is not None + assert author_field["value"][0]["authorName"]["value"] == "Jane Depositor" + + def test_adds_default_contact_email_when_missing(self): + """Test that default contact email is added when missing""" + dataset_json = { + "datasetVersion": {"metadataBlocks": {"citation": {"fields": []}}} + } + row = pd.Series({}) + defaults = {"email": "default@example.com"} + + ensure_required_fields(dataset_json, row, defaults) + + fields = dataset_json["datasetVersion"]["metadataBlocks"]["citation"]["fields"] + contact_field = next( + (f for f in fields if f["typeName"] == "datasetContact"), None + ) + + assert contact_field is not None + assert ( + contact_field["value"][0]["datasetContactEmail"]["value"] + == "default@example.com" + ) + + def test_adds_default_description_when_missing(self): + """Test that default description is added when missing""" + dataset_json = { + "datasetVersion": {"metadataBlocks": {"citation": {"fields": []}}} + } + row = pd.Series({}) + defaults = {"description": "Default description"} + + ensure_required_fields(dataset_json, row, defaults) + + fields = dataset_json["datasetVersion"]["metadataBlocks"]["citation"]["fields"] + desc_field = next( + (f for f in fields if f["typeName"] == "dsDescription"), None + ) + + assert desc_field is not None + assert ( + desc_field["value"][0]["dsDescriptionValue"]["value"] + == "Default description" + ) + + +class TestIntegration: + """Integration tests for the full conversion process""" + + def test_full_conversion_with_all_fields(self): + """Test complete conversion with various field types""" + csv_content = """title,subtitle,"author: authorName; authorAffiliation",subject,"keyword: keywordValue; keywordVocabulary" +"Research Dataset","Subtitle","John Smith; University | Jane Doe; MIT","Social Sciences","research; | data;" +""" + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, encoding="utf-8" + ) as f: + f.write(csv_content) + csv_path = f.name + + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False + ) as f: + json_path = f.name + + try: + csv_to_dataverse_json(csv_path, json_path) + + with open(json_path, "r", encoding="utf-8") as f: + result = json.load(f) + + # Verify top-level structure + assert "id" in result + assert "identifier" in result + assert "datasetVersion" in result + + # Verify citation metadata + fields = result["datasetVersion"]["metadataBlocks"]["citation"]["fields"] + + # Check title + title_field = next((f for f in fields if f["typeName"] == "title"), None) + assert title_field is not None + assert title_field["value"] == "Research Dataset" + + # Check subtitle + subtitle_field = next( + (f for f in fields if f["typeName"] == "subtitle"), None + ) + assert subtitle_field is not None + assert subtitle_field["value"] == "Subtitle" + + # Check authors + author_field = next((f for f in fields if f["typeName"] == "author"), None) + assert author_field is not None + assert len(author_field["value"]) == 2 + assert author_field["value"][0]["authorName"]["value"] == "John Smith" + assert ( + author_field["value"][0]["authorAffiliation"]["value"] == "University" + ) + + # Check subject + subject_field = next( + (f for f in fields if f["typeName"] == "subject"), None + ) + assert subject_field is not None + assert "Social Sciences" in subject_field["value"] + + finally: + os.unlink(csv_path) + os.unlink(json_path) + + def test_conversion_with_defaults(self): + """Test conversion with default values""" + csv_content = """title +"Minimal Dataset" +""" + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, encoding="utf-8" + ) as f: + f.write(csv_content) + csv_path = f.name + + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False + ) as f: + json_path = f.name + + try: + defaults = { + "author": "Test Author", + "email": "test@example.com", + "description": "Test description", + } + csv_to_dataverse_json(csv_path, json_path, defaults) + + with open(json_path, "r", encoding="utf-8") as f: + result = json.load(f) + + fields = result["datasetVersion"]["metadataBlocks"]["citation"]["fields"] + + # Check that defaults were applied + author_field = next((f for f in fields if f["typeName"] == "author"), None) + assert author_field is not None + assert author_field["value"][0]["authorName"]["value"] == "Test Author" + + contact_field = next( + (f for f in fields if f["typeName"] == "datasetContact"), None + ) + assert contact_field is not None + assert ( + contact_field["value"][0]["datasetContactEmail"]["value"] + == "test@example.com" + ) + + finally: + os.unlink(csv_path) + os.unlink(json_path) + + def test_multiple_rows_conversion(self): + """Test converting CSV with multiple rows""" + csv_content = """title,subject +"Dataset 1","Social Sciences" +"Dataset 2","Medicine, Health and Life Sciences" +""" + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, encoding="utf-8" + ) as f: + f.write(csv_content) + csv_path = f.name + + with tempfile.NamedTemporaryFile( + mode="w", suffix=".json", delete=False + ) as f: + json_path = f.name + + try: + csv_to_dataverse_json(csv_path, json_path) + + with open(json_path, "r", encoding="utf-8") as f: + result = json.load(f) + + # Should be a list of datasets + assert isinstance(result, list) + assert len(result) == 2 + + # Check first dataset + fields1 = result[0]["datasetVersion"]["metadataBlocks"]["citation"][ + "fields" + ] + title1 = next((f for f in fields1 if f["typeName"] == "title"), None) + assert title1["value"] == "Dataset 1" + + # Check second dataset + fields2 = result[1]["datasetVersion"]["metadataBlocks"]["citation"][ + "fields" + ] + title2 = next((f for f in fields2 if f["typeName"] == "title"), None) + assert title2["value"] == "Dataset 2" + + finally: + os.unlink(csv_path) + os.unlink(json_path) + + +class TestGeospatialBlock: + """Test geospatial metadata block creation""" + + def test_create_geospatial_block_with_coverage(self): + """Test creating geospatial block with geographic coverage""" + row = pd.Series({"geographicCoverage": "Canada | United States"}) + + result = create_geospatial_block(row) + + assert result is not None + assert result["name"] == "geospatial" + assert "fields" in result + # Check that geographicCoverage field exists + geo_field = next( + (f for f in result["fields"] if f["typeName"] == "geographicCoverage"), + None, + ) + assert geo_field is not None + assert len(geo_field["value"]) == 2 + assert geo_field["value"][0]["country"]["value"] == "Canada" + + def test_create_geospatial_block_returns_none_when_empty(self): + """Test that None is returned when no geospatial fields present""" + row = pd.Series({"title": "Test Dataset"}) + + result = create_geospatial_block(row) + + assert result is None + + +class TestSocialscienceBlock: + """Test social science metadata block creation""" + + def test_create_socialscience_block_with_unit(self): + """Test creating social science block with unit of analysis""" + row = pd.Series({"unitOfAnalysis": "Individual"}) + + result = create_socialscience_block(row) + + assert result is not None + assert result["name"] == "socialscience" + assert "fields" in result + # Check that unitOfAnalysis field exists + unit_field = next( + (f for f in result["fields"] if f["typeName"] == "unitOfAnalysis"), None + ) + assert unit_field is not None + assert "Individual" in unit_field["value"]