feat(uploader): Implement chapter upload module (Epic 6)

- Add UploadClient for server communication with retry logic
- Add CUCalculator for CU fee calculation (ceil(chars/100))
- Add ContentChecker for compliance checking (sensitive words, format, placeholders)
- Add UploadManager integrating all upload features
- Add MockUploadClient for testing without server
- Add data models: UploadResult, BatchUploadResult, ContentCheckResult, QuotaInfo

Features:
- Single and batch chapter upload
- Content compliance checking (sensitive words, format, placeholders)
- CU calculation: ceil(character_count / 100)
- Customizable sensitive words list
- Strict mode for content validation
- Dry run mode for testing
- Quota checking and management
- Upload workflow with progress tracking
- Pre-upload validation

Tests: 38 tests passing, >80% coverage

Part of Epic 6: Upload Module (28SP)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

src/uploader/__init__.py

@@ -0,0 +1,73 @@
+"""
+Uploader module for chapter upload functionality.
+
+This module provides chapter upload, CU calculation, content
+checking, and upload management.
+"""
+
+from .models import (
+    UploadStatus,
+    CheckStatus,
+    ChapterUploadData,
+    UploadResult,
+    BatchUploadResult,
+    ContentCheckIssue,
+    ContentCheckResult,
+    QuotaInfo,
+    UploadWorkflowResult
+)
+from .client import (
+    UploadClient,
+    MockUploadClient,
+    UploadClientError,
+    AuthenticationError,
+    QuotaExceededError,
+    ServerError,
+    ClientConfig
+)
+from .cu_calculator import CUCalculator
+from .content_checker import (
+    ContentChecker,
+    RuleBasedContentChecker,
+    ContentRule
+)
+from .manager import (
+    UploadManager,
+    create_upload_manager,
+    create_mock_upload_manager
+)
+
+__all__ = [
+    # Models
+    "UploadStatus",
+    "CheckStatus",
+    "ChapterUploadData",
+    "UploadResult",
+    "BatchUploadResult",
+    "ContentCheckIssue",
+    "ContentCheckResult",
+    "QuotaInfo",
+    "UploadWorkflowResult",
+
+    # Client
+    "UploadClient",
+    "MockUploadClient",
+    "UploadClientError",
+    "AuthenticationError",
+    "QuotaExceededError",
+    "ServerError",
+    "ClientConfig",
+
+    # CU Calculator
+    "CUCalculator",
+
+    # Content Checker
+    "ContentChecker",
+    "RuleBasedContentChecker",
+    "ContentRule",
+
+    # Manager
+    "UploadManager",
+    "create_upload_manager",
+    "create_mock_upload_manager",
+]

src/uploader/client.py

@@ -0,0 +1,453 @@
+"""
+Upload client for communicating with the upload server.
+
+This module provides HTTP client functionality for uploading
+chapters to the remote server.
+"""
+
+import json
+import time
+from typing import Dict, Any, List, Optional
+from dataclasses import dataclass
+
+try:
+    import requests
+    REQUESTS_AVAILABLE = True
+except ImportError:
+    REQUESTS_AVAILABLE = False
+
+from .models import (
+    UploadResult,
+    BatchUploadResult,
+    QuotaInfo,
+    ChapterUploadData,
+    UploadStatus
+)
+
+
+class UploadClientError(Exception):
+    """Base exception for upload client errors."""
+    pass
+
+
+class AuthenticationError(UploadClientError):
+    """Raised when authentication fails."""
+    pass
+
+
+class QuotaExceededError(UploadClientError):
+    """Raised when quota is exceeded."""
+    pass
+
+
+class ServerError(UploadClientError):
+    """Raised when server returns an error."""
+    pass
+
+
+@dataclass
+class ClientConfig:
+    """Configuration for the upload client."""
+
+    api_base: str
+    api_key: str
+    timeout: int = 30  # seconds
+    max_retries: int = 3
+    retry_delay: float = 1.0  # seconds
+
+
+class UploadClient:
+    """
+    HTTP client for uploading chapters to the server.
+
+    This client handles:
+    - Single chapter upload
+    - Batch chapter upload
+    - Quota checking
+    - Authentication
+    - Error handling and retries
+
+    Example:
+        >>> client = UploadClient("https://api.example.com", "your-api-key")
+        >>> result = client.upload_chapter("work123", {
+        ...     "title": "Chapter 1",
+        ...     "content": "This is the content...",
+        ...     "index": 0
+        ... })
+        >>> print(result.success, result.cu_consumed)
+    """
+
+    DEFAULT_HEADERS = {
+        "Content-Type": "application/json",
+        "User-Agent": "BmadTranslator/1.0"
+    }
+
+    def __init__(
+        self,
+        api_base: str,
+        api_key: str,
+        config: Optional[ClientConfig] = None
+    ):
+        """
+        Initialize the upload client.
+
+        Args:
+            api_base: Base URL of the API (e.g., "https://api.example.com")
+            api_key: API key for authentication
+            config: Optional client configuration
+        """
+        if not REQUESTS_AVAILABLE:
+            raise ImportError(
+                "requests library is required but not installed. "
+                "Install it with: pip install requests"
+            )
+
+        self.api_base = api_base.rstrip("/")
+        self.api_key = api_key
+        self.config = config or ClientConfig(api_base, api_key)
+
+        # Initialize session
+        self.session = requests.Session()
+        self.session.headers.update(self.DEFAULT_HEADERS)
+        self.session.headers.update({
+            "Authorization": f"Bearer {self.api_key}"
+        })
+
+    def upload_chapter(
+        self,
+        work_id: str,
+        chapter: Dict[str, Any]
+    ) -> UploadResult:
+        """
+        Upload a single chapter.
+
+        Args:
+            work_id: Work item ID
+            chapter: Chapter data with keys: title, content, index
+
+        Returns:
+            UploadResult with status and details
+
+        Raises:
+            AuthenticationError: If authentication fails
+            QuotaExceededError: If quota is exceeded
+            ServerError: If server returns an error
+        """
+        url = f"{self.api_base}/api/v1/works/{work_id}/chapters"
+
+        payload = {
+            "title": chapter.get("title", ""),
+            "content": chapter.get("content", ""),
+            "index": chapter.get("index", 0),
+            "original_text": chapter.get("original_text")
+        }
+
+        try:
+            response = self._make_request("POST", url, json=payload)
+            data = response.json()
+
+            if response.status_code == 201:
+                return UploadResult(
+                    success=True,
+                    chapter_id=chapter.get("chapter_id", ""),
+                    server_chapter_id=data.get("chapter_id"),
+                    cu_consumed=data.get("cu_consumed", 0)
+                )
+            else:
+                return UploadResult(
+                    success=False,
+                    chapter_id=chapter.get("chapter_id", ""),
+                    error_message=data.get("error", "Upload failed")
+                )
+
+        except requests.exceptions.Timeout:
+            return UploadResult(
+                success=False,
+                chapter_id=chapter.get("chapter_id", ""),
+                error_message="Request timed out"
+            )
+        except requests.exceptions.RequestException as e:
+            return UploadResult(
+                success=False,
+                chapter_id=chapter.get("chapter_id", ""),
+                error_message=str(e)
+            )
+
+    def batch_upload(
+        self,
+        work_id: str,
+        chapters: List[Dict[str, Any]],
+        chunk_size: int = 10
+    ) -> BatchUploadResult:
+        """
+        Upload multiple chapters.
+
+        Args:
+            work_id: Work item ID
+            chapters: List of chapter dictionaries
+            chunk_size: Number of chapters to upload per request
+
+        Returns:
+            BatchUploadResult with overall status and individual results
+        """
+        from datetime import datetime
+
+        result = BatchUploadResult(
+            work_id=work_id,
+            total_chapters=len(chapters),
+            status=UploadStatus.UPLOADING,
+            started_at=datetime.now()
+        )
+
+        # Upload in chunks
+        for i in range(0, len(chapters), chunk_size):
+            chunk = chapters[i:i + chunk_size]
+
+            for chapter in chunk:
+                upload_result = self.upload_chapter(work_id, chapter)
+                result.results.append(upload_result)
+
+                if upload_result.success:
+                    result.successful += 1
+                    result.total_cu_consumed += upload_result.cu_consumed
+                else:
+                    result.failed += 1
+
+        # Set final status
+        result.completed_at = datetime.now()
+        if result.failed == 0:
+            result.status = UploadStatus.COMPLETED
+        elif result.successful == 0:
+            result.status = UploadStatus.FAILED
+        else:
+            result.status = UploadStatus.PARTIAL
+
+        return result
+
+    def check_quota(self) -> QuotaInfo:
+        """
+        Check remaining API quota.
+
+        Returns:
+            QuotaInfo with quota details
+
+        Raises:
+            AuthenticationError: If authentication fails
+            ServerError: If server returns an error
+        """
+        url = f"{self.api_base}/api/v1/quota"
+
+        response = self._make_request("GET", url)
+        data = response.json()
+
+        return QuotaInfo(
+            total_cu=data.get("total_cu", 0),
+            used_cu=data.get("used_cu", 0),
+            remaining_cu=data.get("remaining_cu", 0),
+            reset_at=data.get("reset_at")
+        )
+
+    def get_upload_status(self, task_id: str) -> Dict[str, Any]:
+        """
+        Get status of an upload task.
+
+        Args:
+            task_id: Upload task ID
+
+        Returns:
+            Dictionary with task status
+        """
+        url = f"{self.api_base}/api/v1/upload-tasks/{task_id}"
+
+        response = self._make_request("GET", url)
+        return response.json()
+
+    def cancel_upload(self, task_id: str) -> bool:
+        """
+        Cancel an ongoing upload task.
+
+        Args:
+            task_id: Upload task ID
+
+        Returns:
+            True if cancellation was successful
+        """
+        url = f"{self.api_base}/api/v1/upload-tasks/{task_id}/cancel"
+
+        try:
+            response = self._make_request("POST", url)
+            return response.status_code == 200
+        except Exception:
+            return False
+
+    def _make_request(
+        self,
+        method: str,
+        url: str,
+        **kwargs
+    ) -> "requests.Response":
+        """
+        Make an HTTP request with retries.
+
+        Args:
+            method: HTTP method
+            url: Request URL
+            **kwargs: Arguments passed to requests.request
+
+        Returns:
+            Response object
+
+        Raises:
+            AuthenticationError: If authentication fails
+            QuotaExceededError: If quota is exceeded
+            ServerError: If server returns an error
+        """
+        kwargs.setdefault("timeout", self.config.timeout)
+
+        for attempt in range(self.config.max_retries):
+            try:
+                response = self.session.request(method, url, **kwargs)
+
+                # Handle authentication errors
+                if response.status_code == 401:
+                    raise AuthenticationError("Invalid API key or authentication failed")
+
+                # Handle quota errors
+                if response.status_code == 402:
+                    raise QuotaExceededError("Quota exceeded")
+
+                # Handle server errors that might be transient
+                if response.status_code >= 500:
+                    if attempt < self.config.max_retries - 1:
+                        time.sleep(self.config.retry_delay * (attempt + 1))
+                        continue
+                    else:
+                        raise ServerError(f"Server error: {response.status_code}")
+
+                # Handle other error responses
+                if response.status_code >= 400:
+                    try:
+                        error_data = response.json()
+                        message = error_data.get("error", error_data.get("message", "Unknown error"))
+                    except ValueError:
+                        message = f"HTTP {response.status_code}: {response.text[:100]}"
+
+                    raise ServerError(message)
+
+                return response
+
+            except requests.exceptions.Timeout:
+                if attempt == self.config.max_retries - 1:
+                    raise UploadClientError(f"Request timed out after {self.config.max_retries} attempts")
+                time.sleep(self.config.retry_delay)
+
+            except requests.exceptions.ConnectionError:
+                if attempt == self.config.max_retries - 1:
+                    raise UploadClientError(f"Connection failed after {self.config.max_retries} attempts")
+                time.sleep(self.config.retry_delay)
+
+        # Should not reach here
+        raise UploadClientError("Unexpected error in _make_request")
+
+    def health_check(self) -> bool:
+        """
+        Check if the API server is accessible.
+
+        Returns:
+            True if server is healthy, False otherwise
+        """
+        try:
+            url = f"{self.api_base}/api/v1/health"
+            response = self.session.get(url, timeout=5)
+            return response.status_code == 200
+        except Exception:
+            return False
+
+    def close(self) -> None:
+        """Close the HTTP session."""
+        self.session.close()
+
+    def __enter__(self):
+        """Context manager entry."""
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """Context manager exit."""
+        self.close()
+
+
+class MockUploadClient(UploadClient):
+    """
+    Mock upload client for testing.
+
+    Simulates server responses without making actual HTTP requests.
+    """
+
+    def __init__(self, api_base: str = "mock://api", api_key: str = "mock-key"):
+        """Initialize mock client."""
+        # Don't call parent init to avoid requests import
+        self.api_base = api_base
+        self.api_key = api_key
+        self.upload_count = 0
+        self.uploaded_chapters: List[Dict[str, Any]] = []
+
+    def upload_chapter(
+        self,
+        work_id: str,
+        chapter: Dict[str, Any]
+    ) -> UploadResult:
+        """Mock upload that always succeeds."""
+        self.upload_count += 1
+        self.uploaded_chapters.append({**chapter, "work_id": work_id})
+
+        # Calculate CU from content length
+        content_length = len(chapter.get("content", ""))
+        cu_consumed = (content_length + 99) // 100  # Ceiling division
+
+        return UploadResult(
+            success=True,
+            chapter_id=chapter.get("chapter_id", f"ch_{self.upload_count}"),
+            server_chapter_id=f"server_ch_{self.upload_count}",
+            cu_consumed=cu_consumed
+        )
+
+    def batch_upload(
+        self,
+        work_id: str,
+        chapters: List[Dict[str, Any]],
+        chunk_size: int = 10
+    ) -> BatchUploadResult:
+        """Mock batch upload."""
+        from datetime import datetime
+
+        results = []
+        total_cu = 0
+
+        for chapter in chapters:
+            result = self.upload_chapter(work_id, chapter)
+            results.append(result)
+            total_cu += result.cu_consumed
+
+        return BatchUploadResult(
+            work_id=work_id,
+            total_chapters=len(chapters),
+            successful=len(chapters),
+            failed=0,
+            results=results,
+            total_cu_consumed=total_cu,
+            status=UploadStatus.COMPLETED,
+            started_at=datetime.now(),
+            completed_at=datetime.now()
+        )
+
+    def check_quota(self) -> QuotaInfo:
+        """Mock quota check with high values."""
+        return QuotaInfo(
+            total_cu=100000,
+            used_cu=self.upload_count * 10,
+            remaining_cu=100000 - (self.upload_count * 10)
+        )
+
+    def health_check(self) -> bool:
+        """Mock health check - always returns True."""
+        return True

src/uploader/content_checker.py

@@ -0,0 +1,414 @@
+"""
+Content compliance checker for uploaded chapters.
+
+This module provides content checking for sensitive words,
+format issues, and other compliance requirements.
+"""
+
+import re
+from typing import List, Dict, Any, Optional, Set
+from dataclasses import dataclass
+
+from .models import ContentCheckResult, ContentCheckIssue, CheckStatus
+
+
+class ContentChecker:
+    """
+    Check chapter content for compliance issues.
+
+    This checker validates:
+    - Sensitive/prohibited words
+    - Format issues
+    - Length constraints
+    - Character encoding issues
+    - Placeholder issues (unreplaced glossary terms)
+    """
+
+    # Default sensitive words (can be customized)
+    DEFAULT_SENSITIVE_WORDS: Set[str] = {
+        "暴力",
+        "血腥",
+        "色情",
+        "裸露",
+        "性交",
+        "毒品",
+        "赌博",
+        "恐怖主义",
+        "自杀",
+        "自残",
+    }
+
+    # Patterns for detecting format issues
+    FORMAT_PATTERNS = {
+        "excessive_whitespace": r"\s{5,}",
+        "multiple_line_breaks": r"\n{4,}",
+        "mixed_quotes": r'[""''``]{2,}',
+        "broken_punctuation": r'([，。！？；：][a-zA-Z]|[a-zA-Z][，。！？；：])',
+    }
+
+    # Patterns for detecting unreplaced placeholders
+    PLACEHOLDER_PATTERNS = [
+        r"__\w{3,}__",  # __en__ style placeholders
+        r"\[TERM_\w+\]",  # [TERM_xxx] style placeholders
+        r"\{\w+_TERM\}",  # {xxx_TERM} style placeholders
+    ]
+
+    def __init__(
+        self,
+        sensitive_words: Optional[Set[str]] = None,
+        max_chapter_length: int = 100000,
+        min_chapter_length: int = 10,
+        check_placeholders: bool = True,
+        check_format: bool = True
+    ):
+        """
+        Initialize the content checker.
+
+        Args:
+            sensitive_words: Set of prohibited words (uses defaults if None)
+            max_chapter_length: Maximum allowed chapter length
+            min_chapter_length: Minimum allowed chapter length
+            check_placeholders: Whether to check for unreplaced placeholders
+            check_format: Whether to check for format issues
+        """
+        self.sensitive_words = sensitive_words or self.DEFAULT_SENSITIVE_WORDS
+        self.max_chapter_length = max_chapter_length
+        self.min_chapter_length = min_chapter_length
+        self.check_placeholders = check_placeholders
+        self.check_format = check_format
+
+        # Compile regex patterns
+        self._format_patterns = {
+            name: re.compile(pattern)
+            for name, pattern in self.FORMAT_PATTERNS.items()
+        }
+        self._placeholder_patterns = [
+            re.compile(pattern) for pattern in self.PLACEHOLDER_PATTERNS
+        ]
+
+    def check_chapter(
+        self,
+        title: str,
+        content: str,
+        strict_mode: bool = False
+    ) -> ContentCheckResult:
+        """
+        Check a single chapter for compliance issues.
+
+        Args:
+            title: Chapter title
+            content: Chapter content
+            strict_mode: If True, warnings are treated as errors
+
+        Returns:
+            ContentCheckResult with all issues found
+        """
+        issues = []
+        warnings = []
+
+        # Check title
+        title_issues = self._check_title(title)
+        issues.extend(title_issues)
+
+        # Check sensitive words
+        sensitive_issues = self._check_sensitive_words(title, content)
+        issues.extend(sensitive_issues)
+
+        # Check length constraints
+        length_issues = self._check_length(title, content)
+        issues.extend(length_issues)
+
+        # Check for placeholders
+        if self.check_placeholders:
+            placeholder_issues = self._check_placeholders(content)
+            if placeholder_issues:
+                # Placeholders are typically warnings unless in strict mode
+                if strict_mode:
+                    issues.extend(placeholder_issues)
+                else:
+                    warnings.extend(placeholder_issues)
+
+        # Check format issues
+        if self.check_format:
+            format_issues = self._check_format(content)
+            # Format issues are warnings unless in strict mode
+            if strict_mode:
+                issues.extend(format_issues)
+            else:
+                warnings.extend(format_issues)
+
+        # Determine overall status
+        has_errors = len(issues) > 0
+        has_warnings = len(warnings) > 0
+
+        if has_errors:
+            status = CheckStatus.FAILED
+        elif has_warnings:
+            status = CheckStatus.WARNING
+        else:
+            status = CheckStatus.PASSED
+
+        return ContentCheckResult(
+            passed=not has_errors,
+            status=status,
+            issues=issues,
+            warnings=warnings
+        )
+
+    def check_all_chapters(
+        self,
+        chapters: List[Dict[str, Any]],
+        strict_mode: bool = False
+    ) -> Dict[str, ContentCheckResult]:
+        """
+        Check multiple chapters for compliance.
+
+        Args:
+            chapters: List of chapter dictionaries with "title" and "content"
+            strict_mode: If True, warnings are treated as errors
+
+        Returns:
+            Dictionary mapping chapter_id to check results
+        """
+        results = {}
+
+        for i, chapter in enumerate(chapters):
+            chapter_id = chapter.get("chapter_id") or f"ch_{i}"
+            title = chapter.get("title") or chapter.get("chapter_title") or ""
+            content = chapter.get("content") or chapter.get("text") or ""
+
+            results[chapter_id] = self.check_chapter(title, content, strict_mode)
+
+        return results
+
+    def _check_title(self, title: str) -> List[ContentCheckIssue]:
+        """Check chapter title for issues."""
+        issues = []
+
+        if not title or not title.strip():
+            issues.append(ContentCheckIssue(
+                type="empty_title",
+                severity="error",
+                message="Chapter title is empty",
+                location="title"
+            ))
+        elif len(title) > 200:
+            issues.append(ContentCheckIssue(
+                type="title_too_long",
+                severity="warning",
+                message=f"Chapter title is too long ({len(title)} characters)",
+                location="title"
+            ))
+
+        return issues
+
+    def _check_sensitive_words(
+        self,
+        title: str,
+        content: str
+    ) -> List[ContentCheckIssue]:
+        """Check for sensitive/prohibited words."""
+        issues = []
+
+        # Check title
+        for word in self.sensitive_words:
+            if word in title:
+                issues.append(ContentCheckIssue(
+                    type="sensitive_word",
+                    severity="error",
+                    message=f"Sensitive word found in title: {word}",
+                    location="title",
+                    snippet=word
+                ))
+
+        # Check content
+        for word in self.sensitive_words:
+            if word in content:
+                # Count occurrences
+                count = content.count(word)
+                # Find first occurrence location
+                index = content.find(word)
+                line_num = content[:index].count("\n") + 1
+
+                issues.append(ContentCheckIssue(
+                    type="sensitive_word",
+                    severity="error",
+                    message=f"Sensitive word found in content ({count} occurrence(s)): {word}",
+                    location=f"line {line_num}",
+                    snippet=word
+                ))
+
+        return issues
+
+    def _check_length(
+        self,
+        title: str,
+        content: str
+    ) -> List[ContentCheckIssue]:
+        """Check content length constraints."""
+        issues = []
+        content_length = len(content)
+
+        if content_length < self.min_chapter_length:
+            issues.append(ContentCheckIssue(
+                type="content_too_short",
+                severity="error",
+                message=f"Content is too short ({content_length} < {self.min_chapter_length})",
+                location="content"
+            ))
+
+        if content_length > self.max_chapter_length:
+            issues.append(ContentCheckIssue(
+                type="content_too_long",
+                severity="error",
+                message=f"Content is too long ({content_length} > {self.max_chapter_length})",
+                location="content"
+            ))
+
+        return issues
+
+    def _check_placeholders(self, content: str) -> List[ContentCheckIssue]:
+        """Check for unreplaced glossary placeholders."""
+        issues = []
+
+        for pattern in self._placeholder_patterns:
+            matches = pattern.finditer(content)
+            for match in matches:
+                placeholder = match.group()
+                index = match.start()
+                line_num = content[:index].count("\n") + 1
+
+                issues.append(ContentCheckIssue(
+                    type="unreplaced_placeholder",
+                    severity="warning",
+                    message=f"Unreplaced placeholder found: {placeholder}",
+                    location=f"line {line_num}",
+                    snippet=placeholder
+                ))
+
+        return issues
+
+    def _check_format(self, content: str) -> List[ContentCheckIssue]:
+        """Check for format issues."""
+        issues = []
+
+        for pattern_name, pattern in self._format_patterns.items():
+            matches = pattern.finditer(content)
+            for match in matches:
+                text = match.group()
+                index = match.start()
+                line_num = content[:index].count("\n") + 1
+
+                # Get line snippet
+                line_start = content.rfind("\n", 0, index) + 1
+                line_end = content.find("\n", index)
+                if line_end == -1:
+                    line_end = len(content)
+                snippet = content[line_start:line_end].strip()
+
+                issues.append(ContentCheckIssue(
+                    type=pattern_name,
+                    severity="warning",
+                    message=f"Format issue detected: {pattern_name}",
+                    location=f"line {line_num}",
+                    snippet=snippet[:50]  # First 50 chars
+                ))
+
+        return issues
+
+    def add_sensitive_word(self, word: str) -> None:
+        """Add a word to the sensitive words list."""
+        self.sensitive_words.add(word)
+
+    def remove_sensitive_word(self, word: str) -> None:
+        """Remove a word from the sensitive words list."""
+        self.sensitive_words.discard(word)
+
+    def set_sensitive_words(self, words: Set[str]) -> None:
+        """Replace the sensitive words list."""
+        self.sensitive_words = words
+
+    def get_sensitive_words(self) -> Set[str]:
+        """Get the current sensitive words list."""
+        return self.sensitive_words.copy()
+
+
+@dataclass
+class ContentRule:
+    """A content checking rule."""
+
+    name: str
+    pattern: str
+    severity: str  # "error", "warning", "info"
+    message: str
+    enabled: bool = True
+
+
+class RuleBasedContentChecker(ContentChecker):
+    """
+    Content checker with custom rules.
+
+    Allows defining custom checking rules beyond the built-in checks.
+    """
+
+    def __init__(self, *args, custom_rules: Optional[List[ContentRule]] = None, **kwargs):
+        """
+        Initialize with custom rules.
+
+        Args:
+            *args: Arguments passed to ContentChecker
+            custom_rules: List of custom checking rules
+            **kwargs: Keyword arguments passed to ContentChecker
+        """
+        super().__init__(*args, **kwargs)
+        self.custom_rules = custom_rules or []
+        self._compile_custom_rules()
+
+    def _compile_custom_rules(self) -> None:
+        """Compile custom rule patterns."""
+        self._compiled_rules = []
+        for rule in self.custom_rules:
+            if rule.enabled:
+                try:
+                    compiled = re.compile(rule.pattern)
+                    self._compiled_rules.append((rule, compiled))
+                except re.error as e:
+                    print(f"Warning: Invalid pattern for rule {rule.name}: {e}")
+
+    def check_chapter(
+        self,
+        title: str,
+        content: str,
+        strict_mode: bool = False
+    ) -> ContentCheckResult:
+        """Check chapter with custom rules applied."""
+        result = super().check_chapter(title, content, strict_mode)
+
+        # Apply custom rules
+        for rule, pattern in self._compiled_rules:
+            matches = pattern.finditer(content)
+            for match in matches:
+                index = match.start()
+                line_num = content[:index].count("\n") + 1
+
+                issue = ContentCheckIssue(
+                    type=f"rule_{rule.name}",
+                    severity=rule.severity,
+                    message=rule.message,
+                    location=f"line {line_num}",
+                    snippet=match.group()[:50]
+                )
+
+                if rule.severity == "error" or (strict_mode and rule.severity == "warning"):
+                    result.issues.append(issue)
+                else:
+                    result.warnings.append(issue)
+
+        # Update status based on new issues
+        if result.issues:
+            result.passed = False
+            result.status = CheckStatus.FAILED
+        elif result.warnings:
+            result.status = CheckStatus.WARNING
+
+        return result

src/uploader/cu_calculator.py

@@ -0,0 +1,204 @@
+"""
+CU (Chapter Unit) fee calculator.
+
+This module provides CU calculation for chapter uploads.
+CU = chapter character count ÷ 100, rounded up.
+"""
+
+import math
+from typing import List, Dict, Any
+
+
+class CUCalculator:
+    """
+    Calculate CU (Chapter Unit) fees for chapter uploads.
+
+    CU is calculated as: ceil(character_count / 100)
+    This means:
+    - 1-100 characters = 1 CU
+    - 101-200 characters = 2 CU
+    - 201-300 characters = 3 CU
+    etc.
+    """
+
+    CU_DIVISOR = 100  # Characters per CU
+
+    def __init__(self, divisor: int = None):
+        """
+        Initialize the CU calculator.
+
+        Args:
+            divisor: Number of characters per CU (default: 100)
+        """
+        self.divisor = divisor or self.CU_DIVISOR
+
+    def calculate_cu(self, content: str) -> int:
+        """
+        Calculate CU for a single chapter.
+
+        Args:
+            content: Chapter content
+
+        Returns:
+            CU cost (minimum 1)
+
+        Examples:
+            >>> calc = CUCalculator()
+            >>> calc.calculate_cu("A" * 50)
+            1
+            >>> calc.calculate_cu("A" * 123)
+            2
+            >>> calc.calculate_cu("A" * 300)
+            3
+            >>> calc.calculate_cu("")  # Empty content
+            0
+        """
+        if not content:
+            return 0
+
+        char_count = len(content)
+        return math.ceil(char_count / self.divisor)
+
+    def calculate_cu_from_count(self, char_count: int) -> int:
+        """
+        Calculate CU from character count.
+
+        Args:
+            char_count: Number of characters
+
+        Returns:
+            CU cost
+        """
+        if char_count <= 0:
+            return 0
+        return math.ceil(char_count / self.divisor)
+
+    def calculate_batch(self, contents: List[str]) -> int:
+        """
+        Calculate total CU for multiple chapters.
+
+        Args:
+            contents: List of chapter contents
+
+        Returns:
+            Total CU cost
+
+        Examples:
+            >>> calc = CUCalculator()
+            >>> calc.calculate_batch(["A" * 50, "A" * 150])
+            3  # (50/100→1) + (150/100→2) = 3
+        """
+        return sum(self.calculate_cu(content) for content in contents)
+
+    def calculate_chapters_cu(self, chapters: List[Dict[str, Any]]) -> Dict[str, int]:
+        """
+        Calculate CU for a list of chapter dictionaries.
+
+        Args:
+            chapters: List of chapters with "content" or "text" key
+
+        Returns:
+            Dictionary mapping chapter_id to CU cost
+        """
+        results = {}
+        for chapter in chapters:
+            chapter_id = chapter.get("chapter_id") or chapter.get("id", str(len(results)))
+            content = chapter.get("content") or chapter.get("text", "")
+            results[chapter_id] = self.calculate_cu(content)
+        return results
+
+    def estimate_cu(self, chapter_count: int, avg_char_count: int) -> int:
+        """
+        Estimate CU for a work based on chapter count and average length.
+
+        Args:
+            chapter_count: Number of chapters
+            avg_char_count: Average characters per chapter
+
+        Returns:
+            Estimated total CU
+        """
+        if chapter_count <= 0 or avg_char_count <= 0:
+            return 0
+        return chapter_count * self.calculate_cu_from_count(avg_char_count)
+
+    def get_breakdown(self, content: str) -> Dict[str, Any]:
+        """
+        Get detailed breakdown of CU calculation.
+
+        Args:
+            content: Chapter content
+
+        Returns:
+            Dictionary with calculation details
+        """
+        char_count = len(content)
+        cu = self.calculate_cu(content)
+
+        # Calculate efficiency (how close to full CU usage)
+        if cu > 0:
+            efficiency = (char_count / (cu * self.divisor)) * 100
+        else:
+            efficiency = 0
+
+        # Calculate remaining characters in current CU
+        if cu > 0:
+            remaining_in_cu = (cu * self.divisor) - char_count
+        else:
+            remaining_in_cu = 0
+
+        return {
+            "character_count": char_count,
+            "cu_cost": cu,
+            "cu_divisor": self.divisor,
+            "efficiency_percentage": round(efficiency, 2),
+            "remaining_characters_in_cu": remaining_in_cu,
+            "formula": f"ceil({char_count} / {self.divisor}) = {cu}"
+        }
+
+    def reverse_calculate_max_chars(self, cu_limit: int) -> int:
+        """
+        Calculate maximum characters for a given CU limit.
+
+        Args:
+            cu_limit: Maximum CU allowed
+
+        Returns:
+            Maximum character count
+
+        Examples:
+            >>> calc = CUCalculator()
+            >>> calc.reverse_calculate_max_chars(1)
+            100
+            >>> calc.reverse_calculate_max_chars(5)
+            500
+        """
+        return cu_limit * self.divisor
+
+    def calculate_cost_summary(self, contents: List[str]) -> Dict[str, Any]:
+        """
+        Calculate a detailed cost summary for multiple chapters.
+
+        Args:
+            contents: List of chapter contents
+
+        Returns:
+            Dictionary with cost summary
+        """
+        individual_costs = [self.calculate_cu(c) for c in contents]
+        total_cu = sum(individual_costs)
+
+        # Calculate statistics
+        cu_distribution = {}
+        for cost in individual_costs:
+            cu_distribution[cost] = cu_distribution.get(cost, 0) + 1
+
+        return {
+            "total_chapters": len(contents),
+            "total_cu": total_cu,
+            "min_cu": min(individual_costs) if individual_costs else 0,
+            "max_cu": max(individual_costs) if individual_costs else 0,
+            "avg_cu": total_cu / len(individual_costs) if individual_costs else 0,
+            "cu_distribution": cu_distribution,
+            "individual_costs": individual_costs
+        }

src/uploader/manager.py

@@ -0,0 +1,409 @@
+"""
+Upload manager integrating all upload functionality.
+
+This module provides the main workflow for uploading chapters,
+including content checking, CU calculation, and batch upload.
+"""
+
+from typing import List, Dict, Any, Optional, Callable
+from datetime import datetime
+from pathlib import Path
+
+from .client import UploadClient, MockUploadClient, UploadClientError
+from .cu_calculator import CUCalculator
+from .content_checker import ContentChecker, RuleBasedContentChecker
+from .models import (
+    UploadWorkflowResult,
+    ContentCheckResult,
+    ChapterUploadData,
+    UploadResult,
+    BatchUploadResult,
+    UploadStatus
+)
+
+
+class UploadManager:
+    """
+    Upload manager integrating all upload functionality.
+
+    This manager orchestrates:
+    1. Content compliance checking
+    2. CU fee calculation
+    3. Batch upload to server
+    4. Progress tracking and reporting
+
+    Example:
+        >>> client = UploadClient("https://api.example.com", "api-key")
+        >>> manager = UploadManager(client)
+        >>> result = manager.upload_workflow("work123", chapters)
+    """
+
+    def __init__(
+        self,
+        api_client: UploadClient,
+        content_checker: Optional[ContentChecker] = None,
+        cu_calculator: Optional[CUCalculator] = None,
+        strict_mode: bool = False,
+        dry_run: bool = False
+    ):
+        """
+        Initialize the upload manager.
+
+        Args:
+            api_client: Upload client for server communication
+            content_checker: Optional content checker
+            cu_calculator: Optional CU calculator
+            strict_mode: If True, warnings block upload
+            dry_run: If True, simulate uploads without contacting server
+        """
+        self.client = api_client
+        self.content_checker = content_checker or ContentChecker()
+        self.cu_calculator = cu_calculator or CUCalculator()
+        self.strict_mode = strict_mode
+        self.dry_run = dry_run
+
+    def upload_workflow(
+        self,
+        work_id: str,
+        chapters: List[Dict[str, Any]],
+        progress_callback: Optional[Callable[[int, int], None]] = None
+    ) -> UploadWorkflowResult:
+        """
+        Execute the complete upload workflow.
+
+        Workflow steps:
+        1. Content compliance check
+        2. CU fee calculation
+        3. Batch upload (or dry run)
+        4. Return result statistics
+
+        Args:
+            work_id: Work item ID
+            chapters: List of chapter dictionaries
+            progress_callback: Optional callback for progress updates
+
+        Returns:
+            UploadWorkflowResult with all details
+        """
+        result = UploadWorkflowResult(
+            work_id=work_id,
+            status=UploadStatus.PENDING,
+            started_at=datetime.now()
+        )
+
+        try:
+            # Step 1: Content checking
+            if progress_callback:
+                progress_callback(0, len(chapters))
+
+            check_results = self.content_checker.check_all_chapters(
+                chapters,
+                strict_mode=self.strict_mode
+            )
+
+            result.content_check_results = check_results
+
+            # Count total issues
+            total_issues = sum(
+                len(r.issues) + len(r.warnings)
+                for r in check_results.values()
+            )
+            result.total_issues = total_issues
+
+            # Check if any chapters failed content check
+            failed_chapters = [
+                chapter_id for chapter_id, check_result in check_results.items()
+                if not check_result.passed
+            ]
+
+            if failed_chapters and self.strict_mode:
+                result.status = UploadStatus.FAILED
+                result.completed_at = datetime.now()
+                return result
+
+            # Step 2: Prepare chapters for upload (filter failed ones if strict)
+            chapters_to_upload = []
+            for i, chapter in enumerate(chapters):
+                chapter_id = chapter.get("chapter_id") or f"ch_{i}"
+
+                if chapter_id in failed_chapters:
+                    continue
+
+                upload_data = ChapterUploadData(
+                    chapter_id=chapter_id,
+                    work_id=work_id,
+                    title=chapter.get("title", ""),
+                    content=chapter.get("content", ""),
+                    index=chapter.get("index", i),
+                    original_text=chapter.get("original_text")
+                )
+                chapters_to_upload.append(upload_data)
+
+            if not chapters_to_upload:
+                result.status = UploadStatus.FAILED
+                result.completed_at = datetime.now()
+                return result
+
+            # Step 3: Calculate CU
+            total_cu = self.cu_calculator.calculate_batch([
+                c.content for c in chapters_to_upload
+            ])
+            result.total_cu_consumed = total_cu
+
+            # Step 4: Upload
+            if self.dry_run:
+                # Use mock upload for dry run
+                mock_client = MockUploadClient()
+                batch_result = mock_client.batch_upload(
+                    work_id,
+                    [self._chapter_to_dict(c) for c in chapters_to_upload]
+                )
+            else:
+                batch_result = self.client.batch_upload(
+                    work_id,
+                    [self._chapter_to_dict(c) for c in chapters_to_upload]
+                )
+
+            result.upload_results = batch_result.results
+            result.status = batch_result.status
+            result.completed_at = datetime.now()
+
+            if progress_callback:
+                progress_callback(len(chapters), len(chapters))
+
+            return result
+
+        except UploadClientError as e:
+            result.status = UploadStatus.FAILED
+            result.completed_at = datetime.now()
+            raise e
+
+    def _chapter_to_dict(self, chapter: ChapterUploadData) -> Dict[str, Any]:
+        """Convert ChapterUploadData to dictionary for upload."""
+        return {
+            "chapter_id": chapter.chapter_id,
+            "title": chapter.title,
+            "content": chapter.content,
+            "index": chapter.index,
+            "original_text": chapter.original_text
+        }
+
+    def upload_single_chapter(
+        self,
+        work_id: str,
+        chapter: Dict[str, Any]
+    ) -> UploadResult:
+        """
+        Upload a single chapter with pre-checking.
+
+        Args:
+            work_id: Work item ID
+            chapter: Chapter data
+
+        Returns:
+            UploadResult
+        """
+        # Content check
+        title = chapter.get("title", "")
+        content = chapter.get("content", "")
+        check_result = self.content_checker.check_chapter(
+            title,
+            content,
+            strict_mode=self.strict_mode
+        )
+
+        if not check_result.passed:
+            return UploadResult(
+                success=False,
+                chapter_id=chapter.get("chapter_id", ""),
+                error_message=f"Content check failed: {len(check_result.issues)} issues"
+            )
+
+        # Upload
+        if self.dry_run:
+            mock = MockUploadClient()
+            return mock.upload_chapter(work_id, chapter)
+
+        return self.client.upload_chapter(work_id, chapter)
+
+    def estimate_cu(self, chapters: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """
+        Estimate CU cost for chapters without uploading.
+
+        Args:
+            chapters: List of chapter data
+
+        Returns:
+            Dictionary with CU estimation details
+        """
+        contents = [c.get("content", "") for c in chapters]
+        return self.cu_calculator.calculate_cost_summary(contents)
+
+    def preview_upload(
+        self,
+        work_id: str,
+        chapters: List[Dict[str, Any]]
+    ) -> Dict[str, Any]:
+        """
+        Preview upload without actually uploading.
+
+        Shows:
+        - Content check results
+        - CU estimation
+        - Potential issues
+
+        Args:
+            work_id: Work item ID
+            chapters: List of chapter data
+
+        Returns:
+            Dictionary with preview information
+        """
+        check_results = self.content_checker.check_all_chapters(chapters)
+        cu_estimate = self.estimate_cu(chapters)
+
+        return {
+            "work_id": work_id,
+            "total_chapters": len(chapters),
+            "content_check": {
+                "passed": sum(1 for r in check_results.values() if r.passed),
+                "failed": sum(1 for r in check_results.values() if not r.passed),
+                "warnings": sum(1 for r in check_results.values() if r.has_warnings),
+                "total_issues": sum(len(r.issues) + len(r.warnings) for r in check_results.values())
+            },
+            "cu_estimate": cu_estimate,
+            "dry_run": self.dry_run,
+            "strict_mode": self.strict_mode
+        }
+
+    def get_upload_status(self, task_id: str) -> Dict[str, Any]:
+        """
+        Get status of an upload task.
+
+        Args:
+            task_id: Upload task ID
+
+        Returns:
+            Task status dictionary
+        """
+        if self.dry_run:
+            return {
+                "task_id": task_id,
+                "status": "completed",
+                "dry_run": True
+            }
+
+        return self.client.get_upload_status(task_id)
+
+    def check_quota(self) -> Dict[str, Any]:
+        """
+        Check API quota status.
+
+        Returns:
+            Dictionary with quota information
+        """
+        quota = self.client.check_quota()
+
+        return {
+            "total_cu": quota.total_cu,
+            "used_cu": quota.used_cu,
+            "remaining_cu": quota.remaining_cu,
+            "usage_percentage": quota.usage_percentage * 100,
+            "is_exceeded": quota.is_exceeded
+        }
+
+    def validate_before_upload(
+        self,
+        chapters: List[Dict[str, Any]]
+    ) -> Dict[str, Any]:
+        """
+        Validate chapters before upload.
+
+        Checks:
+        - All chapters have required fields
+        - Content is not empty
+        - Titles are present
+
+        Args:
+            chapters: List of chapter data
+
+        Returns:
+            Dictionary with validation results
+        """
+        errors = []
+        warnings = []
+        valid_chapters = []
+
+        for i, chapter in enumerate(chapters):
+            chapter_errors = []
+            chapter_warnings = []
+
+            # Check required fields
+            if "title" not in chapter or not chapter["title"]:
+                chapter_errors.append("Missing or empty title")
+
+            if "content" not in chapter:
+                chapter_errors.append("Missing content field")
+            elif not chapter["content"]:
+                chapter_warnings.append("Empty content")
+
+            # Check index
+            if "index" not in chapter:
+                chapter_warnings.append("Missing index, will use sequential")
+
+            if chapter_errors:
+                errors.append({
+                    "chapter_index": i,
+                    "errors": chapter_errors
+                })
+            elif chapter_warnings:
+                warnings.append({
+                    "chapter_index": i,
+                    "warnings": chapter_warnings
+                })
+                valid_chapters.append(chapter)
+            else:
+                valid_chapters.append(chapter)
+
+        return {
+            "total_chapters": len(chapters),
+            "valid_chapters": len(valid_chapters),
+            "has_errors": len(errors) > 0,
+            "has_warnings": len(warnings) > 0,
+            "errors": errors,
+            "warnings": warnings
+        }
+
+
+def create_upload_manager(
+    api_base: str,
+    api_key: str,
+    **kwargs
+) -> UploadManager:
+    """
+    Factory function to create an upload manager.
+
+    Args:
+        api_base: API base URL
+        api_key: API key
+        **kwargs: Additional arguments for UploadManager
+
+    Returns:
+        Configured UploadManager instance
+    """
+    client = UploadClient(api_base, api_key)
+    return UploadManager(client, **kwargs)
+
+
+def create_mock_upload_manager(**kwargs) -> UploadManager:
+    """
+    Create an upload manager with mock client for testing.
+
+    Args:
+        **kwargs: Additional arguments for UploadManager
+
+    Returns:
+        UploadManager with MockUploadClient
+    """
+    mock_client = MockUploadClient()
+    return UploadManager(mock_client, **kwargs)

src/uploader/models.py

@@ -0,0 +1,252 @@
+"""
+Data models for the uploader module.
+
+This module defines the core data structures for chapter upload,
+CU calculation, and content checking.
+"""
+
+from dataclasses import dataclass, field
+from typing import Optional, List, Dict, Any
+from datetime import datetime
+from enum import Enum
+
+
+class UploadStatus(Enum):
+    """Status of an upload operation."""
+
+    PENDING = "pending"
+    UPLOADING = "uploading"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    PARTIAL = "partial"
+    CANCELLED = "cancelled"
+
+
+class CheckStatus(Enum):
+    """Status of content compliance check."""
+
+    PASSED = "passed"
+    FAILED = "failed"
+    WARNING = "warning"
+
+
+@dataclass
+class ChapterUploadData:
+    """
+    Data for uploading a chapter.
+
+    Attributes:
+        chapter_id: Unique chapter identifier
+        work_id: Work item ID
+        title: Chapter title
+        content: Translated content
+        index: Chapter index
+        original_text: Original Chinese text (optional)
+    """
+
+    chapter_id: str
+    work_id: str
+    title: str
+    content: str
+    index: int
+    original_text: Optional[str] = None
+
+    @property
+    def content_length(self) -> int:
+        """Get the length of the content."""
+        return len(self.content)
+
+    @property
+    def word_count(self) -> int:
+        """Estimate word count."""
+        # Simple count: characters + spaces
+        return len(self.content.replace("\n", "").replace(" ", ""))
+
+
+@dataclass
+class UploadResult:
+    """
+    Result of a chapter upload.
+
+    Attributes:
+        success: Whether the upload succeeded
+        chapter_id: The chapter ID
+        server_chapter_id: ID assigned by server
+        cu_consumed: CU consumed by this upload
+        error_message: Error message if failed
+        timestamp: Upload timestamp
+    """
+
+    success: bool
+    chapter_id: str
+    server_chapter_id: Optional[str] = None
+    cu_consumed: int = 0
+    error_message: Optional[str] = None
+    timestamp: datetime = field(default_factory=datetime.now)
+
+
+@dataclass
+class BatchUploadResult:
+    """
+    Result of a batch upload operation.
+
+    Attributes:
+        work_id: Work item ID
+        total_chapters: Total number of chapters in batch
+        successful: Number of successful uploads
+        failed: Number of failed uploads
+        results: List of individual upload results
+        total_cu_consumed: Total CU consumed
+        status: Overall upload status
+        started_at: Batch start timestamp
+        completed_at: Batch completion timestamp
+    """
+
+    work_id: str
+    total_chapters: int
+    successful: int = 0
+    failed: int = 0
+    results: List[UploadResult] = field(default_factory=list)
+    total_cu_consumed: int = 0
+    status: UploadStatus = UploadStatus.PENDING
+    started_at: Optional[datetime] = None
+    completed_at: Optional[datetime] = None
+
+    @property
+    def success_rate(self) -> float:
+        """Get success rate (0.0 to 1.0)."""
+        if self.total_chapters == 0:
+            return 0.0
+        return self.successful / self.total_chapters
+
+    @property
+    def is_complete(self) -> bool:
+        """Check if batch upload is complete."""
+        return self.status in (UploadStatus.COMPLETED, UploadStatus.FAILED, UploadStatus.PARTIAL, UploadStatus.CANCELLED)
+
+    @property
+    def duration(self) -> Optional[float]:
+        """Get batch duration in seconds."""
+        if self.started_at and self.completed_at:
+            return (self.completed_at - self.started_at).total_seconds()
+        return None
+
+
+@dataclass
+class ContentCheckIssue:
+    """
+    An issue found during content checking.
+
+    Attributes:
+        type: Type of issue (sensitive_word, format, length, etc.)
+        severity: Severity level (error, warning, info)
+        message: Human-readable issue description
+        location: Location in content (line number, position, etc.)
+        snippet: Snippet of content that triggered the issue
+    """
+
+    type: str
+    severity: str  # "error", "warning", "info"
+    message: str
+    location: Optional[str] = None
+    snippet: Optional[str] = None
+
+
+@dataclass
+class ContentCheckResult:
+    """
+    Result of content compliance check.
+
+    Attributes:
+        passed: Whether content passed all checks
+        status: Overall check status
+        issues: List of issues found
+        warnings: List of warnings found
+        checked_at: Check timestamp
+    """
+
+    passed: bool
+    status: CheckStatus
+    issues: List[ContentCheckIssue] = field(default_factory=list)
+    warnings: List[ContentCheckIssue] = field(default_factory=list)
+    checked_at: datetime = field(default_factory=datetime.now)
+
+    @property
+    def has_errors(self) -> bool:
+        """Check if there are any error-level issues."""
+        return any(issue.severity == "error" for issue in self.issues)
+
+    @property
+    def has_warnings(self) -> bool:
+        """Check if there are any warnings."""
+        return len(self.warnings) > 0 or any(
+            issue.severity == "warning" for issue in self.issues
+        )
+
+    @property
+    def total_issues(self) -> int:
+        """Get total number of issues."""
+        return len(self.issues) + len(self.warnings)
+
+
+@dataclass
+class QuotaInfo:
+    """
+    Information about API quota.
+
+    Attributes:
+        total_cu: Total CU available
+        used_cu: CU already used
+        remaining_cu: Remaining CU
+        reset_at: When quota resets
+    """
+
+    total_cu: int
+    used_cu: int
+    remaining_cu: int
+    reset_at: Optional[datetime] = None
+
+    @property
+    def usage_percentage(self) -> float:
+        """Get usage percentage (0.0 to 1.0)."""
+        if self.total_cu == 0:
+            return 0.0
+        return self.used_cu / self.total_cu
+
+    @property
+    def is_exceeded(self) -> bool:
+        """Check if quota is exceeded."""
+        return self.remaining_cu <= 0
+
+
+@dataclass
+class UploadWorkflowResult:
+    """
+    Result of the complete upload workflow.
+
+    Attributes:
+        work_id: Work item ID
+        status: Overall workflow status
+        content_check_results: Content check results per chapter
+        upload_results: Upload results per chapter
+        total_cu_consumed: Total CU consumed
+        total_issues: Total content issues found
+        started_at: Workflow start timestamp
+        completed_at: Workflow completion timestamp
+    """
+
+    work_id: str
+    status: UploadStatus
+    content_check_results: Dict[str, ContentCheckResult] = field(default_factory=dict)
+    upload_results: List[UploadResult] = field(default_factory=list)
+    total_cu_consumed: int = 0
+    total_issues: int = 0
+    started_at: Optional[datetime] = None
+    completed_at: Optional[datetime] = None
+
+    @property
+    def duration(self) -> Optional[float]:
+        """Get workflow duration in seconds."""
+        if self.started_at and self.completed_at:
+            return (self.completed_at - self.started_at).total_seconds()
+        return None

tests/uploader/test_upload.py

@@ -0,0 +1,448 @@
+"""
+Unit tests for uploader module.
+"""
+
+import pytest
+from datetime import datetime
+
+from src.uploader import (
+    CUCalculator,
+    ContentChecker,
+    MockUploadClient,
+    UploadManager,
+    create_mock_upload_manager,
+    ChapterUploadData,
+    UploadResult,
+    UploadStatus,
+    CheckStatus,
+    QuotaInfo
+)
+
+
+class TestCUCalculator:
+    """Test suite for CUCalculator."""
+
+    def test_calculate_cu_basic(self):
+        """Test basic CU calculation."""
+        calc = CUCalculator()
+
+        # Minimum CU (1-100 chars = 1 CU)
+        assert calc.calculate_cu("A" * 50) == 1
+        assert calc.calculate_cu("A" * 100) == 1
+
+        # Second CU (101-200 chars = 2 CU)
+        assert calc.calculate_cu("A" * 101) == 2
+        assert calc.calculate_cu("A" * 200) == 2
+
+        # Third CU (201-300 chars = 3 CU)
+        assert calc.calculate_cu("A" * 201) == 3
+        assert calc.calculate_cu("A" * 300) == 3
+
+    def test_calculate_cu_examples(self):
+        """Test examples from the spec."""
+        calc = CUCalculator()
+
+        # "123字" → 2 CU (123/100=1.23→向上取整2)
+        assert calc.calculate_cu("A" * 123) == 2
+
+        # "300字" → 3 CU
+        assert calc.calculate_cu("A" * 300) == 3
+
+        # "99字" → 1 CU
+        assert calc.calculate_cu("A" * 99) == 1
+
+    def test_calculate_cu_empty(self):
+        """Test CU calculation for empty content."""
+        calc = CUCalculator()
+        assert calc.calculate_cu("") == 0
+
+    def test_calculate_cu_from_count(self):
+        """Test CU calculation from character count."""
+        calc = CUCalculator()
+        assert calc.calculate_cu_from_count(0) == 0
+        assert calc.calculate_cu_from_count(150) == 2
+        assert calc.calculate_cu_from_count(350) == 4
+
+    def test_calculate_batch(self):
+        """Test batch CU calculation."""
+        calc = CUCalculator()
+
+        # (50/100→1) + (150/100→2) = 3
+        result = calc.calculate_batch(["A" * 50, "A" * 150])
+        assert result == 3
+
+    def test_estimate_cu(self):
+        """Test CU estimation."""
+        calc = CUCalculator()
+
+        # 10 chapters × 150 chars each = 10 × 2 CU = 20 CU
+        result = calc.estimate_cu(10, 150)
+        assert result == 20
+
+    def test_get_breakdown(self):
+        """Test CU calculation breakdown."""
+        calc = CUCalculator()
+        breakdown = calc.get_breakdown("A" * 123)
+
+        assert breakdown["character_count"] == 123
+        assert breakdown["cu_cost"] == 2
+        assert breakdown["efficiency_percentage"] == pytest.approx(61.5, 0.1)
+        assert breakdown["remaining_characters_in_cu"] == 77
+
+    def test_reverse_calculate_max_chars(self):
+        """Test reverse calculation of max chars."""
+        calc = CUCalculator()
+
+        assert calc.reverse_calculate_max_chars(1) == 100
+        assert calc.reverse_calculate_max_chars(5) == 500
+
+    def test_calculate_cost_summary(self):
+        """Test cost summary calculation."""
+        calc = CUCalculator()
+
+        contents = ["A" * 50, "A" * 150, "A" * 250]
+        summary = calc.calculate_cost_summary(contents)
+
+        assert summary["total_chapters"] == 3
+        assert summary["total_cu"] == 6  # 1 + 2 + 3 = 6
+        assert summary["min_cu"] == 1
+        assert summary["max_cu"] == 3
+
+    def test_custom_divisor(self):
+        """Test calculator with custom divisor."""
+        calc = CUCalculator(divisor=50)
+
+        # With divisor=50: 1-50=1 CU, 51-100=2 CU
+        assert calc.calculate_cu("A" * 25) == 1
+        assert calc.calculate_cu("A" * 50) == 1
+        assert calc.calculate_cu("A" * 51) == 2
+        assert calc.calculate_cu("A" * 100) == 2
+
+
+class TestContentChecker:
+    """Test suite for ContentChecker."""
+
+    def test_check_normal_content(self):
+        """Test checking normal, compliant content."""
+        checker = ContentChecker()
+        result = checker.check_chapter("Test Title", "This is normal content.")
+
+        assert result.passed is True
+        assert result.status == CheckStatus.PASSED
+        assert len(result.issues) == 0
+
+    def test_check_empty_title(self):
+        """Test checking chapter with empty title."""
+        checker = ContentChecker()
+        result = checker.check_chapter("", "Content here")
+
+        assert result.passed is False
+        assert len(result.issues) > 0
+        assert result.issues[0].type == "empty_title"
+
+    def test_check_sensitive_words(self):
+        """Test checking for sensitive words."""
+        checker = ContentChecker()
+        result = checker.check_chapter("Normal Title", "This contains 暴力 content.")
+
+        assert result.passed is False
+        assert any(i.type == "sensitive_word" for i in result.issues)
+
+    def test_check_content_too_short(self):
+        """Test checking content that's too short."""
+        checker = ContentChecker(min_chapter_length=50)
+        result = checker.check_chapter("Title", "Short")
+
+        assert result.passed is False
+        assert any(i.type == "content_too_short" for i in result.issues)
+
+    def test_check_content_too_long(self):
+        """Test checking content that's too long."""
+        checker = ContentChecker(max_chapter_length=100)
+        result = checker.check_chapter("Title", "A" * 101)
+
+        assert result.passed is False
+        assert any(i.type == "content_too_long" for i in result.issues)
+
+    def test_check_placeholders(self):
+        """Test checking for unreplaced placeholders."""
+        checker = ContentChecker()
+        result = checker.check_chapter("Title", "This has [TERM_test] placeholder.")
+
+        # Bracket-style placeholders are checked
+        assert len(result.warnings) > 0 or len(result.issues) > 0
+
+    def test_check_strict_mode(self):
+        """Test strict mode where warnings become errors."""
+        checker = ContentChecker()
+        result = checker.check_chapter("Title", "This has [TERM_test] placeholder.", strict_mode=True)
+
+        # In strict mode, warnings become errors
+        assert result.passed is False or len(result.issues) > 0
+
+    def test_check_all_chapters(self):
+        """Test checking multiple chapters."""
+        checker = ContentChecker()
+
+        chapters = [
+            {"title": "Chapter 1", "content": "Normal content."},
+            {"title": "Chapter 2", "content": "Content with 暴力."}
+        ]
+
+        results = checker.check_all_chapters(chapters)
+
+        assert len(results) == 2
+        assert results["ch_0"].passed is True
+        assert results["ch_1"].passed is False
+
+    def test_add_sensitive_word(self):
+        """Test adding custom sensitive word."""
+        checker = ContentChecker()
+        checker.add_sensitive_word("testword")
+
+        assert "testword" in checker.sensitive_words
+
+    def test_remove_sensitive_word(self):
+        """Test removing sensitive word."""
+        checker = ContentChecker()
+        checker.remove_sensitive_word("暴力")
+
+        assert "暴力" not in checker.sensitive_words
+
+    def test_set_sensitive_words(self):
+        """Test replacing sensitive words list."""
+        checker = ContentChecker()
+        checker.set_sensitive_words({"custom1", "custom2"})
+
+        assert "custom1" in checker.sensitive_words
+        assert "暴力" not in checker.sensitive_words
+
+
+class TestMockUploadClient:
+    """Test suite for MockUploadClient."""
+
+    def test_upload_single_chapter(self):
+        """Test uploading a single chapter."""
+        client = MockUploadClient()
+
+        result = client.upload_chapter(
+            "work123",
+            {
+                "title": "Chapter 1",
+                "content": "A" * 150,
+                "index": 0
+            }
+        )
+
+        assert result.success is True
+        assert result.cu_consumed == 2  # 150/100 → 2 CU
+
+    def test_batch_upload(self):
+        """Test batch upload."""
+        client = MockUploadClient()
+
+        chapters = [
+            {"title": "Ch1", "content": "A" * 50, "index": 0},
+            {"title": "Ch2", "content": "A" * 150, "index": 1}
+        ]
+
+        result = client.batch_upload("work123", chapters)
+
+        assert result.total_chapters == 2
+        assert result.successful == 2
+        assert result.failed == 0
+        assert result.status == UploadStatus.COMPLETED
+
+    def test_check_quota(self):
+        """Test quota checking."""
+        client = MockUploadClient()
+        quota = client.check_quota()
+
+        assert quota.total_cu > 0
+        assert quota.remaining_cu > 0
+
+    def test_health_check(self):
+        """Test health check."""
+        client = MockUploadClient()
+        assert client.health_check() is True
+
+
+class TestUploadManager:
+    """Test suite for UploadManager."""
+
+    def test_upload_workflow_dry_run(self):
+        """Test upload workflow in dry run mode."""
+        mock_client = MockUploadClient()
+        manager = UploadManager(mock_client, dry_run=True)
+
+        chapters = [
+            {"title": "Chapter 1", "content": "This is chapter one.", "index": 0},
+            {"title": "Chapter 2", "content": "This is chapter two.", "index": 1}
+        ]
+
+        result = manager.upload_workflow("work123", chapters)
+
+        assert result.status == UploadStatus.COMPLETED
+        assert len(result.upload_results) == 2
+
+    def test_upload_workflow_with_content_check_fail(self):
+        """Test workflow with content check failure."""
+        mock_client = MockUploadClient()
+        manager = UploadManager(mock_client, strict_mode=True)
+
+        # Content with sensitive word will fail
+        chapters = [
+            {"title": "", "content": "Content with empty title causes failure", "index": 0}
+        ]
+
+        result = manager.upload_workflow("work123", chapters)
+
+        # Empty title causes content check failure in strict mode
+        assert result.status == UploadStatus.FAILED or len(result.upload_results) == 0
+
+    def test_estimate_cu(self):
+        """Test CU estimation."""
+        mock_client = MockUploadClient()
+        manager = UploadManager(mock_client)
+
+        chapters = [
+            {"content": "A" * 50},
+            {"content": "A" * 150}
+        ]
+
+        estimate = manager.estimate_cu(chapters)
+
+        assert estimate["total_cu"] == 3  # 1 + 2
+
+    def test_preview_upload(self):
+        """Test upload preview."""
+        mock_client = MockUploadClient()
+        manager = UploadManager(mock_client)
+
+        chapters = [
+            {"title": "Ch1", "content": "Content 1"},
+            {"title": "Ch2", "content": "Content 2"}
+        ]
+
+        preview = manager.preview_upload("work123", chapters)
+
+        assert preview["total_chapters"] == 2
+        assert "content_check" in preview
+        assert "cu_estimate" in preview
+
+    def test_validate_before_upload(self):
+        """Test pre-upload validation."""
+        mock_client = MockUploadClient()
+        manager = UploadManager(mock_client)
+
+        # Invalid chapters (missing title, empty content)
+        chapters = [
+            {"content": "Content without title"},
+            {"title": "Title without content", "content": ""}
+        ]
+
+        validation = manager.validate_before_upload(chapters)
+
+        # Should have errors or warnings
+        assert validation["has_errors"] is True or validation["has_warnings"] is True
+        assert len(validation["errors"]) + len(validation["warnings"]) >= 2
+
+    def test_check_quota(self):
+        """Test checking quota through manager."""
+        mock_client = MockUploadClient()
+        manager = UploadManager(mock_client)
+
+        quota_info = manager.check_quota()
+
+        assert "total_cu" in quota_info
+        assert "remaining_cu" in quota_info
+
+
+class TestChapterUploadData:
+    """Test suite for ChapterUploadData model."""
+
+    def test_content_length(self):
+        """Test content_length property."""
+        data = ChapterUploadData(
+            "ch1",
+            "work1",
+            "Title",
+            "Content here",
+            0
+        )
+
+        assert data.content_length == 12
+
+    def test_word_count(self):
+        """Test word_count property."""
+        data = ChapterUploadData(
+            "ch1",
+            "work1",
+            "Title",
+            "Content here",
+            0
+        )
+
+        # word_count removes newlines and spaces
+        assert data.word_count > 0  # Should count non-whitespace characters
+
+
+class TestUploadResult:
+    """Test suite for UploadResult model."""
+
+    def test_success_result(self):
+        """Test successful upload result."""
+        result = UploadResult(
+            success=True,
+            chapter_id="ch1",
+            server_chapter_id="srv_ch1",
+            cu_consumed=2
+        )
+
+        assert result.success is True
+        assert result.cu_consumed == 2
+
+    def test_failed_result(self):
+        """Test failed upload result."""
+        result = UploadResult(
+            success=False,
+            chapter_id="ch1",
+            error_message="Upload failed"
+        )
+
+        assert result.success is False
+        assert result.error_message == "Upload failed"
+
+
+class TestQuotaInfo:
+    """Test suite for QuotaInfo model."""
+
+    def test_quota_calculations(self):
+        """Test quota calculation properties."""
+        quota = QuotaInfo(
+            total_cu=1000,
+            used_cu=300,
+            remaining_cu=700
+        )
+
+        assert quota.usage_percentage == 0.3
+        assert quota.is_exceeded is False
+
+    def test_quota_exceeded(self):
+        """Test exceeded quota."""
+        quota = QuotaInfo(
+            total_cu=1000,
+            used_cu=1000,
+            remaining_cu=0
+        )
+
+        assert quota.is_exceeded is True
+
+
+def test_create_mock_upload_manager():
+    """Test factory function for mock upload manager."""
+    manager = create_mock_upload_manager()
+
+    assert isinstance(manager, UploadManager)
+    # Check that it uses a mock client (has upload_count attribute)
+    assert hasattr(manager.client, "upload_count")