ypshidifu
a41cf92379
- Win DB 实锤：目标会话 ses_eb074c0daffe 的 session_v2.directory = C:/Users/ypshidifu（截图报错来源） - FIX_WORKTREE_SRC 扩展：project.worktree + session.directory + session_v2.directory 三表全洗（幂等，每次无条件跑） - 推送侧转换统一 _conv_win_path，session/session_v2 行的 directory 列同步转换
446 lines
17 KiB
Python
446 lines
17 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"]
|
||
# ★ project.worktree 路径转换（Win 绝对路径 -> WSL 绝对路径）
|
||
# Win 侧 session 的 project.worktree = C:/Users/ypshidifu/... （OC 启动 cwd）
|
||
# 推给 WSL 后原样是 Win 路径 → WSL 侧 session 面板提示
|
||
# 「Session location unavailable / C:/Users/ypshidifu / Choose another directory」。
|
||
# 必须把 C:/Users/ypshidifu 前缀转成 WSL 的 /root。
|
||
_WIN_HOME_PREFIX_SL = "C:/Users/ypshidifu"
|
||
_WSL_HOME_PREFIX = "/root"
|
||
|
||
|
||
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))
|
||
'''
|
||
|
||
# ★ 存量 Win 路径洗白（2026-10-20 二次实锤：OC 2.x 的 "Session location" 读的是
|
||
# session/session_v2 表自身的 directory 列，不只是 project.worktree！
|
||
# 目标会话 ses_eb074c0daffe 的 session_v2.directory = "C:/Users/ypshidifu" 即截图来源。
|
||
# 且 IMPORT 用 INSERT OR IGNORE —— WSL 已存在的行永远不会被 UPDATE，
|
||
# 推送侧转换只对「新插入」的行生效。此脚本每次同步无条件跑（幂等 UPDATE），
|
||
# 把三张表里所有 Win home 前缀的路径洗成 /root。
|
||
FIX_WORKTREE_SRC = r'''
|
||
import sqlite3
|
||
BK = chr(92)
|
||
PRE_SL = "C:/Users/ypshidifu"
|
||
PRE_BK = "C:" + BK + "Users" + BK + "ypshidifu"
|
||
res = {"fixed": {}, "err": None}
|
||
def conv(p):
|
||
if not p or not isinstance(p, str):
|
||
return None
|
||
if p.startswith(PRE_SL):
|
||
return "/root" + p[len(PRE_SL):]
|
||
if p.startswith(PRE_BK):
|
||
return "/root" + p[len(PRE_BK):]
|
||
return None
|
||
try:
|
||
con = sqlite3.connect("/root/.local/share/opencode/opencode.db")
|
||
cur = con.cursor()
|
||
# project.worktree
|
||
n = 0
|
||
for pid, wt in cur.execute("select id, worktree from project").fetchall():
|
||
nw = conv(wt)
|
||
if nw:
|
||
cur.execute("update project set worktree=? where id=?", (nw, pid))
|
||
n += 1
|
||
res["fixed"]["project.worktree"] = n
|
||
# session.directory（列存在才跑）
|
||
def fix_dir(table):
|
||
cols = [r[1] for r in cur.execute("PRAGMA table_info(%s)" % table)]
|
||
if "directory" not in cols:
|
||
return -1
|
||
n = 0
|
||
for sid, d in cur.execute("select id, directory from %s" % table).fetchall():
|
||
nw = conv(d)
|
||
if nw:
|
||
cur.execute("update %s set directory=? where id=?" % table, (nw, sid))
|
||
n += 1
|
||
return n
|
||
res["fixed"]["session.directory"] = fix_dir("session")
|
||
res["fixed"]["session_v2.directory"] = fix_dir("session_v2")
|
||
con.commit()
|
||
con.close()
|
||
except Exception as e:
|
||
res["err"] = str(e)
|
||
open("/root/.oc_fix_wt_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 {}
|
||
|
||
# ★ 存量 worktree 洗白：每次无条件跑（幂等，修 INSERT OR IGNORE 洗不掉存量行的洞）
|
||
try:
|
||
rc_fw, _ = wsl_py(FIX_WORKTREE_SRC, timeout=60)
|
||
if rc_fw == 0:
|
||
fw = json.loads(wsl_cat("/root/.oc_fix_wt_result.json"))
|
||
if fw.get("fixed"):
|
||
log("worktree fixed: %d rows -> /root" % fw["fixed"])
|
||
if fw.get("err"):
|
||
log("worktree fix err: " + str(fw["err"]))
|
||
except Exception as e:
|
||
log("worktree fix exception: " + str(e))
|
||
# 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),
|
||
)
|
||
]
|
||
# ★ Win 路径转换（C:/Users/ypshidifu / C:\Users\ypshidifu -> /root）
|
||
# 修 WSL 侧「Session location unavailable / C:/Users/ypshidifu」
|
||
# 2026-10-20 二次实锤：OC 2.x 的 location 显示读 session/session_v2 自己的
|
||
# directory 列，project.worktree 也要转，三处一起。
|
||
def _conv_win_path(p):
|
||
if not isinstance(p, str):
|
||
return p
|
||
if p.startswith(_WIN_HOME_PREFIX_SL):
|
||
return _WSL_HOME_PREFIX + p[len(_WIN_HOME_PREFIX_SL):]
|
||
if p.startswith("C:" + chr(92) + "Users" + chr(92) + "ypshidifu"):
|
||
return _WSL_HOME_PREFIX + p[len("C:" + chr(92) + "Users" + chr(92) + "ypshidifu"):]
|
||
return p
|
||
|
||
if tables.get("project"):
|
||
for _pr in tables["project"]:
|
||
_pr["worktree"] = _conv_win_path(_pr.get("worktree"))
|
||
for _t in ("session", "session_v2"):
|
||
for _r in tables.get(_t) or []:
|
||
if "directory" in _r:
|
||
_r["directory"] = _conv_win_path(_r.get("directory"))
|
||
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)
|