83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
将 MongoDB 导出的文档转为 MySQL `wz_documents` 的 INSERT SQL（不依赖第三方库）。
|
||
|
||
步骤：
|
||
1) mongoexport（与 old 集合名一致）：
|
||
mongoexport --uri="$MONGO_URI" --collection=streamers --jsonArray -o streamers.json
|
||
2) 生成 SQL：
|
||
python mongo_to_wz_documents.py --coll streamers --file streamers.json --out streamers.sql
|
||
3) 导入：
|
||
mysql -h ... -u ... -p dbname < streamers.sql
|
||
|
||
校验：SELECT coll, COUNT(*) FROM wz_documents GROUP BY coll; 与 Mongo count 对照。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
|
||
|
||
def oid_of(doc: dict) -> str:
|
||
_id = doc.get("_id")
|
||
if isinstance(_id, dict) and "$oid" in _id:
|
||
return str(_id["$oid"])
|
||
if isinstance(_id, str):
|
||
return _id
|
||
raise ValueError("无法解析 _id")
|
||
|
||
|
||
def load_docs(path: str) -> list[dict]:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
raw = f.read().strip()
|
||
if not raw:
|
||
return []
|
||
if raw[0] == "[":
|
||
data = json.loads(raw)
|
||
if not isinstance(data, list):
|
||
raise ValueError("JSON 根须为数组")
|
||
return data
|
||
return [json.loads(line) for line in raw.splitlines() if line.strip()]
|
||
|
||
|
||
def sql_escape(s: str) -> str:
|
||
return s.replace("\\", "\\\\").replace("'", "''")
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--coll", required=True)
|
||
ap.add_argument("--file", required=True)
|
||
ap.add_argument("--out", required=True)
|
||
args = ap.parse_args()
|
||
|
||
docs = load_docs(args.file)
|
||
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
lines = [
|
||
"-- generated by mongo_to_wz_documents.py",
|
||
"SET NAMES utf8mb4;",
|
||
"START TRANSACTION;",
|
||
]
|
||
for d in docs:
|
||
eid = oid_of(d)
|
||
body = json.dumps(d, ensure_ascii=False, separators=(",", ":"))
|
||
lines.append(
|
||
"INSERT INTO `wz_documents` (`coll`,`ext_id`,`body`,`created_at`,`updated_at`) VALUES ("
|
||
f"'{sql_escape(args.coll)}','{sql_escape(eid)}','{sql_escape(body)}','{now}','{now}')"
|
||
" ON DUPLICATE KEY UPDATE `body`=VALUES(`body`), `updated_at`=VALUES(`updated_at`);"
|
||
)
|
||
lines.append("COMMIT;")
|
||
|
||
with open(args.out, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(lines) + "\n")
|
||
|
||
print(len(docs), "rows ->", args.out, file=sys.stderr)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|