如何通过Python脚本在两个Firestore项目间复制集合文档并保持ID一致

现有代码核心问题 #

• 写入时调用无参document()方法自动生成随机ID，未复用生产环境原有文档ID
• 文档ID和文档数据分开存储为两个独立列表，索引容易错位
• 硬编码指定单文档子集合，无法适配多文档、多层级子集合的复制需求
• 没有重复判断逻辑，每次运行都会重复写入数据

优化后完整代码 #

import firebase_admin
from firebase_admin import credentials, firestore

# 全局配置
export_collection_name = "Resources"
import_collection_name = "Resources"
# 单次批量操作上限，最高可设置为500
BATCH_SIZE = 50

export_db = None
import_db = None
all_docs = []
count = 0


def batch_data(iterable, n=1):
    l = len(iterable)
    for ndx in range(0, l, n):
        yield iterable[ndx:min(ndx + n, l)]


def init_export_db():
    global export_db
    # 替换为生产环境服务账号JSON路径
    cred = credentials.Certificate("./prod-service-account.json")
    source_app = firebase_admin.initialize_app(cred, name="source")
    export_db = firestore.client(app=source_app)


def init_import_db():
    global import_db
    # 替换为开发环境服务账号JSON路径
    cred = credentials.Certificate("./dev-service-account.json")
    destination_app = firebase_admin.initialize_app(cred, name="destination")
    import_db = firestore.client(app=destination_app)


def read_collection(coll_ref, parent_coll_path):
    """递归读取集合及所有子集合下的文档"""
    global all_docs, count
    for doc in coll_ref.stream():
        doc_data = doc.to_dict()
        # 存储文档的目标集合路径、ID、内容
        all_docs.append({
            "coll_path": parent_coll_path,
            "doc_id": doc.id,
            "data": doc_data
        })
        count += 1
        # 读取当前文档的所有子集合，递归处理
        for sub_coll in doc.collections():
            sub_coll_path = f"{parent_coll_path}/{doc.id}/{sub_coll.id}"
            read_collection(sub_coll, sub_coll_path)


def write_to_firestore():
    """批量写入数据，保持ID一致，跳过已存在文档"""
    global all_docs, import_db
    print(f"开始上传，共{len(all_docs)}条文档")
    for batched_docs in batch_data(all_docs, BATCH_SIZE):
        batch = import_db.batch()
        for doc_item in batched_docs:
            # 按照路径定位目标集合
            target_coll = import_db
            for path_segment in doc_item["coll_path"].split("/"):
                target_coll = target_coll.collection(path_segment)
            # 指定原有文档ID创建引用
            doc_ref = target_coll.document(doc_item["doc_id"])
            # 跳过已存在的文档，避免重复写入
            if doc_ref.get().exists:
                continue
            batch.set(doc_ref, doc_item["data"])
        batch.commit()
    print("开发环境数据库更新完成")


if __name__ == '__main__':
    init_export_db()
    # 从指定根集合开始递归读取所有数据
    root_coll = export_db.collection(export_collection_name)
    read_collection(root_coll, import_collection_name)
    init_import_db()
    write_to_firestore()
    print(f"操作完成，共处理{count}条文档")

功能说明 #

• 自动保持所有文档ID和生产环境完全一致
• 支持任意层级的子集合自动递归复制，不需要硬编码路径
• 自动跳过目标环境已存在的文档，重复运行脚本也不会重复复制数据
• 兼容Firestore批量操作限制，可根据需求调整BATCH_SIZE参数提升复制效率

内容的提问来源于stack exchange，提问作者Devansh Shukla