355 lines
12 KiB
Python
355 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
oc-session-sync.py - 对话数据同步: 本机(Win) -> WSL。
|
||
收尸链路 step 7 调用。设计(2026-10-02 首版; 2026-10-12 重建, 适配 2.0.27 schema):
|
||
- 幂等: .oc_sync_state.json 存 {session_id: time_updated} 快照, 无变化秒退。
|
||
- 并集语义: 绝不删 WSL 数据, 只补缺:
|
||
* WSL 缺的 session -> 整行插 + 其全部 message/part
|
||
* 两边都有但本机 time_updated 更新 -> 只补 WSL 缺的 message/part;
|
||
time_updated 只增不减 (max 更新)。
|
||
- 传输: export gzip -> `wsl -- bash -c "cat > file"` stdin 管道(绕 UNC OSError 22);
|
||
WSL 侧脚本一律经 stdin 喂 python3 -, 脚本只读本地文件(避免 stdin 冲突)。
|
||
- 跨版本 schema: pragma table_info 取目标表列交集, INSERT OR IGNORE。
|
||
- 预算 150s; 异常吞掉只写日志([sync] 前缀), 绝不抛出阻塞退出链路。
|
||
"""
|
||
import gzip
|
||
import io
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
|
||
try:
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
|
||
sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
pass
|
||
|
||
TAG = "[sync]"
|
||
HOME = os.path.expanduser("~")
|
||
WIN_DB = os.path.join(HOME, ".local", "share", "opencode", "opencode.db")
|
||
STATE = os.path.join(HOME, ".config", "opencode", ".oc_sync_state.json")
|
||
LOG = os.path.join(HOME, ".config", "opencode", "oc-session-all.log")
|
||
REMOTE_IMPORT = "/root/.oc_sync_import.json.gz"
|
||
REMOTE_PROBE = "/root/.oc_sync_probe.json"
|
||
REMOTE_RESULT = "/root/.oc_sync_result.json"
|
||
DISTRO = "Ubuntu-24.04"
|
||
BUDGET = 150
|
||
# 同步的表(WSL 侧缺表/缺列自动跳过)
|
||
TABLES = ["project", "session", "session_v2", "session_message", "part"]
|
||
|
||
|
||
def log(m):
|
||
try:
|
||
with open(LOG, "a", encoding="utf-8") as f:
|
||
f.write("[%s] %s %s\n" % (time.strftime("%Y-%m-%dT%H:%M:%S"), TAG, m))
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def wsl_py(src, timeout=120):
|
||
"""经 stdin 喂脚本给 wsl python3 - (脚本内禁止再读 stdin)。返回 (rc, out_bytes)。"""
|
||
r = subprocess.run(
|
||
["wsl.exe", "-d", DISTRO, "--", "python3", "-"],
|
||
input=src.encode("utf-8"),
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
)
|
||
return r.returncode, r.stdout
|
||
|
||
|
||
def wsl_push(data, remote_path, timeout=120):
|
||
cmd = 'cat > "%s"' % remote_path
|
||
r = subprocess.run(
|
||
["wsl.exe", "-d", DISTRO, "--", "bash", "-c", cmd],
|
||
input=data,
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
)
|
||
return r.returncode
|
||
|
||
|
||
def wsl_cat(remote_path, timeout=60):
|
||
r = subprocess.run(
|
||
["wsl.exe", "-d", DISTRO, "--", "cat", remote_path],
|
||
capture_output=True,
|
||
timeout=timeout,
|
||
)
|
||
raw = r.stdout or b""
|
||
if raw[:1] == b"{":
|
||
return raw.decode("utf-8", errors="replace")
|
||
try:
|
||
t = raw.decode("utf-16-le")
|
||
if "{" in t:
|
||
return t[t.index("{"):]
|
||
except Exception:
|
||
pass
|
||
return raw.decode("utf-8", errors="replace")
|
||
|
||
|
||
PROBE_SRC = r'''
|
||
import json, os, sqlite3, glob
|
||
out = {"ok": False}
|
||
try:
|
||
cands = ["/root/.local/share/opencode/opencode.db"]
|
||
cands += glob.glob("/root/.local/share/**/opencode.db", recursive=True)
|
||
db = next((c for c in cands if os.path.exists(c)), None)
|
||
if not db:
|
||
out["err"] = "wsl db not found"
|
||
else:
|
||
con = sqlite3.connect(db)
|
||
def tu(t):
|
||
try:
|
||
return {r[0]: (r[1] or 0) for r in con.execute("select id, time_updated from " + t)}
|
||
except Exception:
|
||
return {}
|
||
out["db"] = db
|
||
# 2026-10-20：双表驱动（新版 OC 活跃会话写 session_v2，旧版写 session）
|
||
out["sessions"] = tu("session")
|
||
out["sessions_v2"] = tu("session_v2")
|
||
out["msgs"] = tu("session_message")
|
||
out["ok"] = True
|
||
con.close()
|
||
except Exception as e:
|
||
out["err"] = str(e)
|
||
open("/root/.oc_sync_probe.json", "w", encoding="utf-8").write(json.dumps(out))
|
||
'''
|
||
|
||
IMPORT_SRC = r'''
|
||
import gzip, json, os, sqlite3
|
||
res = {"ok": False}
|
||
try:
|
||
raw = gzip.decompress(open("/root/.oc_sync_import.json.gz", "rb").read())
|
||
payload = json.loads(raw.decode("utf-8"))
|
||
con = sqlite3.connect("/root/.local/share/opencode/opencode.db")
|
||
cur = con.cursor()
|
||
stats = {}
|
||
for t, rows in payload.get("tables", {}).items():
|
||
try:
|
||
tgt_cols = [r[1] for r in cur.execute("pragma table_info(%s)" % t)]
|
||
except Exception:
|
||
tgt_cols = []
|
||
if not tgt_cols:
|
||
stats[t] = "missing-table"
|
||
continue
|
||
ins = 0
|
||
upd = 0
|
||
for row in rows:
|
||
cols = [c for c in tgt_cols if c in row]
|
||
if not cols or "id" not in cols:
|
||
continue
|
||
try:
|
||
cur.execute(
|
||
"INSERT OR IGNORE INTO %s (%s) VALUES (%s)"
|
||
% (t, ",".join(cols), ",".join("?" * len(cols))),
|
||
[row[c] for c in cols],
|
||
)
|
||
if cur.rowcount > 0:
|
||
ins += cur.rowcount
|
||
elif "time_updated" in tgt_cols and "time_updated" in row and row["time_updated"]:
|
||
cur.execute(
|
||
"UPDATE %s SET time_updated=? WHERE id=? AND (time_updated IS NULL OR time_updated<?)"
|
||
% t,
|
||
[row["time_updated"], row["id"], row["time_updated"]],
|
||
)
|
||
upd += cur.rowcount
|
||
except Exception:
|
||
pass
|
||
stats[t] = {"ins": ins, "upd": upd}
|
||
con.commit()
|
||
con.close()
|
||
res = {"ok": True, "stats": stats}
|
||
except Exception as e:
|
||
res = {"err": str(e)}
|
||
open("/root/.oc_sync_result.json", "w", encoding="utf-8").write(json.dumps(res))
|
||
'''
|
||
|
||
|
||
def main():
|
||
t0 = time.time()
|
||
if not os.path.exists(WIN_DB):
|
||
log("win db missing: " + WIN_DB)
|
||
return
|
||
# 2026-10-20：双表驱动。新版 OC(2.0.27) 活跃会话写 session_v2，旧表 session 是历史遗留。
|
||
# 只盯 session 旧表会导致「Win 新建会话全在 v2 → 旧表无变化 → 永远 skip → WSL 读不到」。
|
||
win = sqlite3.connect("file:%s?mode=ro" % WIN_DB.replace("\\", "/"), uri=True)
|
||
try:
|
||
win_sessions = {r[0]: (r[1] or 0) for r in win.execute("select id, time_updated from session")}
|
||
except Exception as e:
|
||
log("win session read ERR: " + str(e))
|
||
return
|
||
finally:
|
||
win.close()
|
||
|
||
try:
|
||
rc, _ = wsl_py(PROBE_SRC, timeout=90)
|
||
if rc != 0:
|
||
log("wsl probe rc=" + str(rc) + " (vm cold-boot or python missing?)")
|
||
return
|
||
probe = json.loads(wsl_cat(REMOTE_PROBE))
|
||
if not probe.get("ok"):
|
||
log("wsl probe ERR: " + str(probe.get("err")))
|
||
return
|
||
except Exception as e:
|
||
log("wsl probe exception: " + str(e))
|
||
return
|
||
|
||
wsl_sessions = probe.get("sessions") or {}
|
||
wsl_sessions_v2 = probe.get("sessions_v2") or {}
|
||
wsl_msgs = probe.get("msgs") or {}
|
||
# WSL 侧合并双表（id -> 最新 time_updated）
|
||
wsl_all = dict(wsl_sessions)
|
||
for k, v in (wsl_sessions_v2 or {}).items():
|
||
wsl_all[k] = max(wsl_all.get(k, 0), v)
|
||
|
||
# 读 Win session_v2（新版主表；缺表当空）
|
||
try:
|
||
win = sqlite3.connect("file:%s?mode=ro" % WIN_DB.replace("\\", "/"), uri=True)
|
||
win_sessions_v2 = {r[0]: (r[1] or 0) for r in win.execute("select id, time_updated from session_v2")}
|
||
except Exception:
|
||
win_sessions_v2 = {}
|
||
finally:
|
||
win.close()
|
||
# Win 侧合并双表
|
||
win_all = dict(win_sessions)
|
||
for k, v in win_sessions_v2.items():
|
||
win_all[k] = max(win_all.get(k, 0), v)
|
||
|
||
# 幂等快照（按合并后的双表快照判断，任一表变都触发重推）
|
||
snapshot = {
|
||
"session": win_sessions,
|
||
"session_v2": win_sessions_v2,
|
||
}
|
||
try:
|
||
state = json.load(open(STATE, encoding="utf-8"))
|
||
except Exception:
|
||
state = {}
|
||
# 兼容旧版快照（旧版直接存 {id: time} 平铺）
|
||
if not isinstance(state, dict) or "session" not in state:
|
||
state = {"session": state if isinstance(state, dict) else {}, "session_v2": {}}
|
||
if state == snapshot:
|
||
log("unchanged, skip (session=%d v2=%d)" % (len(win_sessions), len(win_sessions_v2)))
|
||
return
|
||
|
||
# diff：合并双表后的缺失/过期
|
||
need_sessions = [s for s in win_all if s not in wsl_all]
|
||
stale = [s for s in win_all if s in wsl_all and win_all[s] > wsl_all[s]]
|
||
targets = set(need_sessions) | set(stale)
|
||
if not targets:
|
||
json.dump(snapshot, open(STATE, "w", encoding="utf-8"))
|
||
log("no missing/stale sessions; state refreshed (session=%d v2=%d)" %
|
||
(len(win_sessions), len(win_sessions_v2)))
|
||
return
|
||
|
||
win = sqlite3.connect("file:%s?mode=ro" % WIN_DB.replace("\\", "/"), uri=True)
|
||
win.row_factory = sqlite3.Row
|
||
try:
|
||
# collect message ids per target session (need full set for new, missing-only handled at import by OR IGNORE)
|
||
msg_rows = []
|
||
for r in win.execute(
|
||
"select * from session_message where session_id in (%s)"
|
||
% ",".join("?" * len(targets)),
|
||
list(targets),
|
||
):
|
||
msg_rows.append(dict(r))
|
||
msg_ids = [m["id"] for m in msg_rows if m["id"] not in wsl_msgs]
|
||
missing_msg = [m for m in msg_rows if m["id"] not in wsl_msgs]
|
||
|
||
tables = {}
|
||
# project rows referenced by target sessions（双表并集查 project_id）
|
||
pids = set()
|
||
q1 = "select distinct project_id from session where id in (%s)" % ",".join("?" * len(targets))
|
||
for r in win.execute(q1, list(targets)):
|
||
if r[0]:
|
||
pids.add(r[0])
|
||
try:
|
||
for r in win.execute(
|
||
"select distinct project_id from session_v2 where id in (%s)" % ",".join("?" * len(targets)),
|
||
list(targets),
|
||
):
|
||
if r[0]:
|
||
pids.add(r[0])
|
||
except Exception:
|
||
pass
|
||
if pids:
|
||
tables["project"] = [
|
||
dict(r)
|
||
for r in win.execute(
|
||
"select * from project where id in (%s)" % ",".join("?" * len(pids)),
|
||
list(pids),
|
||
)
|
||
]
|
||
tables["session"] = [
|
||
dict(r)
|
||
for r in win.execute(
|
||
"select * from session where id in (%s)" % ",".join("?" * len(targets)),
|
||
list(targets),
|
||
)
|
||
]
|
||
try:
|
||
tables["session_v2"] = [
|
||
dict(r)
|
||
for r in win.execute(
|
||
"select * from session_v2 where id in (%s)" % ",".join("?" * len(targets)),
|
||
list(targets),
|
||
)
|
||
]
|
||
except Exception:
|
||
tables["session_v2"] = []
|
||
tables["session_message"] = missing_msg
|
||
# parts of missing messages
|
||
if msg_ids:
|
||
parts = []
|
||
chunk = 400
|
||
for i in range(0, len(msg_ids), chunk):
|
||
sub = msg_ids[i : i + chunk]
|
||
for r in win.execute(
|
||
"select * from part where message_id in (%s)" % ",".join("?" * len(sub)),
|
||
sub,
|
||
):
|
||
parts.append(dict(r))
|
||
tables["part"] = parts
|
||
else:
|
||
tables["part"] = []
|
||
except Exception as e:
|
||
win.close()
|
||
log("win export ERR: " + str(e))
|
||
return
|
||
win.close()
|
||
|
||
payload = json.dumps({"tables": tables}, ensure_ascii=False).encode("utf-8")
|
||
blob = gzip.compress(payload, 1)
|
||
nmsg = len(tables.get("session_message") or [])
|
||
npart = len(tables.get("part") or [])
|
||
log(
|
||
"diff: new=%d stale=%d msgs=%d parts=%d payload=%dKB"
|
||
% (len(need_sessions), len(stale), nmsg, npart, len(blob) // 1024)
|
||
)
|
||
|
||
try:
|
||
rc = wsl_push(blob, REMOTE_IMPORT, timeout=120)
|
||
if rc != 0:
|
||
log("push rc=" + str(rc))
|
||
return
|
||
rc, _ = wsl_py(IMPORT_SRC, timeout=120)
|
||
if rc != 0:
|
||
log("import rc=" + str(rc))
|
||
return
|
||
res = json.loads(wsl_cat(REMOTE_RESULT))
|
||
log("import result: " + json.dumps(res, ensure_ascii=False)[:300])
|
||
json.dump(snapshot, open(STATE, "w", encoding="utf-8"))
|
||
log("state updated (session=%d v2=%d) in %ds" %
|
||
(len(win_sessions), len(win_sessions_v2), int(time.time() - t0)))
|
||
except Exception as e:
|
||
log("sync exception: " + str(e))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except Exception as e:
|
||
log("FATAL: " + str(e))
|
||
sys.exit(0)
|