feat(file): report upload failures clearly and verify before storing

src/osw/controller/file/wiki.py

@@ -2,13 +2,80 @@
 import os
 from typing import IO, Any, Dict, List, Optional
 
+import mwclient.errors
+
 from osw.controller.file.base import FileController
 from osw.controller.file.remote import RemoteFileController
 from osw.core import OSW, model
 from osw.utils.wiki import get_namespace, get_title
 from osw.wtsite import WtSite
 
 
+def format_allowed_extensions(allowed: WtSite.AllowedFileExtensionsResult) -> str:
+    """Renders the allowed-extensions hint for a rejected-upload error message"""
+    if not allowed.extensions or allowed.fetched_at is None:
+        return ""
+    ttl_hours = int(WtSite.ALLOWED_FILE_EXTENSIONS_TTL.total_seconds() // 3600)
+    return (
+        f" Extensions allowed there, read {allowed.fetched_at:%Y-%m-%d %H:%M:%S} and "
+        f"cached for {ttl_hours} h: {', '.join(sorted(allowed.extensions))}. "
+        "Call osw.site.clear_allowed_file_extensions_cache() to re-read the list "
+        "if the wiki configuration changed since."
+    )
+
+
+def reraise_upload_error(
+    error: mwclient.errors.APIError, site: WtSite, title: str, suffix: Optional[str]
+) -> None:
+    """Turns a rejected file extension into a readable error, re-raises the rest
+
+    MediaWiki reports a rejected extension as filetype-banned,
+    filetype-banned-type or filetype-badtype, depending on the version.
+    """
+    if "filetype" not in str(getattr(error, "code", "")):
+        raise error
+    allowed = site.get_allowed_file_extensions()
+    hint = format_allowed_extensions(allowed)
+    raise ValueError(
+        f"Upload of '{title}' was rejected because the file extension "
+        f"'{suffix}' is not allowed on {site.mw_site.host}.{hint}"
+    ) from error
+
+
+def assert_upload_success(result: Any, title: str, host: str) -> None:
+    """Raises unless the upload API reports that it stored the file
+
+    mwclient raises only when the response carries an 'error' key. A file that
+    MediaWiki declined for any other reason comes back as a normal return value
+    with a result other than 'Success', so without this check the upload would
+    fail silently.
+    """
+    status = result.get("result") if isinstance(result, dict) else None
+    if status == "Success":
+        return
+    warnings = result.get("warnings") if isinstance(result, dict) else None
+    detail = f" Warnings: {warnings}." if warnings else ""
+    raise ValueError(
+        f"Upload of '{title}' to {host} did not succeed (result: {status}).{detail}"
+    )
+
+
+def store_params_for_upload(
+    se_params: Dict[str, Any], page_existed: bool
+) -> Dict[str, Any]:
+    """Picks the overwrite policy for the entity that accompanies an upload
+
+    An upload creates the file page when it was not there yet. store_entity
+    would then see an existing page and, under the default 'keep existing',
+    leave the metadata unwritten. So the entity has to replace what the upload
+    put there. A page that was already there keeps whatever the caller asked
+    for.
+    """
+    if page_existed:
+        return se_params
+    return {**se_params, "overwrite": "replace remote"}
+
+
 class WikiFileController(model.WikiFile, RemoteFileController):
     """File controller for wiki files"""
 
@@ -131,20 +198,29 @@ def put(self, file: IO, **kwargs: Dict[str, Any]):
         }
         for key in ["entities", "namespace"]:
             se_params.pop(key, None)  # avoid duplicated kwargs
+        # Upload before storing the entity: MediaWiki offers no transaction
+        # across the two writes, so one of them can be left standing. A file
+        # page without metadata is visibly incomplete, while metadata without a
+        # file looks like a valid entity until someone tries to download it.
+        page_existed = self.osw.mw_site.pages[f"{self.namespace}:{self.title}"].exists
+        try:
+            result = self.osw.mw_site.upload(
+                file=file,
+                filename=self.title,
+                # comment="",
+                # description="",
+                ignore=True,
+            )
+        except mwclient.errors.APIError as e:
+            reraise_upload_error(e, self.osw.site, self.title, self.suffix)
+        assert_upload_success(result, self.title, self.osw.mw_site.host)
         self.osw.store_entity(
             OSW.StoreEntityParam(
                 entities=[self.cast(model.WikiFile, **wf_params)],
                 namespace=self.namespace,
-                **se_params,
+                **store_params_for_upload(se_params, page_existed),
             )
         )
-        self.osw.mw_site.upload(
-            file=file,
-            filename=self.title,
-            # comment="",
-            # description="",
-            ignore=True,
-        )
 
     def put_from(self, other: FileController, **kwargs: Dict[str, Any]):
         # if isinstance(file, LocalFileController) and self.suffix is None:

src/osw/core.py

@@ -59,12 +59,14 @@
     get_full_title,
     get_namespace,
     get_title,
-    get_uuid,
     is_empty,
     namespace_from_full_title,
     remove_empty,
     title_from_full_title,
 )
+from osw.utils.wiki import (
+    get_uuid as get_uuid_from_osw_id,
+)
 from osw.wiki_tools import SearchParam
 from osw.wtsite import WtPage, WtSite
 
@@ -197,18 +199,21 @@ def get_osw_id(uuid: Union[str, UUID]) -> str:
 
     @staticmethod
     def get_uuid(osw_id: str) -> UUID:
-        """Returns the uuid for a given OSW-ID
+        """Returns the uuid for a given OSW-ID. Kept for backwards compatibility,
+        the implementation lives in osw.utils.wiki.get_uuid()
 
         Parameters
         ----------
         osw_id
-            OSW-ID string, e.g. OSW2ea5b605c91f4e5a95593dff79fdd4a5
+            OSW-ID string, e.g. OSW2ea5b605c91f4e5a95593dff79fdd4a5, with or
+            without file suffixes, e.g.
+            OSW2ea5b605c91f4e5a95593dff79fdd4a5.drawio.png
 
         Returns
         -------
             uuid object, e.g. UUID("2ea5b605-c91f-4e5a-9559-3dff79fdd4a5")
         """
-        return UUID(osw_id.replace("OSW", ""))
+        return get_uuid_from_osw_id(osw_id)
 
     class SortEntitiesResult(OswBaseModel):
         by_name: Dict[str, List[OswBaseModel]]
@@ -1348,7 +1353,7 @@ def validate_entity(cls, entity, values):
             if jsondata is None:  # Guard clause
                 title = title_from_full_title(page.title)
                 try:
-                    uuid_from_title = get_uuid(title)
+                    uuid_from_title = get_uuid_from_osw_id(title)
                 except ValueError:
                     print(
                         f"Error: UUID could not be determined from title: '{title}', "

src/osw/utils/wiki.py

@@ -1,9 +1,17 @@
+import re
 from copy import deepcopy
 from uuid import UUID
 
 # Legacy imports:
 from opensemantic.v1 import get_full_title, get_namespace, get_title  # noqa: F401
 
+OSW_ID_PATTERN = re.compile(
+    r"^(?:OSW)?"  # the prefix, absent when a bare uuid is passed
+    r"([0-9a-fA-F]{32})"  # the uuid, in the hex form an OSW-ID carries it
+    r"(?:\.[\w-]+)*$"  # file suffixes, e.g. '.png' or '.drawio.png'
+)
+"""Matches an OSW-ID with an optional prefix and any number of file suffixes"""
+
 
 def get_osw_id(uuid: UUID) -> str:
     """Generates a OSW-ID based on the given uuid by prefixing "OSW" and removing
@@ -21,19 +29,33 @@ def get_osw_id(uuid: UUID) -> str:
     return "OSW" + str(uuid).replace("-", "")
 
 
-def get_uuid(osw_id) -> UUID:
-    """Returns the uuid for a given OSW-ID. Duplicate of OSW.get_uuid() from src/sw/core/osw.py
+def get_uuid(osw_id: str) -> UUID:
+    """Returns the uuid for a given OSW-ID. The single implementation, wrapped by
+    OSW.get_uuid() from src/osw/core.py
+
+    A file page keeps its file extension in the title, so the OSW-ID of a file is
+    followed by one or more suffixes that are not part of the uuid. These are
+    ignored, as is the OSW prefix.
 
     Parameters
     ----------
     osw_id
-        OSW-ID string, e.g., OSW2ea5b605c91f4e5a95593dff79fdd4a5
+        OSW-ID string, e.g., OSW2ea5b605c91f4e5a95593dff79fdd4a5, with or without
+        file suffixes, e.g., OSW2ea5b605c91f4e5a95593dff79fdd4a5.drawio.png
 
     Returns
     -------
         uuid object, e.g., UUID("2ea5b605-c91f-4e5a-9559-3dff79fdd4a5")
+
+    Raises
+    ------
+    ValueError
+        If no OSW-ID can be read from the given string.
     """
-    return UUID(osw_id.replace("OSW", ""))
+    match = OSW_ID_PATTERN.match(osw_id)
+    if match is None:
+        raise ValueError(f"No OSW-ID could be read from '{osw_id}'")
+    return UUID(match.group(1))
 
 
 def namespace_from_full_title(full_title: str) -> str:

src/osw/wtsite.py

@@ -10,7 +10,7 @@
 import warnings
 import xml.etree.ElementTree as et
 from copy import deepcopy
-from datetime import datetime
+from datetime import datetime, timedelta
 from io import StringIO
 from pathlib import Path
 from pprint import pprint
@@ -79,6 +79,9 @@ class WtSiteLegacyConfig(OswBaseModel):
         class Config:
             arbitrary_types_allowed = True
 
+    ALLOWED_FILE_EXTENSIONS_TTL = timedelta(hours=24)
+    """how long a fetched list of accepted file extensions stays valid"""
+
     def __init__(self, config: Union[WtSiteConfig, WtSiteLegacyConfig]):
         """creates a new WtSite instance from a WtSiteConfig
 
@@ -161,6 +164,10 @@ def __init__(self, config: Union[WtSiteConfig, WtSiteLegacyConfig]):
         self._page_cache = {}
         self._cache_enabled = False
 
+        # The file extensions the wiki accepts, read on demand and reused for
+        #  ALLOWED_FILE_EXTENSIONS_TTL
+        self._allowed_file_extensions = None
+
     def _get_session_lock(self) -> threading.RLock:
         """Return the session lock, lazily creating it if absent.
 
@@ -492,6 +499,68 @@ def clear_cache(self):
         del self._page_cache
         self._page_cache = {}
 
+    class AllowedFileExtensionsResult(OswBaseModel):
+        """The file extensions a wiki accepts and when that list was read"""
+
+        extensions: Optional[List[str]]
+        """the accepted extensions, None if the lookup failed"""
+        fetched_at: Optional[datetime]
+        """when the list was read from the wiki, None if the lookup failed"""
+
+    def get_allowed_file_extensions(
+        self, refresh: bool = False
+    ) -> AllowedFileExtensionsResult:
+        """Returns the file extensions the wiki accepts, from cache if possible
+
+        The list only changes when the wiki configuration changes, so it is
+        cached for ALLOWED_FILE_EXTENSIONS_TTL instead of being read on every
+        call. A failed lookup is never cached, so the next call retries; a
+        failed refresh leaves any previously cached list in place.
+
+        Parameters
+        ----------
+        refresh:
+            if True, bypasses the cache and re-reads the list from the wiki
+
+        Returns
+        -------
+            an AllowedFileExtensionsResult with the accepted extensions and
+            when they were read, or with both fields None if the lookup failed
+        """
+        cached = getattr(self, "_allowed_file_extensions", None)
+        if (
+            cached is not None
+            and not refresh
+            and cached.fetched_at is not None
+            and datetime.now() - cached.fetched_at < self.ALLOWED_FILE_EXTENSIONS_TTL
+        ):
+            return cached
+        try:
+            result = self.mw_site.api(
+                "query", meta="siteinfo", siprop="fileextensions", formatversion=2
+            )
+            extensions = [
+                entry["ext"]
+                for entry in result.get("query", {}).get("fileextensions", [])
+                if "ext" in entry
+            ]
+        except Exception:
+            # only used to enrich an error message, never worth failing over
+            return WtSite.AllowedFileExtensionsResult(extensions=None, fetched_at=None)
+        self._allowed_file_extensions = WtSite.AllowedFileExtensionsResult(
+            extensions=extensions, fetched_at=datetime.now()
+        )
+        return self._allowed_file_extensions
+
+    def clear_allowed_file_extensions_cache(self):
+        """Clears the cached list of file extensions the wiki accepts
+
+        The next get_allowed_file_extensions() call reads the list from the
+        wiki again. Use this when the wiki's upload configuration changed
+        within ALLOWED_FILE_EXTENSIONS_TTL.
+        """
+        self._allowed_file_extensions = None
+
     def _clear_cookies(self):
         # see https://github.com/mwclient/mwclient/issues/221
         # Iterate a snapshot (list(...)) so mutating the jar mid-loop is safe, and