Openclaw迁移功能（时间简史）初步方案

astrbot/cli/__main__.py

@@ -5,7 +5,7 @@
 import click
 
 from . import __version__
-from .commands import conf, init, plug, run
+from .commands import conf, init, migrate, plug, run
 
 logo_tmpl = r"""
      ___           _______.___________..______      .______     ______   .___________.
@@ -54,6 +54,7 @@ def help(command_name: str | None) -> None:
 cli.add_command(help)
 cli.add_command(plug)
 cli.add_command(conf)
+cli.add_command(migrate)
 
 if __name__ == "__main__":
     cli()

astrbot/cli/commands/__init__.py

@@ -1,6 +1,7 @@
 from .cmd_conf import conf
 from .cmd_init import init
+from .cmd_migrate import migrate
 from .cmd_plug import plug
 from .cmd_run import run
 
-__all__ = ["conf", "init", "plug", "run"]
+__all__ = ["conf", "init", "migrate", "plug", "run"]

astrbot/cli/commands/cmd_migrate.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import click
+
+from ..utils import get_astrbot_root
+from ..utils.openclaw_migrate import run_openclaw_migration
+
+
+@click.group(name="migrate")
+def migrate() -> None:
+    """Data migration utilities for external runtimes."""
+
+
+@migrate.command(name="openclaw")
+@click.option(
+    "--source",
+    "source_path",
+    type=click.Path(path_type=Path, file_okay=False, resolve_path=True),
+    default=None,
+    help="Path to OpenClaw root directory (default: ~/.openclaw).",
+)
+@click.option(
+    "--target",
+    "target_path",
+    type=click.Path(path_type=Path, file_okay=False, resolve_path=False),
+    default=None,
+    help=(
+        "Custom output directory. If omitted, writes to "
+        "data/migrations/openclaw/run-<timestamp>."
+    ),
+)
+@click.option(
+    "--dry-run",
+    is_flag=True,
+    default=False,
+    help="Preview migration candidates without writing files.",
+)
+def migrate_openclaw(
+    source_path: Path | None,
+    target_path: Path | None,
+    dry_run: bool,
+) -> None:
+    """Migrate OpenClaw workspace snapshots into AstrBot migration artifacts."""
+
+    astrbot_root = get_astrbot_root()
+    source_root = source_path or (Path.home() / ".openclaw")
+
+    report = run_openclaw_migration(
+        source_root=source_root,
+        astrbot_root=astrbot_root,
+        dry_run=dry_run,
+        target_dir=target_path,
+    )
+
+    click.echo("OpenClaw migration report:")
+    click.echo(f"  Source root:        {report.source_root}")
+    click.echo(f"  Source workspace:   {report.source_workspace}")
+    click.echo(f"  Dry run:            {report.dry_run}")
+    click.echo(f"  Memory entries:     {report.memory_entries_total}")
+    click.echo(f"    - sqlite:         {report.memory_entries_from_sqlite}")
+    click.echo(f"    - markdown:       {report.memory_entries_from_markdown}")
+    click.echo(f"  Workspace files:    {report.workspace_files_total}")
+    click.echo(f"  Workspace size:     {report.workspace_bytes_total} bytes")
+    click.echo(f"  Config found:       {report.config_found}")
+
+    if dry_run:
+        click.echo("")
+        click.echo("Dry-run mode: no files were written.")
+        if target_path is not None:
+            click.echo("Note: --target is ignored when --dry-run is enabled.")
+        click.echo("Run without --dry-run to perform migration.")
+        return
+
+    click.echo("")
+    click.echo(f"Migration output:     {report.target_dir}")
+    click.echo(f"  Copied files:       {report.copied_workspace_files}")
+    click.echo(f"  Imported memories:  {report.copied_memory_entries}")
+    click.echo(f"  Timeline written:   {report.wrote_timeline}")
+    click.echo(f"  Config TOML written: {report.wrote_config_toml}")
+    click.echo("Done.")
+
+
+__all__ = ["migrate"]

astrbot/cli/utils/openclaw_artifacts.py

@@ -0,0 +1,169 @@
+from __future__ import annotations
+
+import datetime as dt
+import json
+import os
+import shutil
+from pathlib import Path
+from typing import Any
+
+import click
+
+from .openclaw_models import MemoryEntry
+from .openclaw_toml import json_to_toml
+
+
+def _is_within(path: Path, parent: Path) -> bool:
+    try:
+        path.resolve().relative_to(parent.resolve())
+        return True
+    except (OSError, ValueError):
+        return False
+
+
+def collect_workspace_files(
+    workspace_dir: Path, *, exclude_dir: Path | None = None
+) -> list[Path]:
+    files: list[Path] = []
+    exclude_resolved = exclude_dir.resolve() if exclude_dir is not None else None
+
+    for root, dirnames, filenames in os.walk(
+        workspace_dir, topdown=True, followlinks=False
+    ):
+        root_path = Path(root)
+
+        pruned_dirs: list[str] = []
+        for dirname in dirnames:
+            dir_path = root_path / dirname
+            if dir_path.is_symlink():
+                continue
+            if exclude_resolved is not None and _is_within(dir_path, exclude_resolved):
+                continue
+            pruned_dirs.append(dirname)
+        dirnames[:] = pruned_dirs
+
+        for filename in filenames:
+            path = root_path / filename
+            if path.is_symlink() or not path.is_file():
+                continue
+            if exclude_resolved is not None and _is_within(path, exclude_resolved):
+                continue
+            files.append(path)
+
+    return sorted(files)
+
+
+def workspace_total_size(files: list[Path]) -> int:
+    total_bytes = 0
+    for path in files:
+        try:
+            total_bytes += path.stat().st_size
+        except OSError:
+            # Best-effort accounting: files may disappear or become unreadable
+            # during migration scans.
+            continue
+    return total_bytes
+
+
+def _write_jsonl(path: Path, entries: list[MemoryEntry]) -> None:
+    with path.open("w", encoding="utf-8") as fp:
+        for entry in entries:
+            fp.write(
+                json.dumps(
+                    {
+                        "key": entry.key,
+                        "content": entry.content,
+                        "category": entry.category,
+                        "timestamp": entry.timestamp,
+                        "source": entry.source,
+                    },
+                    ensure_ascii=False,
+                )
+                + "\n"
+            )
+
+
+def _write_timeline(path: Path, entries: list[MemoryEntry], source_root: Path) -> None:
+    ordered = sorted(entries, key=lambda e: (e.timestamp or "", e.order))
+
+    lines: list[str] = []
+    lines.append("# OpenClaw Migration - Time Brief History")
+    lines.append("")
+    lines.append("> 时间简史（初步方案）：按时间汇总可迁移记忆条目。")
+    lines.append("")
+    lines.append(f"- Generated at: {dt.datetime.now(dt.timezone.utc).isoformat()}")
+    lines.append(f"- Source: `{source_root}`")
+    lines.append(f"- Total entries: {len(ordered)}")
+    lines.append("")
+    lines.append("## Timeline")
+    lines.append("")
+
+    for entry in ordered:
+        ts = entry.timestamp or "unknown"
+        snippet = entry.content.replace("\n", " ").strip()
+        if len(snippet) > 160:
+            snippet = snippet[:157] + "..."
+        safe_key = (entry.key or "").replace("`", "\\`")
+        safe_snippet = snippet.replace("`", "\\`")
+        lines.append(f"- [{ts}] ({entry.category}) `{safe_key}`: {safe_snippet}")
+
+    lines.append("")
+    path.write_text("\n".join(lines), encoding="utf-8")
+
+
+def write_migration_artifacts(
+    *,
+    workspace_dir: Path,
+    workspace_files: list[Path],
+    resolved_target: Path,
+    source_root: Path,
+    memory_entries: list[MemoryEntry],
+    config_obj: dict[str, Any] | None,
+    config_json_path: Path | None,
+) -> tuple[int, int, bool, bool]:
+    workspace_target = resolved_target / "workspace"
+    workspace_target.mkdir(parents=True, exist_ok=True)
+
+    copied_workspace_files = 0
+    for src_file in workspace_files:
+        rel_path = src_file.relative_to(workspace_dir)
+        dst_file = workspace_target / rel_path
+        dst_file.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(src_file, dst_file)
+        copied_workspace_files += 1
+
+    copied_memory_entries = 0
+    wrote_timeline = False
+    if memory_entries:
+        _write_jsonl(resolved_target / "memory_entries.jsonl", memory_entries)
+        copied_memory_entries = len(memory_entries)
+        _write_timeline(
+            resolved_target / "time_brief_history.md",
+            memory_entries,
+            source_root,
+        )
+        wrote_timeline = True
+
+    wrote_config_toml = False
+    if config_obj is not None:
+        (resolved_target / "config.original.json").write_text(
+            json.dumps(config_obj, ensure_ascii=False, indent=2),
+            encoding="utf-8",
+        )
+        try:
+            converted_toml = json_to_toml(config_obj)
+        except ValueError as exc:
+            source_hint = str(config_json_path) if config_json_path else "config JSON"
+            raise click.ClickException(
+                f"Failed to convert {source_hint} to TOML: {exc}"
+            ) from exc
+        (resolved_target / "config.migrated.toml").write_text(
+            converted_toml,
+            encoding="utf-8",
+        )
+        wrote_config_toml = True
+
+    return copied_workspace_files, copied_memory_entries, wrote_timeline, wrote_config_toml
+
+
+__all__ = ["collect_workspace_files", "workspace_total_size", "write_migration_artifacts"]