{"id":1670,"date":"2026-01-10T14:58:27","date_gmt":"2026-01-10T06:58:27","guid":{"rendered":"https:\/\/notes.coremix.net\/?p=1670"},"modified":"2026-01-10T21:55:55","modified_gmt":"2026-01-10T13:55:55","slug":"python%e7%a1%ae%e6%9d%83%e5%b7%a5%e5%85%b7%ef%bc%88%e5%b8%a6%e8%8d%89%e7%a8%bf%e7%a1%ae%e6%9d%83%ef%bc%89","status":"publish","type":"post","link":"https:\/\/notes.coremix.net\/?p=1670","title":{"rendered":"python\u786e\u6743\u5de5\u5177\uff08\u5e26\u8349\u7a3f\u786e\u6743\uff09"},"content":{"rendered":"

 <\/p>\n

\u7a81\u7136\u60f3\u8d77\u6765\u4e00\u4e2a\u786e\u6743\u7684\u4e8b\u60c5\uff0c\u4e8e\u662f\u5199\u4e86\u4e00\u4e2a\u533a\u5757\u94fe\u786e\u6743\u5de5\u5177\uff08wordpress\u4e0a\u6709\u63d2\u4ef6\u7684\uff0c\u4f46\u662f\u542f\u7528\u624d500\uff09\uff0c\u56e0\u6b64\u81ea\u5df1\u5199\u4e86\u4e00\u4e2a\uff0c\u539f\u7406\u662f\uff1a<\/p>\n

    \n
  1. \u6bcf\u6b21\u68c0\u6d4b\u6587\u7ae0\u6570\u636e\u5e93\u8868\uff08wp_posts\uff09\u5e76\u83b7\u53d6\u6700\u65b048\u5c0f\u65f6\u7684\u8349\u7a3f\u3001\u6587\u7ae0\u5185\u5bb9\u5230json\u548csql\u6587\u4ef6<\/li>\n
  2. \u901a\u8fc7python\u8ba1\u7b97sha256\u548cmd5.<\/li>\n
  3. \u6700\u540e\u5199\u5165\u6587\u4ef6\u5e76\u52a0\u4e0a\u7248\u6743\u4fe1\u606f\uff08\u57df\u540d\u3001RSA\uff09<\/li>\n
  4. \u6700\u540e\u4f7f\u7528ots stamp filename \u83b7\u53d6\u6bd4\u7279\u5e01\u7684\u533a\u5757\u8ba4\u8bc1\uff08\u53ef\u5ba1\u8ba1\uff09<\/li>\n<\/ol>\n

    \u5982\u679c\u62c5\u5fc3\u6709\u5b9e\u65f6\u722c\u866b\uff08\u9664\u975e\u540d\u58f0\u5f88\u5927\uff0c\u672c\u7ad9\u534a\u5e74\u90fd\u6ca1\u4eba\u770b\u5b8c\u5168\u4e0d\u62c5\u5fc3\uff09\u53ef\u4ee5\u4fdd\u5b58\u5230\u8349\u7a3f\u5ef6\u8fdf\u4e00\u5929\u53d1\u5e03\u3002<\/p>\n

    \u540e\u671f\u53ef\u4ee5\u6dfb\u52a0\u9644\u4ef6\uff08\u56fe\u7247\u76ee\u5f55\u548c\u9644\u4ef6\u76ee\u5f55\uff09\u7684\u54c8\u5e0c\u4ee5\u53ca\u5199\u5165\u6700\u7ec8ots\u4e0a\u53bb\u3002<\/p>\n

    \u4e0b\u9762\u662fpython\u4ee3\u7801\u548cjson\u6587\u4ef6\uff1a<\/p>\n

    \u00a0\r\n#!\/usr\/bin\/env python3\r\n# -*- coding: utf-8 -*-\r\n\r\nimport argparse\r\nimport datetime as dt\r\nimport hashlib\r\nimport json\r\nimport os\r\nimport subprocess\r\nimport sys\r\nfrom pathlib import Path\r\nfrom typing import Any, Dict, List, Tuple\r\n\r\nimport pymysql\r\n\r\n\r\nCOPYRIGHT_BLOCK = """\\\r\n\u00a9 Xiao. All rights reserved.\r\nRSA Fingerprint: 35023C1E8DEB09435105647FFE38FE1D457E8CB0\r\nSource: coremix.net\r\n"""\r\n\r\n\r\ndef utcnow() -> dt.datetime:\r\n    return dt.datetime.now(dt.timezone.utc)\r\n\r\n\r\ndef ensure_dir(p: Path) -> None:\r\n    p.mkdir(parents=True, exist_ok=True)\r\n\r\n\r\ndef load_json(p: Path) -> Dict[str, Any]:\r\n    with p.open("r", encoding="utf-8") as f:\r\n        return json.load(f)\r\n\r\n\r\ndef save_text(p: Path, s: str) -> None:\r\n    with p.open("w", encoding="utf-8") as f:\r\n        f.write(s)\r\n\r\n\r\ndef save_json(p: Path, obj: Any) -> None:\r\n    with p.open("w", encoding="utf-8") as f:\r\n        json.dump(obj, f, ensure_ascii=False, indent=2, default=str)\r\n\r\n\r\ndef file_hashes(p: Path) -> Tuple[str, str]:\r\n    md5 = hashlib.md5()\r\n    sha = hashlib.sha256()\r\n    with p.open("rb") as f:\r\n        for chunk in iter(lambda: f.read(1024 * 1024), b""):\r\n            md5.update(chunk)\r\n            sha.update(chunk)\r\n    return md5.hexdigest(), sha.hexdigest()\r\n\r\n\r\ndef mysql_connect(mysql_cfg: Dict[str, Any], db: str):\r\n    return pymysql.connect(\r\n        host=mysql_cfg["host"],\r\n        port=int(mysql_cfg.get("port", 3306)),\r\n        user=mysql_cfg["user"],\r\n        password=mysql_cfg["password"],\r\n        database=db,\r\n        charset=mysql_cfg.get("charset", "utf8mb4"),\r\n        cursorclass=pymysql.cursors.DictCursor,\r\n        autocommit=True,\r\n    )\r\n\r\n\r\ndef detect_first_run(state_path: Path) -> bool:\r\n    return not state_path.exists()\r\n\r\n\r\ndef load_state(state_path: Path) -> Dict[str, Any]:\r\n    if not state_path.exists():\r\n        return {}\r\n    return load_json(state_path)\r\n\r\n\r\ndef save_state(state_path: Path, state: Dict[str, Any]) -> None:\r\n    save_json(state_path, state)\r\n\r\n\r\ndef sql_escape(val: Any, conn) -> str:\r\n    # Use pymysql escape where possible\r\n    if val is None:\r\n        return "NULL"\r\n    if isinstance(val, (int, float)):\r\n        return str(val)\r\n    if isinstance(val, (dt.datetime, dt.date)):\r\n        return "'" + str(val) + "'"\r\n    s = str(val)\r\n    return "'" + conn.escape_string(s) + "'"\r\n\r\n\r\ndef rows_to_insert_sql(conn, table: str, rows: List[Dict[str, Any]]) -> str:\r\n    if not rows:\r\n        return f"-- No rows for {table}\\n"\r\n    cols = list(rows[0].keys())\r\n    col_list = ", ".join([f"`{c}`" for c in cols])\r\n    lines = [f"-- Exported from {table}", "SET NAMES utf8mb4;", ""]\r\n    for r in rows:\r\n        values = ", ".join([sql_escape(r.get(c), conn) for c in cols])\r\n        lines.append(f"INSERT INTO `{table}` ({col_list}) VALUES ({values});")\r\n    lines.append("")\r\n    return "\\n".join(lines)\r\n\r\n\r\ndef fetch_posts_and_meta(conn, prefix: str, full: bool, hours: int) -> Dict[str, Any]:\r\n    posts_table = f"{prefix}posts"\r\n    postmeta_table = f"{prefix}postmeta"\r\n\r\n    with conn.cursor() as cur:\r\n        if full:\r\n            # \u5168\u91cf\uff1apost\/page\/revision\uff08\u4f60\u4e5f\u53ef\u4ee5\u6309\u9700\u52a0 attachment\uff09\r\n            cur.execute(\r\n                f"""\r\n                SELECT *\r\n                FROM `{posts_table}`\r\n                WHERE post_type IN ('post','page','revision')\r\n                """\r\n            )\r\n        else:\r\n            cutoff = utcnow() - dt.timedelta(hours=hours)\r\n            # \u4f7f\u7528 post_modified_gmt \u66f4\u53ef\u9760\uff08\u8de8\u65f6\u533a\/\u8de8\u673a\u5668\uff09\r\n            cur.execute(\r\n                f"""\r\n                SELECT *\r\n                FROM `{posts_table}`\r\n                WHERE post_type IN ('post','page','revision')\r\n                  AND (\r\n                    post_modified_gmt >= %s\r\n                    OR post_date_gmt >= %s\r\n                  )\r\n                """,\r\n                (cutoff.strftime("%Y-%m-%d %H:%M:%S"), cutoff.strftime("%Y-%m-%d %H:%M:%S")),\r\n            )\r\n        posts = cur.fetchall()\r\n\r\n        post_ids = [p["ID"] for p in posts if "ID" in p]\r\n        meta: List[Dict[str, Any]] = []\r\n        if post_ids:\r\n            # \u6279\u91cf\u62c9\u53d6\u76f8\u5173 postmeta\uff08\u53ea\u53d6\u8fd9\u4e9b\u6587\u7ae0\/\u4fee\u8ba2\u7684\u5143\u6570\u636e\uff09\r\n            # IN \u5207\u7247\u907f\u514d\u8fc7\u957f\r\n            chunk = 1000\r\n            for i in range(0, len(post_ids), chunk):\r\n                part = post_ids[i : i + chunk]\r\n                placeholders = ",".join(["%s"] * len(part))\r\n                cur.execute(\r\n                    f"""\r\n                    SELECT *\r\n                    FROM `{postmeta_table}`\r\n                    WHERE post_id IN ({placeholders})\r\n                    """,\r\n                    part,\r\n                )\r\n                meta.extend(cur.fetchall())\r\n\r\n    return {\r\n        "posts_table": posts_table,\r\n        "postmeta_table": postmeta_table,\r\n        "posts": posts,\r\n        "postmeta": meta,\r\n        "exported_at_utc": utcnow().isoformat(),\r\n        "mode": "full" if full else f"last_{hours}_hours",\r\n    }\r\n\r\n\r\ndef run_ots_stamp(manifest_path: Path) -> Tuple[bool, str]:\r\n    try:\r\n        r = subprocess.run(\r\n            ["\/root\/.local\/bin\/ots", "stamp", str(manifest_path)],\r\n            check=False,\r\n            capture_output=True,\r\n            text=True,\r\n        )\r\n        ok = (r.returncode == 0)\r\n        out = (r.stdout or "") + (r.stderr or "")\r\n        return ok, out.strip()\r\n    except FileNotFoundError:\r\n        return False, "ots command not found (install OpenTimestamps client and ensure 'ots' in PATH)."\r\n\r\n\r\ndef main():\r\n    ap = argparse.ArgumentParser(description="Export WP posts + revisions + meta for OTS proof, hash, and stamp.")\r\n    ap.add_argument("--config", required=True, help="Path to sites.json")\r\n    ap.add_argument("--hours", type=int, default=48, help="Incremental window in hours (default: 48)")\r\n    ap.add_argument("--state", default=".wp_ots_state.json", help="State file path (default: .wp_ots_state.json)")\r\n    ap.add_argument("--no-ots", action="store_true", help="Do not run 'ots stamp' automatically")\r\n    args = ap.parse_args()\r\n\r\n    cfg_path = Path(args.config).expanduser().resolve()\r\n    cfg = load_json(cfg_path)\r\n\r\n    out_dir = Path(cfg["export"]["output_dir"]).expanduser()\r\n    ensure_dir(out_dir)\r\n\r\n    state_path = Path(args.state).expanduser().resolve()\r\n    state = load_state(state_path)\r\n    first_run = detect_first_run(state_path) or state.get("initialized") is not True\r\n\r\n    run_id = utcnow().strftime("%Y%m%dT%H%M%SZ")\r\n    run_dir = out_dir \/ run_id\r\n    ensure_dir(run_dir)\r\n\r\n    manifest_lines: List[str] = []\r\n    manifest_lines.append(f"# WordPress Content Proof Manifest")\r\n    manifest_lines.append(f"- Run UTC: {utcnow().isoformat()}")\r\n    manifest_lines.append(f"- Mode: {'FULL (first run)' if first_run else f'INCREMENTAL last {args.hours} hours'}")\r\n    manifest_lines.append("")\r\n\r\n    all_files: List[Path] = []\r\n\r\n    mysql_cfg = cfg["mysql"]\r\n\r\n    for site in cfg["sites"]:\r\n        name = site["name"]\r\n        db = site["db"]\r\n        prefix = site.get("prefix", "wp_")\r\n        site_url = site.get("site_url", "")\r\n\r\n        site_dir = run_dir \/ name\r\n        ensure_dir(site_dir)\r\n\r\n        try:\r\n            conn = mysql_connect(mysql_cfg, db)\r\n        except Exception as e:\r\n            manifest_lines.append(f"## {name}")\r\n            manifest_lines.append(f"- DB: {db}")\r\n            manifest_lines.append(f"- ERROR: failed to connect: {e}")\r\n            manifest_lines.append("")\r\n            continue\r\n\r\n        try:\r\n            data = fetch_posts_and_meta(conn, prefix=prefix, full=first_run, hours=args.hours)\r\n\r\n            # JSON export (\u6700\u9002\u5408\u786e\u6743\u5185\u5bb9\u6838\u9a8c\uff1a\u6b63\u6587\u3001\u4fee\u8ba2\u3001meta \u539f\u6837\u4fdd\u7559)\r\n            json_path = site_dir \/ f"{name}_{data['mode']}_{run_id}.json"\r\n            payload = {\r\n                "site": {\r\n                    "name": name,\r\n                    "db": db,\r\n                    "prefix": prefix,\r\n                    "site_url": site_url,\r\n                },\r\n                "export": {\r\n                    "exported_at_utc": data["exported_at_utc"],\r\n                    "mode": data["mode"],\r\n                    "hours": None if first_run else args.hours,\r\n                },\r\n                "tables": {\r\n                    data["posts_table"]: data["posts"],\r\n                    data["postmeta_table"]: data["postmeta"],\r\n                },\r\n            }\r\n            save_json(json_path, payload)\r\n\r\n            # SQL export (\u53ef\u7528\u4e8e\u5feb\u901f\u56de\u704c\/\u5ba1\u8ba1\uff1b\u6ce8\u610f\uff1a\u8fd9\u662f INSERT\uff0c\u4e0d\u542b DROP\/CREATE)\r\n            sql_path = site_dir \/ f"{name}_{data['mode']}_{run_id}.sql"\r\n            sql_parts = []\r\n            sql_parts.append(f"-- Site: {name}  DB: {db}  Prefix: {prefix}")\r\n            sql_parts.append(f"-- Exported at UTC: {data['exported_at_utc']}")\r\n            sql_parts.append("")\r\n            sql_parts.append(rows_to_insert_sql(conn, data["posts_table"], data["posts"]))\r\n            sql_parts.append(rows_to_insert_sql(conn, data["postmeta_table"], data["postmeta"]))\r\n            save_text(sql_path, "\\n".join(sql_parts))\r\n\r\n            all_files.extend([json_path, sql_path])\r\n\r\n            # Manifest section\r\n            manifest_lines.append(f"## {name}")\r\n            manifest_lines.append(f"- DB: `{db}`  Prefix: `{prefix}`  Site: `{site_url}`")\r\n            manifest_lines.append(f"- Posts exported: `{len(data['posts'])}`")\r\n            manifest_lines.append(f"- Postmeta exported: `{len(data['postmeta'])}`")\r\n            manifest_lines.append("")\r\n        finally:\r\n            try:\r\n                conn.close()\r\n            except Exception:\r\n                pass\r\n\r\n    # Hash all export files\r\n    manifest_lines.append("# Hashes")\r\n    manifest_lines.append("")\r\n    manifest_lines.append("| File | MD5 | SHA256 |")\r\n    manifest_lines.append("|---|---|---|")\r\n\r\n    for p in sorted(all_files):\r\n        md5h, sha = file_hashes(p)\r\n        rel = p.relative_to(run_dir)\r\n        manifest_lines.append(f"| `{rel}` | `{md5h}` | `{sha}` |")\r\n\r\n    manifest_lines.append("")\r\n    manifest_lines.append("---")\r\n    manifest_lines.append(COPYRIGHT_BLOCK.rstrip())\r\n    manifest_lines.append("")\r\n\r\n    manifest_name = cfg["export"].get("manifest_name", "manifest.md")\r\n    manifest_path = run_dir \/ manifest_name\r\n    save_text(manifest_path, "\\n".join(manifest_lines))\r\n\r\n    # update state\r\n    state["initialized"] = True\r\n    state["last_run_utc"] = utcnow().isoformat()\r\n    state["last_run_id"] = run_id\r\n    save_state(state_path, state)\r\n\r\n    print(f"[OK] Exports written to: {run_dir}")\r\n    print(f"[OK] Manifest: {manifest_path}")\r\n\r\n    if not args.no_ots:\r\n        ok, out = run_ots_stamp(manifest_path)\r\n        if ok:\r\n            print(f"[OK] ots stamp success: {manifest_path}.ots should be created")\r\n        else:\r\n            print(f"[WARN] ots stamp failed: {out}")\r\n\r\n    return 0\r\n\r\n\r\nif __name__ == "__main__":\r\n    sys.exit(main())\r\n# python3 \/data\/wp\/wp_ots_proof.py   --config \/data\/wp\/sites.json   --state \/data\/wp\/.wp_ots_state.json   --hours 48\r\n\r\n# python3 \/data\/wp\/wp_ots_proof.py \\\r\n#   --config \/data\/wp\/sites.json \\\r\n#   --state \/data\/wp\/.wp_ots_state.json \\\r\n#   --hours 48\r\n\r\n# crontab -e\r\n# \u6dfb\u52a0\u8fd9\u53e5\uff1a\r\n# 20 3 * * * \/usr\/bin\/python3 \/data\/wp\/wp_ots_proof.py --config \/data\/wp\/sites.json --state \/data\/wp\/.wp_ots_state.json --hours 48 >> \/data\/wp\/logs\/wp_ots_proof.log 2>&1\r\n# \u9a8c\u8bc1\uff1a\r\n# crontab -l\r\n# \u7b2c\u4e8c\u5929\u68c0\u67e5\uff1a\r\n# ls -lt \/data\/wp\/exports | head\r\n# tail -n 200 \/data\/wp\/logs\/wp_ots_proof.log\r\n <\/pre>\n
    json\u6587\u4ef6\uff1a<\/p>\n
     {\r\n  "mysql": {\r\n    "host": "127.0.0.1",\r\n    "port": 3306,\r\n    "user": "root",\r\n    "password": "xxxxxx",\r\n    "charset": "utf8mb4"\r\n  },\r\n  "sites": [\r\n    {\r\n      "name": "eait",\r\n      "db": "wpdb",\r\n      "prefix": "wp_",\r\n      "site_url": "https:\/\/notes.eait.co"\r\n    },\r\n   \r\n  ],\r\n  "export": {\r\n    "output_dir": "\/data\/wp",\r\n    "manifest_name": "manifest.md"\r\n  }\r\n}\r\n<\/pre>\n","protected":false},"excerpt":{"rendered":"
      \u7a81\u7136\u60f3\u8d77\u6765\u4e00\u4e2a\u786e\u6743\u7684\u4e8b\u60c5\uff0c\u4e8e\u662f\u5199\u4e86\u4e00\u4e2a\u533a\u5757\u94fe\u786e\u6743\u5de5\u5177\uff08wordpress\u4e0a\u6709\u63d2\u4ef6\u7684\uff0c\u4f46\u662f\u542f\u7528\u624d5 […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-1670","post","type-post","status-publish","format-standard","hentry","category-uncategorized"],"blocksy_meta":[],"_links":{"self":[{"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/posts\/1670","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=1670"}],"version-history":[{"count":5,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/posts\/1670\/revisions"}],"predecessor-version":[{"id":1675,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=\/wp\/v2\/posts\/1670\/revisions\/1675"}],"wp:attachment":[{"href":"https:\/\/notes.coremix.net\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=1670"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=1670"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/notes.coremix.net\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=1670"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}