# Zulip's main Markdown implementation.  See docs/subsystems/markdown.md for
# detailed documentation on our Markdown syntax.
import logging
import re
import time
from collections import deque
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timezone
from email.message import EmailMessage
from functools import lru_cache
from re import Match, Pattern
from typing import Any, Generic, Optional, TypeAlias, TypedDict, TypeVar, cast
from urllib.parse import parse_qs, parse_qsl, urlencode, urljoin, urlsplit, urlunsplit
from xml.etree.ElementTree import Element, SubElement

import ahocorasick
import dateutil.parser
import dateutil.tz
import lxml.etree
import markdown
import markdown.blockprocessors
import markdown.inlinepatterns
import markdown.postprocessors
import markdown.preprocessors
import markdown.treeprocessors
import markdown.util
import re2
import regex
import requests
import uri_template
import urllib3.exceptions
from django.conf import settings
from markdown.blockparser import BlockParser
from markdown.extensions import codehilite, nl2br, sane_lists, tables
from tlds import tld_set
from typing_extensions import NotRequired, Self, override

from zerver.lib import mention
from zerver.lib.camo import get_camo_url
from zerver.lib.emoji import EMOTICON_RE, codepoint_to_name, name_to_codepoint, translate_emoticons
from zerver.lib.emoji_utils import emoji_to_hex_codepoint, unqualify_emoji
from zerver.lib.exceptions import MarkdownRenderingError
from zerver.lib.markdown import fenced_code
from zerver.lib.markdown.fenced_code import FENCE_RE
from zerver.lib.mention import (
    BEFORE_MENTION_ALLOWED_REGEX,
    ChannelTopicInfo,
    FullNameInfo,
    MentionBackend,
    MentionData,
    get_user_group_mention_display_name,
)
from zerver.lib.mime_types import AUDIO_INLINE_MIME_TYPES, guess_type
from zerver.lib.outgoing_http import OutgoingSession
from zerver.lib.subdomains import is_static_or_current_realm_url
from zerver.lib.tex import render_tex
from zerver.lib.thumbnail import (
    AttachmentData,
    manifest_and_get_user_upload_previews,
    rewrite_thumbnailed_images,
)
from zerver.lib.timeout import unsafe_timeout
from zerver.lib.timezone import common_timezones
from zerver.lib.types import LinkifierDict
from zerver.lib.url_encoding import encode_channel, encode_hash_component
from zerver.lib.url_preview.types import UrlEmbedData, UrlOEmbedData
from zerver.models import Message, Realm, UserProfile
from zerver.models.linkifiers import linkifiers_for_realm
from zerver.models.realm_emoji import EmojiInfo, get_name_keyed_dict_for_active_realm_emoji

ReturnT = TypeVar("ReturnT")


# Taken from
# https://html.spec.whatwg.org/multipage/system-state.html#safelisted-scheme
html_safelisted_schemes = (
    "bitcoin",
    "geo",
    "im",
    "irc",
    "ircs",
    "magnet",
    "mailto",
    "matrix",
    "mms",
    "news",
    "nntp",
    "openpgp4fpr",
    "sip",
    "sms",
    "smsto",
    "ssh",
    "tel",
    "urn",
    "webcal",
    "wtai",
    "xmpp",
)
allowed_schemes = ("http", "https", "ftp", "file", *html_safelisted_schemes)


class LinkInfo(TypedDict):
    parent: Element
    title: str | None
    index: int | None
    remove: Element | None


@dataclass
class MessageRenderingResult:
    rendered_content: str
    mentions_topic_wildcard: bool
    mentions_stream_wildcard: bool
    mentions_user_ids: set[int]
    mentions_user_group_ids: set[int]
    alert_words: set[str]
    links_for_preview: set[str]
    user_ids_with_alert_words: set[int]
    potential_attachment_path_ids: list[str]
    thumbnail_spinners: set[str]


@dataclass
class DbData:
    mention_data: MentionData
    realm_url: str
    realm_alert_words_automaton: ahocorasick.Automaton | None
    active_realm_emoji: dict[str, EmojiInfo]
    sent_by_bot: bool
    stream_names: dict[str, int]
    topic_info: dict[ChannelTopicInfo, int | None]
    translate_emoticons: bool
    user_upload_previews: AttachmentData


# Format version of the Markdown rendering; stored along with rendered
# messages so that we can efficiently determine what needs to be re-rendered
version = 1

_T = TypeVar("_T")
ElementStringNone: TypeAlias = Element | str | None

EMOJI_REGEX = r"(?P<syntax>:[\w\-\+]+:)"


def verbose_compile(pattern: str) -> Pattern[str]:
    return re.compile(
        rf"^(.*?){pattern}(.*?)$",
        re.DOTALL | re.VERBOSE,
    )


STREAM_LINK_REGEX = rf"""
                     {BEFORE_MENTION_ALLOWED_REGEX} # Start after whitespace or specified chars
                     \#\*\*                         # and after hash sign followed by double asterisks
                         (?P<stream_name>[^\*]+)    # stream name can contain anything
                     \*\*                           # ends by double asterisks
                    """


@lru_cache(None)
def get_compiled_stream_link_regex() -> Pattern[str]:
    # Not using verbose_compile as it adds ^(.*?) and
    # (.*?)$ which cause extra overhead of matching
    # pattern which is not required.
    # With new InlineProcessor these extra patterns
    # are not required.
    return re.compile(
        STREAM_LINK_REGEX,
        re.DOTALL | re.VERBOSE,
    )


STREAM_TOPIC_LINK_REGEX = rf"""
                     {BEFORE_MENTION_ALLOWED_REGEX}  # Start after whitespace or specified chars
                     \#\*\*                          # and after hash sign followed by double asterisks
                         (?P<stream_name>[^\*>]+)    # stream name can contain anything except >
                         >                           # > acts as separator
                         (?P<topic_name>[^\*]*)      # topic name can be an empty string or contain anything
                     \*\*                            # ends by double asterisks
                   """


@lru_cache(None)
def get_compiled_stream_topic_link_regex() -> Pattern[str]:
    # Not using verbose_compile as it adds ^(.*?) and
    # (.*?)$ which cause extra overhead of matching
    # pattern which is not required.
    # With new InlineProcessor these extra patterns
    # are not required.
    return re.compile(
        STREAM_TOPIC_LINK_REGEX,
        re.DOTALL | re.VERBOSE,
    )


STREAM_TOPIC_MESSAGE_LINK_REGEX = rf"""
                     {BEFORE_MENTION_ALLOWED_REGEX}  # Start after whitespace or specified chars
                     \#\*\*                          # and after hash sign followed by double asterisks
                         (?P<stream_name>[^\*>]+)    # stream name can contain anything except >
                         >                           # > acts as separator
                         (?P<topic_name>[^\*]*)      # topic name can be an empty string or contain anything
                         @
                         (?P<message_id>\d+)         # message id
                     \*\*                            # ends by double asterisks
                   """


@lru_cache(None)
def get_compiled_stream_topic_message_link_regex() -> Pattern[str]:
    # Not using verbose_compile as it adds ^(.*?) and
    # (.*?)$ which cause extra overhead of matching
    # pattern which is not required.
    # With new InlineProcessor these extra patterns
    # are not required.
    return re.compile(
        STREAM_TOPIC_MESSAGE_LINK_REGEX,
        re.DOTALL | re.VERBOSE,
    )


@lru_cache(None)
def get_web_link_regex() -> Pattern[str]:
    # We create this one time, but not at startup.  So the
    # first message rendered in any process will have some
    # extra costs.  It's roughly 75ms to run this code, so
    # caching the value is super important here.

    tlds = r"|".join(list_of_tlds())

    # A link starts at a word boundary, and ends at space, punctuation, or end-of-input.
    #
    # We detect a URL either by the `https?://` or by building around the TLD.

    # In lieu of having a recursive regex (which python doesn't support) to match
    # arbitrary numbers of nested matching parenthesis, we manually build a regexp that
    # can match up to six
    # The inner_paren_contents chunk matches the innermore non-parenthesis-holding text,
    # and the paren_group matches text with, optionally, a matching set of parens
    inner_paren_contents = r"[^\s()\"]*"
    paren_group = r"""
                    [^\s()\"]*?            # Containing characters that won't end the URL
                    (?: \( %s \)           # and more characters in matched parens
                        [^\s()\"]*?        # followed by more characters
                    )*                     # zero-or-more sets of paired parens
                   """
    nested_paren_chunk = paren_group
    for i in range(6):
        nested_paren_chunk %= (paren_group,)
    nested_paren_chunk %= (inner_paren_contents,)

    file_links = r"| (?:file://(/[^/ ]*)+/?)" if settings.ENABLE_FILE_LINKS else r""
    REGEX = rf"""
        (?<![^\s'"\(,:<])    # Start after whitespace or specified chars
                             # (Double-negative lookbehind to allow start-of-string)
        (?P<url>             # Main group
            (?:(?:           # Domain part
                https?://[\w.:@-]+?   # If it has a protocol, anything goes.
               |(?:                   # Or, if not, be more strict to avoid false-positives
                    (?:[\w-]+\.)+     # One or more domain components, separated by dots
                    (?:{tlds})        # TLDs
                )
            )
            (?:
                (?:/ {nested_paren_chunk} )      # A path, beginning with /; zero-to-6 sets of paired parens
                |
                (?: \? (?![)\"\s]|\Z) {nested_paren_chunk} ) # Query starting with ? (must not be trailing punctuation)
            )?)              # Path is optional
            | (?:[\w.-]+\@[\w.-]+\.[\w]+) # Email is separate, since it can't have a path
            {file_links}               # File path start with file:///, enable by setting ENABLE_FILE_LINKS=True
            | (?:bitcoin:[13][a-km-zA-HJ-NP-Z1-9]{{25,34}})  # Bitcoin address pattern, see https://mokagio.github.io/tech-journal/2014/11/21/regex-bitcoin.html
        )
        (?=                            # URL must be followed by (not included in group)
            [!:;\?\),\.\'\"\>]*         # Optional punctuation characters
            (?:\Z|\s)                  # followed by whitespace or end of string
        )
        """
    return verbose_compile(REGEX)


def clear_web_link_regex_for_testing() -> None:
    # The link regex never changes in production, but our tests
    # try out both sides of ENABLE_FILE_LINKS, so we need
    # a way to clear it.
    get_web_link_regex.cache_clear()


markdown_logger = logging.getLogger()


def rewrite_local_links_to_relative(db_data: DbData | None, link: str) -> str:
    """If the link points to a local destination (e.g. #narrow/...),
    generate a relative link that will open it in the current window.
    """

    if db_data:
        realm_url_prefix = db_data.realm_url + "/"
        if link.startswith((realm_url_prefix + "#", realm_url_prefix + "user_uploads/")):
            return link.removeprefix(realm_url_prefix)

    return link


def maybe_add_attachment_path_id(url: str, zmd: "ZulipMarkdown") -> None:
    # Due to rewrite_local_links_to_relative, we need to
    # handle both relative URLs beginning with
    # `/user_uploads` and beginning with `user_uploads`.
    # This urllib construction converts the latter into
    # the former.
    parsed_url = urlsplit(urljoin("/", url))
    host = parsed_url.netloc

    if host != "" and (zmd.zulip_realm is None or host != zmd.zulip_realm.host):
        return

    if not parsed_url.path.startswith("/user_uploads/"):
        return

    path_id = parsed_url.path.removeprefix("/user_uploads/")
    zmd.zulip_rendering_result.potential_attachment_path_ids.append(path_id)


def url_embed_preview_enabled(
    message: Message | None = None, realm: Realm | None = None, no_previews: bool = False
) -> bool:
    if not settings.INLINE_URL_EMBED_PREVIEW:
        return False

    if no_previews:
        return False

    if realm is None and message is not None:
        realm = message.get_realm()

    if realm is None:
        # realm can be None for odd use cases
        # like generating documentation or running
        # test code
        return True

    return realm.inline_url_embed_preview


def image_preview_enabled(
    message: Message | None = None, realm: Realm | None = None, no_previews: bool = False
) -> bool:
    if not settings.INLINE_IMAGE_PREVIEW:
        return False

    if no_previews:
        return False

    if realm is None and message is not None:
        realm = message.get_realm()

    if realm is None:
        # realm can be None for odd use cases
        # like generating documentation or running
        # test code
        return True

    return realm.inline_image_preview


def list_of_tlds() -> list[str]:
    # Skip a few overly-common false-positives from file extensions
    common_false_positives = {"java", "md", "mov", "py", "zip"}
    return sorted(tld_set - common_false_positives, key=len, reverse=True)


def walk_tree(
    root: Element, processor: Callable[[Element], _T | None], stop_after_first: bool = False
) -> list[_T]:
    results = []
    queue = deque([root])

    while queue:
        currElement = queue.popleft()
        for child in currElement:
            queue.append(child)

            result = processor(child)
            if result is not None:
                results.append(result)
                if stop_after_first:
                    return results

    return results


@dataclass
class ElementFamily:
    grandparent: Element | None
    parent: Element
    child: Element
    in_blockquote: bool


T = TypeVar("T")


class ResultWithFamily(Generic[T]):
    family: ElementFamily
    result: T

    def __init__(self, family: ElementFamily, result: T) -> None:
        self.family = family
        self.result = result


class ElementPair:
    parent: Optional["ElementPair"]
    value: Element

    def __init__(self, parent: Optional["ElementPair"], value: Element) -> None:
        self.parent = parent
        self.value = value


def walk_tree_with_family(
    root: Element,
    processor: Callable[[Element], _T | None],
) -> list[ResultWithFamily[_T]]:
    results = []

    queue = deque([ElementPair(parent=None, value=root)])
    while queue:
        currElementPair = queue.popleft()
        for child in currElementPair.value:
            queue.append(ElementPair(parent=currElementPair, value=child))
            result = processor(child)
            if result is not None:
                if currElementPair.parent is not None:
                    grandparent_element = currElementPair.parent
                    grandparent: Element | None = grandparent_element.value
                else:
                    grandparent = None
                family = ElementFamily(
                    grandparent=grandparent,
                    parent=currElementPair.value,
                    child=child,
                    in_blockquote=has_blockquote_ancestor(currElementPair),
                )

                results.append(
                    ResultWithFamily(
                        family=family,
                        result=result,
                    )
                )

    return results


def has_blockquote_ancestor(element_pair: ElementPair | None) -> bool:
    if element_pair is None:
        return False
    elif element_pair.value.tag == "blockquote":
        return True
    else:
        return has_blockquote_ancestor(element_pair.parent)


class OpenGraphSession(OutgoingSession):
    def __init__(self) -> None:
        super().__init__(role="markdown", timeout=1)


def fetch_open_graph_image(url: str) -> dict[str, Any] | None:
    og: dict[str, str | None] = {"image": None, "title": None, "desc": None}

    try:
        with OpenGraphSession().get(
            url, headers={"Accept": "text/html,application/xhtml+xml"}, stream=True
        ) as res:
            if res.status_code != requests.codes.ok:
                return None

            m = EmailMessage()
            m["Content-Type"] = res.headers.get("Content-Type")
            mimetype = m.get_content_type()
            if mimetype not in ("text/html", "application/xhtml+xml"):
                return None
            html = mimetype == "text/html"

            res.raw.decode_content = True
            for event, element in lxml.etree.iterparse(
                res.raw, events=("start",), no_network=True, remove_comments=True, html=html
            ):
                parent = element.getparent()
                if parent is not None:
                    # Reduce memory usage.
                    parent.text = None
                    parent.remove(element)

                if element.tag in ("body", "{http://www.w3.org/1999/xhtml}body"):
                    break
                elif element.tag in ("meta", "{http://www.w3.org/1999/xhtml}meta"):
                    if element.get("property") == "og:image":
                        content = element.get("content")
                        if content is not None:
                            og["image"] = urljoin(res.url, content)
                    elif element.get("property") == "og:title":
                        og["title"] = element.get("content")
                    elif element.get("property") == "og:description":
                        og["desc"] = element.get("content")

    except (requests.RequestException, urllib3.exceptions.HTTPError):
        return None

    return None if og["image"] is None else og


class InlineImageProcessor(markdown.treeprocessors.Treeprocessor):
    """
    Rewrite inline img tags to serve external content via Camo.

    This rewrites all images, except ones that are served from the current
    realm or global STATIC_URL. This is to ensure that each realm only loads
    images that are hosted on that realm or by the global installation,
    avoiding information leakage to external domains or between realms. We need
    to disable proxying of images hosted on the same realm, because otherwise
    we will break images in /user_uploads/, which require authorization to
    view.
    """

    def __init__(self, zmd: "ZulipMarkdown") -> None:
        super().__init__(zmd)
        self.zmd = zmd

    @override
    def run(self, root: Element) -> None:
        # Get all URLs from the blob
        found_imgs = walk_tree(root, lambda e: e if e.tag == "img" else None)
        for img in found_imgs:
            url = img.get("src")
            assert url is not None
            if is_static_or_current_realm_url(url, self.zmd.zulip_realm):
                # Don't rewrite images on our own site (e.g. emoji, user uploads).
                continue
            img.set("src", get_camo_url(url))


class InlineVideoProcessor(markdown.treeprocessors.Treeprocessor):
    """
    Rewrite inline video tags to serve external content via Camo.

    This rewrites all video, except ones that are served from the current
    realm or global STATIC_URL. This is to ensure that each realm only loads
    videos that are hosted on that realm or by the global installation,
    avoiding information leakage to external domains or between realms. We need
    to disable proxying of videos hosted on the same realm, because otherwise
    we will break videos in /user_uploads/, which require authorization to
    view.
    """

    def __init__(self, zmd: "ZulipMarkdown") -> None:
        super().__init__(zmd)
        self.zmd = zmd

    @override
    def run(self, root: Element) -> None:
        # Get all URLs from the blob
        found_videos = walk_tree(root, lambda e: e if e.tag == "video" else None)
        for video in found_videos:
            url = video.get("src")
            assert url is not None
            if is_static_or_current_realm_url(url, self.zmd.zulip_realm):
                # Don't rewrite videos on our own site (e.g. user uploads).
                continue
            # The href= is still set to the non-Camo'd version, which
            # is used by clients to play the full video without
            # imposing load on the Camo server; this src is used to
            # load the initial thumbnail and metadata (through Camo)
            video.set("src", get_camo_url(url))


class BacktickInlineProcessor(markdown.inlinepatterns.BacktickInlineProcessor):
    """Return a `<code>` element containing the matching text."""

    @override
    def handleMatch(  # type: ignore[override] # https://github.com/python/mypy/issues/10197
        self, m: Match[str], data: str
    ) -> tuple[Element | str | None, int | None, int | None]:
        # Let upstream's implementation do its job as it is, we'll
        # just replace the text to not strip the group because it
        # makes it impossible to put leading/trailing whitespace in
        # an inline code span.
        el, _start, _end = ret = super().handleMatch(m, data)
        if el is not None and m.group(3):
            assert isinstance(el, Element)
            # upstream's code here is: m.group(3).strip() rather than m.group(3).
            el.text = markdown.util.AtomicString(markdown.util.code_escape(m.group(3)))
        return ret


# List from https://support.google.com/chromeos/bin/answer.py?hl=en&answer=183093
IMAGE_EXTENSIONS = [".bmp", ".gif", ".jpe", ".jpeg", ".jpg", ".png", ".webp"]


class DropboxMediaInfo(TypedDict):
    is_image: bool
    is_video: bool
    media_url: str
    title: NotRequired[str]
    desc: NotRequired[str]


class InlineInterestingLinkProcessor(markdown.treeprocessors.Treeprocessor):
    INLINE_PREVIEW_LIMIT_PER_MESSAGE = 24

    def __init__(self, zmd: "ZulipMarkdown") -> None:
        super().__init__(zmd)
        self.zmd = zmd

    def add_a(
        self,
        root: Element,
        image_url: str,
        link: str,
        title: str | None = None,
        desc: str | None = None,
        class_attr: str = "message_inline_image",
        data_id: str | None = None,
        insertion_index: int | None = None,
        already_thumbnailed: bool = False,
    ) -> None:
        desc = desc if desc is not None else ""

        # Update message.has_image attribute.
        if "message_inline_image" in class_attr and self.zmd.zulip_message:
            self.zmd.zulip_message.has_image = True

        if insertion_index is not None:
            div = Element("div")
            root.insert(insertion_index, div)
        else:
            div = SubElement(root, "div")

        div.set("class", class_attr)
        a = SubElement(div, "a")
        a.set("href", link)
        if title is not None:
            a.set("title", title)
        if data_id is not None:
            a.set("data-id", data_id)
        img = SubElement(a, "img")
        if image_url.startswith("/user_uploads/") and self.zmd.zulip_db_data:
            path_id = image_url.removeprefix("/user_uploads/")

            # We should have pulled the preview data for this image
            # (even if that's "no preview yet") from the database
            # before rendering; is_image should have enforced that.
            assert path_id in self.zmd.zulip_db_data.user_upload_previews.image_metadata
            metadata = self.zmd.zulip_db_data.user_upload_previews.image_metadata[path_id]

            # Insert a placeholder image spinner.  We post-process
            # this content (see rewrite_thumbnailed_images in
            # zerver.lib.thumbnail), looking specifically for this
            # tag, and may re-write it into the thumbnail URL if it
            # already exists when the message is sent.
            img.set("class", "image-loading-placeholder")
            img.set("src", "/static/images/loading/loader-black.svg")
            img.set(
                "data-original-dimensions",
                f"{metadata.original_width_px}x{metadata.original_height_px}",
            )
            if metadata.original_content_type:
                img.set(
                    "data-original-content-type",
                    metadata.original_content_type,
                )
        else:
            img.set("src", image_url)

    def add_oembed_data(self, root: Element, link: str, extracted_data: UrlOEmbedData) -> None:
        if extracted_data.image is None:
            # Don't add an embed if an image is not found
            return

        if extracted_data.type == "photo":
            self.add_a(
                root,
                image_url=extracted_data.image,
                link=link,
                title=extracted_data.title,
            )

        elif extracted_data.type == "video":
            self.add_a(
                root,
                image_url=extracted_data.image,
                link=link,
                title=extracted_data.title,
                desc=extracted_data.description,
                class_attr="embed-video message_inline_image",
                data_id=extracted_data.html,
                already_thumbnailed=True,
            )

    def add_embed(self, root: Element, link: str, extracted_data: UrlEmbedData) -> None:
        if isinstance(extracted_data, UrlOEmbedData):
            self.add_oembed_data(root, link, extracted_data)
            return

        if extracted_data.image is None:
            # Don't add an embed if an image is not found
            return

        container = SubElement(root, "div")
        container.set("class", "message_embed")

        img_link = get_camo_url(extracted_data.image)
        img = SubElement(container, "a")
        img.set(
            "style",
            'background-image: url("'
            + img_link.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\a ")
            + '")',
        )
        img.set("href", link)
        img.set("class", "message_embed_image")

        data_container = SubElement(container, "div")
        data_container.set("class", "data-container")

        if extracted_data.title:
            title_elm = SubElement(data_container, "div")
            title_elm.set("class", "message_embed_title")
            a = SubElement(title_elm, "a")
            a.set("href", link)
            a.set("title", extracted_data.title)
            a.text = extracted_data.title
        if extracted_data.description:
            description_elm = SubElement(data_container, "div")
            description_elm.set("class", "message_embed_description")
            description_elm.text = extracted_data.description

    def get_actual_image_url(self, url: str) -> str:
        # Add specific per-site cases to convert image-preview URLs to image URLs.
        # See https://github.com/zulip/zulip/issues/4658 for more information
        parsed_url = urlsplit(url)
        if parsed_url.netloc == "github.com" or parsed_url.netloc.endswith(".github.com"):
            # https://github.com/zulip/zulip/blob/main/static/images/logo/zulip-icon-128x128.png ->
            # https://raw.githubusercontent.com/zulip/zulip/main/static/images/logo/zulip-icon-128x128.png
            split_path = parsed_url.path.split("/")
            if len(split_path) > 3 and split_path[3] == "blob":
                return urljoin(
                    "https://raw.githubusercontent.com", "/".join(split_path[0:3] + split_path[4:])
                )

        return url

    def is_image(self, url: str) -> bool:
        if not self.zmd.image_preview_enabled:
            return False
        parsed_url = urlsplit(url)
        # remove HTML URLs which end with image extensions that cannot be shorted
        if parsed_url.netloc == "pasteboard.co":
            return False

        # Check against the previews we generated -- if we didn't have
        # a row for the ImageAttachment, then its header didn't parse
        # as a valid image type which libvips handles.
        if url.startswith("/user_uploads/") and self.zmd.zulip_db_data:
            path_id = url.removeprefix("/user_uploads/")
            return path_id in self.zmd.zulip_db_data.user_upload_previews.image_metadata

        return any(parsed_url.path.lower().endswith(ext) for ext in IMAGE_EXTENSIONS)

    def corrected_image_source(self, url: str) -> str | None:
        # This function adjusts any URLs from linx.li and
        # wikipedia.org to point to the actual image URL.  It's
        # structurally very similar to dropbox_media, and possibly
        # should be rewritten to use open graph, but has some value.
        parsed_url = urlsplit(url)
        if parsed_url.netloc.lower().endswith(".wikipedia.org") and parsed_url.path.startswith(
            "/wiki/File:"
        ):
            # Redirecting from "/wiki/File:" to "/wiki/Special:FilePath/File:"
            # A possible alternative, that avoids the redirect after hitting "Special:"
            # is using the first characters of md5($filename) to generate the URL
            newpath = parsed_url.path.replace("/wiki/File:", "/wiki/Special:FilePath/File:", 1)
            return parsed_url._replace(path=newpath).geturl()
        if parsed_url.netloc == "linx.li":
            return "https://linx.li/s" + parsed_url.path
        return None

    def dropbox_media(self, url: str) -> DropboxMediaInfo | None:
        parsed_url = urlsplit(url)
        if parsed_url.netloc == "dropbox.com" or parsed_url.netloc.endswith(".dropbox.com"):
            # See https://www.dropboxforum.com/discussions/101001012/shared-link--scl-to-s/689070/replies/695266
            # for more info on the URL structure mentioned here.
            # It is not possible to generate /sc/ links which is kind of a showcase
            # for multiple images. We treat it now as a folder instead.
            is_album = parsed_url.path.startswith(
                "/scl/fo/"  # codespell:ignore fo
            ) or parsed_url.path.startswith("/sc/")
            is_file = parsed_url.path.startswith("/scl/fi/")
            if not (is_file or is_album):
                return None

            is_image = self.is_image(url)
            # is_video function uses `mimetypes` library to determine
            # file type. For that we have to get rid of any query params
            # in the url before passing it to `is_video`.
            is_video = self.is_video(urlsplit(url).path)

            # If it is from an album or not an actual image file,
            # just use open graph image.
            if is_album or not (is_image or is_video):
                open_graph_image_info = fetch_open_graph_image(url)
                # Failed to follow link to find an image preview so
                # use placeholder image and guess filename
                if open_graph_image_info is None:
                    return None

                if is_album:
                    title = "Dropbox folder"
                    desc = "Click to open folder."
                else:
                    title = "Dropbox file"
                    desc = "Click to open file."

                return DropboxMediaInfo(
                    title=title,
                    desc=desc,
                    is_image=is_image,
                    is_video=is_video,
                    media_url=open_graph_image_info["image"],
                )

            # Adding raw=1 as query param will give us the URL of the
            # actual image instead of the dropbox image preview page.
            query_params = dict(parse_qsl(parsed_url.query))
            query_params["raw"] = "1"
            query = urlencode(query_params)

            return DropboxMediaInfo(
                is_image=is_image,
                is_video=is_video,
                media_url=parsed_url._replace(query=query).geturl(),
            )
        return None

    def youtube_id(self, url: str) -> str | None:
        if not self.zmd.image_preview_enabled:
            return None

        id = None
        split_url = urlsplit(url)
        if split_url.scheme in ("http", "https"):
            if split_url.hostname in (
                "m.youtube.com",
                "www.youtube.com",
                "www.youtube-nocookie.com",
                "youtube.com",
                "youtube-nocookie.com",
            ):
                query = parse_qs(split_url.query)
                if split_url.path in ("/watch", "/watch_popup") and "v" in query:
                    id = query["v"][0]
                elif split_url.path == "/watch_videos" and "video_ids" in query:
                    id = query["video_ids"][0].split(",", 1)[0]
                elif split_url.path.startswith(("/embed/", "/shorts/", "/v/")):
                    id = split_url.path.split("/", 3)[2]
