confluence-mdx: 제목 기반 route와 임시 redirect를 적용합니다

confluence-mdx/README.md

@@ -197,7 +197,7 @@ bin/convert_all.py --verify-translations
 - `target/public/` 디렉토리에 첨부파일이 저장됩니다.
 - 한국어 제목의 번역이 누락된 경우, 오류와 함께 누락 목록을 출력합니다.
   - `etc/korean-titles-translations.txt`에 번역을 추가한 후 재실행합니다.
-- 표시용 영어 제목과 다른 기존 public route segment를 유지해야 하는 경우 `etc/content-slug-overrides.yaml`에 Confluence content ID와 canonical slug를 추가합니다.
+- Public route는 현재 영어 제목 번역을 slugify하여 생성합니다. 제목 변경으로 route가 바뀌면 conversion manifest가 이전 route를 감지해 `src/content-route-redirects.yaml`에 기본 8주 임시 redirect를 기록합니다.
 
 ## Confluence xhtml 을 Markdown 으로 변환하기
 

confluence-mdx/bin/content_redirects.py

@@ -0,0 +1,233 @@
+"""Lifecycle management for title-derived public content redirects."""
+
+import os
+import tempfile
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path
+from typing import Any, Dict, Mapping, Sequence
+
+import yaml
+
+
+REDIRECT_RETENTION_DAYS = 56
+
+
+class ContentRedirectError(RuntimeError):
+    """Raised when the content redirect registry is invalid."""
+
+
+def _parse_iso_date(value: Any, field: str) -> date:
+    if not isinstance(value, str):
+        raise ContentRedirectError(f"{field} must be an ISO date string")
+    try:
+        parsed = date.fromisoformat(value)
+    except ValueError as exc:
+        raise ContentRedirectError(
+            f"{field} must use YYYY-MM-DD: {value!r}"
+        ) from exc
+    if parsed.isoformat() != value:
+        raise ContentRedirectError(
+            f"{field} must use canonical YYYY-MM-DD: {value!r}"
+        )
+    return parsed
+
+
+def _validate_route(value: Any, field: str) -> str:
+    if not isinstance(value, str) or not value.startswith("/"):
+        raise ContentRedirectError(f"{field} must be a root-relative route")
+    if value == "/" or value.endswith("/") or "//" in value:
+        raise ContentRedirectError(f"{field} is not a canonical content route: {value!r}")
+    if any(part in ("", ".", "..") for part in value.split("/")[1:]):
+        raise ContentRedirectError(f"{field} contains an unsafe segment: {value!r}")
+    return value
+
+
+def _validate_redirects(data: Any) -> list[Dict[str, str]]:
+    if data is None:
+        return []
+    if not isinstance(data, list):
+        raise ContentRedirectError("Content redirect registry must be a list")
+
+    validated: list[Dict[str, str]] = []
+    seen_sources: set[str] = set()
+    for index, item in enumerate(data):
+        if not isinstance(item, dict):
+            raise ContentRedirectError(
+                f"Content redirect at index {index} must be a mapping"
+            )
+        source = _validate_route(item.get("source"), "source")
+        destination = _validate_route(item.get("destination"), "destination")
+        if source == destination:
+            raise ContentRedirectError(
+                f"Content redirect source equals destination: {source}"
+            )
+        if source in seen_sources:
+            raise ContentRedirectError(
+                f"Duplicate content redirect source: {source}"
+            )
+        created_on = _parse_iso_date(item.get("created_on"), "created_on")
+        expires_on = _parse_iso_date(item.get("expires_on"), "expires_on")
+        if expires_on <= created_on:
+            raise ContentRedirectError(
+                f"expires_on must be later than created_on for {source}"
+            )
+        seen_sources.add(source)
+        validated.append({
+            "source": source,
+            "destination": destination,
+            "created_on": created_on.isoformat(),
+            "expires_on": expires_on.isoformat(),
+        })
+    return validated
+
+
+def load_content_redirects(path: Path) -> list[Dict[str, str]]:
+    """Load and validate the persisted redirect registry."""
+    if not path.exists():
+        return []
+    try:
+        data = yaml.safe_load(path.read_text(encoding="utf-8"))
+    except yaml.YAMLError as exc:
+        raise ContentRedirectError(
+            f"Invalid YAML in content redirect registry {path}: {exc}"
+        ) from exc
+    return _validate_redirects(data)
+
+
+def _mdx_routes_by_content_id(
+    outputs: Sequence[Mapping[str, str]],
+) -> Dict[str, str]:
+    routes: Dict[str, str] = {}
+    for entry in outputs:
+        if entry.get("kind") != "mdx":
+            continue
+        content_id = str(entry.get("page_id") or "").strip()
+        relative_path = entry.get("path")
+        if not content_id or not isinstance(relative_path, str):
+            raise ContentRedirectError(
+                f"Invalid MDX manifest entry: {entry!r}"
+            )
+        if not relative_path.endswith(".mdx"):
+            raise ContentRedirectError(
+                f"MDX manifest path must end with .mdx: {relative_path}"
+            )
+        route = _validate_route(f"/{relative_path[:-4]}", "manifest route")
+        if content_id in routes:
+            raise ContentRedirectError(
+                f"Duplicate MDX output for content ID: {content_id}"
+            )
+        routes[content_id] = route
+    return routes
+
+
+def reconcile_content_redirects(
+    existing: Sequence[Mapping[str, str]],
+    previous_outputs: Sequence[Mapping[str, str]],
+    current_outputs: Sequence[Mapping[str, str]],
+    current_date: date,
+) -> list[Dict[str, str]]:
+    """Prune expired redirects and apply current content route moves."""
+    redirects = [
+        dict(item)
+        for item in _validate_redirects(list(existing))
+        if _parse_iso_date(item["expires_on"], "expires_on") > current_date
+    ]
+    previous_routes = _mdx_routes_by_content_id(previous_outputs)
+    current_routes = _mdx_routes_by_content_id(current_outputs)
+    live_routes = set(current_routes.values())
+
+    redirects = [
+        item for item in redirects
+        if item["source"] not in live_routes
+    ]
+    preexisting_redirects = list(redirects)
+
+    moves = sorted(
+        (
+            content_id,
+            previous_routes[content_id],
+            current_routes[content_id],
+        )
+        for content_id in previous_routes.keys() & current_routes.keys()
+        if previous_routes[content_id] != current_routes[content_id]
+    )
+
+    for _, old_route, new_route in moves:
+        for item in preexisting_redirects:
+            if item["destination"] == old_route:
+                item["destination"] = new_route
+
+        existing_rule = next(
+            (item for item in redirects if item["source"] == old_route),
+            None,
+        )
+        if existing_rule is None:
+            redirects.append({
+                "source": old_route,
+                "destination": new_route,
+                "created_on": current_date.isoformat(),
+                "expires_on": (
+                    current_date + timedelta(days=REDIRECT_RETENTION_DAYS)
+                ).isoformat(),
+            })
+        else:
+            existing_rule["destination"] = new_route
+
+    redirects = [
+        item for item in redirects
+        if item["source"] not in live_routes
+        and item["source"] != item["destination"]
+    ]
+    return sorted(redirects, key=lambda item: item["source"])
+
+
+def _dump_redirects(redirects: Sequence[Mapping[str, str]]) -> str:
+    return yaml.safe_dump(
+        list(redirects),
+        allow_unicode=True,
+        sort_keys=False,
+    )
+
+
+def update_content_redirects(
+    path: Path,
+    previous_outputs: Sequence[Mapping[str, str]],
+    current_outputs: Sequence[Mapping[str, str]],
+    current_date: date | None = None,
+) -> list[Dict[str, str]]:
+    """Atomically update the redirect registry for a successful conversion."""
+    effective_date = current_date or datetime.now(timezone.utc).date()
+    resolved_path = path.resolve()
+    existing = load_content_redirects(resolved_path)
+    redirects = reconcile_content_redirects(
+        existing,
+        previous_outputs,
+        current_outputs,
+        effective_date,
+    )
+    serialized = _dump_redirects(redirects)
+    if (
+        resolved_path.exists()
+        and resolved_path.read_text(encoding="utf-8") == serialized
+    ):
+        return redirects
+
+    resolved_path.parent.mkdir(parents=True, exist_ok=True)
+    temp_path: Path | None = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=resolved_path.parent,
+            prefix=f".{resolved_path.name}.",
+            suffix=".tmp",
+            delete=False,
+        ) as temp_file:
+            temp_file.write(serialized)
+            temp_path = Path(temp_file.name)
+        temp_path.chmod(0o644)
+        os.replace(temp_path, resolved_path)
+    finally:
+        if temp_path is not None and temp_path.exists():
+            temp_path.unlink()
+    return redirects

confluence-mdx/bin/convert_all.py

@@ -17,6 +17,7 @@
 import subprocess
 import sys
 import tempfile
+from datetime import date
 from pathlib import Path
 from typing import Any, Dict, List, Mapping, Sequence
 from urllib.parse import quote, urlsplit
@@ -36,6 +37,7 @@
     sys.path.insert(0, str(_SCRIPT_DIR))
 
 from fetch.sync_profiles import SYNC_PROFILES
+from content_redirects import update_content_redirects
 
 
 def _resolve(rel: str) -> str:
@@ -555,8 +557,10 @@ def finalize_manifest(
     sync_code: str,
     current_outputs: Sequence[Mapping[str, str]],
     output_base_dir: Path,
+    redirects_path: Path | None = None,
+    redirect_date: date | None = None,
 ) -> None:
-    """Remove exclusively owned stale files and atomically replace the manifest."""
+    """Update route redirects, remove stale files, and replace the manifest."""
     output_root = output_base_dir.resolve()
     previous_outputs = _manifest_outputs(manifest_path, sync_code)
 
@@ -581,6 +585,14 @@ def finalize_manifest(
         sync_code,
         output_root,
     )
+    if redirects_path is not None:
+        update_content_redirects(
+            redirects_path,
+            previous_outputs,
+            current_outputs,
+            redirect_date,
+        )
+
     for stale_relative_path in sorted(
         set(previous_by_path) - set(current_by_path) - other_profile_paths,
         reverse=True,
@@ -623,7 +635,8 @@ def convert_all(pages: List[Dict], var_dir: str, output_base_dir: str, public_di
                 log_level: str, pages_yaml: str = '',
                 manifest_path: str = '', sync_code: str = 'qm',
                 base_url: str = _DEFAULT_CONFLUENCE_BASE_URL,
-                space_key: str = '') -> int:
+                space_key: str = '', redirects_path: str = '',
+                redirect_date: date | None = None) -> int:
     """Convert typed catalog nodes and return the number of failures."""
     # Skip the root page
     root_page_id = pages[0]['page_id'] if pages else None
@@ -724,6 +737,8 @@ def convert_all(pages: List[Dict], var_dir: str, output_base_dir: str, public_di
                 sync_code,
                 generated_outputs,
                 output_base_path,
+                Path(redirects_path) if redirects_path else None,
+                redirect_date,
             )
         except Exception as exc:
             failures += 1
@@ -748,6 +763,11 @@ def main():
                         help='Public assets directory (default: target/public)')
     parser.add_argument('--translations', default='etc/korean-titles-translations.txt',
                         help='Path to translations file')
+    parser.add_argument(
+        '--redirects-file',
+        default='target/content-route-redirects.yaml',
+        help='Path to temporary content route redirects registry',
+    )
     parser.add_argument('--base-url', default=_DEFAULT_CONFLUENCE_BASE_URL,
                         help='Confluence base URL for generated folder links')
     parser.add_argument('--space-key', default=None,
@@ -769,6 +789,7 @@ def main():
     args.output_dir = _resolve(args.output_dir)
     args.public_dir = _resolve(args.public_dir)
     args.translations = _resolve(args.translations)
+    args.redirects_file = _resolve(args.redirects_file)
     manifest_path = os.path.join(
         args.var_dir,
         "convert-manifests",
@@ -805,7 +826,8 @@ def main():
                            manifest_path=manifest_path,
                            sync_code=args.sync_code,
                            base_url=args.base_url,
-                           space_key=space_key)
+                           space_key=space_key,
+                           redirects_path=args.redirects_file)
 
     if failures:
         print(f"\nCompleted with {failures} failure(s) out of {len(pages)} pages", file=sys.stderr)

confluence-mdx/bin/fetch/config.py

@@ -29,7 +29,6 @@ class Config:
     default_output_dir: str = "var"
     cache_dir: str = "cache"
     translations_file: str = "etc/korean-titles-translations.txt"
-    slug_overrides_file: str = "etc/content-slug-overrides.yaml"
     email: Optional[str] = None
     api_token: Optional[str] = None
     download_attachments: bool = False
@@ -51,7 +50,6 @@ def __post_init__(self):
             'default_output_dir',
             'cache_dir',
             'translations_file',
-            'slug_overrides_file',
         ):
             value = getattr(self, field)
             if not os.path.isabs(value):

confluence-mdx/bin/fetch/processor.py

@@ -27,7 +27,6 @@ def __init__(self, config: Config, logger: logging.Logger):
         self.file_manager = FileManager(logger)
         self.translation_service = TranslationService(
             config.translations_file,
-            config.slug_overrides_file,
             logger,
         )
 
@@ -39,7 +38,6 @@ def __init__(self, config: Config, logger: logging.Logger):
 
         # Load translations
         self.translation_service.load_translations()
-        self.translation_service.load_slug_overrides()
 
     def process_page_complete(
         self,