"""Seed 5 collections and ~40 sample documents with authors and QA state.""" from __future__ import annotations import logging import random from io import BytesIO from django.core.files.base import ContentFile from django.db import transaction from django.utils import timezone logger = logging.getLogger(__name__) COLLECTIONS = [ { "slug": "theses", "name": "Theses & Dissertations", "description": "Masters and PhD theses from RUFORUM-supported scholarships.", "visibility": "public_open", "qa_workflow_enabled": True, }, { "slug": "policy-briefs", "name": "Policy Briefs", "description": "Short policy briefs distilling research into actionable recommendations.", "visibility": "public_open", "qa_workflow_enabled": True, }, { "slug": "datasets", "name": "Datasets", "description": "Research datasets published under open licences.", "visibility": "public_open", "qa_workflow_enabled": True, }, { "slug": "reports", "name": "Technical & Project Reports", "description": "Project and technical reports from funded RUFORUM programmes.", "visibility": "authenticated", "qa_workflow_enabled": True, }, { "slug": "innovation", "name": "Innovation Documents", "description": "Technical documentation for innovations and SME-Hub outputs.", "visibility": "authenticated", "qa_workflow_enabled": True, }, { "slug": "rims-closeout", "name": "RIMS Grant Close-outs", "description": "Outputs pushed automatically at grant close-out.", "visibility": "restricted", "qa_workflow_enabled": True, }, ] SAMPLE_AUTHORS = [ {"first_name": "Amina", "last_name": "Kato", "orcid": "0000-0001-1234-5678", "country": "UG"}, {"first_name": "Joseph", "last_name": "Mutua", "orcid": "0000-0002-2345-6789", "country": "KE"}, {"first_name": "Fatima", "last_name": "Diallo", "orcid": "0000-0003-3456-7890", "country": "SN"}, {"first_name": "Peter", "last_name": "Mwangi", "orcid": None, "country": "KE"}, {"first_name": "Sarah", "last_name": "Okonkwo", "orcid": "0000-0004-4567-8901", "country": "NG"}, {"first_name": "Tendai", "last_name": "Moyo", "orcid": None, "country": "ZW"}, {"first_name": "Grace", "last_name": "Nyambura", "orcid": "0000-0005-5678-9012", "country": "KE"}, {"first_name": "Abdoulaye", "last_name": "Ba", "orcid": None, "country": "SN"}, ] DOCUMENT_TEMPLATES = [ ("thesis", "Smallholder climate adaptation in highland maize systems", "This thesis examines climate adaptation strategies among smallholder maize farmers in the East African highlands.", "cc_by", "theses", 2024), ("thesis", "Agroforestry impacts on soil nitrogen in West African savannas", "We evaluate long-term nitrogen balance changes under four agroforestry systems across 3 watersheds.", "cc_by", "theses", 2024), ("thesis", "Groundwater recharge dynamics under shifting rainfall regimes", "A multi-catchment assessment of rainfall variability and its effect on shallow aquifer recharge.", "cc_by_sa", "theses", 2023), ("thesis", "Post-harvest loss reduction in maize using hermetic storage", "Field trial of hermetic storage bags across 200 smallholders in Uganda and Tanzania.", "cc_by", "theses", 2023), ("policy_brief", "Scaling drought-tolerant seed systems across RUFORUM member states", "Policy recommendations for harmonising drought-tolerant seed certification across 12 member states.", "cc_by", "policy-briefs", 2025), ("policy_brief", "Youth employment in agro-processing value chains", "Brief outlining pathways for youth employment within African agro-processing value chains.", "cc_by", "policy-briefs", 2025), ("policy_brief", "Gender mainstreaming in agricultural extension services", "Evidence and recommendations for gender-inclusive extension service design.", "cc_by", "policy-briefs", 2024), ("dataset", "Maize yield panel dataset (10 countries, 2010–2024)", "A harmonised panel of district-level maize yields across 10 RUFORUM member countries.", "cc0", "datasets", 2025), ("dataset", "Smallholder dairy productivity survey (2023)", "Cross-sectional smallholder dairy productivity survey across Kenya, Uganda, and Rwanda.", "cc_by", "datasets", 2023), ("dataset", "Rainfall anomaly raster stack 1990–2024", "Monthly rainfall anomaly rasters at 10 km resolution for East and Southern Africa.", "cc_by", "datasets", 2024), ("technical_report", "Post-harvest loss assessment for cassava in Ghana", "Technical assessment of cassava post-harvest losses and intervention cost-effectiveness.", "cc_by_nc", "reports", 2024), ("technical_report", "Soil fertility status atlas — Eastern Africa", "Comprehensive soil fertility atlas with interpretive maps for 6 countries.", "cc_by", "reports", 2023), ("project_report", "PRESSA mid-term evaluation report", "Mid-term evaluation of PRESSA programme — outputs, outcomes, and learning.", "all_rights_reserved", "reports", 2024), ("innovation_document", "Solar-powered cold room for fish traders", "Technical documentation, bill of materials, and deployment guidance for solar cold rooms.", "cc_by_sa", "innovation", 2025), ("innovation_document", "Biochar production kiln — open design", "Open-source kiln design with performance data from 40 pilot sites.", "cc_by", "innovation", 2024), ("publication", "Remote sensing of maize phenology across agroecological zones", "Peer-reviewed publication on MODIS-based maize phenology detection.", "cc_by", "theses", 2023), ] def _dummy_file(title: str, idx: int, kind: str = "pdf") -> ContentFile: """Create a tiny binary blob that passes magic-byte validation for PDFs. The title + idx is embedded in the body so every sample file hashes to a unique value — otherwise the duplicate-detection service would swallow most of the seed inserts. """ if kind == "pdf": body = ( b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n1 0 obj<>endobj\n" b"2 0 obj<>endobj\n" + f"3 0 obj({idx}: {title})\n".encode("utf-8") + b"trailer<<>>\n%%EOF\n" ) buf = BytesIO(body) else: buf = BytesIO(f"{idx}: {title}".encode("utf-8")) return ContentFile(buf.getvalue(), name=f"{title[:40].replace(' ', '_')}_{idx}.{kind}") @transaction.atomic def seed(count: int = 40) -> dict: from apps.core.authentication.models import Institution from apps.core.countries import to_country_name from apps.core.permissions.roles import UserRole from apps.repository.documents import services from apps.repository.documents.models import ( Author, Collection, Document, QARecord, ) from django.contrib.auth import get_user_model User = get_user_model() # Collections. col_objs: dict[str, Collection] = {} for col in COLLECTIONS: obj, _ = Collection.objects.update_or_create( slug=col["slug"], defaults={ "name": col["name"], "description": col["description"], "visibility": col["visibility"], "qa_workflow_enabled": col["qa_workflow_enabled"], "qa_turnaround_days": 14, }, ) col_objs[col["slug"]] = obj # Authors. authors: list[Author] = [] for a in SAMPLE_AUTHORS: # Institution.country now stores full names — canonicalise the seed code # ("UG") to "Uganda" so the lookup resolves instead of silently matching nothing. institution = Institution.objects.filter(country=to_country_name(a["country"])).first() # Look up by ORCID when present (unique), otherwise by name — # multiple NULL ORCIDs would otherwise collide on update_or_create. if a["orcid"]: lookup = {"orcid": a["orcid"]} else: lookup = { "first_name": a["first_name"], "last_name": a["last_name"], "orcid__isnull": True, } author, _ = Author.objects.update_or_create( **lookup, defaults={ "first_name": a["first_name"], "last_name": a["last_name"], "orcid": a["orcid"], "institution": institution, "affiliation_text": institution.name if institution else "", "verified_at": timezone.now() if a["orcid"] else None, }, ) authors.append(author) # Seed / reuse a REPO_MANAGER + submitter. submitter, _ = User.objects.get_or_create( email="repo.submitter@ruforum.example", defaults={ "first_name": "Repo", "last_name": "Submitter", "role": UserRole.SCHOLAR, "is_active": True, }, ) manager, _ = User.objects.get_or_create( email="repo.manager@ruforum.example", defaults={ "first_name": "Repo", "last_name": "Manager", "role": UserRole.REPO_MANAGER, "is_active": True, }, ) for col in col_objs.values(): col.managed_by.add(manager) created = 0 skipped = 0 templates = DOCUMENT_TEMPLATES.copy() # Produce up to `count` distinct documents by cycling templates with suffixes. for i in range(count): tpl = templates[i % len(templates)] kind, title, abstract, license_type, collection_slug, year = tpl title_variant = title if i < len(templates) else f"{title} (vol. {i // len(templates) + 1})" collection = col_objs[collection_slug] try: doc = services.ingest_document( file=_dummy_file(title_variant, i), title=title_variant, collection=collection, document_type=kind, abstract=abstract, license_type=license_type, year=year, authors=random.sample(authors, k=min(3, len(authors))), submitted_by=submitter, source_system="seed", actor=submitter, ) except services.DuplicateDocumentError: skipped += 1 continue # Randomly advance the document through the QA lifecycle. outcome = random.choices( ["published", "under_qa", "revision", "rejected", "withdrawn"], weights=[60, 20, 5, 5, 10], )[0] services.submit_for_qa(doc, actor=submitter) qa = QARecord.objects.filter(document=doc).first() if qa is None: continue if outcome == "published": services.decide_qa( qa, decision=QARecord.Decision.APPROVE, actor=manager, checklist=services.full_qa_checklist(), ) elif outcome == "revision": services.decide_qa(qa, decision=QARecord.Decision.REVISION, comments="Please expand methods.", actor=manager) elif outcome == "rejected": services.decide_qa( qa, decision=QARecord.Decision.REJECT, comments="Out of scope.", actor=manager ) elif outcome == "withdrawn": services.decide_qa( qa, decision=QARecord.Decision.APPROVE, actor=manager, checklist=services.full_qa_checklist(), ) services.withdraw_document(doc, reason="Seeded withdrawal for testing.", actor=manager) created += 1 return { "collections": len(col_objs), "authors": len(authors), "documents_created": created, "documents_skipped_duplicate": skipped, }