西门子能源：基于 AWS 的事件驱动、AI 赋能的数据管道

# construct_object_key.py — 确定性的、可查询的 S3 键生成

   class Capability(str, Enum):
       EDAA = "edaa"
       CUSTOMER_FACING = "customer-facing"
       MANUFACTURING = "manufacturing"
       QUALITY = "quality"
       # ...

   def construct_object_key(
       capability: Capability | str | None = None,
       file_name: str = "",
       sub_capability: str | None = None,
       year: str | None = None,
       month: str | None = None,
       day: str | None = None,
   ) -> str:
       now = datetime.now()
       generated_hash = hashlib.md5(
           f"{capability}{sub_capability}{file_name}".encode()
       ).hexdigest()[:2]

       return (
           f"capability={capability}/"
           f"sub-capability={sub_capability or 'unknown'}/"
           f"year={year or now.year}/"
           f"month={month or f'{now.month:02d}'}/"
           f"day={day or f'{now.day:02d}'}/"
           f"hash={generated_hash}/"
           f"{file_name}"
       )

   # access_grants.py — 通过 S3 Access Grants 实现按用户的分区隔离

   CfnAccessGrant(
       self,
       f"AccessGrantUser{user_idx}",
       access_grants_location_id=user_location.ref,
       permission="READWRITE",
       grantee=CfnAccessGrant.GranteeProperty(
           grantee_identifier=user_role.role_arn,
           grantee_type="IAM",
       ),
   )

# get_cached_or_exchange_credentials.py — 带缓存的令牌交换
 
def get_cached_or_exchange_credentials(id_token: str) -> CredentialsTypeDef:
    key = _cache_key_from_token(id_token)
    now = time.time()
    with _CACHE_LOCK:
        entry = _CACHE.get(key)
        if entry and entry["expires_at"] > now + _SKEW_SECONDS:
            return entry["credentials"]
 
    new_credentials = _exchange_and_assume_with_expiry(id_token)
 
    with _CACHE_LOCK:
        _CACHE[key] = {
            "credentials": new_credentials,
            "expires_at": new_credentials["Expiration"].timestamp(),
        }
    return new_credentials

# workflow.py — 使用 Bedrock Data Automation 的 Step Functions 编排
 
definition_body = sfn.DefinitionBody.from_chainable(
    extract_event_data
    .next(generate_blueprint)          # 基于元数据要求的动态 blueprint
    .next(start_data_automation_job)   # Bedrock Data Automation
    .next(wait_for_job_completion)     # 使用任务令牌的异步等待
    .next(normalize_event_data)        # 经过 EventBridge 模式校验的输出
    .next(publish_output_event)        # EventBridge 完成事件
    .next(sfn.Succeed(self, "FileMetadataEnrichmentCompletion"))
)

# bedrock_data_automation_blueprint_generator.py
 
def generate_blueprint_schema(enrichments, blueprint_type):
    properties = {}
 
    # 基础属性——始终提取
    properties["data_classification"] = {
        "type": "string",
        "instruction": "The data classification level (public, internal, confidential, restricted)",
    }
    properties["summary"] = {
        "type": "string",
        "instruction": "A brief summary of the file content",
    }
    properties["keywords"] = {
        "type": "array",
        "items": {"type": "string"},
        "instruction": "Key terms and keywords extracted from the document",
    }
 
    # 特定于模态的增强
    if enrichments.get("audio", {}).get("transcribe"):
        properties["transcript"] = {
            "type": "string",
            "instruction": "Full transcript of the audio content",
        }
    if enrichments.get("video", {}).get("scenes"):
        properties["scenes"] = {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "scene_number": {"type": "number"},
                    "start_time": {"type": "number"},
                    "end_time": {"type": "number"},
                    "description": {"type": "string"},
                },
            },
            "instruction": "Scene changes detected in the video with timestamps",
        }
 
    return {"class": f"{blueprint_type.capitalize()}Metadata", "properties": properties}

# multi_chunk.py — 预签名分段上传编排
 
def initiate_multi_chunk_upload_presigned(body):
    chunk_size, num_chunks = _calculate_chunk_size(request.fileSize)
 
    response = s3.create_multipart_upload(
        Bucket=bucket, Key=key, ServerSideEncryption="aws:kms"
    )
 
    presigned_urls = []
    for chunk_number in range(1, num_chunks + 1):
        presigned_url = s3.generate_presigned_url(
            "upload_part",
            Params={
                "Bucket": bucket, "Key": key,
                "UploadId": response["UploadId"],
                "PartNumber": chunk_number,
            },
            ExpiresIn=expires_in,
        )
        presigned_urls.append({
            "chunkNumber": chunk_number,
            "url": presigned_url,
            "startByte": (chunk_number - 1) * chunk_size,
            "endByte": min(chunk_number * chunk_size - 1, request.fileSize - 1),
        })
 
    return {"uploadId": response["UploadId"], "chunks": presigned_urls}

// api.service.ts — 自动上传策略选择
 
async upload(file: File, request: InitiateUploadRequest, onProgress?) {
  if (file.size > FILE_SIZE_THRESHOLD) {
    return this.multiChunkUploadService.uploadFile(file, request, { onProgress });
  }
 
  const base64Content = await readFileAsBase64(file);
  return this.httpService.request('/files', 'POST', {
    body: { content: base64Content, fileName: request.fileName },
  });
}

// storage-browser.provider.tsx — 自定义存储浏览器，带有基于 API 的操作
 
const { StorageBrowser } = createStorageBrowser({
  config: {
    registerAuthListener: async (onAuthStateChange) => {
      const authService = getAuthService();
      authService.registerAuthListener(onAuthStateChange);
    },
    listLocations: async ({ options }) => {
      return await apiService.getLocations({
        options: { pageSize: 30, nextToken: options?.nextToken },
      });
    },
  },
  actions: actionsBuilder.buildActions(),
});

# file_metadata_enrichment_processor.py — 跨账户事件发布
 
def put_event(bucket, key, size=None, etag=None, enrichments=None):
    detail = create_event_detail(bucket, key, size, etag, enrichments)
 
    return events_client.put_events(Entries=[{
        "Source": "com.siemens-energy.pmdp.file-metadata-enrichment",
        "DetailType": "FileMetadataEnrichmentRequest",
        "Detail": json.dumps(detail),
        "EventBusName": EVENT_BUS_ARN,  # 跨账户 ARN
    }])

# config.py — 类型安全的、特定于环境的配置
 
config_environment = make_config(
    env=zf(cdk.Environment),
    environment=zf(Literal["sandbox", "dev", "uat", "prd"]),
    s3explorer=zf(config_s3explorer),
    access_grants=zf(config_access_grants),
    project_name=zf(str, default="mfg-product-measurement-data-pipeline"),
    file_metadata_enrichment_event_bus_arn=zf(str),
)