Spaces:
Runtime error
Runtime error
| 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: "<resource 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 | |