# Copyright The IETF Trust 2010-2026, All Rights Reserved

# -*- coding: utf-8 -*-

from collections import namedtuple

import datetime

import logging

import os

import django.db

import rfc2html

from io import BufferedReader

from pathlib import Path

from django.core.exceptions import ValidationError

from django.db.models import Q

from lxml import etree

from typing import Optional, Protocol, TYPE_CHECKING, Union

from weasyprint import HTML as wpHTML

from weasyprint.text.fonts import FontConfiguration

from django.db import models

from django.core import checks

from django.core.files.base import File

from django.core.cache import caches

from django.core.validators import (

URLValidator,

RegexValidator,

ProhibitNullCharactersValidator,

)

from django.urls import reverse as urlreverse

from django.contrib.contenttypes.models import ContentType

from django.conf import settings

from django.utils import timezone

from django.utils.encoding import force_str

from django.utils.html import mark_safe # type:ignore

from django.contrib.staticfiles import finders

import debug # pyflakes:ignore

from ietf.group.models import Group

from ietf.doc.storage_utils import (

store_str as utils_store_str,

store_bytes as utils_store_bytes,

store_file as utils_store_file

)

from ietf.name.models import ( DocTypeName, DocTagName, StreamName, IntendedStdLevelName, StdLevelName,

DocRelationshipName, DocReminderTypeName, BallotPositionName, ReviewRequestStateName, ReviewAssignmentStateName, FormalLanguageName,

DocUrlTagName, ExtResourceName)

from ietf.person.models import Email, Person

from ietf.person.utils import get_active_balloters

from ietf.utils import log

from ietf.utils.decorators import memoize

from ietf.utils.text import decode_document_content

from ietf.utils.validators import validate_no_control_chars

from ietf.utils.mail import formataddr

from ietf.utils.models import ForeignKey

from ietf.utils.timezone import date_today, RPC_TZINFO, DEADLINE_TZINFO

if TYPE_CHECKING:

# importing other than for type checking causes errors due to cyclic imports

from ietf.meeting.models import ProceedingsMaterial, Session

logger = logging.getLogger('django')

class StateType(models.Model):

slug = models.CharField(primary_key=True, max_length=30) # draft, draft-iesg, charter, ...

label = models.CharField(max_length=255, help_text="Label that should be used (e.g. in admin) for state drop-down for this type of state") # State, IESG state, WG state, ...

def __str__(self):

return self.slug

@checks.register('db-consistency')

def check_statetype_slugs(app_configs, **kwargs):

errors = []

try:

state_type_slugs = [ t.slug for t in StateType.objects.all() ]

except django.db.ProgrammingError:

# When running initial migrations on an empty DB, attempting to retrieve StateType will raise a

# ProgrammingError. Until Django 3, there is no option to skip the checks.

return []

else:

for type in DocTypeName.objects.all():

if not type.slug in state_type_slugs:

errors.append(checks.Error(

"The document type '%s (%s)' does not have a corresponding entry in the doc.StateType table" % (type.name, type.slug),

hint="You should add a doc.StateType entry with a slug '%s' to match the DocTypeName slug."%(type.slug),

obj=type,

id='datatracker.doc.E0015',

))

return errors

class State(models.Model):

type = ForeignKey(StateType)

slug = models.SlugField()

name = models.CharField(max_length=255)

used = models.BooleanField(default=True)

desc = models.TextField(blank=True)

order = models.IntegerField(default=0)

next_states = models.ManyToManyField('doc.State', related_name="previous_states", blank=True)

def __str__(self):

return self.name

class Meta:

ordering = ["type", "order"]

IESG_BALLOT_ACTIVE_STATES = ("lc", "writeupw", "goaheadw", "iesg-eva", "defer")

IESG_CHARTER_ACTIVE_STATES = ("intrev", "extrev", "iesgrev")

IESG_STATCHG_CONFLREV_ACTIVE_STATES = ("iesgeval", "defer")

IESG_SUBSTATE_TAGS = ('ad-f-up', 'need-rev', 'extpty')

def validate_doc_keywords(value):

if (

not isinstance(value, list | tuple | set)

or not all(isinstance(elt, str) for elt in value)

):

raise ValidationError("Value must be an array of strings")

class DocumentInfo(models.Model):

"""Any kind of document. Draft, RFC, Charter, IPR Statement, Liaison Statement"""

time = models.DateTimeField(default=timezone.now) # should probably have auto_now=True

type = ForeignKey(DocTypeName, blank=True, null=True) # Draft, Agenda, Minutes, Charter, Discuss, Guideline, Email, Review, Issue, Wiki, External ...

title = models.CharField(

max_length=255,

validators=[

ProhibitNullCharactersValidator(),

validate_no_control_chars,

],

)

states = models.ManyToManyField(State, blank=True) # plain state (Active/Expired/...), IESG state, stream state

tags = models.ManyToManyField(DocTagName, blank=True) # Revised ID Needed, ExternalParty, AD Followup, ...

stream = ForeignKey(StreamName, blank=True, null=True) # IETF, IAB, IRTF, Independent Submission, Editorial

group = ForeignKey(Group, blank=True, null=True) # WG, RG, IAB, IESG, Edu, Tools

abstract = models.TextField(blank=True)

rev = models.CharField(verbose_name="revision", max_length=16, blank=True)

pages = models.IntegerField(blank=True, null=True)

words = models.IntegerField(blank=True, null=True)

formal_languages = models.ManyToManyField(FormalLanguageName, blank=True, help_text="Formal languages used in document")

intended_std_level = ForeignKey(IntendedStdLevelName, verbose_name="Intended standardization level", blank=True, null=True)

std_level = ForeignKey(StdLevelName, verbose_name="Standardization level", blank=True, null=True)

ad = ForeignKey(Person, verbose_name="area director", related_name='ad_%(class)s_set', blank=True, null=True)

shepherd = ForeignKey(Email, related_name='shepherd_%(class)s_set', blank=True, null=True)

expires = models.DateTimeField(blank=True, null=True)

notify = models.TextField(max_length=1023, blank=True)

external_url = models.URLField(blank=True)

uploaded_filename = models.TextField(blank=True)

note = models.TextField(blank=True)

rfc_number = models.PositiveIntegerField(blank=True, null=True) # only valid for type="rfc"

keywords = models.JSONField(

default=list,

max_length=1000,

validators=[validate_doc_keywords],

blank=True,

)

@property

def doi(self) -> str | None:

if self.type_id == "rfc" and self.rfc_number is not None:

return f"{settings.IETF_DOI_PREFIX}/RFC{self.rfc_number}"

return None

def file_extension(self):

if not hasattr(self, '_cached_extension'):

if self.uploaded_filename:

_, ext= os.path.splitext(self.uploaded_filename)

self._cached_extension = ext.lstrip(".").lower()

else:

self._cached_extension = "txt"

return self._cached_extension

def get_file_path(self):

if not hasattr(self, '_cached_file_path'):

if self.type_id == "rfc":

self._cached_file_path = settings.RFC_PATH

elif self.type_id == "draft":

if self.is_dochistory():

self._cached_file_path = settings.INTERNET_ALL_DRAFTS_ARCHIVE_DIR

else:

# This could be simplified since anything in INTERNET_DRAFT_PATH is also already in INTERNET_ALL_DRAFTS_ARCHIVE_DIR

draft_state = self.get_state('draft')

if draft_state and draft_state.slug == 'active':

self._cached_file_path = settings.INTERNET_DRAFT_PATH

else:

self._cached_file_path = settings.INTERNET_ALL_DRAFTS_ARCHIVE_DIR

elif self.meeting_related() and self.type_id in (

"agenda", "minutes", "narrativeminutes", "slides", "bluesheets", "procmaterials", "chatlog", "polls"

):

meeting = self.get_related_meeting()

if meeting is not None:

self._cached_file_path = os.path.join(meeting.get_materials_path(), self.type_id) + "/"

else:

self._cached_file_path = ""

elif self.type_id == "charter":

self._cached_file_path = settings.CHARTER_PATH

elif self.type_id == "conflrev":

self._cached_file_path = settings.CONFLICT_REVIEW_PATH

elif self.type_id == "statchg":

self._cached_file_path = settings.STATUS_CHANGE_PATH

elif self.type_id == "bofreq": # TODO: This is probably unneeded, as is the separate path setting

self._cached_file_path = settings.BOFREQ_PATH

else:

self._cached_file_path = settings.DOCUMENT_PATH_PATTERN.format(doc=self)

return self._cached_file_path

def get_base_name(self):

if not hasattr(self, '_cached_base_name'):

if self.uploaded_filename:

self._cached_base_name = self.uploaded_filename

elif self.type_id == 'rfc':

self._cached_base_name = "%s.txt" % self.name

elif self.type_id == 'draft':

if self.is_dochistory():

self._cached_base_name = "%s-%s.txt" % (self.doc.name, self.rev)

else:

self._cached_base_name = "%s-%s.txt" % (self.name, self.rev)

elif self.type_id in ["slides", "agenda", "minutes", "bluesheets", "procmaterials", ] and self.meeting_related():

ext = 'pdf' if self.type_id == 'procmaterials' else 'txt'

self._cached_base_name = f'{self.name}-{self.rev}.{ext}'

elif self.type_id == 'review':

# TODO: This will be wrong if a review is updated on the same day it was created (or updated more than once on the same day)

self._cached_base_name = "%s.txt" % self.name

elif self.type_id in ['bofreq', 'statement']:

self._cached_base_name = "%s-%s.md" % (self.name, self.rev)

else:

if self.rev:

self._cached_base_name = "%s-%s.txt" % (self.name, self.rev)

else:

self._cached_base_name = "%s.txt" % (self.name, )

return self._cached_base_name

def get_file_name(self):

if not hasattr(self, '_cached_file_name'):

self._cached_file_name = os.path.join(self.get_file_path(), self.get_base_name())

return self._cached_file_name

def revisions_by_dochistory(self):

revisions = []

if self.type_id != "rfc":

for h in self.history_set.order_by("time", "id"):

if h.rev and not h.rev in revisions:

revisions.append(h.rev)

if not self.rev in revisions:

revisions.append(self.rev)

return revisions

def revisions_by_newrevisionevent(self):

revisions = []

if self.type_id != "rfc":

doc = self.doc if isinstance(self, DocHistory) else self

for e in doc.docevent_set.filter(type='new_revision').distinct():

if e.rev and not e.rev in revisions:

revisions.append(e.rev)

if not doc.rev in revisions:

revisions.append(doc.rev)

revisions.sort()

return revisions

def get_href(self, meeting=None):

return self._get_ref(meeting=meeting, versioned=True)

def get_versionless_href(self, meeting=None):

return self._get_ref(meeting=meeting, versioned=False)

def _get_ref(self, meeting=None, versioned=True):

"""

Returns an url to the document text. This differs from .get_absolute_url(),

which returns an url to the datatracker page for the document.

"""

# If self.external_url truly is an url, use it. This is a change from

# the earlier resolution order, but there's at the moment one single

# instance which matches this (with correct results), so we won't

# break things all over the place.

cache_attr = "_cached_href" if versioned else "_cached_versionless_href"

if not hasattr(self, cache_attr):

validator = URLValidator()

if self.external_url and self.external_url.split(':')[0] in validator.schemes:

validator(self.external_url)

return self.external_url

meeting_doc_refs = (

settings.MEETING_DOC_HREFS if versioned else settings.MEETING_DOC_GREFS

)

if self.type_id in settings.DOC_HREFS and self.type_id in meeting_doc_refs:

if self.meeting_related():

self.is_meeting_related = True

format = meeting_doc_refs[self.type_id]

else:

self.is_meeting_related = False

format = settings.DOC_HREFS[self.type_id]

elif self.type_id in settings.DOC_HREFS:

self.is_meeting_related = False

if self.type_id == "rfc":

format = settings.DOC_HREFS['rfc']

else:

format = settings.DOC_HREFS[self.type_id]

elif self.type_id in meeting_doc_refs:

self.is_meeting_related = True

else:

return None

if self.is_meeting_related:

if not meeting:

meeting = self.get_related_meeting()

if meeting is None:

return ''

# After IETF 96, meeting materials acquired revision

# handling, and the document naming changed.

if meeting.proceedings_format_version == 1:

format = settings.MEETING_DOC_OLD_HREFS[self.type_id]

else:

# This branch includes interims

format = meeting_doc_refs[self.type_id]

info = dict(doc=self, meeting=meeting)

else:

info = dict(doc=self)

href = format.format(**info)

# For slides that are not meeting-related, we need to know the file extension.

# Assume we have access to the same files as settings.DOC_HREFS["slides"] and

# see what extension is available

if self.type_id == "slides" and not self.meeting_related() and not href.endswith("/"):

filepath = Path(self.get_file_path()) / self.get_base_name() # start with this

if not filepath.exists():

# Look for other extensions - grab the first one, sorted for stability

for existing in sorted(filepath.parent.glob(f"{filepath.stem}.*")):

filepath = filepath.with_suffix(existing.suffix)

break

href += filepath.suffix # tack on the extension

if href.startswith('/'):

href = settings.IDTRACKER_BASE_URL + href

setattr(self, cache_attr, href)

return getattr(self, cache_attr)

def refresh_from_db(self, using=None, fields=None, **kwargs):

super().refresh_from_db(using=using, fields=fields, **kwargs)

self.state_cache = None

self._cached_state_slug = {}

def set_state(self, state):

"""Switch state type implicit in state to state. This just

sets the state, doesn't log the change."""

already_set = self.states.filter(type=state.type)

others = [s for s in already_set if s != state]

if others:

self.states.remove(*others)

if state not in already_set:

self.states.add(state)

if state.type and state.type.slug == 'draft-iesg':

iesg_state = self.states.get(type_id="draft-iesg") # pyflakes:ignore

log.assertion('iesg_state', note="A document's 'draft-iesg' state should never be unset'. Failed for %s"%self.name)

self.state_cache = None # invalidate cache

self._cached_state_slug = {}

def unset_state(self, state_type):

"""Unset state of type so no state of that type is any longer set."""

log.assertion('state_type != "draft-iesg"')

self.states.remove(*self.states.filter(type=state_type))

self.state_cache = None # invalidate cache

self._cached_state_slug = {}

def get_state(self, state_type=None):

"""Get state of type, or default state for document type if

not specified. Uses a local cache to speed multiple state

reads up."""

if self.pk == None: # states is many-to-many so not in database implies no state

return None

if state_type == None:

state_type = self.type_id

if not hasattr(self, "state_cache") or self.state_cache == None:

state_cache = {}

for s in self.states.all():

state_cache[s.type_id] = s

self.state_cache = state_cache

return self.state_cache.get(state_type, None)

def get_state_slug(self, state_type=None):

"""Get state of type, or default if not specified, returning

the slug of the state or None. This frees the caller of having

to check against None before accessing the slug for a

comparison."""

if not hasattr(self, '_cached_state_slug'):

self._cached_state_slug = {}

if not state_type in self._cached_state_slug:

s = self.get_state(state_type)

self._cached_state_slug[state_type] = s.slug if s else None

return self._cached_state_slug[state_type]

def rfc_editor_queue_status(self):

"""Human-readable RPC publication queue "Status" for the document, or None.

While a document is in the RFC Editor queue (draft-rfceditor state

"in_progress" or "blocked"), this is the status text pushed by the RFC

Production Center, matching what the queue website shows. Displays of the RFC

Editor state show it in place of the state name. It is None for a document whose

draft-rfceditor state predates the queue integration, which leaves those displays

showing the state name itself.

Costs a query per call. Tables rendering many documents replace this with a

precomputed value; see ietf.doc.utils_search.fill_in_rfc_editor_queue_status.

"""

if self.get_state_slug("draft-rfceditor") not in ("in_progress", "blocked"):

return None

event = self.latest_event(RpcAssignmentDocEvent, type="changed_rpc_assignments")

return event.assignments if event else None

def friendly_state(self, label_iesg_state=True):

""" Return a concise text description of the document's current state.

For a draft in the RFC Editor queue that description is "IESG: RFC Ed Queue",

labeled because displays of it sit next to the RFC Editor's own status for the

document. Every other state stands on its own and is returned unlabeled. Callers

rendering the description somewhere already labeled as the IESG's pass

label_iesg_state=False.

"""

state = self.get_state()

if not state:

return "Unknown state"

if self.type_id == "rfc":

return f"RFC {self.rfc_number} ({self.std_level})"

elif self.type_id == 'draft':

iesg_state = self.get_state("draft-iesg")

iesg_state_summary = None

if iesg_state:

iesg_substate = [t for t in self.tags.all() if t.slug in IESG_SUBSTATE_TAGS]

# There really shouldn't be more than one tag in iesg_substate, but this will do something sort-of-sensible if there is

iesg_state_summary = iesg_state.name

if iesg_substate:

iesg_state_summary = iesg_state_summary + "::"+"::".join(tag.name for tag in iesg_substate)

rfc = self.became_rfc()

if rfc:

return f"Became RFC {rfc.rfc_number} ({rfc.std_level})"

elif state.slug == "repl":

rs = self.related_that("replaces")

if rs:

return mark_safe("Replaced by " + ", ".join("<a href=\"%s\">%s</a>" % (urlreverse('ietf.doc.views_doc.document_main', kwargs=dict(name=related.name)), related) for related in rs))

else:

return "Replaced"

elif state.slug == "active":

if iesg_state:

if iesg_state.slug == "dead":

# Many drafts in the draft-iesg "Dead" state are not dead

# in other state machines; they're just not currently under

# IESG processing. Show them as "I-D Exists (IESG: Dead)" instead...

return "I-D Exists (IESG: %s)" % iesg_state_summary

elif iesg_state.slug == "lc":

e = self.latest_event(LastCallDocEvent, type="sent_last_call")

if e:

return iesg_state_summary + " (ends %s)" % e.expires.astimezone(DEADLINE_TZINFO).date().isoformat()

elif label_iesg_state and iesg_state.slug == "rfcqueue":

# The only state whose displays sit next to the RFC Editor's own

# status for the document, where a bare state name is ambiguous.

return "IESG: %s" % iesg_state_summary

return iesg_state_summary

else:

return "I-D Exists"

else:

if iesg_state and iesg_state.slug == "dead":

return state.name + " (IESG: %s)" % iesg_state_summary

# Expired/Withdrawn by Submitter/IETF

return state.name

else:

return state.name

def author_names(self):

"""Author names as a list of strings"""

names = []

if self.type_id == "rfc" and self.rfcauthor_set.exists():

for author in self.rfcauthor_set.select_related("person"):

if author.person:

names.append(author.person.name)

else:

# titlepage_name cannot be blank

names.append(author.titlepage_name)

else:

names = [

author.person.name

for author in self.documentauthor_set.select_related("person")

]

return names

def author_persons_or_names(self):

"""Authors as a list of named tuples with person and/or titlepage_name"""

Author = namedtuple("Author", "person titlepage_name")

persons_or_names = []

if self.type_id=="rfc" and self.rfcauthor_set.exists():

for author in self.rfcauthor_set.select_related("person"):

persons_or_names.append(Author(person=author.person, titlepage_name=author.titlepage_name))

else:

for author in self.documentauthor_set.select_related("person"):

persons_or_names.append(Author(person=author.person, titlepage_name=""))

return persons_or_names

def author_persons(self):

"""Authors as a list of Persons

Omits any RfcAuthors with a null person field.

"""

if self.type_id == "rfc" and self.rfcauthor_set.exists():

authors_qs = self.rfcauthor_set.filter(person__isnull=False)

else:

authors_qs = self.documentauthor_set.all()

return [a.person for a in authors_qs.select_related("person")]

def author_list(self):

"""List of author emails"""

if self.type_id == "rfc" and self.rfcauthor_set.exists():

author_qs = self.rfcauthor_set.select_related("person").order_by("order")

else:

author_qs = self.documentauthor_set.select_related("email").order_by(

"order"

)

best_addresses = []

for author in author_qs:

if author.email:

if author.email.active or not author.email.person:

best_addresses.append(author.email.address)

else:

best_addresses.append(author.email.person.email_address())

return ", ".join(best_addresses)

# This, and several other ballot related functions here, assume that there is only one active ballot for a document at any point in time.

# If that assumption is violated, they will only expose the most recently created ballot

def ballot_open(self, ballot_type_slug):

e = self.latest_event(BallotDocEvent, ballot_type__slug=ballot_type_slug)

return e if e and not e.type == "closed_ballot" else None

def latest_ballot(self):

"""Returns the most recently created ballot"""

ballot = self.latest_event(BallotDocEvent, type__in=("created_ballot", "closed_ballot"))

return ballot

def active_ballot(self):

"""Returns the most recently created ballot if it isn't closed."""

ballot = self.latest_ballot()

if ballot and ballot.type == "created_ballot":

return ballot

else:

return None

def has_rfc_editor_note(self):

e = self.latest_event(WriteupDocEvent, type="changed_rfc_editor_note_text")

return e != None and (e.text != "")

def meeting_related(self):

if self.type_id in ("agenda","minutes", "narrativeminutes", "bluesheets","slides","recording","procmaterials","chatlog","polls"):

return self.type_id != "slides" or self.get_state_slug('reuse_policy')=='single'

return False

def get_related_session(self) -> Optional['Session']:

"""Get the meeting session related to this document

Return None if there is no related session.

Must define this in DocumentInfo subclasses.

"""

raise NotImplementedError(f'Class {self.__class__} must define get_related_session()')

def get_related_proceedings_material(self) -> Optional['ProceedingsMaterial']:

"""Get the proceedings material related to this document

Return None if there is no related proceedings material.

Must define this in DocumentInfo subclasses.

"""

raise NotImplementedError(f'Class {self.__class__} must define get_related_proceedings_material()')

def get_related_meeting(self):

"""Get the meeting this document relates to"""

if not self.meeting_related():

return None # no related meeting if not meeting_related!

# get an item that links this doc to a meeting

item = self.get_related_session() or self.get_related_proceedings_material()

return getattr(item, 'meeting', None)

def relations_that(self, relationship):

"""Return the related-document objects that describe a given relationship targeting self."""

if isinstance(relationship, str):

relationship = ( relationship, )

if not isinstance(relationship, tuple):

raise TypeError("Expected a string or tuple, received %s" % type(relationship))

if isinstance(self, Document):

return RelatedDocument.objects.filter(target=self, relationship__in=relationship).select_related('source')

elif isinstance(self, DocHistory):

return RelatedDocHistory.objects.filter(target=self.doc, relationship__in=relationship).select_related('source')

else:

raise TypeError("Expected method called on Document or DocHistory")

def all_relations_that(self, relationship, related=None):

if not related:

related = tuple([])

rels = self.relations_that(relationship)

for r in rels:

if not r in related:

related += ( r, )

related = r.source.all_relations_that(relationship, related)

return related

def relations_that_doc(self, relationship):

"""Return the related-document objects that describe a given relationship from self to other documents."""

if isinstance(relationship, str):

relationship = ( relationship, )

if not isinstance(relationship, tuple):

raise TypeError("Expected a string or tuple, received %s" % type(relationship))

if isinstance(self, Document):

return RelatedDocument.objects.filter(source=self, relationship__in=relationship).select_related('target')

elif isinstance(self, DocHistory):

return RelatedDocHistory.objects.filter(source=self, relationship__in=relationship).select_related('target')

else:

raise TypeError("Expected method called on Document or DocHistory")

def all_relations_that_doc(self, relationship, related=None):

if not related:

related = tuple([])

rels = self.relations_that_doc(relationship)

for r in rels:

if not r in related:

related += ( r, )

related = r.target.all_relations_that_doc(relationship, related)

return related

def related_that(self, relationship):

# _cached_related_that is populated in bulk by callers that render many

# documents at once (see ietf.doc.utils_search.fill_in_document_relations);

# without it each document costs a query per relationship it displays.

cached = getattr(self, "_cached_related_that", None)

if cached is not None and relationship in cached:

return cached[relationship]

return list(set([x.source for x in self.relations_that(relationship)]))

def all_related_that(self, relationship, related=None):

return list(set([x.source for x in self.all_relations_that(relationship)]))

def related_that_doc(self, relationship):

cached = getattr(self, "_cached_related_that_doc", None)

if cached is not None and relationship in cached:

return cached[relationship]

return list(set([x.target for x in self.relations_that_doc(relationship)]))

def all_related_that_doc(self, relationship, related=None):

return list(set([x.target for x in self.all_relations_that_doc(relationship)]))

def replaces(self):

return self.related_that_doc("replaces")

def replaced_by(self):

return set([ r.document for r in self.related_that("replaces") ])

def _text_path(self):

path = self.get_file_name()

root, ext = os.path.splitext(path)

txtpath = root+'.txt'

if ext != '.txt' and os.path.exists(txtpath):

path = txtpath

return path

def text_exists(self):

path = Path(self._text_path())

return path.exists()

def text(self, size = -1):

path = Path(self._text_path())

if not path.exists():

return None

try:

with path.open('rb') as file:

raw = file.read(size)

except IOError as e:

log.log(f"Error reading text for {path}: {e}")

return None

return decode_document_content(raw)

def text_or_error(self):

return self.text() or "Error; cannot read '%s'"%self.get_base_name()

def html_body(self, classes=""):

if self.type_id == "rfc":

try:

html = Path(

os.path.join(settings.RFC_PATH, self.name + ".html")

).read_text()

except (IOError, UnicodeDecodeError):

return None

else:

try:

html = Path(

os.path.join(

settings.INTERNET_ALL_DRAFTS_ARCHIVE_DIR,

self.name + "-" + self.rev + ".html",

)

).read_text()

except (IOError, UnicodeDecodeError):

return None

# If HTML was generated by rfc2html, do not return it. Caller

# will use htmlize() to use a more current rfc2html to

# generate an HTMLized version. TODO: There should be a

# better way to determine how an HTML format was generated.

if html.startswith("<pre>"):

return None

# get body

etree_html = etree.HTML(html)

if etree_html is None:

return None

body = etree_html.xpath("//body")[0]

body.tag = "div"

if classes:

body.attrib["class"] = classes

# remove things

for tag in ["script"]:

for t in body.xpath(f"//{tag}"):

t.getparent().remove(t)

html = etree.tostring(body, encoding=str, method="html")

return html

def htmlized(self):

name = self.get_base_name()

text = self.text()

if name.endswith('.html'):

return text

if not name.endswith('.txt'):

return None

html = ""

if text:

cache = caches['htmlized']

cache_key = name.split('.')[0]

try:

html = cache.get(cache_key)

except EOFError:

html = None

if not html:

# The path here has to match the urlpattern for htmlized

# documents in order to produce correct intra-document links

html = rfc2html.markup(text, path=settings.HTMLIZER_URL_PREFIX)

html = f'<div class="rfcmarkup">{html}</div>'

if html:

cache.set(cache_key, html, settings.HTMLIZER_CACHE_TIME)

return html

def pdfized(self):

name = self.get_base_name()

text = self.html_body(classes="rfchtml")

stylesheets = [finders.find("ietf/css/document_html_referenced.css")]

if text:

stylesheets.append(finders.find("ietf/css/document_html_txt.css"))

else:

text = self.htmlized()

stylesheets.append(f'{settings.STATIC_IETF_ORG_INTERNAL}/fonts/noto-sans-mono/import.css')

cache = caches["pdfized"]

cache_key = name.split(".")[0]

try:

pdf = cache.get(cache_key)

except EOFError:

pdf = None

if not pdf:

try:

font_config = FontConfiguration()

pdf = wpHTML(

string=text, base_url=settings.IDTRACKER_BASE_URL

).write_pdf(

stylesheets=stylesheets,

font_config=font_config,

presentational_hints=True,

optimize_images=True,

)

except AssertionError:

pdf = None

except Exception as e:

log.log('weasyprint failed:'+str(e))

raise

if pdf:

cache.set(cache_key, pdf, settings.PDFIZER_CACHE_TIME)

return pdf

def references(self):

return self.relations_that_doc(('refnorm','refinfo','refunk','refold'))

def referenced_by(self):

return self.relations_that(("refnorm", "refinfo", "refunk", "refold")).filter(

models.Q(

source__type__slug="draft",

source__states__type__slug="draft",

source__states__slug="active",

)

| models.Q(source__type__slug="rfc")

).distinct()

def referenced_by_rfcs(self):

"""Get refs to this doc from RFCs"""

return self.relations_that(("refnorm", "refinfo", "refunk", "refold")).filter(

source__type__slug="rfc"

)

def became_rfc(self):

if not hasattr(self, "_cached_became_rfc"):

doc = self if isinstance(self, Document) else self.doc

self._cached_became_rfc = next(iter(doc.related_that_doc("became_rfc")), None)

return self._cached_became_rfc

def came_from_draft(self):

if not hasattr(self, "_cached_came_from_draft"):

doc = self if isinstance(self, Document) else self.doc

self._cached_came_from_draft = next(iter(doc.related_that("became_rfc")), None)

return self._cached_came_from_draft

def contains(self):

return self.related_that_doc("contains")

def part_of(self):

return self.related_that("contains")

def referenced_by_rfcs_as_rfc_or_draft(self):

"""Get refs to this doc, or a draft/rfc it came from, from an RFC"""

refs_to = self.referenced_by_rfcs()

if self.type_id == "rfc" and self.came_from_draft():

refs_to |= self.came_from_draft().referenced_by_rfcs()

return refs_to

def sent_to_rfc_editor_event(self):

if self.stream_id == "ietf":

return self.docevent_set.filter(type="iesg_approved").order_by("-time").first()

elif self.stream_id in ["editorial", "iab", "irtf", "ise"]:

return self.docevent_set.filter(type="requested_publication").order_by("-time").first()

else:

return None

class Meta:

abstract = True

class HasNameRevAndTypeIdProtocol(Protocol):

"""Typing Protocol describing a class that has name, rev, and type_id properties"""

@property

def name(self) -> str: ...

@property

def rev(self) -> str: ...

@property

def type_id(self) -> str: ...

class StorableMixin:

"""Mixin that adds storage helpers to a DocumentInfo subclass"""

def store_str(

self: HasNameRevAndTypeIdProtocol,

name: str,

content: str,

allow_overwrite: bool = False

) -> None:

return utils_store_str(self.type_id, name, content, allow_overwrite, self.name, self.rev)

def store_bytes(

self: HasNameRevAndTypeIdProtocol,

name: str,

content: bytes,

allow_overwrite: bool = False,

doc_name: Optional[str] = None,

doc_rev: Optional[str] = None

) -> None:

return utils_store_bytes(self.type_id, name, content, allow_overwrite, self.name, self.rev)

def store_file(

self: HasNameRevAndTypeIdProtocol,

name: str,

file: Union[File, BufferedReader],

allow_overwrite: bool = False,

doc_name: Optional[str] = None,

doc_rev: Optional[str] = None

) -> None:

return utils_store_file(self.type_id, name, file, allow_overwrite, self.name, self.rev)

STATUSCHANGE_RELATIONS = ('tops','tois','tohist','toinf','tobcp','toexp')

class RelatedDocument(models.Model):

source = ForeignKey('Document')

target = ForeignKey('Document', related_name='targets_related')

relationship = ForeignKey(DocRelationshipName)

originaltargetaliasname = models.CharField(max_length=255, null=True, blank=True)

def action(self):

return self.relationship.name

def __str__(self):

return u"%s %s %s" % (self.source.name, self.relationship.name.lower(), self.target.name)

def is_downref(self):

if self.source.type_id not in ["draft","rfc"] or self.relationship.slug not in [

"refnorm",

"refold",

"refunk",

]:

return None

if self.source.type_id == "rfc":

source_lvl = self.source.std_level_id

elif self.source.type_id in ["bcp","std"]:

source_lvl = self.source.type_id

else:

source_lvl = self.source.intended_std_level_id

if source_lvl not in ["bcp", "ps", "ds", "std", "unkn"]:

return None

if self.target.type_id == 'rfc':

if not self.target.std_level:

target_lvl = 'unkn'

else:

target_lvl = self.target.std_level_id

elif self.target.type_id in ["bcp", "std"]:

target_lvl = self.target.type_id

else:

if not self.target.intended_std_level:

target_lvl = 'unkn'

else:

target_lvl = self.target.intended_std_level_id

if self.relationship.slug not in ["refnorm", "refunk"]:

return None

if source_lvl in ["inf", "exp"]:

return None

pos_downref = (

"Downref" if self.relationship_id != "refunk" else "Possible Downref"

)

if source_lvl in ["bcp", "ps", "ds", "std"] and target_lvl in ["inf", "exp"]:

return pos_downref

if source_lvl == "ds" and target_lvl == "ps":

return pos_downref

if source_lvl == "std" and target_lvl in ["ps", "ds"]:

return pos_downref

if source_lvl not in ["inf", "exp"] and target_lvl == "unkn":

return "Possible Downref"

if source_lvl == "unkn" and target_lvl in ["ps", "ds"]:

return "Possible Downref"

return None

def is_approved_downref(self):

if self.target.type_id == 'rfc':

if RelatedDocument.objects.filter(relationship_id='downref-approval', target=self.target).exists():

return "Approved Downref"

return False

class RfcAuthor(models.Model):

"""Captures the authors of an RFC as represented on the RFC title page.

This deviates from DocumentAuthor in that it does not get moved into the DocHistory

hierarchy as documents are saved. It will attempt to preserve email, country, and affiliation

from the DocumentAuthor objects associated with the draft leading to this RFC (which

may be wrong if the author moves or changes affiliation while the document is in the

queue).

It does not, at this time, attempt to capture the authors from anything _but_ the title

page. The datatracker may know more about such authors based on information from the draft

leading to the RFC, and future work may take that into account.

Once doc.rfcauthor_set.exists() for a doc of type `rfc`, doc.documentauthor_set should be

ignored.

"""

document = ForeignKey(

"Document",

on_delete=models.CASCADE,

limit_choices_to={"type_id": "rfc"}, # only affects ModelForms (e.g., admin)

)

titlepage_name = models.CharField(max_length=128, blank=False)

is_editor = models.BooleanField(default=False)

person = ForeignKey(Person, null=True, blank=True, on_delete=models.PROTECT)

affiliation = models.CharField(max_length=100, blank=True, help_text="Organization/company used by author for submission")

country = models.CharField(max_length=255, blank=True, help_text="Country used by author for submission")

order = models.IntegerField(default=1)

def __str__(self):

return u"%s %s (%s)" % (self.document.name, self.person, self.order)