graph LR
    A[视频会议平台<br/>Zoom/Teams/腾讯会议/自建] -->|Webhook/定时拉取| B(接入层<br/>对象存储/消息队列)
    B --> C{计算层<br/>异步任务编排}
    C --> D[ASR 语音识别<br/>Whisper/FunASR/商业API]
    C --> E[OCV 视觉识别<br/>屏幕共享/水印/人脸]
    D & E --> F[多模态融合<br/>时间轴对齐]
    F --> G(策略层<br/>规则引擎/Drools/自研DSL)
    G --> H[分级标记<br/>公开/内部/机密/绝密]
    G --> I[脱敏执行<br/>静音/马赛克/字符替换]
    H & I --> J[应用层<br/>合规归档/审计报表/下发业务系统]

{
  "meeting_id": "string (UUID)",
  "platform": "enum: [zoom, teams, tencent, feishu, self_hosted]",
  "start_time": "timestamp (UTC)",
  "duration_sec": "integer",
  "participants": [
    {"user_id": "string", "name": "string", "dept": "string", "role": "host/attendee"}
  ],
  "storage_uri": "s3://bucket/raw/2024/05/20/meeting_xxx.mp4",
  "recording_type": "enum: [cloud, local, hybrid]",
  "watermark_info": {"user_id": "string", "timestamp": "timestamp"} // 用于溯源
}

# dags/video_audit_pipeline.py
from airflow.decorators import dag, task
from airflow.providers.cncf.kubernetes.operators.pod import KubernetesPodOperator
from datetime import datetime, timedelta
import json

@dag(
    dag_id="video_compliance_audit",
    schedule=None,  # 由 Event Trigger 触发
    start_date=datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=500,  # 高并发支撑
    default_args={"retries": 2, "retry_delay": timedelta(minutes=5)},
    tags=["compliance", "video", "pii"],
)
def video_audit_pipeline():

    @task(task_id="parse_metadata")
    def parse_metadata(**context):
        # 从 Kafka/Webhook 消息体解析统一元数据
        conf = context["dag_run"].conf
        return {
            "meeting_id": conf["meeting_id"],
            "source_uri": conf["storage_uri"],
            "platform": conf["platform"],
            "participants": conf["participants"],
            "priority": "high" if conf.get("dept") in ["finance", "legal", "rd_core"] else "normal"
        }

    # 1. ASR 任务：动态请求 GPU 资源，支持 Spot 实例容忍
    asr_task = KubernetesPodOperator(
        task_id="asr_transcribe",
        namespace="ai-compute",
        image="registry.internal/asr-worker:funasr-v2.1.0",
        arguments=["--input_uri", "{{ ti.xcom_pull(task_ids='parse_metadata')['source_uri'] }}",
                   "--output_uri", "s3://bucket/asr_raw/{{ ti.xcom_pull(task_ids='parse_metadata')['meeting_id'] }}.json",
                   "--hotword_list", "s3://bucket/config/hotwords.txt"],
        container_resources={
            "request_memory": "8Gi", "limit_memory": "12Gi",
            "request_nvidia_com_gpu": "1", "limit_nvidia_com_gpu": "1"
        },
        tolerations=[{"key": "nvidia.com/gpu", "operator": "Exists", "effect": "NoSchedule"}],
        node_selector={"node-type": "gpu-inference", "lifecycle": "spot"},  # 成本优化关键
        get_logs=True,
        is_delete_operator_pod=True,
    )

    # 2. 视觉并行分支
    ocr_task = KubernetesPodOperator(...) # 类似配置，处理屏幕共享关键帧
    watermark_task = KubernetesPodOperator(...) # 水印提取

    # 3. 多模态融合与 NER (CPU 密集型，可用 Graviton/ARM 实例降本)
    @task.kubernetes(
        task_id="fusion_ner",
        image="registry.internal/nlp-worker:bert-ner-v3.4",
        container_resources={"request_memory": "4Gi", "limit_memory": "6Gi"},
        node_selector={"arch": "arm64"} # 利用 ARM 实例价格优势
    )
    def fusion_ner(asr_result, ocr_result, watermark_result, metadata):
        # 核心逻辑：时间轴对齐 -> 实体抽取 -> 置信度校准
        # 返回: {"entities": [...], "risk_score": 0.92, "suggested_level": "L3"}
        pass

    # 4. 策略引擎决策 (规则即数据，热加载)
    @task(task_id="policy_decision")
    def policy_decision(ner_result, metadata):
        # 调用规则引擎服务
        # 返回: {"level": "L3", "actions": ["mute_audio:10-15", "mosaic_video:200-210", "replace_text:ID_CARD"]}
        pass

    # 5. 脱敏执行 (IO 密集，FFmpeg 复杂滤镜链)
    desensitize_task = KubernetesPodOperator(
        task_id="desensitize_media",
        image="registry.internal/ffmpeg-worker:6.1-libfdk-aac",
        # 动态生成 filter_complex 脚本
        arguments=["--script", "{{ ti.xcom_pull(task_ids='policy_decision')['ffmpeg_script'] }}"],
        container_resources={"request_cpu": "2", "limit_cpu": "4"},
    )

    # 6. 归档与回调
    @task(task_id="archive_and_callback")
    def archive_and_callback(desensitize_result, policy_result, metadata):
        # 1. 原始文件 -> 冷存储
        # 2. 脱敏文件 -> 业务桶
        # 3. 审计日志 -> ClickHouse/ES
        # 4. 回调业务系统/发送工单
        pass

    # 依赖链定义
    meta = parse_metadata()
    asr_out = asr_task(meta)
    ocr_out = ocr_task(meta)
    wm_out = watermark_task(meta)
    ner_out = fusion_ner(asr_out, ocr_out, wm_out, meta)
    policy_out = policy_decision(ner_out, meta)
    desens_out = desensitize_task(policy_out, meta)
    archive_and_callback(desens_out, policy_out, meta)

dag = video_audit_pipeline()

# utils/ffmpeg_filter_builder.py
class FilterGraphBuilder:
    def __init__(self, video_duration, width, height):
        self.filters = []
        self.audio_filters = []
        self.video_filters = []
        self.map_idx = 0

    def add_audio_mute(self, start_sec, end_sec, beep_freq=1000):
        # 方案 A: 静音
        # self.audio_filters.append(f"volume=enable='between(t,{start_sec},{end_sec})':volume=0")
        # 方案 B: 插入提示音 (合规提示)
        self.audio_filters.append(
            f"sine=frequency={beep_freq}:duration={end_sec-start_sec}:start_time={start_sec}[beep];"
            f"[0:a][beep]amix=inputs=2:duration=first:dropout_transition=0[a{self.map_idx}]"
        )
        self.map_idx += 1

    def add_video_mosaic(self, start_sec, end_sec, x, y, w, h, block_size=20):
        # 关键：使用 sendcmd 配合 timeline editing 实现动态区域马赛克
        # 需配合 drawbox 或 gblur/removegrain
        # 此处简化为全程马赛克，生产环境需结合目标检测坐标序列生成动态 filter
        self.video_filters.append(
            f"[0:v]crop={w}:{h}:{x}:{y},gblur=sigma=20[mosaic{self.map_idx}];"
            f"[0:v][mosaic{self.map_idx}]overlay=x={x}:y={y}:enable='between(t,{start_sec},{end_sec})'[v{self.map_idx}]"
        )
        self.map_idx += 1

    def add_subtitle_redact(self, subtitle_path, rules):
        # 字幕流单独处理：ass/srt -> 正则替换 -> 重新烧录或软字幕挂载
        pass

    def build_complex_filter(self):
        # 拼接音视频滤镜链，处理链式依赖
        return ";".join(self.audio_filters + self.video_filters)

    def get_output_args(self):
        return ["-filter_complex", self.build_complex_filter(), "-map", "[v_out]", "-map", "[a_out]"]

graph LR
    A[线上推理<br/>低置信度/高风险样本] --> B{不确定性采样<br/>Entropy / Margin Sampling}
    B --> C[人工标注平台<br/>Label Studio / CVAT]
    C --> D[增量训练集<br/>版本化管理 DVC/Git-LFS]
    D --> E[自动化训练流水线<br/>Kubeflow Pipelines]
    E --> F[模型评测<br/>回归集 F1/P/R 对比]
    F -->|通过| G[灰度发布<br/>Canary 5% 流量]
    G -->|指标达标| H[全量发布<br/>模型注册表版本+1]
    H --> A

[总部合规中台] <-- 加密控制面/模型下发 --> [分支机构/海外节点 边缘计算集群]
      |                                                    |
      |-- 策略配置/模型权重/热词表 (单向下发)              |-- 原始视频/音频 (绝不上传)
      |-- 审计日志/脱敏结果/模型指标 (单向上收)            |-- 脱敏后文件/结构化报告 (上收)

# OSS/ S3 Lifecycle Configuration 示例
Rules:
  - ID: "RawVideo_Tiering"
    Status: "Enabled"
    Filter: { Prefix: "raw/" }
    Transitions:
      - Days: 1, StorageClass: "IA" (低频访问)      # 会议结束 1 天后降冷
      - Days: 30, StorageClass: "Archive" (归档)     # 30 天后归档
      - Days: 365, StorageClass: "DeepArchive" (深度归档) # 1 年后深冷
    Expiration:
      Days: 2555 # 7 年合规保留期满自动删除 (L3/L4 级)
  
  - ID: "Desensitized_Video"
    Filter: { Prefix: "desensitized/" }
    Transitions:
      - Days: 90, StorageClass: "Archive"
    Expiration:
      Days: 1095 # 脱敏版保留 3 年

paths:
  /api/v1/forensics/trace:
    post:
      summary: "泄露片段溯源定责"
      requestBody:
        content:
          application/json:
            schema:
              type: object
              required: [leaked_content_hash, leaked_media_segment]
              properties:
                leaked_content_hash: { type: string, description: "泄露文件/截图 SHA256" }
                leaked_media_segment: 
                  type: object
                  properties:
                    video_md5: { type: string }
                    time_range: { type: array, items: {type: number}, minItems: 2, maxItems: 2 }
      responses:
        '200':
          description: "溯源报告"
          content:
            application/json:
              schema:
                type: object
                properties:
                  source_meeting_id: { type: string }
                  leaker_watermark: { type: string, description: "提取的水印工号/IP" }
                  leak_time: { type: string, format: date-time }
                  original_level: { type: string, enum: [L1, L2, L3, L4] }
                  desensitized_version_uri: { type: string }
                  audit_log_integrity_proof: { type: string, description: "区块链存证/默克尔树根哈希" }