如何获取Google Cloud Storage DLP检查作业的完整明细结果

扫描本地/外部文件时调用的是同步inspect_content接口，接口会直接在响应中返回全量检测明细。扫描GCS存储文件走的是异步DLP作业机制，get_dlp_job接口返回的info_type_stats字段是设计好的汇总统计值，默认不会在作业对象中携带单条命中的quote、位置、置信度等明细字段，不是代码写法问题。

• 创建GCS扫描的DLP作业时，必须在作业配置中添加SaveFindings动作，指定将全量检测明细写入BigQuery表，这是异步存储类扫描作业获取全量明细的官方标准路径。
• 等Pub/Sub收到作业完成的通知后，不需要从get_dlp_job返回结果里翻明细，直接从指定的BigQuery表中拉取对应作业ID关联的所有记录，就能拿到和本地扫描完全一致的明细字段。
• 不要尝试直接从DLP作业对象中提取全量finding，异步作业的返回结构本身就不承载全量明细数据，所有单条命中结果只会写入你配置的输出位置。

1. 创建DLP作业时添加BigQuery输出配置 #

from google.cloud import dlp_v2

dlp_client = dlp_v2.DlpServiceClient()

# 构造作业配置时新增保存明细到BigQuery的动作
job_config = {
    "inspect_config": inspect_config, # 替换为你之前定义的检测规则配置
    "storage_config": {
        "cloud_storage_options": {
            "file_set": {
                "url": "gs://your-bucket-name/your-scan-path/*" # 替换为你的GCS路径
            }
        }
    },
    "actions": [
        {
            "save_findings": {
                "output_config": {
                    "table": {
                        "project_id": "your-gcp-project-id", # 替换为你的GCP项目ID
                        "dataset_id": "your-bq-dataset-name", # 替换为你已创建的BQ数据集名
                        "table_id": "dlp_scan_findings" # 表不存在时DLP会自动创建
                    }
                }
            }
        }
    ]
}

# 提交创建扫描作业
operation = dlp_client.create_dlp_job(
    parent=f"projects/your-gcp-project-id/locations/global",
    inspect_job=job_config
)

2. 修改Pub/Sub回调逻辑，从BigQuery拉取明细 #

from google.cloud import bigquery

def callback(message):
    try:
        if message.attributes["DlpJobName"] == operation.name:
            message.ack()
            job = dlp_client.get_dlp_job(request={"name": operation.name})
            
            # 初始化BQ客户端拉取当前作业的全量明细
            bq_client = bigquery.Client()
            query = """
                SELECT 
                    quote,
                    info_type.name AS info_type,
                    likelihood,
                    location.byte_range.start AS location_start,
                    location.byte_range.end AS location_end
                FROM `your-gcp-project-id.your-bq-dataset-name.dlp_scan_findings`
                WHERE job_name = @job_name
            """
            bq_job_config = bigquery.QueryJobConfig(
                query_parameters=[
                    bigquery.ScalarQueryParameter("job_name", "STRING", operation.name)
                ]
            )

            results = []
            for row in bq_client.query(query, job_config=bq_job_config):
                results.append(dict(row))
            
            # 此处results结构和你本地扫描拿到的明细格式完全一致
            print(results)

            job_done.set()
        else:
            message.drop()
    except Exception as e:
        print(e)
        raise

权限注意：运行作业前需要给DLP对应的服务账号授予目标BigQuery数据集的写入权限，否则作业执行完成后不会写入明细数据。如果扫描的文件命中量超过1000条，BigQuery是唯一稳定支持全量明细存储的输出渠道，其他输出方式存在截断风险。

内容的提问来源于stack exchange，提问作者Elisius Legodi