Use OAuth sessions for web API requests

add_links/api.py

@@ -1,3 +1,5 @@
+import os
+from pathlib import Path
 import re
 import sys
 import typing
@@ -8,6 +10,7 @@ from simplejson.scanner import JSONDecodeError
 
 from .language import get_current_language
 from .util import is_disambig
+from .wikimedia_api_logging import WikimediaApiLogConfig, logged_request
 
 StrDict = dict[str, typing.Any]
 
@@ -15,6 +18,15 @@ ua = (
     "find-link/2.2 "
     + "(https://github.com/EdwardBetts/find_link; contact: edward@4angle.com)"
 )
+wikimedia_log_config = WikimediaApiLogConfig(
+    tool="missinglink",
+    log_path=Path(
+        os.environ.get(
+            "MISSINGLINK_WIKIMEDIA_API_LOG", "/var/log/missinglink/wikimedia-api.jsonl"
+        )
+    ),
+    user_agent=ua,
+)
 re_disambig = re.compile(r"^(.*) \((.*)\)$")
 
 
@@ -73,23 +85,40 @@ webpage_error = (
 )
 
 
-def _get_active_session() -> requests.sessions.Session:
-    """Return OAuth session if one is available in Flask context, else plain session."""
+def _get_request_oauth_session() -> requests.sessions.Session | None:
+    """Return the logged-in user's OAuth session in a Flask request, if present."""
     try:
-        from flask import g
-        if hasattr(g, "oauth_session") and g.oauth_session is not None:
-            return g.oauth_session  # type: ignore[return-value]
+        from flask import g, has_request_context
+
+        if not has_request_context():
+            return None
+
+        oauth_session = getattr(g, "oauth_session", None)
+        if oauth_session is not None:
+            return typing.cast(requests.sessions.Session, oauth_session)
+
+        from . import mediawiki_oauth
+
+        return typing.cast(
+            requests.sessions.Session | None, mediawiki_oauth.get_oauth_session()
+        )
     except RuntimeError:
-        pass
+        return None
+
+
+def _get_active_session() -> tuple[requests.sessions.Session, str]:
+    """Return active session and auth mode."""
+    if oauth_session := _get_request_oauth_session():
+        return oauth_session, "oauth"
+
     print("WARNING: using unauthenticated session", file=sys.stderr)
-    return get_session()
+    return get_session(), "anon"
 
 
 def api_get(params: StrDict) -> StrDict:
     """Make call to Wikipedia API."""
-    s = _get_active_session()
-
-    r = s.get(get_query_url(), params=params)
+    s, _auth = _get_active_session()
+    r = logged_request(s, wikimedia_log_config, "GET", get_query_url(), params=params)
     try:
         ret: StrDict = r.json()
     except JSONDecodeError:
@@ -287,8 +316,8 @@ def call_get_diff(title: str, section_num: int, section_text: str) -> str:
         "rvdifftotext": section_text.strip(),
     }
 
-    s = _get_active_session()
-    r = s.post(get_query_url(), data=data)
+    s, _auth = _get_active_session()
+    r = logged_request(s, wikimedia_log_config, "POST", get_query_url(), data=data)
     try:
         ret = r.json()
     except JSONDecodeError:

add_links/match.py

@@ -116,10 +116,15 @@ def find_cite_template_spans(text: str) -> list[tuple[int, int]]:
 def parse_cite(text: str) -> typing.Iterator[tuple[str, str]]:
     """Parse citations yielding (type, chunk) tuples, skipping ref tags, cite templates, and external links."""
     regions = [(m.start(), m.end()) for m in re_cite.finditer(text)]
+    link_spans = [(m.start(), m.end()) for m in re_link_in_text.finditer(text)]
     regions.extend(find_cite_template_spans(text))
     regions.extend((m.start(), m.end()) for m in re_no_param_template.finditer(text))
     regions.extend((m.start(), m.end()) for m in re_external_link.finditer(text))
-    regions.extend((m.start(), m.end()) for m in re_italic.finditer(text))
+    regions.extend(
+        (m.start(), m.end())
+        for m in re_italic.finditer(text)
+        if not any(start <= m.start() and m.end() <= end for start, end in link_spans)
+    )
     regions.extend((m.start(), m.end()) for m in re_bullet_with_url.finditer(text))
     regions.sort()
 

add_links/mediawiki_api.py

@@ -1,7 +1,6 @@
 """Interface with the mediawiki API."""
 
 import typing
-from pprint import pprint
 from typing import Any, cast
 
 import requests
@@ -12,6 +11,17 @@ from . import mediawiki_oauth
 class APIError(Exception):
     """Unexpected response from the MediaWiki API."""
 
+
+def _format_api_error(data: dict[str, Any]) -> str:
+    """Format a MediaWiki API error response."""
+    error = data.get("error")
+    if isinstance(error, dict):
+        message = error.get("info") or error.get("code")
+        if message:
+            return str(message)
+    return f"Unexpected MediaWiki API response: {data!r}"
+
+
 wiki_hostname = "en.wikipedia.org"
 wiki_api_php = f"https://{wiki_hostname}/w/api.php"
 user_agent = "add-links/0.1"
@@ -103,8 +113,5 @@ def edit_page(
     }
     ret = call(params, timeout=30)
     if "edit" not in ret:
-        print("params")
-        pprint(params)
-        print()
-        pprint(ret)
+        raise APIError(_format_api_error(ret))
     return typing.cast(str, ret["edit"])

add_links/mediawiki_api_old.py

@@ -4,6 +4,9 @@ from typing import Any
 
 import requests
 
+from .api import wikimedia_log_config
+from .wikimedia_api_logging import logged_request
+
 wiki_hostname = "en.wikipedia.org"
 wiki_api_php = f"https://{wiki_hostname}/w/api.php"
 user_agent = "dab-mechanic/0.1"
@@ -27,8 +30,10 @@ def parse_page(enwiki: str) -> dict[str, Any]:
 
 def get(params: dict[str, str | int]) -> dict[str, Any]:
     """Make GET request to mediawiki API."""
-    data: dict[str, Any] = requests.get(
-        wiki_api_php, headers={"User-Agent": user_agent}, params=params
+    session = requests.Session()
+    session.headers = {"User-Agent": user_agent}
+    data: dict[str, Any] = logged_request(
+        session, wikimedia_log_config, "GET", wiki_api_php, params=params
     ).json()
     return data
 

add_links/mediawiki_oauth.py

@@ -9,7 +9,8 @@ import requests
 from flask import current_app, session
 from requests_oauthlib import OAuth1Session
 
-from .api import ua
+from .api import ua, wikimedia_log_config
+from .wikimedia_api_logging import logged_request
 
 wiki_hostname = "en.wikipedia.org"
 api_url = f"https://{wiki_hostname}/w/api.php"
@@ -44,7 +45,15 @@ def api_post_request(params: dict[str, str | int], timeout: int = 4) -> requests
     )
     oauth.headers.update({"User-Agent": ua})
     proxies = get_edit_proxy()
-    return oauth.post(api_url, data=params, timeout=timeout, proxies=proxies)
+    return logged_request(
+        oauth,
+        wikimedia_log_config,
+        "POST",
+        api_url,
+        data=params,
+        timeout=timeout,
+        proxies=proxies,
+    )
 
 
 def raw_request(params: typing.Mapping[str, str | int]) -> requests.Response:
@@ -63,8 +72,13 @@ def raw_request(params: typing.Mapping[str, str | int]) -> requests.Response:
     )
     oauth.headers.update({"User-Agent": ua})
     proxies = get_edit_proxy()
-    return oauth.get(
-        api_url + "?" + urllib.parse.urlencode(params), timeout=4, proxies=proxies
+    return logged_request(
+        oauth,
+        wikimedia_log_config,
+        "GET",
+        api_url + "?" + urllib.parse.urlencode(params),
+        timeout=4,
+        proxies=proxies,
     )
 
 

add_links/wikimedia_api_logging.py

@@ -0,0 +1,222 @@
+"""JSONL logging helpers for Wikimedia API request metrics."""
+
+import json
+import logging
+import os
+import socket
+import time
+from collections.abc import Mapping
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from types import TracebackType
+from urllib.parse import parse_qs, urlparse
+
+import requests
+
+
+@dataclass(frozen=True)
+class WikimediaApiLogConfig:
+    """Configuration for Wikimedia API request logging."""
+
+    tool: str
+    log_path: Path
+    user_agent: str
+
+
+@dataclass(frozen=True)
+class WikimediaApiRequestMetric:
+    """Details of one Wikimedia API request."""
+
+    tool: str
+    url: str
+    method: str
+    status_code: int | None
+    elapsed_ms: int
+    user_agent: str
+    request_params: Mapping[str, object] | None = None
+    error: str | None = None
+
+
+_logger_cache: dict[Path, logging.Logger] = {}
+
+
+def setup_wikimedia_api_logger(log_path: Path) -> logging.Logger:
+    """Create a JSONL logger for Wikimedia API request metrics."""
+    if log_path in _logger_cache:
+        return _logger_cache[log_path]
+
+    logger_name = f"wikimedia_api_metrics.{log_path}"
+    logger = logging.getLogger(logger_name)
+    logger.setLevel(logging.INFO)
+    logger.propagate = False
+
+    if not logger.handlers:
+        try:
+            log_path.parent.mkdir(parents=True, exist_ok=True)
+            handler: logging.Handler = logging.FileHandler(log_path)
+        except OSError:
+            handler = logging.NullHandler()
+        handler.setFormatter(logging.Formatter("%(message)s"))
+        logger.addHandler(handler)
+
+    _logger_cache[log_path] = logger
+    return logger
+
+
+def get_mediawiki_action(
+    url: str, request_params: Mapping[str, object] | None = None
+) -> str | None:
+    """Extract the MediaWiki API action from params or a URL, if present."""
+    if request_params is not None:
+        action = request_params.get("action")
+        if isinstance(action, str):
+            return action
+
+    parsed = urlparse(url)
+    query = parse_qs(parsed.query)
+    values = query.get("action")
+
+    if not values:
+        return None
+
+    return values[0]
+
+
+def get_flask_request_context() -> tuple[str | None, str | None]:
+    """Return Flask endpoint and browser URL when running in a request."""
+    try:
+        from flask import has_request_context, request
+
+        if not has_request_context():
+            return None, None
+        return request.endpoint, request.url
+    except RuntimeError:
+        return None, None
+
+
+def build_log_record(metric: WikimediaApiRequestMetric) -> dict[str, object]:
+    """Build a JSON-serialisable log record for one API request."""
+    parsed = urlparse(metric.url)
+    flask_endpoint, flask_url = get_flask_request_context()
+
+    record: dict[str, object] = {
+        "ts": datetime.now(UTC).isoformat(),
+        "tool": metric.tool,
+        "host": socket.gethostname(),
+        "pid": os.getpid(),
+        "method": metric.method,
+        "api_host": parsed.netloc,
+        "path": parsed.path,
+        "action": get_mediawiki_action(metric.url, metric.request_params),
+        "status_code": metric.status_code,
+        "elapsed_ms": metric.elapsed_ms,
+        "user_agent": metric.user_agent,
+        "flask_endpoint": flask_endpoint,
+        "flask_url": flask_url,
+    }
+
+    if metric.error is not None:
+        record["error"] = metric.error
+
+    return record
+
+
+def log_wikimedia_api_request(
+    logger: logging.Logger,
+    metric: WikimediaApiRequestMetric,
+) -> None:
+    """Write one Wikimedia API request metric as a JSONL log line."""
+    record = build_log_record(metric)
+    logger.info(json.dumps(record, separators=(",", ":"), sort_keys=True))
+
+
+class WikimediaRequestTimer:
+    """Context manager for timing and logging a Wikimedia API request."""
+
+    def __init__(
+        self,
+        config: WikimediaApiLogConfig,
+        method: str,
+        url: str,
+        request_params: Mapping[str, object] | None = None,
+    ) -> None:
+        self.config = config
+        self.method = method
+        self.url = url
+        self.request_params = request_params
+        self.started = 0.0
+        self.logger = setup_wikimedia_api_logger(config.log_path)
+
+    def __enter__(self) -> "WikimediaRequestTimer":
+        """Start timing a request."""
+        self.started = time.monotonic()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> bool:
+        """Log failed requests when an exception escapes."""
+        if exc is None:
+            return False
+
+        elapsed_ms = int((time.monotonic() - self.started) * 1000)
+
+        log_wikimedia_api_request(
+            self.logger,
+            WikimediaApiRequestMetric(
+                tool=self.config.tool,
+                url=self.url,
+                method=self.method,
+                status_code=None,
+                elapsed_ms=elapsed_ms,
+                user_agent=self.config.user_agent,
+                request_params=self.request_params,
+                error=type(exc).__name__,
+            ),
+        )
+
+        return False
+
+    def log_response(self, status_code: int, final_url: str | None = None) -> None:
+        """Log a completed request."""
+        elapsed_ms = int((time.monotonic() - self.started) * 1000)
+
+        log_wikimedia_api_request(
+            self.logger,
+            WikimediaApiRequestMetric(
+                tool=self.config.tool,
+                url=final_url or self.url,
+                method=self.method,
+                status_code=status_code,
+                elapsed_ms=elapsed_ms,
+                user_agent=self.config.user_agent,
+                request_params=self.request_params,
+            ),
+        )
+
+
+def logged_request(
+    session: requests.sessions.Session,
+    config: WikimediaApiLogConfig,
+    method: str,
+    url: str,
+    **kwargs: object,
+) -> requests.Response:
+    """Make a Wikimedia API request and log one JSONL metric line."""
+    request_params = kwargs.get("params") or kwargs.get("data")
+    if not isinstance(request_params, Mapping):
+        request_params = None
+
+    with WikimediaRequestTimer(
+        config, method.upper(), url, request_params=request_params
+    ) as timer:
+        response = getattr(session, method.lower())(url, **kwargs)
+        final_url = getattr(response, "url", None)
+        timer.log_response(
+            response.status_code, final_url if isinstance(final_url, str) else None
+        )
+        return response

templates/article.html

@@ -85,6 +85,12 @@
   const redirectTo = {{ redirect_to | tojson }};
   const apiUrl = {{ url_for('api_valid_hit') | tojson }};
   const pageUrl = new URL(window.location.href);
+  const cleanPageUrl = new URL(pageUrl);
+  cleanPageUrl.searchParams.delete('title');
+  cleanPageUrl.searchParams.delete('after');
+  cleanPageUrl.searchParams.delete('skip');
+  const storageKey = 'missinglink.skipped.' + linkTo;
+  const skippedTitles = [];
 
   const elProgress = document.getElementById('search-progress');
   const elStatus   = document.getElementById('search-status');
@@ -101,8 +107,39 @@
     if (elCount) elCount.textContent = elList.children.length;
   }
 
+  function loadSkippedTitles() {
+    try {
+      const stored = JSON.parse(localStorage.getItem(storageKey) || '[]');
+      if (Array.isArray(stored)) {
+        for (const title of stored) rememberSkipped(title, false);
+      }
+    } catch (e) {
+      try { localStorage.removeItem(storageKey); } catch (e) {}
+    }
+
+    for (const title of {{ skipped_titles | tojson }}) rememberSkipped(title, false);
+    saveSkippedTitles();
+  }
+
+  function saveSkippedTitles() {
+    try {
+      localStorage.setItem(storageKey, JSON.stringify(skippedTitles));
+    } catch (e) {}
+  }
+
+  function rememberSkipped(title, save = true) {
+    if (!title || skippedTitles.includes(title)) return;
+    skippedTitles.push(title);
+    if (save) saveSkippedTitles();
+  }
+
   async function search() {
+    loadSkippedTitles();
+    for (const hitTitle of skippedTitles) removeCandidate(hitTitle);
+
     for (const hitTitle of hits) {
+      if (skippedTitles.includes(hitTitle)) continue;
+
       elStatus.textContent = `Checking "${hitTitle}"…`;
 
       let data;
@@ -116,7 +153,11 @@
         continue;
       }
 
-      if (!data.valid) { removeCandidate(hitTitle); continue; }
+      if (!data.valid) {
+        rememberSkipped(hitTitle);
+        removeCandidate(hitTitle);
+        continue;
+      }
 
       elProgress.hidden = true;
 
@@ -126,13 +167,15 @@
       document.getElementById('diff-table').innerHTML = data.diff;
       document.getElementById('hit-input').value = hitTitle;
 
-      const skipUrl = new URL(pageUrl);
-      skipUrl.searchParams.delete('title');
-      skipUrl.searchParams.set('after', hitTitle);
-      document.getElementById('skip-link').href = skipUrl.toString();
+      const elSkipLink = document.getElementById('skip-link');
+      elSkipLink.href = cleanPageUrl.toString();
+      elSkipLink.onclick = function () {
+        rememberSkipped(hitTitle);
+      };
 
       const currentUrl = new URL(pageUrl);
       currentUrl.searchParams.delete('after');
+      currentUrl.searchParams.delete('skip');
       currentUrl.searchParams.set('title', hitTitle);
       history.replaceState(null, '', currentUrl.toString());
 

templates/base.html

@@ -6,15 +6,44 @@
   <meta name="viewport" content="width=device-width, initial-scale=1.0" />
   <link rel="icon" type="image/svg+xml" href="{{ url_for('static', filename='favicon.svg') }}">
   <title>{% block title %}{% endblock %} – Missing Link</title>
+  <style>
+    .site-navbar__tools {
+      margin-left: auto;
+    }
+
+    @media (max-width: 767.98px) {
+      .site-navbar .container {
+        align-items: flex-start;
+        gap: .5rem;
+      }
+
+      .site-navbar__tools {
+        flex-wrap: wrap;
+        margin-left: 0;
+        width: 100%;
+      }
+
+      .site-navbar__search {
+        flex: 1 1 100%;
+        min-width: 0;
+      }
+
+      .site-navbar__search .form-control {
+        flex: 1 1 auto;
+        min-width: 0;
+        width: auto !important;
+      }
+    }
+  </style>
   {% block style %}{% endblock %}
 </head>
 
 <body>
-  <nav class="navbar navbar-expand-md navbar-dark bg-dark mb-4">
+  <nav class="navbar navbar-expand-md navbar-dark bg-dark mb-4 site-navbar">
     <div class="container">
       <a class="navbar-brand fw-semibold" href="{{ url_for('index') }}">Missing Link</a>
-      <div class="d-flex align-items-center gap-2">
-        <form class="d-flex" action="{{ url_for('index') }}">
+      <div class="d-flex align-items-center gap-2 site-navbar__tools">
+        <form class="d-flex site-navbar__search" action="{{ url_for('index') }}">
           <input class="form-control form-control-sm me-2" name="q" placeholder="Article title…" style="width:240px">
           <button class="btn btn-outline-light btn-sm" type="submit">Go</button>
         </form>

templates/error.html

@@ -8,7 +8,7 @@
     <div class="col-md-7">
       <div class="alert alert-danger">
         <h4 class="alert-heading">Something went wrong</h4>
-        <p class="mb-0"><code>{{ message | e | replace('\n', '<br>\n') | safe }}</code></p>
+        <code class="d-block mb-0" style="white-space: pre-line">{{ message }}</code>
       </div>
       <a href="{{ url_for('index') }}" class="btn btn-outline-secondary btn-sm">Back to home</a>
     </div>

test_api.py

@@ -3,6 +3,7 @@ from unittest.mock import Mock, patch
 
 from simplejson.scanner import JSONDecodeError
 
+import web_view
 from add_links import api
 
 
@@ -19,12 +20,25 @@ class ApiGetTests(unittest.TestCase):
         session = Mock()
         session.get.return_value = response
 
-        with patch("add_links.api._get_active_session", return_value=session):
+        with patch("add_links.api._get_active_session", return_value=(session, "anon")):
             with self.assertRaises(api.MediawikiError) as ctx:
                 api.api_get({"action": "query"})
 
         self.assertEqual(str(ctx.exception), response.text)
 
+    def test_active_session_uses_logged_in_request_session(self) -> None:
+        oauth_session = Mock()
+
+        with web_view.app.test_request_context("/"):
+            with patch(
+                "add_links.mediawiki_oauth.get_oauth_session",
+                return_value=oauth_session,
+            ):
+                session, auth_mode = api._get_active_session()
+
+        self.assertIs(session, oauth_session)
+        self.assertEqual(auth_mode, "oauth")
+
 
 if __name__ == "__main__":
     unittest.main()

test_match.py

@@ -70,6 +70,26 @@ class FindLinkInContentTests(unittest.TestCase):
         self.assertEqual(replacement, "electoral fraud|ballot stuffing")
         self.assertEqual(replaced_text, "ballot stuffing")
 
+    def test_italic_text_inside_existing_link_does_not_split_link(self) -> None:
+        content = (
+            'For instance, "[[capitalism]]" is the name for the '
+            "[[Capitalist mode of production (Marxist theory)|capitalist "
+            "''mode of production'']] in which the means of production"
+        )
+
+        new_content, replacement, replaced_text = find_link_in_content(
+            "capitalist mode of production", content, "Means of production"
+        )
+
+        self.assertEqual(
+            new_content,
+            'For instance, "[[capitalism]]" is the name for the '
+            "[[Capitalist mode of production (Marxist theory)|capitalist "
+            "''mode of production'']] in which the [[means of production]]",
+        )
+        self.assertEqual(replacement, "means of production")
+        self.assertEqual(replaced_text, "means of production")
+
 
 if __name__ == "__main__":
     unittest.main()

test_mediawiki_api.py

@@ -0,0 +1,46 @@
+import unittest
+from unittest.mock import patch
+
+from add_links import mediawiki_api
+
+
+class EditPageTests(unittest.TestCase):
+    def test_edit_page_raises_api_error_with_mediawiki_message(self) -> None:
+        with patch(
+            "add_links.mediawiki_api.call",
+            return_value={"error": {"code": "badtoken", "info": "Invalid CSRF token."}},
+        ):
+            with self.assertRaises(mediawiki_api.APIError) as ctx:
+                mediawiki_api.edit_page(
+                    pageid=1,
+                    section=0,
+                    text="text",
+                    summary="summary",
+                    baserevid="123",
+                    token="bad",
+                )
+
+        self.assertEqual(str(ctx.exception), "Invalid CSRF token.")
+
+    def test_edit_page_raises_api_error_for_unexpected_response(self) -> None:
+        response = {"servedby": "mw-api-ext.eqiad.main"}
+
+        with patch("add_links.mediawiki_api.call", return_value=response):
+            with self.assertRaises(mediawiki_api.APIError) as ctx:
+                mediawiki_api.edit_page(
+                    pageid=1,
+                    section=0,
+                    text="text",
+                    summary="summary",
+                    baserevid="123",
+                    token="token",
+                )
+
+        self.assertEqual(
+            str(ctx.exception),
+            "Unexpected MediaWiki API response: {'servedby': 'mw-api-ext.eqiad.main'}",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

test_web_view.py

@@ -0,0 +1,225 @@
+import unittest
+from unittest.mock import patch
+
+import web_view
+
+
+def hit(title: str) -> web_view.Hit:
+    return {
+        "ns": 0,
+        "title": title,
+        "pageid": 1,
+        "size": 1,
+        "wordcount": 1,
+        "snippet": "",
+        "timestamp": "",
+    }
+
+
+class ArticlePageSkipTests(unittest.TestCase):
+    def setUp(self) -> None:
+        web_view.app.config["TESTING"] = True
+        self.client = web_view.app.test_client()
+
+    def test_session_skips_are_left_out_of_candidates(self) -> None:
+        hits = [
+            hit("Mass racial violence in the United States"),
+            hit("National Register of Historic Places listings in Brookhaven (town), New York"),
+            hit("Useful candidate"),
+        ]
+
+        with self.client.session_transaction() as session:
+            session["skipped"] = {
+                "African-American neighborhood": [
+                    "Mass racial violence in the United States",
+                    "National Register of Historic Places listings in Brookhaven (town), New York",
+                ]
+            }
+
+        with (
+            patch("web_view.mediawiki_oauth.get_username", return_value="Test user"),
+            patch("web_view.mediawiki_oauth.get_oauth_session", return_value=object()),
+            patch("web_view.api.get_wiki_info", return_value=None),
+            patch("web_view.search_count", return_value=3),
+            patch("web_view.search_count_with_link", return_value=0),
+            patch("web_view.search_no_link", return_value=(3, hits)),
+        ):
+            response = self.client.get("/link/African-American_neighborhood")
+
+        body = response.get_data(as_text=True)
+        self.assertEqual(response.status_code, 200)
+        self.assertNotIn(
+            'data-title="Mass racial violence in the United States"', body
+        )
+        self.assertNotIn(
+            'data-title="National Register of Historic Places listings in Brookhaven (town), New York"',
+            body,
+        )
+        self.assertIn("Useful candidate", body)
+
+    def test_redirect_target_is_left_out_of_candidates(self) -> None:
+        hits = [
+            hit("Documentary film"),
+            hit("Useful candidate"),
+        ]
+
+        with (
+            patch("web_view.mediawiki_oauth.get_username", return_value="Test user"),
+            patch("web_view.mediawiki_oauth.get_oauth_session", return_value=object()),
+            patch("web_view.api.get_wiki_info", return_value="Documentary film"),
+            patch("web_view.search_count", return_value=2),
+            patch("web_view.search_count_with_link", return_value=0),
+            patch("web_view.search_no_link", return_value=(2, hits)),
+        ):
+            response = self.client.get("/link/DVD_documentary")
+
+        body = response.get_data(as_text=True)
+        self.assertEqual(response.status_code, 200)
+        self.assertNotIn('data-title="Documentary film"', body)
+        self.assertIn("Useful candidate", body)
+
+    def test_redirect_target_case_variant_is_left_out_of_candidates(self) -> None:
+        hits = [hit("documentary film")]
+
+        with (
+            patch("web_view.mediawiki_oauth.get_username", return_value="Test user"),
+            patch("web_view.mediawiki_oauth.get_oauth_session", return_value=object()),
+            patch("web_view.api.get_wiki_info", return_value="Documentary film"),
+            patch("web_view.search_count", return_value=1),
+            patch("web_view.search_count_with_link", return_value=0),
+            patch("web_view.search_no_link", return_value=(1, hits)),
+        ):
+            response = self.client.get("/link/DVD_documentary")
+
+        body = response.get_data(as_text=True)
+        self.assertEqual(response.status_code, 200)
+        self.assertIn("No more candidates found for this article.", body)
+
+
+class ValidHitTests(unittest.TestCase):
+    def setUp(self) -> None:
+        web_view.app.config["TESTING"] = True
+        self.client = web_view.app.test_client()
+
+    def test_redirect_target_is_not_valid_hit(self) -> None:
+        with (
+            patch("web_view.mediawiki_oauth.get_username", return_value="Test user"),
+            patch("web_view.mediawiki_oauth.get_oauth_session", return_value=object()),
+            patch("web_view.get_diff") as get_diff,
+        ):
+            response = self.client.get(
+                "/api/1/valid_hit",
+                query_string={
+                    "link_to": "DVD documentary",
+                    "link_from": "Documentary film",
+                    "redirect_to": "Documentary film",
+                },
+            )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.get_json(), {"valid": False})
+        get_diff.assert_not_called()
+
+
+class LoginRequiredTests(unittest.TestCase):
+    def setUp(self) -> None:
+        web_view.app.config["TESTING"] = True
+        self.client = web_view.app.test_client()
+
+    def test_article_page_redirects_to_login_before_api_calls(self) -> None:
+        with (
+            patch("web_view.mediawiki_oauth.get_username", return_value=None),
+            patch("web_view.mediawiki_oauth.get_oauth_session", return_value=None),
+            patch("web_view.api.get_wiki_info") as get_wiki_info,
+        ):
+            response = self.client.get("/link/DVD_documentary")
+
+        self.assertEqual(response.status_code, 302)
+        self.assertIn("/oauth/start", response.location)
+        get_wiki_info.assert_not_called()
+
+    def test_valid_hit_requires_login_before_diff_call(self) -> None:
+        with (
+            patch("web_view.mediawiki_oauth.get_username", return_value=None),
+            patch("web_view.mediawiki_oauth.get_oauth_session", return_value=None),
+            patch("web_view.get_diff") as get_diff,
+        ):
+            response = self.client.get(
+                "/api/1/valid_hit",
+                query_string={"link_to": "Target", "link_from": "Candidate"},
+            )
+
+        self.assertEqual(response.status_code, 401)
+        self.assertEqual(response.get_json(), {"error": "login required"})
+        get_diff.assert_not_called()
+
+    def test_hits_api_requires_login_before_search(self) -> None:
+        with (
+            patch("web_view.mediawiki_oauth.get_username", return_value=None),
+            patch("web_view.mediawiki_oauth.get_oauth_session", return_value=None),
+            patch("web_view.core.do_search") as do_search,
+        ):
+            response = self.client.get(
+                "/api/1/hits", query_string={"title": "Target"}
+            )
+
+        self.assertEqual(response.status_code, 401)
+        self.assertEqual(response.get_json(), {"error": "login required"})
+        do_search.assert_not_called()
+
+
+class ProtectedCandidateTests(unittest.TestCase):
+    def test_edit_protected_hits_are_filtered_out(self) -> None:
+        hits = [hit("Editable"), hit("Protected")]
+        hits[0]["pageid"] = 10
+        hits[1]["pageid"] = 20
+
+        with patch(
+            "web_view.get_page_info_by_id",
+            return_value={
+                10: {"pageid": 10, "title": "Editable", "protection": []},
+                20: {
+                    "pageid": 20,
+                    "title": "Protected",
+                    "protection": [
+                        {"type": "edit", "level": "sysop", "expiry": "infinity"}
+                    ],
+                },
+            },
+        ):
+            filtered = web_view.filter_edit_protected_hits(hits)
+
+        self.assertEqual([item["title"] for item in filtered], ["Editable"])
+
+    def test_search_no_link_filters_edit_protected_results(self) -> None:
+        hits = [hit("Editable"), hit("Protected")]
+        hits[0]["pageid"] = 10
+        hits[1]["pageid"] = 20
+
+        with (
+            patch(
+                "web_view.run_search",
+                return_value={"searchinfo": {"totalhits": 2}, "search": hits},
+            ),
+            patch(
+                "web_view.get_page_info_by_id",
+                return_value={
+                    10: {"pageid": 10, "title": "Editable", "protection": []},
+                    20: {
+                        "pageid": 20,
+                        "title": "Protected",
+                        "protection": [
+                            {"type": "edit", "level": "sysop", "expiry": "infinity"}
+                        ],
+                    },
+                },
+            ),
+        ):
+            totalhits, filtered = web_view.search_no_link("Target")
+
+        self.assertEqual(totalhits, 2)
+        self.assertEqual([item["title"] for item in filtered], ["Editable"])
+
+
+if __name__ == "__main__":
+    unittest.main()

web_view.py

@@ -1,7 +1,6 @@
 #!/usr/bin/python3
 
 import html
-import itertools
 import json
 import re
 import sys
@@ -37,6 +36,14 @@ class Hit(typing.TypedDict):
     timestamp: str
 
 
+class PageInfo(typing.TypedDict, total=False):
+    """Wikipedia page metadata."""
+
+    pageid: int
+    title: str
+    protection: list[dict[str, str]]
+
+
 def render_error(message: str) -> str:
     """Render shared error page."""
     return flask.render_template("error.html", message=message)
@@ -97,6 +104,28 @@ def article_url(title: str) -> str:
     return flask.url_for("article_page", url_title=title.replace(" ", "_"))
 
 
+def canonical_title_key(title: str) -> str:
+    """Normalize a Wikipedia article title for same-page comparisons."""
+    title = title.replace("_", " ").strip()
+    return title[:1].casefold() + title[1:]
+
+
+def same_article_title(a: str | None, b: str | None) -> bool:
+    """Compare titles using Wikipedia's first-letter case-insensitive rule."""
+    if not a or not b:
+        return False
+    return canonical_title_key(a) == canonical_title_key(b)
+
+
+def is_self_link_candidate(
+    hit_title: str, from_title: str, redirect_to: str | None = None
+) -> bool:
+    """Return true when a candidate is the source article or its redirect target."""
+    return same_article_title(hit_title, from_title) or same_article_title(
+        hit_title, redirect_to
+    )
+
+
 def get_hit_count(q: str) -> int:
     """Search Wikipedia and return hit count."""
     return typing.cast(int, run_search(q, limit=0)["searchinfo"]["totalhits"])
@@ -115,13 +144,60 @@ def search_count_with_link(q: str, redirect_to: str | None = None) -> int:
     return count
 
 
+def chunks(items: list[int], size: int) -> typing.Iterator[list[int]]:
+    """Yield fixed-size chunks from a list."""
+    for offset in range(0, len(items), size):
+        yield items[offset : offset + size]
+
+
+def get_page_info_by_id(pageids: list[int]) -> dict[int, PageInfo]:
+    """Fetch metadata for pages by page ID."""
+    if not pageids:
+        return {}
+
+    pages: dict[int, PageInfo] = {}
+    for pageid_chunk in chunks(pageids, 50):
+        ret = api.api_get(
+            {
+                "prop": "info",
+                "inprop": "protection",
+                "pageids": "|".join(str(pageid) for pageid in pageid_chunk),
+            }
+        )
+        for page in ret["query"]["pages"]:
+            pageid = typing.cast(int, page["pageid"])
+            pages[pageid] = typing.cast(PageInfo, page)
+    return pages
+
+
+def is_edit_protected(page: PageInfo | None) -> bool:
+    """Check if a page has edit protection."""
+    if not page:
+        return False
+    return any(
+        protection.get("type") == "edit" or protection.get("action") == "edit"
+        for protection in page.get("protection", [])
+    )
+
+
+def filter_edit_protected_hits(hits: list[Hit]) -> list[Hit]:
+    """Remove candidates that cannot be edited because of page protection."""
+    page_info = get_page_info_by_id([hit["pageid"] for hit in hits])
+    return [
+        hit for hit in hits if not is_edit_protected(page_info.get(hit["pageid"]))
+    ]
+
+
 def search_no_link(q: str, redirect_to: str | None = None) -> tuple[int, list[Hit]]:
     """Search for mentions of article title with no link included."""
     exclude = f' -linksto:"{q}"'
     if redirect_to:
         exclude += f' -linksto:"{redirect_to}"'
     query = run_search(article_title_to_search_query(q) + exclude, "max")
-    return (query["searchinfo"]["totalhits"], query["search"])
+    return (
+        query["searchinfo"]["totalhits"],
+        filter_edit_protected_hits(query["search"]),
+    )
 
 
 @app.before_request
@@ -131,6 +207,22 @@ def global_user() -> None:
     flask.g.oauth_session = mediawiki_oauth.get_oauth_session()
 
 
+def is_logged_in() -> bool:
+    """Return true when the current request has a Wikipedia OAuth session."""
+    return flask.g.oauth_session is not None
+
+
+def login_redirect() -> Response:
+    """Redirect the current browser request through Wikipedia OAuth."""
+    next_url = flask.request.full_path.rstrip("?")
+    return flask.redirect(flask.url_for("start_oauth", next=next_url))
+
+
+def login_required_json() -> tuple[werkzeug.wrappers.response.Response, int]:
+    """Return a JSON login-required response without making MediaWiki API calls."""
+    return flask.jsonify(error="login required"), 401
+
+
 @app.route("/")
 def index() -> str | Response:
     """Index page."""
@@ -317,6 +409,9 @@ def match_type(q: str, snippet: str) -> str | None:
 
 def _record_skip(from_title: str, hit_title: str) -> None:
     """Record that a candidate was skipped or saved for this article."""
+    hit_title = hit_title.strip()
+    if not hit_title:
+        return
     skipped: dict[str, list[str]] = flask.session.get("skipped", {})
     article_skipped = skipped.get(from_title, [])
     if hit_title not in article_skipped:
@@ -327,6 +422,10 @@ def _record_skip(from_title: str, hit_title: str) -> None:
 
 def handle_post(url_title: str) -> Response:
     """Handle POST request."""
+    if not is_logged_in():
+        next_url = flask.url_for("article_page", url_title=url_title)
+        return flask.redirect(flask.url_for("start_oauth", next=next_url))
+
     from_title = url_title.replace("_", " ").strip()
     hit_title = flask.request.form["hit"]
     try:
@@ -334,6 +433,8 @@ def handle_post(url_title: str) -> Response:
     except mediawiki_oauth.LoginNeeded:
         next_url = flask.url_for("article_page", url_title=url_title, title=hit_title)
         return flask.redirect(flask.url_for("start_oauth", next=next_url))
+    except NoMatch:
+        return flask.make_response(render_error("No valid unlinked mention found."), 400)
     except (mediawiki_api.APIError, api.MediawikiError) as e:
         return flask.make_response(
             render_mediawiki_error(e, prefix="Save failed"), 502
@@ -355,6 +456,9 @@ def article_page(url_title: str) -> str | Response:
     if flask.request.method == "POST":
         return handle_post(url_title)
 
+    if not is_logged_in():
+        return login_redirect()
+
     from_title = url_title.replace("_", " ").strip()
 
     try:
@@ -371,10 +475,8 @@ def article_page(url_title: str) -> str | Response:
             render_mediawiki_error(e), 502
         )
 
-    # Filter out candidates already processed this session
-    session_skipped: set[str] = set(
-        flask.session.get("skipped", {}).get(from_title, [])
-    )
+    article_skipped = flask.session.get("skipped", {}).get(from_title, [])
+    session_skipped: set[str] = set(article_skipped)
 
     # If a specific candidate was requested, move it to the front
     title_param = flask.request.args.get("title")
@@ -382,14 +484,12 @@ def article_page(url_title: str) -> str | Response:
         hits = [h for h in hits if h["title"] == title_param] + \
                [h for h in hits if h["title"] != title_param]
 
-    # Record and apply explicit skip-past
-    after = flask.request.args.get("after")
-    if after:
-        _record_skip(from_title, after)
-        session_skipped.add(after)
-
-    hits = [h for h in hits if h["title"] not in session_skipped
-            and h["title"] != from_title and h["title"] != case_flip_first(from_title)]
+    hits = [
+        h
+        for h in hits
+        if h["title"] not in session_skipped
+        and not is_self_link_candidate(h["title"], from_title, redirect_to)
+    ]
 
     if not hits:
         return flask.render_template("all_done.html")
@@ -404,19 +504,23 @@ def article_page(url_title: str) -> str | Response:
         with_link=with_link,
         hits=hits,
         url_title=url_title,
+        skipped_titles=article_skipped,
         saves_this_session=saves_this_session,
     )
 
 
 def do_save(title: str, hit_title: str) -> str:
     """Update page on Wikipedia."""
-    token = mediawiki_oauth.get_token()
-
     try:
         redirect_to = api.get_wiki_info(title)
     except (api.MissingPage, api.MultipleRedirects, api.MediawikiError):
         redirect_to = None
 
+    if is_self_link_candidate(hit_title, title, redirect_to):
+        raise NoMatch
+
+    token = mediawiki_oauth.get_token()
+
     found = get_match(title, hit_title, redirect_to)
 
     summary = (
@@ -444,6 +548,9 @@ def save_done() -> str:
 @app.route("/api/1/hits")
 def api_hits() -> werkzeug.wrappers.response.Response:
     """Return candidates for the given article title."""
+    if not is_logged_in():
+        return login_required_json()
+
     title = flask.request.args.get("title")
     assert title
     ret = core.do_search(title)
@@ -456,10 +563,17 @@ def api_hits() -> werkzeug.wrappers.response.Response:
 @app.route("/api/1/valid_hit")
 def api_valid_hit() -> werkzeug.wrappers.response.Response:
     """Check if a candidate article has a valid unlinked mention."""
+    if not is_logged_in():
+        return login_required_json()
+
     link_to = flask.request.args["link_to"]
     link_from = flask.request.args["link_from"]
     redirect_to = flask.request.args.get("redirect_to") or None
 
+    if is_self_link_candidate(link_from, link_to, redirect_to):
+        _record_skip(link_to, link_from)
+        return flask.jsonify(valid=False)
+
     try:
         found = get_diff(link_to, link_from, redirect_to)
     except NoMatch: