zqlit
a74b3c7127
Deploy to Production / pre-check (push) Successful in 58s
Deploy to Production / build (push) Successful in 4m3s
Deploy to Production / deploy-edgeone (push) Successful in 3m48s
Deploy to Production / finalize (push) Successful in 26s
Deploy to Production / notify-failure (push) Skipped
389 lines
14 KiB
Python
389 lines
14 KiB
Python
"""
|
||
增量导入 .artrans —— 只插库里没有的评论/用户/页面，不动已有数据。
|
||
|
||
和 import_artrans.py 的区别：
|
||
import_artrans.py 首次全量导入，头部是 DELETE FROM comments/pages/users/sites（破坏性）
|
||
import_delta.py 增量：先从远端 D1 拉现状（已有 comment id / user / page），
|
||
再只生成「差集」的 INSERT，rid / root_id 依旧按全量 rid 链计算，
|
||
所以新回复挂到已有父评论上时 root_id 也是对的。
|
||
|
||
用法：
|
||
python tools/import_delta.py --src <path.artrans> --site 优世界 # 只生成 SQL，打印计划
|
||
python tools/import_delta.py --src <path.artrans> --site 优世界 --apply # 生成并执行
|
||
|
||
依赖环境变量 CLOUDFLARE_API_TOKEN（wrangler 用）。
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import pathlib
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from collections import OrderedDict
|
||
|
||
DEFAULT_SITE = "优世界"
|
||
DEFAULT_SITE_URL = "https://usj.cc"
|
||
CHUNK = 400
|
||
|
||
ROOT = pathlib.Path(__file__).resolve().parent.parent
|
||
WRANGLER = ROOT / "node_modules" / "wrangler" / "bin" / "wrangler.js"
|
||
|
||
|
||
# ----------------------------------------------------------------- SQL helpers
|
||
def sql_str(v) -> str:
|
||
if v is None:
|
||
return "NULL"
|
||
return "'" + str(v).replace("'", "''") + "'"
|
||
|
||
|
||
def sql_bool(v) -> str:
|
||
return "1" if str(v).lower() in ("true", "1", "yes") else "0"
|
||
|
||
|
||
def sql_int(v, dft=0) -> str:
|
||
try:
|
||
return str(int(str(v).strip()))
|
||
except Exception:
|
||
return str(dft)
|
||
|
||
|
||
def parse_dt(s) -> int:
|
||
"""'2021-09-11 20:33:56 +0800 +0800' -> unix 毫秒"""
|
||
s = (s or "").strip()
|
||
if len(s) < 19:
|
||
return 0
|
||
try:
|
||
import datetime as dt
|
||
|
||
naive = dt.datetime.strptime(s[:19], "%Y-%m-%d %H:%M:%S")
|
||
aware = naive.replace(tzinfo=dt.timezone(dt.timedelta(hours=8)))
|
||
return int(aware.timestamp() * 1000)
|
||
except Exception:
|
||
return 0
|
||
|
||
|
||
# ----------------------------------------------------------------- 远端现状
|
||
def wrangler(args: list, tries: int = 3):
|
||
cmd = ["node", str(WRANGLER)] + args
|
||
last = None
|
||
for i in range(tries):
|
||
try:
|
||
p = subprocess.run(cmd, cwd=ROOT, capture_output=True, text=True, timeout=600)
|
||
if p.returncode == 0:
|
||
return p.stdout
|
||
last = (p.stdout or "") + (p.stderr or "")
|
||
except Exception as e: # noqa: BLE001
|
||
last = str(e)
|
||
print(f" wrangler 第 {i + 1} 次失败，3 秒后重试", file=sys.stderr)
|
||
time.sleep(3)
|
||
raise RuntimeError(f"wrangler 失败：{(last or '')[:400]}")
|
||
|
||
|
||
def parse_json_out(s: str):
|
||
"""wrangler --json 前面可能插一行代理告警，从第一个 [ 或 { 开始截"""
|
||
i = min([x for x in (s.find("["), s.find("{")) if x >= 0] or [0])
|
||
return json.loads(s[i:])
|
||
|
||
|
||
def fetch_state(db: str) -> dict:
|
||
sql = (
|
||
"SELECT id FROM comments;"
|
||
"SELECT id, name, email FROM users;"
|
||
"SELECT id, key FROM pages;"
|
||
"SELECT (SELECT IFNULL(MAX(id),0) FROM comments) mc,"
|
||
" (SELECT IFNULL(MAX(id),0) FROM users) mu,"
|
||
" (SELECT IFNULL(MAX(id),0) FROM pages) mp;"
|
||
)
|
||
out = wrangler(["d1", "execute", db, "--remote", "-y", "--command", sql, "--json"])
|
||
j = parse_json_out(out)
|
||
comment_ids = {int(r["id"]) for r in j[0]["results"]}
|
||
users = {(r["name"], r["email"]): int(r["id"]) for r in j[1]["results"]}
|
||
pages = {r["key"]: int(r["id"]) for r in j[2]["results"]}
|
||
mx = j[3]["results"][0]
|
||
return {
|
||
"comment_ids": comment_ids,
|
||
"users": users,
|
||
"pages": pages,
|
||
"max_comment": int(mx["mc"]),
|
||
"max_user": int(mx["mu"]),
|
||
"max_page": int(mx["mp"]),
|
||
}
|
||
|
||
|
||
# ----------------------------------------------------------------- 主流程
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--src", required=True)
|
||
ap.add_argument("--site", default=DEFAULT_SITE, help="逗号分隔；all = 全导")
|
||
ap.add_argument("--site-url", default=DEFAULT_SITE_URL)
|
||
ap.add_argument("--out", default="tools/delta-out")
|
||
ap.add_argument("--db", default="artalk-cf")
|
||
ap.add_argument("--apply", action="store_true", help="生成后直接执行到远端")
|
||
args = ap.parse_args()
|
||
|
||
src = pathlib.Path(args.src)
|
||
if not src.exists():
|
||
print(f"找不到源文件: {src}", file=sys.stderr)
|
||
return 1
|
||
|
||
data = json.loads(src.read_text(encoding="utf-8"))
|
||
want = (
|
||
None
|
||
if args.site.strip().lower() == "all"
|
||
else [s.strip() for s in args.site.split(",") if s.strip()]
|
||
)
|
||
items = [c for c in data if want is None or c.get("site_name") in want]
|
||
site_name = (want[0] if want else DEFAULT_SITE) or DEFAULT_SITE
|
||
print(f"源文件 {len(data)} 条 → 站点「{site_name}」{len(items)} 条")
|
||
|
||
print("拉取远端现状 …")
|
||
st = fetch_state(args.db)
|
||
print(
|
||
f" 现状：comments {len(st['comment_ids'])}、users {len(st['users'])}、"
|
||
f"pages {len(st['pages'])}"
|
||
)
|
||
|
||
# ---------- 用户差集
|
||
# 按 (昵称,邮箱) 认人；但如果邮箱在库里已存在（昵称变了的情况，比如"小赵同学"→"小赵"），
|
||
# 复用老用户，避免同一个博主被拆成两个账号（新的那条也就不会继承 is_admin / 徽章）。
|
||
user_id = dict(st["users"]) # (name,email) -> id
|
||
by_email = {}
|
||
for (n, e), i in st["users"].items():
|
||
by_email.setdefault(e, i)
|
||
|
||
next_uid = st["max_user"] + 1
|
||
new_users = OrderedDict()
|
||
for c in items:
|
||
nick = (c.get("nick") or "").strip()
|
||
email = (c.get("email") or "").strip()
|
||
if not nick or not email:
|
||
continue
|
||
k = (nick, email)
|
||
if k in user_id:
|
||
continue
|
||
if email in by_email:
|
||
user_id[k] = by_email[email] # 同邮箱复用老 id
|
||
continue
|
||
if k in new_users:
|
||
continue
|
||
new_users[k] = {
|
||
"id": next_uid,
|
||
"nick": nick,
|
||
"email": email,
|
||
"link": c.get("link") or "",
|
||
"badge_name": c.get("badge_name") or "",
|
||
"badge_color": c.get("badge_color") or "",
|
||
"ts": parse_dt(c.get("created_at")),
|
||
}
|
||
user_id[k] = next_uid
|
||
by_email[email] = next_uid
|
||
next_uid += 1
|
||
|
||
# ---------- 页面差集
|
||
page_id = dict(st["pages"])
|
||
next_pid = st["max_page"] + 1
|
||
new_pages = OrderedDict()
|
||
for c in items:
|
||
key = (c.get("page_key") or "").strip()
|
||
if not key or key in page_id:
|
||
continue
|
||
new_pages[key] = {
|
||
"id": next_pid,
|
||
"title": c.get("page_title") or "",
|
||
"admin_only": c.get("page_admin_only"),
|
||
"ts": parse_dt(c.get("created_at")),
|
||
}
|
||
page_id[key] = next_pid
|
||
next_pid += 1
|
||
|
||
# ---------- 评论差集（保留原始 id，rid 关系才不用重算）
|
||
rows = []
|
||
rid_of = {}
|
||
have = st["comment_ids"]
|
||
dup = 0
|
||
for c in items:
|
||
try:
|
||
cid = int(str(c.get("id")))
|
||
except Exception:
|
||
continue
|
||
try:
|
||
rid = int(str(c.get("rid") or 0))
|
||
except Exception:
|
||
rid = 0
|
||
rid_of[cid] = rid
|
||
pk = (c.get("page_key") or "").strip()
|
||
if not pk or pk not in page_id:
|
||
continue
|
||
uk = ((c.get("nick") or "").strip(), (c.get("email") or "").strip())
|
||
if uk not in user_id:
|
||
continue
|
||
if cid in have:
|
||
dup += 1
|
||
continue
|
||
have.add(cid)
|
||
rows.append(
|
||
{
|
||
"id": cid,
|
||
"rid": rid,
|
||
"uid": user_id[uk],
|
||
"pid": page_id[pk],
|
||
"page_key": pk,
|
||
"content": c.get("content") or "",
|
||
"is_collapsed": c.get("is_collapsed"),
|
||
"is_pending": c.get("is_pending"),
|
||
"is_pinned": c.get("is_pinned"),
|
||
"vote_up": c.get("vote_up"),
|
||
"vote_down": c.get("vote_down"),
|
||
"ua": c.get("ua") or "",
|
||
"ip": c.get("ip") or "",
|
||
"created_at": parse_dt(c.get("created_at")),
|
||
"updated_at": parse_dt(c.get("updated_at")),
|
||
}
|
||
)
|
||
|
||
# root_id：沿 rid 链上溯。rid_of 来自**全量导出**，所以父评论在库里也能算对
|
||
root_cache = {}
|
||
|
||
def root_of(cid: int) -> int:
|
||
if cid in root_cache:
|
||
return root_cache[cid]
|
||
chain = []
|
||
cur = cid
|
||
seen = set()
|
||
while cur and cur not in root_cache and cur not in seen:
|
||
seen.add(cur)
|
||
parent = rid_of.get(cur, 0)
|
||
if parent == 0:
|
||
break # cur 就是顶层评论
|
||
chain.append(cur)
|
||
cur = parent
|
||
root = root_cache.get(cur, cur)
|
||
for n in chain:
|
||
root_cache[n] = root
|
||
root_cache[root] = root
|
||
return root
|
||
|
||
print(
|
||
f"\n差集：新用户 {len(new_users)}、新页面 {len(new_pages)}、新评论 {len(rows)}"
|
||
f"（导出里已有 {dup} 条跳过）"
|
||
)
|
||
if not rows and not new_users and not new_pages:
|
||
print("库里已经是最新，无需导入。")
|
||
return 0
|
||
|
||
out_dir = ROOT / args.out
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
for old in out_dir.glob("*.sql"):
|
||
old.unlink()
|
||
|
||
files = []
|
||
|
||
def dump(name: str, stmts: list):
|
||
for i in range(0, len(stmts), CHUNK):
|
||
fn = f"{name}-{i // CHUNK + 1:03d}.sql"
|
||
(out_dir / fn).write_text(
|
||
"\n".join(stmts[i : i + CHUNK]) + "\n", encoding="utf-8", newline="\n"
|
||
)
|
||
files.append(fn)
|
||
|
||
if new_users:
|
||
us = ["-- 新增用户"]
|
||
for u in new_users.values():
|
||
us.append(
|
||
"INSERT INTO users (id, name, email, link, badge_name, badge_color, receive_email, created_at, updated_at) "
|
||
f"VALUES ({u['id']}, {sql_str(u['nick'])}, {sql_str(u['email'])}, {sql_str(u['link'])}, "
|
||
f"{sql_str(u['badge_name'])}, {sql_str(u['badge_color'])}, 1, {u['ts']}, {u['ts']});"
|
||
)
|
||
dump("01-users", us)
|
||
|
||
if new_pages:
|
||
ps = ["-- 新增页面"]
|
||
for key, p in new_pages.items():
|
||
ps.append(
|
||
"INSERT INTO pages (id, key, site_name, title, admin_only, created_at, updated_at) "
|
||
f"VALUES ({p['id']}, {sql_str(key)}, {sql_str(site_name)}, "
|
||
f"{sql_str(p['title'])}, {sql_bool(p['admin_only'])}, {p['ts']}, {p['ts']});"
|
||
)
|
||
dump("02-pages", ps)
|
||
|
||
if rows:
|
||
ins = ["-- 新增评论（先全部按顶层插入，再回填 rid / root_id）"]
|
||
fix = []
|
||
for r in rows:
|
||
ins.append(
|
||
"INSERT INTO comments (id, content, page_key, site_name, user_id, rid, root_id, "
|
||
"is_collapsed, is_pending, is_pinned, is_verified, vote_up, vote_down, ua, ip, created_at, updated_at) "
|
||
f"VALUES ({r['id']}, {sql_str(r['content'])}, {sql_str(r['page_key'])}, {sql_str(site_name)}, "
|
||
f"{r['uid']}, 0, 0, {sql_bool(r['is_collapsed'])}, {sql_bool(r['is_pending'])}, {sql_bool(r['is_pinned'])}, "
|
||
f"0, {sql_int(r['vote_up'])}, {sql_int(r['vote_down'])}, {sql_str(r['ua'])}, {sql_str(r['ip'])}, "
|
||
f"{r['created_at']}, {r['updated_at'] or r['created_at']});"
|
||
)
|
||
if r["rid"]:
|
||
fix.append(
|
||
f"UPDATE comments SET rid = {r['rid']}, root_id = {root_of(r['rid'])} WHERE id = {r['id']};"
|
||
)
|
||
dump("03-comments", ins)
|
||
if fix:
|
||
dump("04-rid-fix", fix)
|
||
|
||
check = [
|
||
"-- 核对",
|
||
"SELECT (SELECT COUNT(*) FROM comments) AS comments,"
|
||
" (SELECT COUNT(*) FROM comments WHERE rid != 0) AS replies,"
|
||
" (SELECT COUNT(*) FROM users) AS users,"
|
||
" (SELECT COUNT(*) FROM pages) AS pages;",
|
||
]
|
||
(out_dir / "99-check.sql").write_text("\n".join(check) + "\n", encoding="utf-8", newline="\n")
|
||
files.append("99-check.sql")
|
||
|
||
print(f"\n生成 {len(files)} 个 SQL 于 {out_dir}")
|
||
if not args.apply:
|
||
print(f"执行： bash {args.out}/run.sh --remote （或加 --apply 自动跑）")
|
||
return 0
|
||
|
||
env = dict(os.environ)
|
||
for f in files:
|
||
print(f"==> {f}")
|
||
ok = False
|
||
for attempt in range(3):
|
||
p = subprocess.run(
|
||
["node", str(WRANGLER), "d1", "execute", args.db, "--remote", "-y", f"--file={out_dir.as_posix()}/{f}"],
|
||
cwd=ROOT,
|
||
capture_output=True,
|
||
text=True,
|
||
env=env,
|
||
)
|
||
if p.returncode == 0:
|
||
ok = True
|
||
break
|
||
print(f" 第 {attempt + 1} 次失败，重试（跨境 fetch 偶发）", file=sys.stderr)
|
||
time.sleep(3)
|
||
if not ok:
|
||
print(f" FAIL {f}\n{(p.stdout or '')[:300]}\n{(p.stderr or '')[:300]}", file=sys.stderr)
|
||
return 1
|
||
print(" ok")
|
||
|
||
out = wrangler(
|
||
[
|
||
"d1",
|
||
"execute",
|
||
args.db,
|
||
"--remote",
|
||
"-y",
|
||
"--command",
|
||
"SELECT (SELECT COUNT(*) FROM comments) AS comments,"
|
||
" (SELECT COUNT(*) FROM comments WHERE rid != 0) AS replies,"
|
||
" (SELECT COUNT(*) FROM users) AS users,"
|
||
" (SELECT COUNT(*) FROM pages) AS pages;",
|
||
"--json",
|
||
]
|
||
)
|
||
print("\n导入后：", json.dumps(parse_json_out(out)[0]["results"][0], ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|