learning-website-django1-4  Exercise 2: The Importer and the import_content Command
================================================================================

Write the importer: walk the content folder, parse each file, and create, update
or skip each page in the database, reporting what changed. Wrap it in a
management command so it runs as  python manage.py import_content.

First add three fields and a helper to the Page model (the migration is created
for you):

  content_hash = models.CharField(max_length=64, blank=True)    # sha256 of the file
  fragment = models.TextField(blank=True)                       # the page body
  imported_at = models.DateTimeField(null=True, blank=True)
  objects = PageQuerySet.as_manager()     # with light() = defer("fragment")
  html property:  return mark_safe(self.fragment)               # the one place it is marked safe

The model now (apps/content/models.py):

from django.core.exceptions import ValidationError
from django.db import models
from django.utils.safestring import mark_safe

from config.sites_config import NoSiteError, SITES, site_for_path

SITE_CHOICES = [(name, info["title"]) for name, info in SITES.items()]


def validate_content_path(value):
    """A content path is relative, uses forward slashes, ends in .html and never climbs out of the tree."""
    if "\\" in value or value.startswith("/") or ".." in value.split("/"):
        raise ValidationError("A content path must be relative, use forward slashes and contain no '..'.")
    if not value.endswith(".html"):
        raise ValidationError("A content path must end in .html.")


class Course(models.Model):
    """A folder of numbered chapters, such as hungary/hungarian-basic-3."""
    site = models.CharField(max_length=40, choices=SITE_CHOICES)
    folder = models.CharField(max_length=400, unique=True)
    name = models.CharField(max_length=200)
    course_no = models.PositiveSmallIntegerField(null=True, blank=True)

    class Meta:
        ordering = ["site", "folder"]

    def __str__(self):
        return self.name


class PageQuerySet(models.QuerySet):
    def light(self):
        """Leave out the page body. Use it for every list, menu and sitemap: a page's HTML averages
        22 KB and the largest is 0.77 MB, and a list of pages does not need it."""
        return self.defer("fragment")


class Page(models.Model):
    class Kind(models.TextChoices):
        COURSE_CHAPTER = "course_chapter", "Course chapter"
        SIDEBAR = "sidebar", "Sidebar page"
        LESSON = "lesson", "Standalone lesson or reference"
        FULL_PAGE = "full_page", "Complete HTML page"
        OTHER = "other", "Other"

    path = models.CharField(max_length=500, unique=True, validators=[validate_content_path])
    site = models.CharField(max_length=40, choices=SITE_CHOICES, db_index=True)
    kind = models.CharField(max_length=20, choices=Kind.choices)
    title = models.CharField(max_length=300)
    course = models.ForeignKey(Course, null=True, blank=True, on_delete=models.SET_NULL, related_name="pages")
    chapter_no = models.PositiveSmallIntegerField(null=True, blank=True)
    created = models.DateField(null=True, blank=True)
    updated = models.DateField(null=True, blank=True)

    # filled in by the importer (Chapter 4)
    content_hash = models.CharField(max_length=64, blank=True)      # sha256 of the file, to detect changes
    fragment = models.TextField(blank=True)                         # the page body, ready to show
    imported_at = models.DateTimeField(null=True, blank=True)

    objects = PageQuerySet.as_manager()

    class Meta:
        # chapter_no is a NUMBER, so chapter 2 sorts before chapter 10 (a text sort would not)
        ordering = ["course_id", "chapter_no", "path"]
        indexes = [models.Index(fields=["site", "kind"])]

    def __str__(self):
        return self.title

    @property
    def html(self):
        """The one place a stored fragment is marked safe to output. It is safe only because the
        fragment came from YOUR files via the importer: never store visitor-supplied text here."""
        return mark_safe(self.fragment)

    @property
    def url_path(self):
        """/hungary/hungarian-basic-3/hungarian_basic_conversation_3_1/  (the live site's address form)"""
        return "/" + self.path[: -len(".html")] + "/"

    def clean(self):
        super().clean()
        try:
            expected = site_for_path(self.path)
        except NoSiteError:
            raise ValidationError({"path": "This path belongs to no site."})
        if self.site != expected:
            raise ValidationError({"site": f"This path belongs to the {expected} site, not {self.site}."})

  python manage.py makemigrations content
  python manage.py migrate

Save as apps/content/importer.py:

"""Import the content folder into the database: create, update, skip unchanged, report missing."""
import hashlib
import os
from dataclasses import dataclass, field

from django.db import transaction
from django.utils import timezone

from config.sites_config import NoSiteError, site_for_path

from .fragments import MalformedDocument, UnsafePath, prepare_fragment, safe_join
from .models import Course, Page, validate_content_path
from .parsing import parse_page

MAX_BYTES = 2_000_000                       # the largest real page is 0.77 MB, so 2 MB is generous
EXCLUDED_PREFIXES = ("japan/japanese-language/reference-materials/kanji",)   # the archive copy, not routed
SKIP_DIRS = {"pdfs", "solutions"}
BATCH = 200

FIELDS = ["site", "kind", "title", "course", "chapter_no", "created", "updated",
          "content_hash", "fragment", "imported_at"]


@dataclass
class ImportResult:
    created: int = 0
    updated: int = 0
    unchanged: int = 0
    missing: list = field(default_factory=list)     # in the database, no longer on disk
    removed: int = 0
    errors: list = field(default_factory=list)      # (path, message)

    @property
    def seen(self):
        return self.created + self.updated + self.unchanged


def iter_content_paths(root):
    """Relative paths (forward slashes) of every page file that should be imported."""
    for dirpath, dirnames, filenames in os.walk(root):
        rel_dir = os.path.relpath(dirpath, root).replace(os.sep, "/")
        rel_dir = "" if rel_dir == "." else rel_dir
        if any(rel_dir == p or rel_dir.startswith(p + "/") for p in EXCLUDED_PREFIXES):
            dirnames[:] = []
            continue
        dirnames[:] = sorted(d for d in dirnames if d not in SKIP_DIRS)
        for name in sorted(filenames):
            if name.endswith(".html") and not name.endswith("_print.html") and not name.startswith("_"):
                yield f"{rel_dir}/{name}" if rel_dir else name


def import_tree(root, *, dry_run=False, prune=False, force=False, only_site=None):
    result = ImportResult()
    existing = {path: (pk, digest) for pk, path, digest in Page.objects.values_list("pk", "path", "content_hash")}
    seen_paths = set()
    to_create, to_update = [], []
    courses = {}

    for rel in iter_content_paths(root):
        try:
            validate_content_path(rel)
            if only_site and site_for_path(rel) != only_site:
                continue                                     # decided from the path alone: no file is read
            full = safe_join(root, rel)
            if os.path.getsize(full) > MAX_BYTES:
                raise ValueError(f"larger than {MAX_BYTES} bytes")
            data = open(full, "rb").read()
            raw = data.decode("utf-8")                       # a file that is not UTF-8 is an error, not a guess
            parsed = parse_page(raw, rel)
            if only_site and parsed.site != only_site:
                continue
            seen_paths.add(rel)
            digest = hashlib.sha256(data).hexdigest()
            if not force and rel in existing and existing[rel][1] == digest:
                result.unchanged += 1
                continue
            fragment = prepare_fragment(raw, rel)
        except (NoSiteError, MalformedDocument, UnsafePath, UnicodeDecodeError, ValueError, OSError) as exc:
            result.errors.append((rel, f"{type(exc).__name__}: {exc}"))
            continue

        course = None
        if parsed.course_folder and not dry_run:
            course = courses.get(parsed.course_folder)
            if course is None:
                course, _ = Course.objects.update_or_create(
                    folder=parsed.course_folder,
                    defaults={"site": parsed.site, "name": parsed.course_name or parsed.course_folder,
                              "course_no": parsed.course_no})
                courses[parsed.course_folder] = course

        values = dict(site=parsed.site, kind=parsed.kind, title=parsed.title[:300], course=course,
                      chapter_no=parsed.chapter_no, created=parsed.created, updated=parsed.updated,
                      content_hash=digest, fragment=fragment, imported_at=timezone.now())
        if rel in existing:
            result.updated += 1
            to_update.append((existing[rel][0], values))
        else:
            result.created += 1
            to_create.append(Page(path=rel, **values))

    on_disk_scope = Page.objects.all() if not only_site else Page.objects.filter(site=only_site)
    result.missing = sorted(set(on_disk_scope.values_list("path", flat=True)) - seen_paths)

    if dry_run:
        return result

    with transaction.atomic():
        Page.objects.bulk_create(to_create, batch_size=BATCH)
        pages = Page.objects.in_bulk([pk for pk, _ in to_update])
        changed = []
        for pk, values in to_update:
            page = pages[pk]
            for name, value in values.items():
                setattr(page, name, value)
            changed.append(page)
        Page.objects.bulk_update(changed, FIELDS, batch_size=BATCH)
        if prune and result.missing:
            result.removed = Page.objects.filter(path__in=result.missing).delete()[0]
    return result

Save as apps/content/management/commands/import_content.py (and create empty
__init__.py files in management/ and management/commands/):

import time

from django.conf import settings
from django.core.management.base import BaseCommand, CommandError

from apps.content.importer import import_tree
from config.sites_config import SITES


class Command(BaseCommand):
    help = "Import the content folder into the database (create, update, skip unchanged)."

    def add_arguments(self, parser):
        parser.add_argument("--root", default=None, help="content folder (default: settings.CONTENT_ROOT)")
        parser.add_argument("--dry-run", action="store_true", help="report what would change; write nothing")
        parser.add_argument("--prune", action="store_true", help="delete pages whose file no longer exists")
        parser.add_argument("--force", action="store_true", help="re-import every page, even if unchanged")
        parser.add_argument("--site", choices=sorted(SITES), help="only this site's pages")

    def handle(self, *args, **opts):
        root = opts["root"] or settings.CONTENT_ROOT
        started = time.perf_counter()
        result = import_tree(root, dry_run=opts["dry_run"], prune=opts["prune"],
                             force=opts["force"], only_site=opts["site"])
        seconds = time.perf_counter() - started

        mode = "DRY RUN (nothing written): " if opts["dry_run"] else ""
        self.stdout.write(f"{mode}created {result.created}, updated {result.updated}, "
                          f"unchanged {result.unchanged}, errors {len(result.errors)}  ({seconds:.1f}s)")
        if result.missing:
            action = f"removed {result.removed}" if result.removed else "kept (use --prune to remove)"
            self.stdout.write(f"in the database but not on disk: {len(result.missing)}, {action}")
            for path in result.missing[:5]:
                self.stdout.write(f"    {path}")
        for path, message in result.errors[:10]:
            self.stderr.write(f"  ERROR {path}: {message}")
        if result.errors:
            raise CommandError(f"{len(result.errors)} page(s) could not be imported")

Save as tests/test_importer.py:

import os
import tempfile
from pathlib import Path

from django.test import TestCase

from apps.content.importer import MAX_BYTES, import_tree, iter_content_paths
from apps.content.models import Course, Page

CHAPTER = """<!-- ============================================================
     Course: Hungarian Basic Conversation 3
     Chapter: {title}
     File: hungarian_basic_conversation_3_{n}.html
     Date Created: 2026-10-05   Date Updated: 2026-10-05
     ============================================================ -->
<div class="hu-lesson"><p>{body}</p><a href="solutions/ex{n}.txt">s</a></div>
"""


class ContentTree:
    """A small content folder in a temporary directory."""

    def __init__(self, testcase):
        self._dir = tempfile.TemporaryDirectory()
        testcase.addCleanup(self._dir.cleanup)
        self.root = Path(self._dir.name)

    def write(self, rel, text, binary=False):
        p = self.root / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        if binary:
            p.write_bytes(text)
        else:
            p.write_text(text, encoding="utf-8")

    def chapter(self, n, body="hello", title=None):
        self.write(f"hungary/hungarian-basic-3/hungarian_basic_conversation_3_{n}.html",
                   CHAPTER.format(n=n, body=body, title=title or f"Chapter {n}"))

    def remove(self, rel):
        (self.root / rel).unlink()


class ImportTests(TestCase):
    def setUp(self):
        self.tree = ContentTree(self)
        for n in (1, 2, 10):
            self.tree.chapter(n)

    def run_import(self, **kw):
        return import_tree(self.tree.root, **kw)

    def test_the_first_import_creates_pages_and_one_course(self):
        r = self.run_import()
        self.assertEqual((r.created, r.updated, r.unchanged, len(r.errors)), (3, 0, 0, 0))
        course = Course.objects.get()
        self.assertEqual((course.name, course.folder, course.course_no), ("Hungarian Basic Conversation 3", "hungary/hungarian-basic-3", 3))
        self.assertEqual([p.chapter_no for p in course.pages.all()], [1, 2, 10])

    def test_the_banner_is_removed_and_the_solution_link_is_rewritten(self):
        self.run_import()
        page = Page.objects.get(chapter_no=1)
        self.assertNotIn("<!--", page.fragment)
        self.assertIn('href="/hungary/hungarian-basic-3/solutions/ex1.txt"', page.fragment)
        self.assertEqual(page.title, "Chapter 1")

    def test_a_second_import_changes_nothing(self):
        self.run_import()
        r = self.run_import()
        self.assertEqual((r.created, r.updated, r.unchanged), (0, 0, 3))

    def test_an_edited_file_is_updated_and_the_others_are_left_alone(self):
        self.run_import()
        before = Page.objects.get(chapter_no=2).imported_at
        self.tree.chapter(1, body="changed text", title="A new title")
        r = self.run_import()
        self.assertEqual((r.created, r.updated, r.unchanged), (0, 1, 2))
        page = Page.objects.get(chapter_no=1)
        self.assertIn("changed text", page.fragment)
        self.assertEqual(page.title, "A new title")
        self.assertEqual(Page.objects.get(chapter_no=2).imported_at, before)

    def test_a_new_file_is_created(self):
        self.run_import()
        self.tree.chapter(3)
        r = self.run_import()
        self.assertEqual((r.created, r.unchanged), (1, 3))

    def test_a_deleted_file_is_reported_and_kept_unless_pruned(self):
        self.run_import()
        self.tree.remove("hungary/hungarian-basic-3/hungarian_basic_conversation_3_10.html")
        r = self.run_import()
        self.assertEqual(r.missing, ["hungary/hungarian-basic-3/hungarian_basic_conversation_3_10.html"])
        self.assertEqual(Page.objects.count(), 3)
        r = self.run_import(prune=True)
        self.assertEqual(r.removed, 1)
        self.assertEqual(Page.objects.count(), 2)

    def test_a_dry_run_writes_nothing(self):
        r = self.run_import(dry_run=True)
        self.assertEqual(r.created, 3)
        self.assertEqual((Page.objects.count(), Course.objects.count()), (0, 0))

    def test_force_reimports_everything(self):
        self.run_import()
        r = self.run_import(force=True)
        self.assertEqual((r.updated, r.unchanged), (3, 0))

    def test_one_bad_file_does_not_stop_the_others(self):
        self.tree.write("hungary/bad_utf8.html", b"<p>caf\xe9</p>", binary=True)          # not valid UTF-8
        self.tree.write("nonsense/page.html", "<p>belongs to no site</p>")
        self.tree.write("hungary/huge.html", "x" * (MAX_BYTES + 1))
        r = self.run_import()
        self.assertEqual(r.created, 3)
        problems = dict(r.errors)
        self.assertIn("UnicodeDecodeError", problems["hungary/bad_utf8.html"])
        self.assertIn("NoSiteError", problems["nonsense/page.html"])
        self.assertIn("larger than", problems["hungary/huge.html"])

    def test_a_non_ascii_file_name_imports_and_has_an_address(self):
        self.tree.write("japan/japanese-language/reference-materials/hiragana/hiragana_あ.html", "<p>a</p>")
        self.run_import()
        page = Page.objects.get(path__endswith="hiragana_あ.html")
        self.assertEqual(page.site, "languages")
        self.assertEqual(page.url_path, "/japan/japanese-language/reference-materials/hiragana/hiragana_あ/")

    def test_only_one_site_can_be_imported(self):
        self.tree.write("linux/vim/vim_1_1.html", "<!-- Course: Vim\n Chapter: Modes\n -->\n<p>x</p>")
        r = self.run_import(only_site="systems")
        self.assertEqual(r.created, 1)
        self.assertEqual(Page.objects.get().site, "systems")

    def test_print_variants_underscore_files_and_solutions_are_skipped(self):
        self.tree.write("hungary/x/page_print.html", "<p>p</p>")
        self.tree.write("hungary/x/_notes.html", "<p>n</p>")
        self.tree.write("hungary/x/solutions/s.html", "<p>s</p>")
        self.tree.write("japan/japanese-language/reference-materials/kanji/kanji_x.html", "<p>archive</p>")
        names = list(iter_content_paths(self.tree.root))
        self.assertEqual(len(names), 3)

    def test_the_page_body_is_marked_safe_in_one_place(self):
        self.run_import()
        self.assertEqual(type(Page.objects.first().html).__name__, "SafeString")

    def test_light_listings_leave_out_the_body(self):
        self.run_import()
        page = Page.objects.light().first()
        self.assertIn("fragment", page.get_deferred_fields())

Run everything (Django 6.1.2):

  python manage.py test tests

Found 76 test(s).
System check identified no issues (0 silenced).
Creating test database for alias 'default'...
............................................................................
----------------------------------------------------------------------
Ran 76 tests in 0.520s
OK
Destroying test database for alias 'default'...

How it behaves
--------------
- Change detection is a SHA-256 hash of the file's bytes. An unchanged file is
  skipped without parsing, and only changed pages are written. The tests check
  that editing one file updates exactly one row and leaves the others' import
  time alone.
- A file that is not valid UTF-8, is larger than 2 MB, belongs to no site, or has
  a body tag with no end is an ERROR for that file, listed at the end. The rest
  still import, and the command exits with an error code so a build can fail.
- A page whose file has gone is reported as missing and KEPT, unless you pass
  --prune. Deleting by default would turn a typo in the content path into an
  empty site.
- --dry-run does everything except write, --force re-imports every page, and
  --site imports one site's pages (decided from the path alone, so the other
  sites' files are never even read).
- Everything is written in one transaction with bulk_create and bulk_update, in
  batches of 200, so a failure half way leaves the database as it was.
- Listing code should use Page.objects.light(), which leaves out the body.

WHY THIS WORKS AS AN ANSWER
---------------------------
An importer you cannot rerun safely is a liability. Hashing makes the second run
a no-op, errors are isolated per file, destructive steps are opt-in, and the
whole thing is tested against a temporary content folder, including the
awkward cases (bad encoding, oversized files, a non-ASCII file name, a deleted
file).