本文使用Dify v1.4.0版本，主要解析了commands.py中的clear_free_plan_tenant_expired_logs、clear_orphaned_file_records和remove_orphaned_files_on_storage等函数的执行逻辑。源码位置：dify\api\commands.py

一.clear_free_plan_tenant_expired_logs()函数

完整的执行命令示例，如下所示：

flask clear-free-plan-tenant-expired-logs --days 30 --batch 100 --tenant_ids 1 2 3

• --days：要清理的天数（如 30）

• --batch：每批处理的数量（如 100）

• --tenant_ids：要清理的租户 ID，可以指定多个（如 1 2 3）

该段代码定义了一个命令行工具命令 clear-free-plan-tenant-expired-logs，用于清理免费套餐租户过期的日志。支持自定义天数、批量大小和租户范围，适合运维人员在命令行下操作。

@click.command("clear-free-plan-tenant-expired-logs", help="Clear free plan tenant expired logs.")
@click.option("--days", prompt=True, help="The days to clear free plan tenant expired logs.", default=30)
@click.option("--batch", prompt=True, help="The batch size to clear free plan tenant expired logs.", default=100)
@click.option(
"--tenant_ids",
    prompt=True,
    multiple=True,
    help="The tenant ids to clear free plan tenant expired logs.",
)
defclear_free_plan_tenant_expired_logs(days: int, batch: int, tenant_ids: list[str]):
"""
    Clear free plan tenant expired logs.
    """
    click.echo(click.style("Starting clear free plan tenant expired logs.", fg="white"))

    ClearFreePlanTenantExpiredLogs.process(days, batch, tenant_ids)

    click.echo(click.style("Clear free plan tenant expired logs completed.", fg="green"))

1.命令定义

使用 @click.command 装饰器注册命令名为 clear-free-plan-tenant-expired-logs，并添加帮助说明。

2.参数定义

• --days：需要用户输入，表示要清理多少天前的日志，默认30天。

• --batch：需要用户输入，表示每批处理的日志数量，默认100。

• --tenant_ids：需要用户输入，可以输入多个租户ID，指定要清理哪些租户的日志。

3.函数实现

• 首先输出”开始清理免费套餐租户过期日志”的提示信息。

• 调用 ClearFreePlanTenantExpiredLogs.process(days, batch, tenant_ids) 方法，执行实际的清理逻辑。该方法会根据传入的天数、批量大小和租户ID列表，处理并清理对应的日志。

• 清理完成后，输出”清理完成”的提示信息。

二.clear_free_plan_tenant_expired_logs.py

该 process 方法用于清理免费套餐租户的过期日志。它支持并发处理多个租户，自动分批，动态调整批次大小。

执行流程：指定租户或自动分批处理所有租户；采用线程池并发处理，提升效率；动态调整批次大小，避免单批过大；只处理免费套餐租户，异常有日志记录；处理进度定期输出。

1.方法入口与初始化

打印开始信息，设置起止时间，初始化计数器和线程池。

@classmethod
defprocess(cls, days: int, batch: int, tenant_ids: list[str]):
"""
    Clear free plan tenant expired logs.
    """

    click.echo(click.style("Clearing free plan tenant expired logs", fg="white"))
    ended_at = datetime.datetime.now()
    started_at = datetime.datetime(2023, 4, 3, 8, 59, 24)
    current_time = started_at

with Session(db.engine) as session:
        total_tenant_count = session.query(Tenant.id).count()

    click.echo(click.style(f"Total tenant count: {total_tenant_count}", fg="white"))

    handled_tenant_count = 0

    thread_pool = ThreadPoolExecutor(max_workers=10)

2.定义租户处理函数

只处理免费（sandbox）租户，调用 process_tenant 进行清理，异常时记录日志，并统计已处理数量。

defprocess_tenant(flask_app: Flask, tenant_id: str) -> None:
try:
if (
not dify_config.BILLING_ENABLED
or BillingService.get_info(tenant_id)["subscription"]["plan"] == "sandbox"
            ):
# only process sandbox tenant
                cls.process_tenant(flask_app, tenant_id, days, batch)
except Exception:
            logger.exception(f"Failed to process tenant {tenant_id}")
finally:
nonlocal handled_tenant_count
            handled_tenant_count += 1
if handled_tenant_count % 100 == 0:
                click.echo(
                    click.style(
f"[{datetime.datetime.now()}] "
f"Processed {handled_tenant_count} tenants "
f"({(handled_tenant_count / total_tenant_count) * 100:.1f}%), "
f"{handled_tenant_count}/{total_tenant_count}",
                        fg="green",
                    )
                )

这段代码定义了一个内部函数 process_tenant，用于处理单个租户的过期日志清理。主要逻辑：

• 首先尝试（try）判断该租户是否为免费租户（sandbox），或者账单功能未启用（BILLING_ENABLED 为 False），只有满足条件才会调用 cls.process_tenant 进行清理操作。

• 如果处理过程中发生异常，会记录异常日志。

• 最后（finally），无论是否成功，都会将已处理租户计数加一，并且每处理 100 个租户会输出一次进度信息，包括当前时间、已处理租户数、总租户数及百分比，便于监控批量处理进度。

3.并发任务列表初始化

    futures = []

4.处理指定租户列表

如果传入了 tenant_ids，则直接并发处理这些租户。

if tenant_ids:
for tenant_id in tenant_ids:
            futures.append(
                thread_pool.submit(
                    process_tenant,
                    current_app._get_current_object(),  # type: ignore[attr-defined]
                    tenant_id,
                )
            )

5.动态分批处理所有租户

如果未指定租户，则按创建时间动态分批，每批目标约 100 个租户，自动调整时间区间，逐批并发处理。

else:
while current_time < ended_at:
            click.echo(
                click.style(f"Current time: {current_time}, Started at: {datetime.datetime.now()}", fg="white")
            )
# Initial interval of 1 day, will be dynamically adjusted based on tenant count
            interval = datetime.timedelta(days=1)
# Process tenants in this batch
with Session(db.engine) as session:
# Calculate tenant count in next batch with current interval
# Try different intervals until we find one with a reasonable tenant count
                test_intervals = [
                    datetime.timedelta(days=1),
                    datetime.timedelta(hours=12),
                    datetime.timedelta(hours=6),
                    datetime.timedelta(hours=3),
                    datetime.timedelta(hours=1),
                ]

for test_interval in test_intervals:
                    tenant_count = (
                        session.query(Tenant.id)
                        .filter(Tenant.created_at.between(current_time, current_time + test_interval))
                        .count()
                    )
if tenant_count <= 100:
                        interval = test_interval
break
else:
# If all intervals have too many tenants, use minimum interval
                    interval = datetime.timedelta(hours=1)

# Adjust interval to target ~100 tenants per batch
if tenant_count > 0:
# Scale interval based on ratio to target count
                    interval = min(
                        datetime.timedelta(days=1),  # Max 1 day
                        max(
                            datetime.timedelta(hours=1),  # Min 1 hour
                            interval * (100 / tenant_count),  # Scale to target 100
                        ),
                    )

                batch_end = min(current_time + interval, ended_at)

                rs = (
                    session.query(Tenant.id)
                    .filter(Tenant.created_at.between(current_time, batch_end))
                    .order_by(Tenant.created_at)
                )

                tenants = []
for row in rs:
                    tenant_id = str(row.id)
try:
                        tenants.append(tenant_id)
except Exception:
                        logger.exception(f"Failed to process tenant {tenant_id}")
continue

                    futures.append(
                        thread_pool.submit(
                            process_tenant,
                            current_app._get_current_object(),  # type: ignore[attr-defined]
                            tenant_id,
                        )
                    )

            current_time = batch_end

执行逻辑，如下所示：

• 外层 while current_time < ended_at: 循环，遍历时间区间，每次处理一段时间内新创建的租户。

• test_intervals 设定了多个时间间隔（1天、12小时、6小时、3小时、1小时），用于动态调整每批次的时间跨度。

• 通过循环 test_intervals，查询每个时间区间内的租户数量，找到第一个不超过 100 个租户的区间，作为本批次的处理区间。如果都超过 100，则用最小的 1 小时区间。

• 如果本批次租户数量大于 0，会根据实际数量动态缩放区间长度，使每批大致 100 个租户（但不超过 1 天，不小于 1 小时）。

• 查询本批次区间内的所有租户，依次将每个租户的处理任务提交到线程池（thread_pool.submit），实现并发处理。

• 每次循环结束后，current_time 前移到本批次区间的结束时间，进入下一个批次。

这样可以高效且均匀地处理大量租户，避免单批次数据量过大导致阻塞或资源浪费。

6.等待所有线程处理完成

# wait for all threads to finish
for future in futures:
        future.result()

注解：

1.with Session(db.engine) as session

使用 SQLAlchemy 的 Session 类，基于 db.engine（数据库引擎）创建一个新的数据库会话（session）对象，并通过 with 上下文管理器自动管理会话的生命周期。这样可以确保在代码块执行完毕后，session 会被正确关闭和释放资源，无论是否发生异常。在该代码块内，可以安全地执行数据库的查询、插入、更新、删除等操作。

2.cls.process_tenant(flask_app, tenant_id, days, batch)

cls 是指当前类 ClearFreePlanTenantExpiredLogs。在 Python 的类方法（用 @classmethod 装饰）中，第一个参数通常命名为 cls，代表类本身（类似于实例方法的 self 代表实例对象）。因此，这里的 cls.process_tenant 实际上等价于 ClearFreePlanTenantExpiredLogs.process_tenant，用于调用类方法。

3.future.result()

该代码的作用是阻塞主线程，直到对应的线程池任务执行完成，并返回其结果。如果该任务在执行过程中抛出异常，这里会重新抛出异常。这样可以确保所有通过 ThreadPoolExecutor 提交的任务都已完成后，主流程才会继续往下执行。常用于等待多线程并发任务全部结束。

三.clear_orphaned_file_records()函数

完整的执行命令示例，如下所示：

flask clear-orphaned-file-records --force

该函数用于 清理数据库中孤立的文件记录，即数据库中存在但未被其它表引用的文件记录。

1.用户确认与风险提示

首先，函数会向用户详细说明操作风险和影响，列出将要处理的表和字段，并要求用户确认是否继续。

# notify user and ask for confirmation
click.echo(
    click.style(
"This command will first find and delete orphaned file records from the message_files table,", fg="yellow"
    )
)
click.echo(
    click.style(
"and then it will find and delete orphaned file records in the following tables:",
        fg="yellow",
    )
)
for files_table in files_tables:
    click.echo(click.style(f"- {files_table['table']}", fg="yellow"))
click.echo(
    click.style("The following tables and columns will be scanned to find orphaned file records:", fg="yellow")
)
for ids_table in ids_tables:
    click.echo(click.style(f"- {ids_table['table']} ({ids_table['column']})", fg="yellow"))
click.echo("")
click.echo(click.style("!!! USE WITH CAUTION !!!", fg="red"))
click.echo(
    click.style(
        (
"Since not all patterns have been fully tested, "
"please note that this command may delete unintended file records."
        ),
        fg="yellow",
    )
)
click.echo(
    click.style("This cannot be undone. Please make sure to back up your database before proceeding.", fg="yellow")
)
click.echo(
    click.style(
        (
"It is also recommended to run this during the maintenance window, "
"as this may cause high load on your instance."
        ),
        fg="yellow",
    )
)
ifnot force:
    click.confirm("Do you want to proceed?", abort=True)

该代码定义了用于清理数据库中孤立文件记录的准备和用户确认流程：

• files_tables 和 ids_tables：分别定义了需要处理的文件表（如 upload_files、tool_files）及其主键/文件键字段，以及需要扫描以查找文件引用的其他表和字段（如 messages、documents 等），包括类型（uuid、text、json）。

• 通过 click.echo 向用户详细说明即将执行的操作，包括会删除哪些表中的孤立文件记录、会扫描哪些表和字段、操作具有不可逆性、建议备份数据库并在维护窗口执行等警告。

• 如果未加 --force 参数，会弹出确认提示，用户确认后才会继续。

• 最后输出”开始清理孤立文件记录”的提示，进入实际清理流程。

这部分代码的目的是在执行危险操作前，充分告知用户风险并获得确认。

2.查找并删除 message_files 表中孤立记录

该代码用于清理 message_files 表中孤立的文件记录（即 message_id 在 messages 表中不存在的记录），并在用户确认后删除。

# clean up the orphaned records in the message_files table where message_id doesn't exist in messages table
try:
    click.echo(
        click.style("- Listing message_files records where message_id doesn't exist in messages table", fg="white")
    )
    query = (
"SELECT mf.id, mf.message_id "
"FROM message_files mf LEFT JOIN messages m ON mf.message_id = m.id "
"WHERE m.id IS NULL"
    )
    orphaned_message_files = []
with db.engine.begin() as conn:
        rs = conn.execute(db.text(query))
for i in rs:
            orphaned_message_files.append({"id": str(i[0]), "message_id": str(i[1])})

if orphaned_message_files:
        click.echo(click.style(f"Found {len(orphaned_message_files)} orphaned message_files records:", fg="white"))
for record in orphaned_message_files:
            click.echo(click.style(f"  - id: {record['id']}, message_id: {record['message_id']}", fg="black"))

ifnot force:
            click.confirm(
                (
f"Do you want to proceed "
f"to delete all {len(orphaned_message_files)} orphaned message_files records?"
                ),
                abort=True,
            )

        click.echo(click.style("- Deleting orphaned message_files records", fg="white"))
        query = "DELETE FROM message_files WHERE id IN :ids"
with db.engine.begin() as conn:
            conn.execute(db.text(query), {"ids": tuple([record["id"] for record in orphaned_message_files])})
        click.echo(
            click.style(f"Removed {len(orphaned_message_files)} orphaned message_files records.", fg="green")
        )
else:
        click.echo(click.style("No orphaned message_files records found. There is nothing to delete.", fg="green"))
except Exception as e:
    click.echo(click.style(f"Error deleting orphaned message_files records: {str(e)}", fg="red"))

执行流程，如下所示：

• 查询所有 message_files 记录，其 message_id 在 messages 表中找不到对应记录（通过 LEFT JOIN 和 WHERE m.id IS NULL 实现）。

• 如果有孤立记录，打印出这些记录的数量和详细信息。

• 如果未加 --force 参数，会再次提示用户确认是否删除这些孤立记录。

• 执行删除操作：将所有孤立的 message_files 记录批量删除。

• 打印删除结果；如果没有孤立记录，则提示无需删除。

• 如果过程中有异常，捕获并输出错误信息。

这样可以帮助数据库保持数据一致性，避免无效的文件引用。

3.查找所有文件表中的文件ID和所有被引用的文件ID

这段代码的作用是在数据库中查找并收集所有files_tables中的文件记录，以及在其它相关表和字段中引用到的文件 ID，然后用于后续比对，找出”孤立文件记录”（即数据库中存在但未被引用的文件）。

# fetch file id and keys from each table
all_files_in_tables = []
for files_table in files_tables:
    click.echo(click.style(f"- Listing file records in table {files_table['table']}", fg="white"))
    query = f"SELECT {files_table['id_column']}, {files_table['key_column']} FROM {files_table['table']}"
with db.engine.begin() as conn:
        rs = conn.execute(db.text(query))
for i in rs:
        all_files_in_tables.append({"table": files_table["table"], "id": str(i[0]), "key": i[1]})
click.echo(click.style(f"Found {len(all_files_in_tables)} files in tables.", fg="white"))

# fetch referred table and columns
guid_regexp = "[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"
all_ids_in_tables = []
for ids_table in ids_tables:
    query = ""
if ids_table["type"] == "uuid":
        click.echo(
            click.style(
f"- Listing file ids in column {ids_table['column']} in table {ids_table['table']}", fg="white"
            )
        )
        query = (
f"SELECT {ids_table['column']} FROM {ids_table['table']} WHERE {ids_table['column']} IS NOT NULL"
        )
with db.engine.begin() as conn:
            rs = conn.execute(db.text(query))
for i in rs:
            all_ids_in_tables.append({"table": ids_table["table"], "id": str(i[0])})
elif ids_table["type"] == "text":
        click.echo(
            click.style(
f"- Listing file-id-like strings in column {ids_table['column']} in table {ids_table['table']}",
                fg="white",
            )
        )
        query = (
f"SELECT regexp_matches({ids_table['column']}, '{guid_regexp}', 'g') AS extracted_id "
f"FROM {ids_table['table']}"
        )
with db.engine.begin() as conn:
            rs = conn.execute(db.text(query))
for i in rs:
for j in i[0]:
                all_ids_in_tables.append({"table": ids_table["table"], "id": j})
elif ids_table["type"] == "json":
        click.echo(
            click.style(
                (
f"- Listing file-id-like JSON string in column {ids_table['column']} "
f"in table {ids_table['table']}"
                ),
                fg="white",
            )
        )
        query = (
f"SELECT regexp_matches({ids_table['column']}::text, '{guid_regexp}', 'g') AS extracted_id "
f"FROM {ids_table['table']}"
        )
with db.engine.begin() as conn:
            rs = conn.execute(db.text(query))
for i in rs:
for j in i[0]:
                all_ids_in_tables.append({"table": ids_table["table"], "id": j})
click.echo(click.style(f"Found {len(all_ids_in_tables)} file ids in tables.", fg="white"))

遍历所有文件表，收集所有文件ID；再遍历所有引用表，收集所有被引用的文件ID。执行流程，如下所示：

• 遍历 files_tables，查询每个文件表（如 upload_files、tool_files），收集所有文件的 id 和 key，存入 all_files_in_tables。

• 遍历 ids_tables，根据字段类型（uuid、text、json）分别用不同 SQL 查询方式，提取所有被引用的文件 id，存入 all_ids_in_tables。

• uuid 类型直接查字段非空的 id。

• text/json 类型用正则表达式提取 GUID 格式的 id。

• 最后输出在各表中找到的文件数量和引用数量。

• 如果过程中出错，会输出错误信息并返回。

4.计算孤立文件ID并提示用户

通过集合差集，找出未被引用的文件ID，并提示用户。

# find orphaned files
all_files = [file["id"] for file in all_files_in_tables]
all_ids = [file["id"] for file in all_ids_in_tables]
orphaned_files = list(set(all_files) - set(all_ids))
ifnot orphaned_files:
    click.echo(click.style("No orphaned file records found. There is nothing to delete.", fg="green"))
return
click.echo(click.style(f"Found {len(orphaned_files)} orphaned file records.", fg="white"))
for file in orphaned_files:
    click.echo(click.style(f"- orphaned file id: {file}", fg="black"))
ifnot force:
    click.confirm(f"Do you want to proceed to delete all {len(orphaned_files)} orphaned file records?", abort=True)

5.删除孤立文件记录

遍历所有files_tables，删除所有孤立文件ID对应的记录。

# delete orphaned records for each file
try:
for files_table in files_tables:
        click.echo(click.style(f"- Deleting orphaned file records in table {files_table['table']}", fg="white"))
        query = f"DELETE FROM {files_table['table']} WHERE {files_table['id_column']} IN :ids"
with db.engine.begin() as conn:
            conn.execute(db.text(query), {"ids": tuple(orphaned_files)})
except Exception as e:
    click.echo(click.style(f"Error deleting orphaned file records: {str(e)}", fg="red"))
return
click.echo(click.style(f"Removed {len(orphaned_files)} orphaned file records.", fg="green"))

四.remove_orphaned_files_on_storage()函数

完整的执行示例，如下所示：

flask remove-orphaned-files-on-storage -f

该命令 remove_orphaned_files_on_storage 主要用于 清理存储中孤立的文件，即那些在数据库表中没有记录、但实际还存在于存储（如对象存储、文件系统）中的文件。

1.命令定义与参数

• 使用 @click.command 注册命令名为 remove-orphaned-files-on-storage。

• 使用 @click.option 增加 --force/-f 参数，控制是否跳过用户确认。

@click.option("-f", "--force", is_flag=True, help="Skip user confirmation and force the command to execute.")
@click.command("remove-orphaned-files-on-storage", help="Remove orphaned files on the storage.")
defremove_orphaned_files_on_storage(force: bool):

2.配置要处理的表和存储路径

• files_tables：数据库中存储文件 key 的表及字段。

• storage_paths：存储中需要扫描的路径。

    files_tables = [
        {"table": "upload_files", "key_column": "key"},
        {"table": "tool_files", "key_column": "file_key"},
    ]
    storage_paths = ["image_files", "tools", "upload_files"]

3.用户提示与确认

输出操作说明、风险提示，若未加 --force，要求用户确认。

    click.echo(click.style("This command will find and remove orphaned files on the storage,", fg="yellow"))
    click.echo(
        click.style("by comparing the files on the storage with the records in the following tables:", fg="yellow")
    )
for files_table in files_tables:
        click.echo(click.style(f"- {files_table['table']}", fg="yellow"))
    click.echo(click.style("The following paths on the storage will be scanned to find orphaned files:", fg="yellow"))
for storage_path in storage_paths:
        click.echo(click.style(f"- {storage_path}", fg="yellow"))
    click.echo("")

    click.echo(click.style("!!! USE WITH CAUTION !!!", fg="red"))
    click.echo(
        click.style(
"Currently, this command will work only for opendal based storage (STORAGE_TYPE=opendal).", fg="yellow"
        )
    )
    click.echo(
        click.style(
"Since not all patterns have been fully tested, please note that this command may delete unintended files.",
            fg="yellow",
        )
    )
    click.echo(
        click.style("This cannot be undone. Please make sure to back up your storage before proceeding.", fg="yellow")
    )
    click.echo(
        click.style(
            (
"It is also recommended to run this during the maintenance window, "
"as this may cause high load on your instance."
            ),
            fg="yellow",
        )
    )
ifnot force:
        click.confirm("Do you want to proceed?", abort=True)

在实际删除存储中孤立文件前，向用户展示即将操作的内容和风险，并要求用户确认，防止误操作。主要功能，如下所示：

• 通过多条 click.echo 输出，详细告知用户该命令会做什么，包括会对哪些表和存储路径进行操作，并强调操作的风险（如可能误删文件、不可恢复、建议备份等）。

• 如果未加 --force 参数，会要求用户手动确认是否继续操作。

4.获取数据库中所有文件 key

遍历 files_tables，查询所有 key，存入 all_files_in_tables。

    all_files_in_tables = []
try:
for files_table in files_tables:
            click.echo(click.style(f"- Listing files from table {files_table['table']}", fg="white"))
            query = f"SELECT {files_table['key_column']} FROM {files_table['table']}"
with db.engine.begin() as conn:
                rs = conn.execute(db.text(query))
for i in rs:
                all_files_in_tables.append(str(i[0]))
        click.echo(click.style(f"Found {len(all_files_in_tables)} files in tables.", fg="white"))
except Exception as e:
        click.echo(click.style(f"Error fetching keys: {str(e)}", fg="red"))

5.获取存储中所有文件

遍历 storage_paths，用 storage.scan 获取所有文件路径，存入 all_files_on_storage。

    all_files_on_storage = []
for storage_path in storage_paths:
try:
            click.echo(click.style(f"- Scanning files on storage path {storage_path}", fg="white"))
            files = storage.scan(path=storage_path, files=True, directories=False)
            all_files_on_storage.extend(files)
except FileNotFoundError as e:
            click.echo(click.style(f"  -> Skipping path {storage_path} as it does not exist.", fg="yellow"))
continue
except Exception as e:
            click.echo(click.style(f"  -> Error scanning files on storage path {storage_path}: {str(e)}", fg="red"))
continue
    click.echo(click.style(f"Found {len(all_files_on_storage)} files on storage.", fg="white"))

6.计算并展示孤立文件

• 用集合差集找出只在存储、但不在数据库的文件（孤立文件）。

• 若无孤立文件则退出，否则输出列表，并再次确认。

    orphaned_files = list(set(all_files_on_storage) - set(all_files_in_tables))
ifnot orphaned_files:
        click.echo(click.style("No orphaned files found. There is nothing to remove.", fg="green"))
return
    click.echo(click.style(f"Found {len(orphaned_files)} orphaned files.", fg="white"))
for file in orphaned_files:
        click.echo(click.style(f"- orphaned file: {file}", fg="black"))
ifnot force:
        click.confirm(f"Do you want to proceed to remove all {len(orphaned_files)} orphaned files?", abort=True)

7.删除孤立文件

遍历孤立文件，调用 storage.delete 删除，统计成功和失败数，输出结果。

    removed_files = 0
    error_files = 0
for file in orphaned_files:
try:
            storage.delete(file)
            removed_files += 1
            click.echo(click.style(f"- Removing orphaned file: {file}", fg="white"))
except Exception as e:
            error_files += 1
            click.echo(click.style(f"- Error deleting orphaned file {file}: {str(e)}", fg="red"))
continue
if error_files == 0:
        click.echo(click.style(f"Removed {removed_files} orphaned files without errors.", fg="green"))
else:
        click.echo(click.style(f"Removed {removed_files} orphaned files, with {error_files} errors.", fg="yellow"))

总结：该命令通过比对数据库和存储中的文件，找出并删除存储中多余的（孤立的）文件，整个流程包含用户提示、数据收集、差集计算、二次确认和实际删除操作。

参考文献

[0] commands.py中的函数解析5：clear_orphaned_file_records等：https://z0yrmerhgi8.feishu.cn/wiki/BgvTwmbY2igm9RkQV1Lcnd6qnmc

[1] click github：https://github.com/pallets/click

[2] click官方文档：https://click.palletsprojects.com/en/stable/

[3] click-extra github：https://github.com/kdeldycke/click-extra

[4] click-extra官方文档：https://kdeldycke.github.io/click-extra/

知识星球服务内容：Dify源码剖析及答疑，Dify对话系统源码，NLP电子书籍报告下载，公众号所有付费资料。加微信buxingtianxia21进NLP工程化资料群。