102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Copy selected legacy Hermes sessions into an isolated user home."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import sqlite3
|
|
|
|
from hermes_state import SessionDB
|
|
|
|
|
|
def _copy_if_missing(source: Path, target: Path) -> None:
|
|
"""Copy one credential/config file without overwriting user state."""
|
|
if source.is_file() and not target.exists():
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(source, target)
|
|
|
|
|
|
def _copy_session(
|
|
source: sqlite3.Connection,
|
|
target: sqlite3.Connection,
|
|
session_id: str,
|
|
user_id: str,
|
|
) -> bool:
|
|
"""Copy a session and its messages, preserving original timestamps."""
|
|
source.row_factory = sqlite3.Row
|
|
session = source.execute(
|
|
"SELECT * FROM sessions WHERE id = ?", (session_id,)
|
|
).fetchone()
|
|
if session is None:
|
|
return False
|
|
|
|
columns = list(session.keys())
|
|
values = [user_id if column == "user_id" else session[column] for column in columns]
|
|
placeholders = ", ".join("?" for _ in columns)
|
|
target.execute(
|
|
f"INSERT OR IGNORE INTO sessions ({', '.join(columns)}) "
|
|
f"VALUES ({placeholders})",
|
|
values,
|
|
)
|
|
|
|
message_columns = [
|
|
row[1] for row in target.execute("PRAGMA table_info(messages)").fetchall()
|
|
]
|
|
selected_columns = [
|
|
column
|
|
for column in message_columns
|
|
if column in {
|
|
row[1]
|
|
for row in source.execute("PRAGMA table_info(messages)").fetchall()
|
|
}
|
|
]
|
|
column_sql = ", ".join(selected_columns)
|
|
target.execute(
|
|
f"INSERT OR IGNORE INTO messages ({column_sql}) "
|
|
f"SELECT {column_sql} FROM source_db.messages WHERE session_id = ?",
|
|
(session_id,),
|
|
)
|
|
return True
|
|
|
|
|
|
def main() -> None:
|
|
"""Create the isolated home and perform the idempotent session copy."""
|
|
source_home = Path(os.environ["HERMES_MIGRATE_SOURCE_HOME"])
|
|
target_home = Path(os.environ["HERMES_HOME"])
|
|
user_id = os.environ["HERMES_MIGRATE_USER_ID"].strip()
|
|
session_ids = [
|
|
value.strip()
|
|
for value in os.environ.get("HERMES_MIGRATE_SESSION_IDS", "").split(",")
|
|
if value.strip()
|
|
]
|
|
|
|
target_home.mkdir(parents=True, exist_ok=True)
|
|
(target_home / "home" / ".local" / "bin").mkdir(parents=True, exist_ok=True)
|
|
(target_home / "workspace" / "skills").mkdir(parents=True, exist_ok=True)
|
|
(target_home / "logs").mkdir(parents=True, exist_ok=True)
|
|
for filename in (".env", "auth.json"):
|
|
_copy_if_missing(source_home / filename, target_home / filename)
|
|
|
|
source_db = source_home / "state.db"
|
|
target_db = target_home / "state.db"
|
|
SessionDB(db_path=target_db).close()
|
|
copied = 0
|
|
if source_db.is_file() and session_ids:
|
|
source = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True)
|
|
target = sqlite3.connect(target_db)
|
|
target.execute("ATTACH DATABASE ? AS source_db", (str(source_db),))
|
|
try:
|
|
with target:
|
|
for session_id in session_ids:
|
|
copied += int(_copy_session(source, target, session_id, user_id))
|
|
finally:
|
|
target.close()
|
|
source.close()
|
|
print(f"isolated Hermes home ready; migrated_sessions={copied}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|