Implement merge.md migration roadmap: Extract super/ to clean architecture

mw_api/auth/__init__.py

@@ -6,8 +6,12 @@
 """
 from .authenticator import Authenticator
 from .token_provider import TokenProvider
+from .cookie_store import CookieStore
+from .session_manager import SessionManager
 
 __all__ = [
     "Authenticator",
     "TokenProvider",
+    "CookieStore",
+    "SessionManager",
 ]

mw_api/auth/cookie_store.py

@@ -0,0 +1,212 @@
+# -*- coding: utf-8 -*-
+"""
+Cookie Store for managing authentication cookies.
+
+Provides the CookieStore class for saving and loading cookies,
+extracted from cookies_bot.py for better separation of concerns.
+"""
+import os
+import stat
+from pathlib import Path
+from datetime import datetime, timedelta
+from typing import Optional
+import requests
+
+from ..core.config import BotConfig, get_default_config
+
+
+class CookieStore:
+    """
+    Store for managing authentication cookies.
+
+    Handles cookie persistence, validation, and cleanup.
+
+    Attributes:
+        cookies_dir: Directory where cookies are stored.
+        config: BotConfig for cookie behavior settings.
+    """
+
+    def __init__(
+        self,
+        cookies_dir: Optional[Path] = None,
+        config: Optional[BotConfig] = None
+    ) -> None:
+        """
+        Initialize the CookieStore.
+
+        Args:
+            cookies_dir: Directory for cookie storage. Defaults to ~/cookies.
+            config: BotConfig instance for cookie behavior.
+        """
+        self._config = config if config is not None else get_default_config()
+
+        # Set up cookies directory
+        if cookies_dir is None:
+            tool = os.getenv("HOME")
+            if not tool:
+                tool = Path(__file__).parent.parent.parent
+            else:
+                tool = Path(tool)
+            self.cookies_dir = tool / "cookies"
+        else:
+            self.cookies_dir = cookies_dir
+
+        # Create directory if it doesn't exist
+        if not self.cookies_dir.exists():
+            self.cookies_dir.mkdir(parents=True, exist_ok=True)
+            statgroup = stat.S_IRWXU | stat.S_IRWXG
+            os.chmod(self.cookies_dir, statgroup)
+
+    def get_cookie_path(self, lang: str, family: str, username: str) -> Path:
+        """
+        Get the path to the cookie file for a specific user.
+
+        Args:
+            lang: Language code.
+            family: Wiki family (e.g., 'wikipedia').
+            username: Username.
+
+        Returns:
+            Path to the cookie file.
+        """
+        if self._config.no_cookies:
+            # Use random filename if cookies are disabled
+            randome = os.urandom(8).hex()
+            return self.cookies_dir / f"{randome}.txt"
+
+        lang = lang.lower()
+        family = family.lower()
+        username = username.lower().replace(" ", "_").split("@")[0]
+
+        return self.cookies_dir / f"{family}_{lang}_{username}.txt"
+
+    def _is_cookie_valid(self, cookie_path: Path) -> bool:
+        """
+        Check if a cookie file is valid.
+
+        Args:
+            cookie_path: Path to the cookie file.
+
+        Returns:
+            True if the cookie is valid, False otherwise.
+        """
+        if not cookie_path.exists():
+            return False
+
+        # Check if file is empty
+        if not cookie_path.stat().st_size:
+            return False
+
+        # Check if file is older than 3 days
+        file_time = datetime.fromtimestamp(cookie_path.stat().st_mtime)
+        if datetime.now() - file_time > timedelta(days=3):
+            return False
+
+        return True
+
+    def load_cookies(self, lang: str, family: str, username: str) -> Optional[str]:
+        """
+        Load cookies from storage.
+
+        Args:
+            lang: Language code.
+            family: Wiki family.
+            username: Username.
+
+        Returns:
+            Cookie string if found and valid, None otherwise.
+        """
+        cookie_path = self.get_cookie_path(lang, family, username)
+
+        if not self._is_cookie_valid(cookie_path):
+            # Clean up invalid cookie file
+            if cookie_path.exists():
+                self._delete_cookie_file(cookie_path)
+            return None
+
+        try:
+            with open(cookie_path, "r", encoding="utf-8") as f:
+                cookies = f.read()
+            return cookies if cookies else None
+        except Exception:
+            return None
+
+    def save_cookies(
+        self,
+        session: requests.Session,
+        lang: str,
+        family: str,
+        username: str
+    ) -> bool:
+        """
+        Save cookies from a session to storage.
+
+        Args:
+            session: The requests Session with cookies to save.
+            lang: Language code.
+            family: Wiki family.
+            username: Username.
+
+        Returns:
+            True if cookies were saved successfully, False otherwise.
+        """
+        cookie_path = self.get_cookie_path(lang, family, username)
+
+        try:
+            # Ensure the file exists with proper permissions
+            if not cookie_path.exists():
+                cookie_path.touch()
+                statgroup = stat.S_IRWXU | stat.S_IRWXG
+                os.chmod(str(cookie_path), statgroup)
+
+            # Convert session cookies to string format
+            cookie_string = self._session_to_cookie_string(session)
+
+            # Save to file
+            with open(cookie_path, "w", encoding="utf-8") as f:
+                f.write(cookie_string)
+
+            return True
+        except Exception:
+            return False
+
+    def _session_to_cookie_string(self, session: requests.Session) -> str:
+        """
+        Convert session cookies to string format.
+
+        Args:
+            session: The requests Session.
+
+        Returns:
+            Cookie string.
+        """
+        # This is a simplified implementation
+        # In a real implementation, you might want to serialize the entire cookie jar
+        cookies = []
+        for cookie in session.cookies:
+            cookies.append(f"{cookie.name}={cookie.value}")
+        return "; ".join(cookies)
+
+    def _delete_cookie_file(self, cookie_path: Path) -> None:
+        """
+        Delete a cookie file.
+
+        Args:
+            cookie_path: Path to the cookie file.
+        """
+        try:
+            cookie_path.unlink(missing_ok=True)
+        except Exception:
+            pass
+
+    def invalidate(self, lang: str, family: str, username: str) -> None:
+        """
+        Invalidate cookies for a specific user.
+
+        Args:
+            lang: Language code.
+            family: Wiki family.
+            username: Username.
+        """
+        cookie_path = self.get_cookie_path(lang, family, username)
+        self._delete_cookie_file(cookie_path)

mw_api/auth/session_manager.py

@@ -0,0 +1,152 @@
+# -*- coding: utf-8 -*-
+"""
+Session Manager for managing login sessions.
+
+Provides the SessionManager class for session caching and reuse,
+extracted from login_wrap.py for better separation of concerns.
+"""
+from typing import Dict, Tuple, Optional
+
+from ..core.protocols import LoginBotProtocol
+
+# Type alias for cache key structure
+CacheKey = Tuple[str, str, str]  # (lang, family, username)
+
+
+class SessionManager:
+    """
+    Manager for login sessions.
+
+    Handles session caching, reuse, and access tracking.
+
+    Attributes:
+        session_cache: Cache of login sessions keyed by (lang, family, username).
+        access_counts: Access counts for each cached session.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the SessionManager."""
+        self._session_cache: Dict[CacheKey, "LoginBotProtocol"] = {}
+        self._access_counts: Dict[CacheKey, int] = {}
+
+    def get_cache_key(
+        self,
+        lang: str,
+        family: str,
+        username: Optional[str] = None
+    ) -> CacheKey:
+        """
+        Get the cache key for a session.
+
+        Args:
+            lang: Language code.
+            family: Wiki family.
+            username: Username (optional).
+
+        Returns:
+            Cache key tuple.
+        """
+        if username:
+            return (lang, family, username)
+        return (lang, family, "")
+
+    def get_session(
+        self,
+        lang: str,
+        family: str,
+        username: Optional[str] = None
+    ) -> Optional[LoginBotProtocol]:
+        """
+        Get a cached session.
+
+        Args:
+            lang: Language code.
+            family: Wiki family.
+            username: Username (optional).
+
+        Returns:
+            Login bot instance if cached, None otherwise.
+        """
+        cache_key = self.get_cache_key(lang, family, username)
+        session = self._session_cache.get(cache_key)
+
+        if session is not None:
+            # Track access
+            self._access_counts.setdefault(cache_key, 0)
+            self._access_counts[cache_key] += 1
+
+        return session
+
+    def add_session(
+        self,
+        lang: str,
+        family: str,
+        session: LoginBotProtocol,
+        username: Optional[str] = None
+    ) -> None:
+        """
+        Add a session to the cache.
+
+        Args:
+            lang: Language code.
+            family: Wiki family.
+            session: Login bot instance.
+            username: Username (optional).
+        """
+        cache_key = self.get_cache_key(lang, family, username)
+        self._session_cache[cache_key] = session
+        self._access_counts[cache_key] = 1
+
+    def invalidate(
+        self,
+        lang: str,
+        family: str,
+        username: Optional[str] = None
+    ) -> None:
+        """
+        Invalidate a cached session.
+
+        Args:
+            lang: Language code.
+            family: Wiki family.
+            username: Username (optional).
+        """
+        cache_key = self.get_cache_key(lang, family, username)
+        if cache_key in self._session_cache:
+            del self._session_cache[cache_key]
+        if cache_key in self._access_counts:
+            del self._access_counts[cache_key]
+
+    def get_access_count(
+        self,
+        lang: str,
+        family: str,
+        username: Optional[str] = None
+    ) -> int:
+        """
+        Get the access count for a session.
+
+        Args:
+            lang: Language code.
+            family: Wiki family.
+            username: Username (optional).
+
+        Returns:
+            Access count.
+        """
+        cache_key = self.get_cache_key(lang, family, username)
+        return self._access_counts.get(cache_key, 0)
+
+    def clear(self) -> None:
+        """Clear all cached sessions."""
+        self._session_cache.clear()
+        self._access_counts.clear()
+
+    def get_all_sessions(self) -> Dict[CacheKey, "LoginBotProtocol"]:
+        """
+        Get all cached sessions.
+
+        Returns:
+            Dictionary of all cached sessions.
+        """
+        return self._session_cache.copy()

mw_api/repositories/__init__.py

@@ -5,7 +5,11 @@
 Contains data access layer classes that abstract API calls.
 """
 from .page_repository import PageRepository
+from .api_repository import ApiRepository
+from .category_repository import CategoryRepository
 
 __all__ = [
     "PageRepository",
+    "ApiRepository",
+    "CategoryRepository",
 ]