superidou
5759c1862e
M0-M7 已完成:核心网关(身份/RBAC/TOTP/OIDC/SAML/Provider/配额/路由/内容策略/审计/定价)+ 资源市场(MCP/Skills/数字员工)。 含 22 个 PostgreSQL 迁移、管理端/门户端前端源码、OpenAPI 契约、部署 compose。 Co-Authored-By: Claude <noreply@anthropic.com>
59 lines
2.5 KiB
Python
59 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Export a legacy SQLite database as deterministic JSON Lines for staging.
|
|
|
|
The default omits credential/ciphertext columns. --include-encrypted retains old
|
|
ciphertext only for a controlled offline transform; it never decrypts or logs it.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import datetime as dt
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
SENSITIVE_MARKERS = ("password", "secret", "api_key", "raw_key", "totp", "backup_codes", "headers_template")
|
|
|
|
|
|
def encode(value):
|
|
if isinstance(value, bytes):
|
|
return {"$binary_base64": base64.b64encode(value).decode("ascii")}
|
|
if isinstance(value, (dt.date, dt.datetime)):
|
|
return value.isoformat()
|
|
return value
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("database", type=Path)
|
|
parser.add_argument("--source-system", default="python-gateway")
|
|
parser.add_argument("--include-encrypted", action="store_true")
|
|
args = parser.parse_args()
|
|
if not args.database.is_file():
|
|
parser.error("SQLite database does not exist")
|
|
connection = sqlite3.connect(f"file:{args.database}?mode=ro", uri=True)
|
|
connection.row_factory = sqlite3.Row
|
|
tables = [row[0] for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")]
|
|
for table in tables:
|
|
columns = [row[1] for row in connection.execute(f'PRAGMA table_info("{table}")')]
|
|
id_column = "id" if "id" in columns else None
|
|
if id_column is None:
|
|
print(f"skip table without id: {table}", file=sys.stderr)
|
|
continue
|
|
selected = [column for column in columns if args.include_encrypted or not any(marker in column.lower() for marker in SENSITIVE_MARKERS)]
|
|
order = f' ORDER BY "{id_column}"'
|
|
query = "SELECT " + ",".join(f'"{column}"' for column in selected) + f' FROM "{table}"' + order
|
|
for row in connection.execute(query):
|
|
data = {column: encode(row[column]) for column in selected}
|
|
record = {"source_system": args.source_system, "entity_type": table, "legacy_id": str(row[id_column]), "data": data}
|
|
canonical = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
record["checksum"] = hashlib.sha256(canonical.encode()).hexdigest()
|
|
print(json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":")))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|