from pathlib import Path import yaml from django.conf import settings from django.core.management.base import BaseCommand, CommandError from django.db import transaction from apps.resources.models import Resource, ResourceCheckpoint, SkillResource from apps.skills.models import Skill RESOURCES_YAML = Path(settings.BASE_DIR) / "seed_data" / "resources.yaml" # Optional sidecar: checkpoint module lists keyed by resource URL. Kept separate # so the curated resources.yaml (comments, ordering) stays pristine. Each entry: # - url: "" # source: jsonld # optional, default manual # titles: ["Module 1", ...] CHECKPOINTS_YAML = Path(settings.BASE_DIR) / "seed_data" / "checkpoints.yaml" REQUIRED_FIELDS = ("title", "provider", "url", "difficulty_level", "duration", "type", "rating", "skills") VALID_DIFFICULTY = {"BEGINNER", "INTERMEDIATE", "ADVANCED"} VALID_TYPE = {"VIDEO", "COURSE", "ARTICLE", "DOCS"} VALID_SOURCES = {value for value, _ in ResourceCheckpoint.SOURCE_CHOICES} class Command(BaseCommand): help = ( "Seed Resource + SkillResource rows from seed_data/resources.yaml. " "Run seed_initial_skills first so all referenced skills exist. " "Idempotent: resources are upserted by url; skill links upserted by " "(skill, resource). --prune deletes rows not present in the YAML." ) def add_arguments(self, parser): parser.add_argument( "--path", default=str(RESOURCES_YAML), help="Path to resources YAML (default: backend/seed_data/resources.yaml)", ) parser.add_argument( "--prune", action="store_true", help="Delete Resource rows whose url is not present in the YAML. " "Cascades SkillResource rows via FK.", ) parser.add_argument( "--checkpoints-path", default=str(CHECKPOINTS_YAML), help="Path to the checkpoints sidecar YAML (default: " "backend/seed_data/checkpoints.yaml). Missing file = skipped.", ) def handle(self, *args, **options): yaml_path = Path(options["path"]) if not yaml_path.exists(): raise CommandError(f"Resources YAML not found: {yaml_path}") with yaml_path.open(encoding="utf-8") as f: data = yaml.safe_load(f) entries = (data or {}).get("resources") or [] if not entries: raise CommandError("No resources found in YAML") self._validate_entries(entries) # Resolve skills up-front so a missing skill aborts before any writes. skill_names = { link["skill_name"] for entry in entries for link in entry["skills"] } skills_by_name = { s.skill_name: s for s in Skill.objects.filter(skill_name__in=skill_names) } missing = sorted(skill_names - set(skills_by_name)) if missing: raise CommandError( "Missing skills in DB (run `python manage.py seed_initial_skills` " "first): " + ", ".join(missing) ) created = updated = 0 links_created = links_updated = links_deleted = 0 resources_deleted = 0 checkpoints_created = 0 with transaction.atomic(): seen_urls: set[str] = set() for entry in entries: resource, was_created = Resource.objects.update_or_create( url=entry["url"], defaults={ "title": entry["title"], "provider": entry["provider"], "difficulty_level": entry["difficulty_level"], "duration": int(entry["duration"]), "type": entry["type"], "rating": float(entry["rating"]), }, ) seen_urls.add(entry["url"]) if was_created: created += 1 else: updated += 1 seen_skill_ids: set[int] = set() for link in entry["skills"]: skill = skills_by_name[link["skill_name"]] seen_skill_ids.add(skill.id) _, lc = SkillResource.objects.update_or_create( resource=resource, skill=skill, defaults={ "relevance_score": float( link.get("relevance_score", 1.0) ), }, ) if lc: links_created += 1 else: links_updated += 1 stale_links = SkillResource.objects.filter( resource=resource ).exclude(skill_id__in=seen_skill_ids) links_deleted += stale_links.count() stale_links.delete() # Checkpoints: create-if-absent only. Never delete/reindex # existing rows — UserCheckpointProgress FKs them with CASCADE, # so a re-seed must never clobber a learner's history. Editing a # populated list happens via the admin inline, not the seed. checkpoint_titles = [ str(t).strip() for t in (entry.get("checkpoints") or []) if str(t).strip() ] if checkpoint_titles and not resource.checkpoints.exists(): source = entry.get("checkpoint_source", "manual") ResourceCheckpoint.objects.bulk_create([ ResourceCheckpoint( resource=resource, order_index=i, title=title[:500], source=source, ) for i, title in enumerate(checkpoint_titles, start=1) ]) checkpoints_created += len(checkpoint_titles) # Apply the optional checkpoints.yaml sidecar (URL-keyed). Same # create-if-absent rule; resources are matched to already-upserted # rows by URL. checkpoints_created += self._apply_checkpoint_sidecar( Path(options["checkpoints_path"]) ) if options["prune"]: stale_resources = Resource.objects.exclude(url__in=seen_urls) resources_deleted = stale_resources.count() stale_resources.delete() msg = ( f"Resources: {created} created, {updated} updated" + (f", {resources_deleted} deleted" if options["prune"] else "") + f". SkillResource links: {links_created} created, " f"{links_updated} updated, {links_deleted} deleted" + f". Checkpoints: {checkpoints_created} created." ) self.stdout.write(self.style.SUCCESS(msg)) def _validate_entries(self, entries: list[dict]) -> None: urls_seen: set[str] = set() for i, entry in enumerate(entries, start=1): missing = [f for f in REQUIRED_FIELDS if f not in entry] if missing: raise CommandError( f"Resource #{i} ({entry.get('title', '?')!r}) missing " f"fields: {missing}" ) if entry["difficulty_level"] not in VALID_DIFFICULTY: raise CommandError( f"Resource #{i} invalid difficulty_level " f"{entry['difficulty_level']!r}" ) if entry["type"] not in VALID_TYPE: raise CommandError( f"Resource #{i} invalid type {entry['type']!r}" ) if entry["url"] in urls_seen: raise CommandError( f"Duplicate url in YAML: {entry['url']}" ) urls_seen.add(entry["url"]) if not entry["skills"]: raise CommandError( f"Resource #{i} ({entry['title']!r}) has no skills — " "every resource must link to ≥1 skill." ) # Optional `checkpoints` (list) + `checkpoint_source`. The create # path uses bulk_create (no model validation), so validate source # against the choices here, like difficulty/type above. if "checkpoints" in entry and not isinstance(entry["checkpoints"], list): raise CommandError( f"Resource #{i} ({entry['title']!r}) `checkpoints` must be a list." ) src = entry.get("checkpoint_source") if src is not None and src not in VALID_SOURCES: raise CommandError( f"Resource #{i} ({entry['title']!r}) invalid " f"checkpoint_source {src!r} (allowed: {sorted(VALID_SOURCES)})" ) def _apply_checkpoint_sidecar(self, path: Path) -> int: """Create checkpoints from the optional checkpoints sidecar YAML. Keyed by resource URL; create-if-absent (skips resources that already have checkpoints). A sidecar URL with no matching resource is skipped with a warning (a stale entry shouldn't abort the whole seed). Returns the number of checkpoint rows created. """ if not path.exists(): return 0 with path.open(encoding="utf-8") as f: data = yaml.safe_load(f) or {} entries = data.get("checkpoints") or [] created = 0 for i, entry in enumerate(entries, start=1): url = entry.get("url") titles = [str(t).strip() for t in (entry.get("titles") or []) if str(t).strip()] source = entry.get("source", "manual") if not url or not titles: raise CommandError( f"checkpoints sidecar entry #{i} needs a url and a non-empty " "titles list." ) if source not in VALID_SOURCES: raise CommandError( f"checkpoints sidecar entry #{i} ({url}) invalid source {source!r}." ) resource = Resource.objects.filter(url=url).first() if resource is None: self.stderr.write( f" checkpoints sidecar: no resource for url {url} — skipped." ) continue if resource.checkpoints.exists(): continue # create-if-absent: never clobber existing rows ResourceCheckpoint.objects.bulk_create([ ResourceCheckpoint( resource=resource, order_index=j, title=title[:500], source=source, ) for j, title in enumerate(titles, start=1) ]) created += len(titles) return created