Adds fixes for Wikibase export for publications

Implements functionality to export publication metadata to a Wikibase instance, including Wikidata

Author: nuest

.claude/settings.local.json

@@ -16,7 +16,16 @@
       "Bash(python manage.py:*)",
       "Bash(python -m py_compile:*)",
       "Bash(python:*)",
-      "Bash(node --check:*)"
+      "Bash(node --check:*)",
+      "Bash(find:*)",
+      "Bash(OPTIMAP_LOGGING_LEVEL=WARNING python manage.py test:*)",
+      "Bash(export OPTIMAP_LOGGING_LEVEL=WARNING)",
+      "Bash(awk:*)",
+      "Bash(chmod:*)",
+      "Bash(bash:*)",
+      "Bash(./create_wikibase_property.sh:*)",
+      "Bash(python3:*)",
+      "Bash(pkill:*)"
     ],
     "deny": [],
     "ask": []

README.md

@@ -284,11 +284,12 @@ python -m smtpd -c DebuggingServer -n localhost:5587
 OPTIMAP_EMAIL_HOST=localhost
 OPTIMAP_EMAIL_PORT=5587
 ```
-### Accessing list of article links
 
-Visit the URL - http://127.0.0.1:8000/articles/links/
+### Accessing list of works
 
-### Harvest Publications from Real Journals
+Visit the URL - <http://127.0.0.1:8000/works/>
+
+### Harvest Publications from real journals
 
 The `harvest_journals` management command allows you to harvest publications from real OAI-PMH journal sources directly into your database. This is useful for:
 

optimap/.env.example

@@ -27,3 +27,10 @@ OPTIMAP_LOGGING_LEVEL=INFO
 DJANGO_LOGGING_LEVEL=ERROR
 
 OPTIMAP_DATA_DUMP_RETENTION=3
+
+WIKIBASE_CONSUMER_TOKEN=your_consumer_token
+WIKIBASE_CONSUMER_SECRET=your_consumer_secret
+WIKIBASE_ACCESS_TOKEN=your_access_token
+WIKIBASE_ACCESS_SECRET=your_access_secret
+WIKIBASE_API_URL=https://your-instance.wikibase.cloud/w/api.php
+

optimap/settings.py

@@ -217,14 +217,27 @@
 EMAIL_IMAP_SENT_FOLDER = env('OPTIMAP_EMAIL_IMAP_SENT_FOLDER', default='')
 OPTIMAP_EMAIL_SEND_DELAY = env("OPTIMAP_EMAIL_SEND_DELAY", default=2)
 BASE_URL = env("BASE_URL", default="http://127.0.0.1:8000")
-OAI_USERNAME = env("OPTIMAP_OAI_USERNAME", default="")
-OAI_PASSWORD = env("OPTIMAP_OAI_PASSWORD", default="")
 EMAIL_SEND_DELAY = 2
 DATA_DUMP_INTERVAL_HOURS = 6
-OPENALEX_MAILTO = "[email protected]"
-WIKIBASE_API_URL = env("WIKIBASE_API_URL")
-WIKIBASE_USERNAME = env("WIKIBASE_USERNAME")
-WIKIBASE_PASSWORD = env("WIKIBASE_PASSWORD")
+
+# Contact email for API user agents (OpenAlex, Wikidata, etc.)
+CONTACT_EMAIL = "[email protected]"
+
+# Wikibase/Wikidata configuration
+WIKIBASE_API_URL = env("WIKIBASE_API_URL", default="")
+
+# OAuth 1.0a authentication (required)
+WIKIBASE_CONSUMER_TOKEN = env("WIKIBASE_CONSUMER_TOKEN", default="")
+WIKIBASE_CONSUMER_SECRET = env("WIKIBASE_CONSUMER_SECRET", default="")
+WIKIBASE_ACCESS_TOKEN = env("WIKIBASE_ACCESS_TOKEN", default="")
+WIKIBASE_ACCESS_SECRET = env("WIKIBASE_ACCESS_SECRET", default="")
+
+# Property auto-creation setting
+WIKIBASE_CREATE_PROPERTIES_IF_MISSING = env("WIKIBASE_CREATE_PROPERTIES_IF_MISSING", default=True, cast=bool)
+
+# Import optimap version for user agent
+import optimap
+WIKIBASE_USER_AGENT = f"OPTIMAP/{optimap.__version__} (https://optimap.science; {CONTACT_EMAIL})"
 
 
 MIDDLEWARE = [

publications/admin.py

@@ -8,7 +8,7 @@
 from leaflet.admin import LeafletGeoAdmin
 from publications.models import Publication, Source, HarvestingEvent, BlockedEmail, BlockedDomain, GlobalRegion
 from import_export.admin import ImportExportModelAdmin
-from publications.models import EmailLog, Subscription, UserProfile
+from publications.models import EmailLog, Subscription, UserProfile, WikidataExportLog
 from publications.tasks import harvest_oai_endpoint, schedule_subscription_email_task, send_monthly_email, schedule_monthly_email_task
 from django_q.models import Schedule
 from django.utils.timezone import now
@@ -17,24 +17,45 @@
 from publications.tasks import regenerate_geopackage_cache
 from django.test import Client
 from django.http import HttpResponse
-from publications.wikidata import export_publications_to_wikidata
+from publications.wikidata import export_publications_to_wikidata, export_publications_to_wikidata_dryrun
 
-@admin.action(description="Create new Wikidata items for selected publications")
+@admin.action(description="Export selected publications to Wikidata/Wikibase")
 def export_to_wikidata(modeladmin, request, queryset):
-    created_count, updated_count, error_records = export_publications_to_wikidata(queryset)
+    stats = export_publications_to_wikidata(queryset)
 
     # Success messages
-    if created_count:
-        messages.success(request, f"{created_count} new Wikidata item(s) created.")
-    if updated_count:
-        messages.success(request, f"{updated_count} existing Wikidata item(s) updated.")
-
-    # Warnings and errors
-    for publication, error_message in error_records:
-        if error_message == "no publicationDate":
-            messages.warning(request, f"Skipping “{publication.title}”: no publication date")
-        else:
-            messages.error(request, f"Failed to export “{publication.title}”: {error_message}")
+    if stats['created']:
+        messages.success(request, f"{stats['created']} new Wikidata item(s) created.")
+    if stats['updated']:
+        messages.success(request, f"{stats['updated']} existing Wikidata item(s) updated.")
+    if stats['skipped']:
+        messages.info(request, f"{stats['skipped']} publication(s) skipped (already exist or duplicate labels).")
+
+    # Error messages
+    if stats['errors']:
+        messages.error(request, f"{stats['errors']} publication(s) failed to export. Check the Wikidata export logs for details.")
+
+    # Summary message
+    messages.info(request, f"Total: {stats['total']} publication(s) processed.")
+
+@admin.action(description="[DRY-RUN] Export selected publications to Wikidata/Wikibase")
+def export_to_wikidata_dryrun(modeladmin, request, queryset):
+    stats = export_publications_to_wikidata_dryrun(queryset)
+
+    # Dry-run summary messages
+    messages.info(request, f"[DRY-RUN] Export simulation complete:")
+
+    if stats['created']:
+        messages.info(request, f"  • Would create {stats['created']} new Wikidata item(s)")
+    if stats['updated']:
+        messages.info(request, f"  • Would update {stats['updated']} existing Wikidata item(s)")
+    if stats['skipped']:
+        messages.info(request, f"  • Would skip {stats['skipped']} publication(s)")
+    if stats['errors']:
+        messages.warning(request, f"  • {stats['errors']} publication(s) have validation errors")
+
+    # Summary message
+    messages.success(request, f"[DRY-RUN] Total: {stats['total']} publication(s) analyzed. No changes were written to Wikibase.")
 
 @admin.action(description="Mark selected publications as published")
 def make_public(modeladmin, request, queryset):
@@ -172,8 +193,9 @@ class PublicationAdmin(LeafletGeoAdmin, ImportExportModelAdmin):
                      "openalex_fulltext_origin", "openalex_is_retracted",
                      "openalex_ids", "openalex_open_access_status")
     readonly_fields = ("created_by", "updated_by", "openalex_link")
-    actions = ["make_public", "make_draft", "regenerate_all_exports",
-               "export_permalinks_csv", "email_permalinks_preview", "export_to_wikidata"]
+    actions = [make_public, make_draft, regenerate_all_exports,
+               "export_permalinks_csv", "email_permalinks_preview",
+               export_to_wikidata, export_to_wikidata_dryrun]
 
     @admin.display(boolean=True, description="Has DOI")
     def has_permalink(self, obj):
@@ -243,12 +265,95 @@ class EmailLogAdmin(admin.ModelAdmin):
         "sent_at",
         "sent_by",
         "trigger_source",
-        "status",  
-        "error_message", 
+        "status",
+        "error_message",
     )
-    list_filter = ("status", "trigger_source", "sent_at")  
-    search_fields = ("recipient_email", "subject", "sent_by__username")  
-    actions = [trigger_monthly_email, trigger_monthly_email_task]  
+    list_filter = ("status", "trigger_source", "sent_at")
+    search_fields = ("recipient_email", "subject", "sent_by__username")
+    actions = [trigger_monthly_email, trigger_monthly_email_task]
+
+@admin.register(WikidataExportLog)
+class WikidataExportLogAdmin(admin.ModelAdmin):
+    """Admin interface for Wikidata export logs."""
+    list_display = (
+        "id",
+        "publication_title",
+        "action",
+        "wikidata_link",
+        "export_date",
+        "fields_count",
+    )
+    list_filter = ("action", "export_date")
+    search_fields = (
+        "publication__title",
+        "publication__doi",
+        "wikidata_qid",
+        "export_summary",
+    )
+    readonly_fields = (
+        "publication",
+        "export_date",
+        "action",
+        "wikidata_qid",
+        "wikidata_url",
+        "wikidata_link_display",
+        "wikibase_endpoint",
+        "exported_fields",
+        "error_message_display",
+        "export_summary",
+    )
+    fields = (
+        "publication",
+        "export_date",
+        "action",
+        "wikibase_endpoint",
+        "wikidata_qid",
+        "wikidata_link_display",
+        "export_summary",
+        "exported_fields",
+        "error_message_display",
+    )
+    ordering = ("-export_date",)
+    date_hierarchy = "export_date"
+
+    @admin.display(description="Publication")
+    def publication_title(self, obj):
+        return obj.publication.title[:60] if obj.publication else "—"
+
+    @admin.display(description="Wikidata")
+    def wikidata_link(self, obj):
+        if obj.wikidata_qid and obj.wikidata_url:
+            return format_html(
+                '<a href="{}" target="_blank" rel="noopener"><i class="fas fa-external-link-alt"></i> {}</a>',
+                obj.wikidata_url,
+                obj.wikidata_qid
+            )
+        return "—"
+
+    @admin.display(description="Wikidata Link")
+    def wikidata_link_display(self, obj):
+        if obj.wikidata_qid and obj.wikidata_url:
+            return format_html(
+                '<a href="{}" target="_blank" rel="noopener">{}</a>',
+                obj.wikidata_url,
+                obj.wikidata_url
+            )
+        return "—"
+
+    @admin.display(description="Fields")
+    def fields_count(self, obj):
+        if obj.exported_fields:
+            return len(obj.exported_fields)
+        return 0
+
+    @admin.display(description="Error Message (Full Traceback)")
+    def error_message_display(self, obj):
+        if obj.error_message:
+            return format_html(
+                '<pre style="white-space: pre-wrap; font-family: monospace; font-size: 12px; background: #f5f5f5; padding: 10px; border: 1px solid #ddd; border-radius: 4px; max-height: 400px; overflow-y: auto;">{}</pre>',
+                obj.error_message
+            )
+        return "—"
 
 @admin.register(Subscription)
 class SubscriptionAdmin(admin.ModelAdmin):

publications/migrations/0004_wikidata_export_log.py

@@ -0,0 +1,37 @@
+# Generated migration for WikidataExportLog model
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('publications', '0002_add_regions_to_subscription'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='WikidataExportLog',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('publication', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='wikidata_exports', to='publications.publication')),
+                ('export_date', models.DateTimeField(auto_now_add=True, db_index=True)),
+                ('action', models.CharField(max_length=20, choices=[('created', 'Created'), ('updated', 'Updated'), ('skipped', 'Skipped'), ('error', 'Error')], db_index=True)),
+                ('wikidata_qid', models.CharField(max_length=50, blank=True, null=True, help_text='Wikidata Q-ID (e.g., Q12345)')),
+                ('wikidata_url', models.URLField(max_length=512, blank=True, null=True, help_text='Full URL to Wikidata item')),
+                ('exported_fields', models.JSONField(blank=True, null=True, help_text='List of fields that were exported')),
+                ('error_message', models.TextField(blank=True, null=True)),
+                ('export_summary', models.TextField(blank=True, null=True, help_text='Summary of what was exported')),
+            ],
+            options={
+                'ordering': ['-export_date'],
+                'verbose_name': 'Wikidata Export Log',
+                'verbose_name_plural': 'Wikidata Export Logs',
+            },
+        ),
+        migrations.AddIndex(
+            model_name='wikidataexportlog',
+            index=models.Index(fields=['wikidata_qid'], name='publications_wikidata_qid_idx'),
+        ),
+    ]

publications/migrations/0005_add_wikibase_endpoint_to_exportlog.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.1.9 on 2025-10-23 20:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("publications", "0004_wikidata_export_log"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="wikidataexportlog",
+            name="wikibase_endpoint",
+            field=models.URLField(
+                blank=True,
+                help_text="Wikibase API endpoint used for this export (e.g., https://www.wikidata.org/w/api.php)",
+                max_length=512,
+                null=True,
+            ),
+        ),
+    ]

publications/models.py

@@ -262,4 +262,66 @@ def save(self, *args, **kwargs):
             name=f"Harvest Source {self.id}",
         )
 
-Journal = Source  
+Journal = Source
+
+
+class WikidataExportLog(models.Model):
+    """
+    Log of Wikidata exports for publications.
+    Tracks when publications were exported, what action was taken,
+    and links to the created/updated Wikidata items.
+    """
+    ACTION_CHOICES = [
+        ('created', 'Created'),
+        ('updated', 'Updated'),
+        ('skipped', 'Skipped'),
+        ('error', 'Error'),
+    ]
+
+    publication = models.ForeignKey(
+        'Publication',
+        on_delete=models.CASCADE,
+        related_name='wikidata_exports'
+    )
+    export_date = models.DateTimeField(auto_now_add=True, db_index=True)
+    action = models.CharField(max_length=20, choices=ACTION_CHOICES, db_index=True)
+    wikidata_qid = models.CharField(
+        max_length=50,
+        blank=True,
+        null=True,
+        help_text='Wikidata Q-ID (e.g., Q12345)'
+    )
+    wikidata_url = models.URLField(
+        max_length=512,
+        blank=True,
+        null=True,
+        help_text='Full URL to Wikidata item'
+    )
+    exported_fields = models.JSONField(
+        blank=True,
+        null=True,
+        help_text='List of fields that were exported'
+    )
+    error_message = models.TextField(blank=True, null=True)
+    export_summary = models.TextField(
+        blank=True,
+        null=True,
+        help_text='Summary of what was exported'
+    )
+    wikibase_endpoint = models.URLField(
+        max_length=512,
+        blank=True,
+        null=True,
+        help_text='Wikibase API endpoint used for this export (e.g., https://www.wikidata.org/w/api.php)'
+    )
+
+    class Meta:
+        ordering = ['-export_date']
+        verbose_name = 'Wikidata Export Log'
+        verbose_name_plural = 'Wikidata Export Logs'
+        indexes = [
+            models.Index(fields=['wikidata_qid'], name='publications_wikidata_qid_idx'),
+        ]
+
+    def __str__(self):
+        return f"{self.action.capitalize()} {self.publication.title[:50]} on {self.export_date.strftime('%Y-%m-%d')}"