#!/usr/bin/env python3
"""Build the Python docs for various branches and various languages.
Without any arguments builds docs for all active versions and
languages.
Environment variables for:
- `SENTRY_DSN` (Error reporting)
- `FASTLY_SERVICE_ID` / `FASTLY_TOKEN` (CDN purges)
- `PYTHON_DOCS_ENABLE_ANALYTICS` (Enable Plausible for online docs)
are read from the site configuration path for your platform
(/etc/xdg/docsbuild-scripts on linux) if available,
and can be overriden by writing a file to the user config dir
for your platform ($HOME/.config/docsbuild-scripts on linux).
The contents of the file is parsed as toml:
```toml
[env]
SENTRY_DSN = "https://0a0a0a0a0a0a0a0a0a0a0a@sentry.io/69420"
FASTLY_SERVICE_ID = "deadbeefdeadbeefdead"
FASTLY_TOKEN = "secureme!"
PYTHON_DOCS_ENABLE_ANALYTICS = "1"
```
Languages are stored in `config.toml` while versions are discovered
from peps.python.org (generated by `python-releases.toml`).
-q selects "quick build", which means to build only HTML.
Translations are fetched from GitHub repositories according to PEP
545. `--languages` allows selecting translations, like `--languages
en` to just build the English documents.
This script was originally created by Georg Brandl in March 2010.
Modified by Benjamin Peterson to do CDN cache invalidation.
Modified by Julien Palard to build translations.
"""
from __future__ import annotations
import argparse
import concurrent.futures
import dataclasses
import datetime as dt
import filecmp
import json
import logging
import logging.handlers
import os
import re
import shlex
import shutil
import stat
import subprocess
import sys
import venv
from bisect import bisect_left as bisect
from contextlib import contextmanager, suppress
from pathlib import Path
from string import Template
from time import perf_counter, sleep
from urllib.parse import urljoin
import jinja2
import platformdirs
import tomlkit
import urllib3
import zc.lockfile
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Collection, Iterator, Sequence, Set
from typing import Literal
try:
from os import EX_OK
from os import EX_SOFTWARE as EX_FAILURE
except ImportError:
EX_OK, EX_FAILURE = 0, 1
try:
import sentry_sdk
except ImportError:
sentry_sdk = None
HERE = Path(__file__).resolve().parent
@dataclasses.dataclass(frozen=True, slots=True)
class Versions:
_seq: Sequence[Version]
def __iter__(self) -> Iterator[Version]:
return iter(self._seq)
def __reversed__(self) -> Iterator[Version]:
return reversed(self._seq)
@classmethod
def from_json(cls, data: dict) -> Versions:
"""Load versions from the devguide's JSON representation."""
permitted = ", ".join(sorted(Version.STATUSES | Version.SYNONYMS.keys()))
versions = []
for name, release in data.items():
branch = release["branch"]
status = release["status"]
if status in Version.SKIP_STATUSES:
logging.info("Skipping %s with status %r", name, status)
continue
status = Version.SYNONYMS.get(status, status)
if status not in Version.STATUSES:
logging.warning(
"Saw invalid version status %r, expected to be one of %s. Context: %s",
status,
permitted,
release,
)
continue
versions.append(Version(name=name, status=status, branch_or_tag=branch))
return cls(sorted(versions, key=Version.as_tuple))
def filter(self, branches: Sequence[str] = ()) -> Sequence[Version]:
"""Filter the given versions.
If *branches* is given, only *versions* matching *branches* are returned.
Else all live versions are returned (this means no EOL and no
security-fixes branches).
"""
if branches:
branches = frozenset(branches)
return [v for v in self if {v.name, v.branch_or_tag} & branches]
return [v for v in self if v.status not in {"EOL", "security-fixes"}]
@property
def current_stable(self) -> Version:
"""Find the current stable CPython version."""
return max((v for v in self if v.status == "stable"), key=Version.as_tuple)
@property
def current_dev(self) -> Version:
"""Find the current CPython version in development."""
return max(self, key=Version.as_tuple)
@dataclasses.dataclass(frozen=True, kw_only=True, slots=True)
class Version:
"""Represents a CPython version and its documentation build dependencies."""
name: str
status: Literal[
"in development",
"pre-release",
"stable",
"security-fixes",
"EOL",
]
branch_or_tag: str
STATUSES = {
"in development",
"pre-release",
"stable",
"security-fixes",
"EOL",
}
# Statuses for versions we don't build docs for at all.
SKIP_STATUSES = {"planned"}
# Those synonyms map branch status vocabulary found in the devguide
# with our vocabulary.
SYNONYMS = {
"feature": "in development",
"bugfix": "stable",
"security": "security-fixes",
"end-of-life": "EOL",
"prerelease": "pre-release",
}
def __eq__(self, other: Version) -> bool:
return self.name == other.name
@property
def requirements(self) -> list[str]:
"""Generate the right requirements for this version.
Since CPython 3.8 a Doc/requirements.txt file can be used.
In case the Doc/requirements.txt is absent or wrong (a
sub-dependency broke), use this function to override it.
See https://github.com/python/cpython/issues/91294
See https://github.com/python/cpython/issues/91483
"""
dependencies = [
"-rrequirements.txt",
"jieba", # To improve zh search.
"PyStemmer~=2.2.0", # To improve performance for word stemming.
]
if self.as_tuple() >= (3, 11):
return dependencies
if self.as_tuple() >= (3, 8):
# Restore the imghdr module for Python 3.8-3.10.
# Use setuptools with pkg_resources
return dependencies + ["standard-imghdr", "setuptools None:
"""Build a venv for the specific Python version.
The venv is created at most once per run, reused by later builds
of the same version, and removed at the end of the run: reusing
a venv across runs can silently keep outdated packages, because
pip considers a requirement satisfied when the installed version
number matches, even if the requirement is a direct URL now
pointing at different code.
"""
venv_name = self.build_meta.venv_name
if self.select_output is not None:
# Never share a venv with a concurrent differently-selected
# build, which may recreate it mid-build.
venv_name += f"-{self.select_output}"
venv_path = self.build_root / venv_name
if venv_path in self.built_venvs:
self.venv = venv_path
return
requirements = list(self.build_meta.dependencies)
if self.includes_html:
# opengraph previews
requirements.append("matplotlib>=3")
venv.create(
venv_path,
symlinks=os.name != "nt",
with_pip=True,
clear=True,
upgrade_deps=True,
)
python = venv_path / "bin" / "python"
if (self.checkout / "Doc" / "pylock.toml").is_file():
requirements.remove("-rrequirements.txt")
run(
(python, "-m", "pip", "install", "-rpylock.toml"),
cwd=self.checkout / "Doc",
)
run(
(python, "-m", "pip", "install", self.theme, *requirements),
cwd=self.checkout / "Doc",
)
run((python, "-m", "pip", "freeze", "--all"))
self.built_venvs.add(venv_path)
self.venv = venv_path
def setup_indexsidebar(self) -> None:
"""Copy indexsidebar.html for Sphinx."""
tmpl_src = HERE / "templates"
tmpl_dst = self.checkout / "Doc" / "tools" / "templates"
dbv_path = tmpl_dst / "_docs_by_version.html"
shutil.copy(tmpl_src / "indexsidebar.html", tmpl_dst / "indexsidebar.html")
if not self.build_meta.is_eol:
dbv_path.write_bytes(self.docs_by_version_content)
else:
shutil.copy(tmpl_src / "_docs_by_version.html", dbv_path)
def copy_build_to_webroot(self, http: urllib3.PoolManager) -> None:
"""Copy a given build to the appropriate webroot with appropriate rights."""
logging.info("Publishing start.")
start_time = perf_counter()
self.www_root.mkdir(parents=True, exist_ok=True)
if not self.build_meta.is_translation:
target = self.www_root / self.build_meta.version
else:
language_dir = self.www_root / self.build_meta.language
language_dir.mkdir(parents=True, exist_ok=True)
chgrp(language_dir, group=self.group, recursive=True)
language_dir.chmod(0o775)
target = language_dir / self.build_meta.version
# Builds run concurrently but may publish the same language/version to
# the same directory, so serialise publishes per target.
# Contention is expected, but brief, so wait instead of dying.
with wait_for_lock(
HERE / f"publish-{self.build_meta.language}-{self.build_meta.version}.lock"
):
target.mkdir(parents=True, exist_ok=True)
try:
target.chmod(0o775)
except PermissionError as err:
logging.warning("Can't change mod of %s: %s", target, str(err))
chgrp(target, group=self.group, recursive=True)
changed = 0
if self.includes_html:
# Copy built HTML files to webroot (default /srv/docs.python.org)
changed += changed_files(
self.checkout / "Doc" / "build" / "html", target
)
logging.info("Copying HTML files to %s", target)
chgrp(
self.checkout / "Doc" / "build" / "html/",
group=self.group,
recursive=True,
)
chmod_make_readable(self.checkout / "Doc" / "build" / "html")
run((
"rsync",
"-a",
"--delete-delay",
"--filter",
"P archives/",
str(self.checkout / "Doc" / "build" / "html") + "/",
target,
))
dist_dir = self.checkout / "Doc" / "dist"
if dist_dir.is_dir():
# Copy archive files to /archives/
logging.debug("Copying dist files.")
chgrp(dist_dir, group=self.group, recursive=True)
chmod_make_readable(dist_dir)
archives_dir = target / "archives"
archives_dir.mkdir(parents=True, exist_ok=True)
archives_dir.chmod(
archives_dir.stat().st_mode | stat.S_IROTH | stat.S_IXOTH
)
chgrp(archives_dir, group=self.group)
changed += 1
for dist_file in dist_dir.iterdir():
shutil.copy2(dist_file, archives_dir / dist_file.name)
changed += 1
logging.info("%s files changed", changed)
if changed and not self.skip_cache_invalidation:
purge_surrogate_key(http, self.build_meta.slug)
logging.info(
"Publishing done (%s).", format_seconds(perf_counter() - start_time)
)
def should_rebuild(self, force: bool) -> str | Literal[False]:
state = self.load_state()
if not state:
logging.info("Should rebuild: no previous state found.")
return "no previous state"
cpython_sha = self.cpython_repo.run("rev-parse", "HEAD").stdout.strip()
if self.build_meta.is_translation:
translation_sha = self.translation_repo.run(
"rev-parse", "HEAD"
).stdout.strip()
if translation_sha != state["translation_sha"]:
logging.info(
"Should rebuild: new translations (from %s to %s)",
state["translation_sha"],
translation_sha,
)
return "new translations"
if cpython_sha != state["cpython_sha"]:
diff = self.cpython_repo.run(
"diff", "--name-only", state["cpython_sha"], cpython_sha
).stdout
if "Doc/" in diff or "Misc/NEWS.d/" in diff:
logging.info(
"Should rebuild: Doc/ has changed (from %s to %s)",
state["cpython_sha"],
cpython_sha,
)
return "Doc/ has changed"
if force:
logging.info("Should rebuild: forced.")
return "forced"
logging.info("Nothing changed, no rebuild needed.")
return False
def load_state(self) -> dict:
if self.select_output is not None:
state_file = self.build_root / f"state-{self.select_output}.toml"
else:
state_file = self.build_root / "state.toml"
try:
return tomlkit.loads(state_file.read_text(encoding="UTF-8"))[
f"/{self.build_meta.slug}/"
]
except (KeyError, FileNotFoundError):
return {}
def save_state(
self, build_start: dt.datetime, build_duration: float, trigger: str
) -> None:
"""Save current CPython sha1 and current translation sha1.
Using this we can deduce if a rebuild is needed or not.
"""
if self.select_output is not None:
state_file = self.build_root / f"state-{self.select_output}.toml"
else:
state_file = self.build_root / "state.toml"
try:
states = tomlkit.parse(state_file.read_text(encoding="UTF-8"))
except FileNotFoundError:
states = tomlkit.document()
key = f"/{self.build_meta.slug}/"
state = {
"last_build_start": build_start,
"last_build_duration": round(build_duration, 0),
"triggered_by": trigger,
"cpython_sha": self.cpython_repo.run("rev-parse", "HEAD").stdout.strip(),
}
if self.build_meta.is_translation:
state["translation_sha"] = self.translation_repo.run(
"rev-parse", "HEAD"
).stdout.strip()
states[key] = state
state_file.write_text(tomlkit.dumps(states), encoding="UTF-8")
table = tomlkit.inline_table()
table |= state
logging.info("Saved new rebuild state for %s: %s", key, table.as_string())
def chgrp(
path: Path,
/,
group: int | str | None,
*,
recursive: bool = False,
follow_symlinks: bool = True,
) -> None:
if sys.platform == "win32":
return
from grp import getgrnam
try:
try:
group_id = int(group)
except ValueError:
group_id = getgrnam(group)[2]
except (LookupError, TypeError, ValueError):
return
try:
os.chown(path, -1, group_id, follow_symlinks=follow_symlinks)
if recursive:
for p in path.rglob("*"):
os.chown(p, -1, group_id, follow_symlinks=follow_symlinks)
except OSError as err:
logging.warning("Can't change group of %s: %s", path, str(err))
def chmod_make_readable(path: Path, /, mode: int = stat.S_IROTH) -> None:
if not path.is_dir():
raise ValueError
path.chmod(path.stat().st_mode | stat.S_IROTH | stat.S_IXOTH) # o+rx
for p in path.rglob("*"):
if p.is_dir():
p.chmod(p.stat().st_mode | stat.S_IROTH | stat.S_IXOTH) # o+rx
else:
p.chmod(p.stat().st_mode | stat.S_IROTH) # o+r
def format_seconds(seconds: float) -> str:
hours, remainder = divmod(seconds, 3600)
minutes, seconds = divmod(remainder, 60)
hours, minutes, seconds = int(hours), int(minutes), round(seconds)
match (hours, minutes, seconds):
case 0, 0, s:
return f"{s}s"
case 0, m, s:
return f"{m}m {s}s"
case h, m, s:
return f"{h}h {m}m {s}s"
raise ValueError("unreachable")
def _checkout_name(select_output: str | None) -> str:
if select_output is not None:
return f"cpython-{select_output}"
return "cpython"
def main() -> int:
"""Script entry point."""
args = parse_args()
setup_logging(args.log_directory, args.select_output)
load_environment_variables()
if sentry_sdk:
sentry_sdk.init()
if args.select_output is None:
return build_docs_with_lock(args, "build_docs.lock")
if args.select_output == "no-html":
# Disable Plausible analytics for download copies
os.environ.pop("PYTHON_DOCS_ENABLE_ANALYTICS", None)
return build_docs_with_lock(args, "build_docs_archives.lock")
if args.select_output == "only-html":
return build_docs_with_lock(args, "build_docs_html.lock")
if args.select_output == "only-html-en":
return build_docs_with_lock(args, "build_docs_html_en.lock")
return EX_FAILURE
def parse_args() -> argparse.Namespace:
"""Parse command-line arguments."""
parser = argparse.ArgumentParser(
description="Runs a build of the Python docs for various branches.",
allow_abbrev=False,
)
parser.suggest_on_error = True
parser.add_argument(
"--select-output",
choices=("no-html", "only-html", "only-html-en"),
help="Choose what outputs to build.",
)
parser.add_argument(
"-q",
"--quick",
action="store_true",
help="Run a quick build (only HTML files).",
)
parser.add_argument(
"-b",
"--branches",
nargs="*",
metavar="3.12",
help="Versions to build (defaults to all maintained branches).",
)
parser.add_argument(
"-r",
"--build-root",
type=Path,
help="Path to a directory containing a checkout per branch.",
default=Path("/srv/docsbuild"),
)
parser.add_argument(
"-w",
"--www-root",
type=Path,
help="Path where generated files will be copied.",
default=Path("/srv/docs.python.org"),
)
parser.add_argument(
"--force",
action="store_true",
help="Always build the chosen languages and versions, "
"regardless of existing state.",
)
parser.add_argument(
"--skip-cache-invalidation",
help="Skip Fastly cache invalidation.",
action="store_true",
)
parser.add_argument(
"--group",
help="Group files on targets and www-root file should get.",
default="docs",
)
parser.add_argument(
"--log-directory",
type=Path,
help="Directory used to store logs.",
default=Path("/var/log/docsbuild/"),
)
parser.add_argument(
"--languages",
nargs="*",
help="Language translation, as a PEP 545 language tag like"
" 'fr' or 'pt-br'. "
"Builds all available languages by default.",
metavar="fr",
)
parser.add_argument(
"--version",
action="store_true",
help="Get build_docs and dependencies version info",
)
parser.add_argument(
"--theme",
default="python-docs-theme",
help="Python package to use for python-docs-theme: Useful to test branches:"
" --theme git+https://github.com/obulat/python-docs-theme@master",
)
args = parser.parse_args()
if args.version:
version_info()
sys.exit(0)
del args.version
if args.log_directory:
args.log_directory = args.log_directory.resolve()
if args.build_root:
args.build_root = args.build_root.resolve()
if args.www_root:
args.www_root = args.www_root.resolve()
return args
def setup_logging(log_directory: Path, select_output: str | None) -> None:
"""Setup logging to stderr if run by a human, or to a file if run from a cron."""
log_format = "%(asctime)s %(levelname)s: %(message)s"
if sys.stderr.isatty() or "CI" in os.environ:
logging.basicConfig(format=log_format, stream=sys.stderr)
else:
log_directory.mkdir(parents=True, exist_ok=True)
if select_output is None:
filename = log_directory / "docsbuild.log"
else:
filename = log_directory / f"docsbuild-{select_output}.log"
handler = logging.handlers.WatchedFileHandler(filename)
handler.setFormatter(logging.Formatter(log_format))
logging.getLogger().addHandler(handler)
logging.getLogger().setLevel(logging.DEBUG)
def load_environment_variables() -> None:
dbs_user_config = platformdirs.user_config_path("docsbuild-scripts")
dbs_site_config = platformdirs.site_config_path("docsbuild-scripts")
if dbs_user_config.is_file():
env_conf_file = dbs_user_config
elif dbs_site_config.is_file():
env_conf_file = dbs_site_config
else:
logging.info(
"No environment variables configured. Configure in %s or %s.",
dbs_site_config,
dbs_user_config,
)
return
logging.info("Reading environment variables from %s.", env_conf_file)
if env_conf_file == dbs_site_config:
logging.info("You can override settings in %s.", dbs_user_config)
elif dbs_site_config.is_file():
logging.info("Overriding %s.", dbs_site_config)
env_config = env_conf_file.read_text(encoding="utf-8")
for key, value in tomlkit.parse(env_config).get("env", {}).items():
logging.debug("Setting %s in environment.", key)
os.environ[key] = value
def build_docs_with_lock(args: argparse.Namespace, lockfile_name: str) -> int:
try:
lock = zc.lockfile.LockFile(HERE / lockfile_name)
except zc.lockfile.LockError:
logging.info("Another builder is running... dying...")
return EX_FAILURE
try:
return build_docs(args)
finally:
lock.close()
def build_docs(args: argparse.Namespace) -> int:
"""Build all docs (each language and each version)."""
logging.info("Full build start.")
start_time = perf_counter()
http = urllib3.PoolManager()
versions = parse_versions_from_peps_site(http)
languages = parse_languages_from_config()
# Reverse languages but not versions, because we take version-language
# pairs from the end of the list, effectively reversing it.
# This runs languages in config.toml order and versions newest first.
todo = [
BuildMetadata(_version=version, _language=language)
for version in versions.filter(args.branches)
for language in reversed(languages.filter(args.languages))
]
del args.branches
del args.languages
force_build = args.force
del args.force
docs_by_version_content = render_docs_by_version(versions).encode()
switchers_content = render_switchers(versions, languages)
build_succeeded = set()
any_build_failed = False
cpython_repo = Repository(
"https://github.com/python/cpython.git",
args.build_root / _checkout_name(args.select_output),
)
built_venvs: set[Path] = set()
try:
while todo:
build_props = todo.pop()
logging.root.handlers[0].setFormatter(
logging.Formatter(
f"%(asctime)s %(levelname)s {build_props.slug}: %(message)s"
)
)
if sentry_sdk:
scope = sentry_sdk.get_isolation_scope()
scope.set_tag("version", build_props.version)
scope.set_tag("language", build_props.language)
cpython_repo.update()
builder = DocBuilder(
build_props,
cpython_repo,
docs_by_version_content,
switchers_content,
built_venvs,
**vars(args),
)
built_successfully = builder.run(http, force_build=force_build)
if built_successfully:
build_succeeded.add(build_props.slug)
elif built_successfully is not None:
any_build_failed = True
finally:
for venv_path in built_venvs:
shutil.rmtree(venv_path, ignore_errors=True)
logging.root.handlers[0].setFormatter(
logging.Formatter("%(asctime)s %(levelname)s: %(message)s")
)
build_sitemap(versions, languages, args.www_root, args.group)
build_404(args.www_root, args.group)
build_robots_txt(
versions,
args.www_root,
args.group,
args.skip_cache_invalidation,
http,
)
make_symlinks(
args.www_root,
args.group,
versions,
languages,
build_succeeded,
args.skip_cache_invalidation,
http,
)
if build_succeeded:
# Only check canonicals if at least one version was built.
proofread_canonicals(args.www_root, args.skip_cache_invalidation, http)
logging.info("Full build done (%s).", format_seconds(perf_counter() - start_time))
return EX_FAILURE if any_build_failed else EX_OK
def parse_versions_from_peps_site(http: urllib3.PoolManager) -> Versions:
releases = http.request(
"GET",
"https://peps.python.org/api/release-cycle.json",
timeout=30,
).json()
return Versions.from_json(releases)
def parse_languages_from_config() -> Languages:
"""Read config.toml to discover languages to build."""
config = tomlkit.parse((HERE / "config.toml").read_text(encoding="UTF-8"))
return Languages.from_json(config["defaults"], config["languages"])
def render_docs_by_version(versions: Versions) -> str:
"""Generate content for _docs_by_version.html."""
links = [f'
' for v in reversed(versions)]
return "\n".join(links)
def render_switchers(versions: Versions, languages: Languages) -> bytes:
language_pairs = sorted((l.tag, l.switcher_label) for l in languages if l.in_prod) # NoQA: E741
version_pairs = [(v.name, v.picker_label) for v in reversed(versions)]
switchers_template_file = HERE / "templates" / "switchers.js"
template = Template(switchers_template_file.read_text(encoding="UTF-8"))
rendered_template = template.safe_substitute(
LANGUAGES=json.dumps(language_pairs),
VERSIONS=json.dumps(version_pairs),
)
return rendered_template.encode("UTF-8")
def build_sitemap(
versions: Versions, languages: Languages, www_root: Path, group: str
) -> None:
"""Build a sitemap with all live versions and translations."""
if not www_root.exists():
logging.info("Skipping sitemap generation (www root does not even exist).")
return
logging.info("Starting sitemap generation...")
template_path = HERE / "templates" / "sitemap.xml"
template = jinja2.Template(template_path.read_text(encoding="UTF-8"))
rendered_template = template.render(languages=languages, versions=versions)
sitemap_path = www_root / "sitemap.xml"
sitemap_path.write_text(rendered_template + "\n", encoding="UTF-8")
sitemap_path.chmod(0o664)
chgrp(sitemap_path, group=group)
def build_404(www_root: Path, group: str) -> None:
"""Build a nice 404 error page to display in case PDFs are not built yet."""
if not www_root.exists():
logging.info("Skipping 404 page generation (www root does not even exist).")
return
logging.info("Copying 404 page...")
not_found_file = www_root / "404.html"
shutil.copyfile(HERE / "templates" / "404.html", not_found_file)
not_found_file.chmod(0o664)
chgrp(not_found_file, group=group)
def build_robots_txt(
versions: Versions,
www_root: Path,
group: str,
skip_cache_invalidation: bool,
http: urllib3.PoolManager,
) -> None:
"""Build robots.txt to www_root."""
if not www_root.exists():
logging.info("Skipping robots.txt generation (www root does not even exist).")
return
logging.info("Starting robots.txt generation...")
template_path = HERE / "templates" / "robots.txt"
template = jinja2.Template(template_path.read_text(encoding="UTF-8"))
rendered_template = template.render(versions=versions)
robots_path = www_root / "robots.txt"
robots_path.write_text(rendered_template, encoding="UTF-8")
robots_path.chmod(0o775)
chgrp(robots_path, group=group)
if not skip_cache_invalidation:
purge(http, "robots.txt")
def make_symlinks(
www_root: Path,
group: str,
versions: Versions,
languages: Languages,
successful_builds: Set[str],
skip_cache_invalidation: bool,
http: urllib3.PoolManager,
) -> None:
"""Maintains the /2/, /3/, and /dev/ symlinks for each language.
Like:
- /2/ /2.7/
- /3/ /3.12/
- /dev/ /3.14/
- /fr/3/ /fr/3.12/
- /es/dev/ /es/3.14/
"""
logging.info("Creating major and development version symlinks...")
for symlink_name, symlink_target in (
("3", versions.current_stable.name),
("2", "2.7"),
("dev", versions.current_dev.name),
):
for language in languages:
if f"{language.tag}/{symlink_target}" in successful_builds:
symlink(
www_root,
language.tag,
symlink_target,
symlink_name,
group,
skip_cache_invalidation,
http,
)
def symlink(
www_root: Path,
language_tag: str,
directory: str,
name: str,
group: str,
skip_cache_invalidation: bool,
http: urllib3.PoolManager,
) -> None:
"""Used by major_symlinks and dev_symlink to maintain symlinks."""
msg = "Creating symlink from /%s/ to /%s/"
if language_tag == "en": # English is rooted on /, no /en/
path = www_root
logging.debug(msg, name, directory)
else:
path = www_root / language_tag
logging.debug(msg, f"{language_tag}/{name}", f"{language_tag}/{directory}")
link = path / name
directory_path = path / directory
if not directory_path.exists():
return # No touching link, dest doc not built yet.
if not link.exists() or os.readlink(link) != directory:
# Link does not exist or points to the wrong target.
link.unlink(missing_ok=True)
link.symlink_to(directory)
chgrp(link, group=group, follow_symlinks=False)
if not skip_cache_invalidation:
surrogate_key = f"{language_tag}/{name}"
purge_surrogate_key(http, surrogate_key)
def proofread_canonicals(
www_root: Path, skip_cache_invalidation: bool, http: urllib3.PoolManager
) -> None:
"""In www_root we check that all canonical links point to existing contents.
It can happen that a canonical is "broken":
- /3.11/whatsnew/3.11.html typically would link to
/3/whatsnew/3.11.html, which may not exist yet.
"""
logging.info("Checking canonical links...")
worker_count = (os.cpu_count() or 1) + 2
with concurrent.futures.ThreadPoolExecutor(worker_count) as executor:
futures = {
executor.submit(_check_canonical_rel, file, www_root)
for file in www_root.glob("**/*.html")
}
paths_to_purge = {
res.relative_to(www_root) # strip the leading /srv/docs.python.org
for fut in concurrent.futures.as_completed(futures)
if (res := fut.result()) is not None
}
if not skip_cache_invalidation:
purge(http, *paths_to_purge)
# Python 3.12 onwards doesn't use self-closing tags for
_canonical_re = re.compile(
b""" Path | None:
# Check for a canonical relation link in the HTML.
# If one exists, ensure that the target exists
# or otherwise remove the canonical link element.
html = file.read_bytes()
canonical = _canonical_re.search(html)
if canonical is None:
return None
target = canonical[1].decode(encoding="UTF-8", errors="surrogateescape")
if (www_root / target).exists():
return None
logging.info("Removing broken canonical from %s to %s", file, target)
start, end = canonical.span()
file.write_bytes(html[:start] + html[end:])
return file
def purge(http: urllib3.PoolManager, *paths: Path | str) -> None:
"""Remove one or many paths from docs.python.org's CDN.
To be used when a file changes, so the CDN fetches the new one.
"""
base = "https://docs.python.org/"
for path in paths:
url = urljoin(base, str(path))
logging.debug("Purging %s from CDN", url)
http.request("PURGE", url, timeout=30)
def purge_surrogate_key(http: urllib3.PoolManager, surrogate_key: str) -> None:
"""Remove paths from docs.python.org's CDN.
All paths matching the given 'Surrogate-Key' will be removed.
This is set by the Nginx server for every language-version pair.
To be used when a directory changes, so the CDN fetches the new one.
https://www.fastly.com/documentation/reference/api/purging/#purge-tag
"""
unset = "__UNSET__"
service_id = os.environ.get("FASTLY_SERVICE_ID", unset)
fastly_key = os.environ.get("FASTLY_TOKEN", unset)
if service_id == unset or fastly_key == unset:
logging.info("CDN secrets not set, skipping Surrogate-Key purge")
return
logging.info("Purging Surrogate-Key '%s' from CDN", surrogate_key)
http.request(
"POST",
f"https://api.fastly.com/service/{service_id}/purge/{surrogate_key}",
headers={"Fastly-Key": fastly_key},
timeout=30,
)
if __name__ == "__main__":
raise SystemExit(main())