import asyncio
import html
import importlib
import json
import mimetypes
import os
import posixpath
import queue
import re
import stat
import sys
import threading
import time
import zipfile
from collections.abc import Callable
from datetime import UTC, datetime
from pathlib import Path
from urllib.parse import quote, urlencode, urljoin
from abx_plugins.plugins.archivewebpage import replay_preview as archivewebpage_replay
from django.contrib.staticfiles import finders
from django.core.handlers.asgi import ASGIRequest
from django.http import Http404, HttpResponse, HttpResponseNotModified, StreamingHttpResponse
from django.template import TemplateDoesNotExist, loader
from django.utils._os import safe_join
from django.utils.http import http_date
from django.utils.translation import gettext as _
from django.views import static
from archivebox.config.common import get_config
from archivebox.misc.logging_util import printable_filesize
_HASHES_CACHE: dict[Path, tuple[float, dict[str, str]]] = {}
IMG_SRC_ATTR_RE = re.compile(r'(]*?\s(?:src|data-src)=["\'])([^"\']+)(["\'])', re.IGNORECASE)
TRANSFORMED_HTML_PREVIEW_STYLE = """"""
def _load_hash_map(snapshot_dir: Path) -> dict[str, str] | None:
hashes_path = snapshot_dir / "hashes" / "hashes.json"
if not hashes_path.exists():
return None
try:
mtime = hashes_path.stat().st_mtime
except OSError:
return None
cached = _HASHES_CACHE.get(hashes_path)
if cached and cached[0] == mtime:
return cached[1]
try:
data = json.loads(hashes_path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError, TypeError, ValueError):
return None
file_map = {str(entry.get("path")): entry.get("hash") for entry in data.get("files", []) if entry.get("path")}
_HASHES_CACHE[hashes_path] = (mtime, file_map)
return file_map
def _hash_for_path(document_root: Path, rel_path: str) -> str | None:
file_map = _load_hash_map(document_root)
if not file_map:
return None
return file_map.get(rel_path)
def _resolve_archive_path(document_root: str | Path, rel_path: str) -> tuple[Path, str]:
rel_path = posixpath.normpath(rel_path).lstrip("/") if rel_path else ""
fullpath = Path(safe_join(document_root, rel_path))
if os.access(fullpath, os.R_OK):
return fullpath, rel_path
root = Path(document_root)
current = root
resolved_parts: list[str] = []
for part in Path(rel_path).parts:
exact = current / part
if os.access(exact, os.R_OK):
current = exact
resolved_parts.append(part)
continue
folded_part = part.casefold()
try:
match = next((child for child in current.iterdir() if child.name.casefold() == folded_part), None)
except OSError:
match = None
if match is None:
return fullpath, rel_path
current = match
resolved_parts.append(match.name)
return current, posixpath.join(*resolved_parts) if resolved_parts else ""
def _cache_policy(config=None, **config_kwargs) -> str:
config = config or get_config(resolve_plugins=False, **config_kwargs)
return "private" if config.PERMISSIONS == "private" else "public"
def _format_direntry_timestamp(stat_result: os.stat_result) -> str:
timestamp = stat_result.st_birthtime if sys.platform == "darwin" else stat_result.st_mtime
return datetime.fromtimestamp(timestamp, tz=UTC).strftime("%Y-%m-%d %H:%M")
def _safe_zip_stem(name: str) -> str:
safe_name = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip("._-")
return safe_name or "archivebox"
class _StreamingQueueWriter:
"""Expose a write-only file-like object so zipfile can stream into a queue."""
def __init__(self, output_queue: queue.Queue[bytes | BaseException | object]) -> None:
self.output_queue = output_queue
self.position = 0
def write(self, data: bytes) -> int:
if data:
self.output_queue.put(data)
self.position += len(data)
return len(data)
def tell(self) -> int:
return self.position
def flush(self) -> None:
return None
def close(self) -> None:
return None
def writable(self) -> bool:
return True
def seekable(self) -> bool:
return False
def _iter_visible_files(root: Path):
"""Yield non-hidden files in a stable order so ZIP output is deterministic."""
for current_root, dirnames, filenames in os.walk(root):
dirnames[:] = sorted(dirname for dirname in dirnames if not dirname.startswith("."))
for filename in sorted(name for name in filenames if not name.startswith(".")):
yield Path(current_root) / filename
def _build_directory_zip_response(
fullpath: Path,
path: str,
*,
is_archive_replay: bool,
use_async_stream: bool,
config=None,
) -> StreamingHttpResponse:
root_name = _safe_zip_stem(fullpath.name or Path(path).name or "archivebox")
sentinel = object()
output_queue: queue.Queue[bytes | BaseException | object] = queue.Queue(maxsize=8)
initial_chunk_target = 64 * 1024
initial_chunk_wait = 0.05
def build_zip() -> None:
# zipfile wants a write-only file object. Feed those bytes straight into
# a queue so the response can stream them out as soon as they are ready.
writer = _StreamingQueueWriter(output_queue)
try:
with zipfile.ZipFile(writer, mode="w", compression=zipfile.ZIP_DEFLATED, compresslevel=6) as zip_file:
for entry in _iter_visible_files(fullpath):
rel_parts = entry.relative_to(fullpath).parts
arcname = Path(root_name, *rel_parts).as_posix()
zip_file.write(entry, arcname)
except (OSError, RuntimeError, TypeError, ValueError, zipfile.BadZipFile) as err:
output_queue.put(err)
finally:
output_queue.put(sentinel)
threading.Thread(target=build_zip, name=f"zip-stream-{root_name}", daemon=True).start()
def iter_zip_chunks():
# Emit a meaningful first chunk quickly so browsers show the download
# immediately instead of waiting on dozens of tiny ZIP header writes.
first_chunk = bytearray()
initial_deadline = time.monotonic() + initial_chunk_wait
while True:
timeout = max(initial_deadline - time.monotonic(), 0) if len(first_chunk) < initial_chunk_target else None
try:
chunk = output_queue.get(timeout=timeout) if timeout is not None else output_queue.get()
except queue.Empty:
if first_chunk:
yield bytes(first_chunk)
first_chunk.clear()
continue
chunk = output_queue.get()
if chunk is sentinel:
if first_chunk:
yield bytes(first_chunk)
break
if isinstance(chunk, BaseException):
raise chunk
if len(first_chunk) < initial_chunk_target:
first_chunk.extend(chunk)
if len(first_chunk) >= initial_chunk_target or time.monotonic() >= initial_deadline:
yield bytes(first_chunk)
first_chunk.clear()
continue
yield chunk
async def stream_zip_async():
# Django ASGI buffers sync StreamingHttpResponse iterators by consuming
# them into a list. Drive the same sync iterator from a worker thread so
# Daphne can send each chunk as it arrives instead of buffering the ZIP.
iterator = iter(iter_zip_chunks())
while True:
chunk = await asyncio.to_thread(next, iterator, None)
if chunk is None:
break
yield chunk
response = StreamingHttpResponse(
stream_zip_async() if use_async_stream else iter_zip_chunks(),
content_type="application/zip",
)
response.headers["Content-Disposition"] = f'attachment; filename="{root_name}.zip"'
response.headers["Cache-Control"] = f"{_cache_policy(config=config)}, max-age=60, stale-while-revalidate=300"
response.headers["Last-Modified"] = http_date(fullpath.stat().st_mtime)
response.headers["X-Accel-Buffering"] = "no"
return _apply_archive_replay_headers(
response,
fullpath=fullpath,
content_type="application/zip",
is_archive_replay=is_archive_replay,
config=config,
)
async def _stream_ranged_file_async(ranged_file: "RangedFileReader"):
iterator = iter(ranged_file)
try:
while True:
chunk = await asyncio.to_thread(next, iterator, None)
if chunk is None:
break
yield chunk
finally:
ranged_file.close()
def _render_directory_index(request, path: str, fullpath: Path) -> HttpResponse:
try:
template = loader.select_template(
[
"static/directory_index.html",
"static/directory_index",
],
)
except TemplateDoesNotExist:
return static.directory_index(path, fullpath)
entries = []
file_list = []
visible_entries = sorted(
(entry for entry in fullpath.iterdir() if not entry.name.startswith(".")),
key=lambda entry: (not entry.is_dir(), entry.name.lower()),
)
for entry in visible_entries:
url = str(entry.relative_to(fullpath))
if entry.is_dir():
url += "/"
file_list.append(url)
stat_result = entry.stat()
entries.append(
{
"name": url,
"url": url,
"is_dir": entry.is_dir(),
"size": "—" if entry.is_dir() else printable_filesize(stat_result.st_size),
"timestamp": _format_direntry_timestamp(stat_result),
},
)
zip_query = request.GET.copy()
zip_query["download"] = "zip"
zip_url = request.path
if zip_query:
zip_url = f"{zip_url}?{zip_query.urlencode()}"
directory_query = "?files=1" if request.GET.get("files") else ""
path_suffix = f"{path.strip('/')}/" if path.strip("/") else ""
snapshot_page_url = request.path
if path_suffix and snapshot_page_url.endswith(path_suffix):
snapshot_page_url = snapshot_page_url[: -len(path_suffix)]
context = {
"directory": f"{path}/",
"file_list": file_list,
"entries": entries,
"zip_url": zip_url,
"directory_query": directory_query,
"snapshot_page_url": snapshot_page_url,
}
return HttpResponse(template.render(context))
# Ensure common web types are mapped consistently across platforms.
mimetypes.add_type("text/html", ".html")
mimetypes.add_type("text/html", ".htm")
mimetypes.add_type("text/css", ".css")
mimetypes.add_type("application/javascript", ".js")
mimetypes.add_type("application/json", ".json")
mimetypes.add_type("application/x-ndjson", ".jsonl")
mimetypes.add_type("text/markdown", ".md")
mimetypes.add_type("text/yaml", ".yml")
mimetypes.add_type("text/yaml", ".yaml")
mimetypes.add_type("text/csv", ".csv")
mimetypes.add_type("text/tab-separated-values", ".tsv")
mimetypes.add_type("application/xml", ".xml")
mimetypes.add_type("image/svg+xml", ".svg")
mimetypes.add_type("multipart/related", ".mhtml")
mimetypes.add_type("multipart/related", ".mht")
try:
_markdown = importlib.import_module("markdown").markdown
except ImportError:
_markdown: Callable[..., str] | None = None
MARKDOWN_INLINE_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)\s]+(?:\([^)]*\)[^)\s]*)*)\)")
MARKDOWN_INLINE_IMAGE_RE = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)")
MARKDOWN_BOLD_RE = re.compile(r"\*\*([^*]+)\*\*")
MARKDOWN_ITALIC_RE = re.compile(r"(?]*>")
HTML_BODY_RE = re.compile(r"]*>(.*)", flags=re.IGNORECASE | re.DOTALL)
RISKY_REPLAY_MIMETYPES = {
"text/html",
"application/xhtml+xml",
"image/svg+xml",
}
RISKY_REPLAY_EXTENSIONS = {".html", ".htm", ".xhtml", ".svg", ".svgz"}
RISKY_REPLAY_MARKERS = (
" str:
candidate = text
body_match = HTML_BODY_RE.search(candidate)
if body_match:
candidate = body_match.group(1)
candidate = re.sub(r"^\s*