meeting:Event (核心会议实体)
├── meeting:InstantMeeting      # 即时会议
├── meeting:ScheduledMeeting    # 预约会议
├── meeting:RecurringMeeting    # 周期性会议
└── meeting:Webinar             # 网络研讨会

meeting:Participant (参会主体)
├── meeting:InternalUser        # 组织内用户
├── meeting:ExternalGuest       # 外部嘉宾
└── meeting:VirtualAssistant    # 智能助手/记录机器人

meeting:Artifact (会议产出物)
├── meeting:Recording           # 云录制文件
├── meeting:Transcript          # 实时字幕/全量转写
├── meeting:Minutes             # 结构化会议纪要
├── meeting:Decision            # 决议事项
└── meeting:ActionItem          # 待办行动项

@prefix meeting: <http://example.org/meeting#> .
@prefix org: <http://example.org/org#> .
@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .

meeting:ScheduledMeeting a owl:Class ;
    rdfs:subClassOf meeting:Event ;
    rdfs:label "预约会议"@zh ;
    owl:equivalentClass [
        a owl:Class ;
        owl:intersectionOf (
            meeting:Event
            [ a owl:Restriction ;
              owl:onProperty meeting:hasSchedule ;
              owl:someValuesFrom meeting:Schedule ]
        )
    ] .

meeting:hasHost a owl:ObjectProperty ;
    rdfs:domain meeting:Event ;
    rdfs:range meeting:Participant ;
    rdfs:label "主持人"@zh ;
    owl:inverseOf meeting:hostsMeeting .

meeting:confidentialityLevel a owl:DatatypeProperty ;
    rdfs:domain meeting:Event ;
    rdfs:range xsd:string ;
    rdfs:label "保密等级"@zh ;
    skos:note "取值受限于 {公开, 内部, 机密, 绝密}"@zh .

{
  "meeting.durationMinutes": "($floor(($millis(meeting.endTime) - $millis(meeting.startTime)) / 60000))",
  "participant.isHost": "$contains(['host', 'co-host', 'organizer'], $lowercase(participant.role))"
}

┌─────────────────────────────────────────────────────────────┐
│                    会议知识图谱构建管线                        │
├─────────────────────────────────────────────────────────────┤
│  数据源层  │  标准化层  │  实体对齐层  │  图构建层  │  服务层   │
│  (多源)   │  (ETL)    │  (Entity    │  (RDF/    │  (GraphQL/│
│           │           │   Resolution)│  Property │  Cypher/  │
│           │           │              │  Graph)   │  REST)    │
└─────────────────────────────────────────────────────────────┘

# 伪代码：关系感知图卷积前向传播
def forward(self, entity_features, edge_index, edge_type):
    # edge_type 编码关系类型：same_as, has_attribute, belongs_to_org 等
    h = entity_features
    for layer in self.layers:
        h = layer(h, edge_index, edge_type)  # 聚合邻居信息
        h = F.relu(h)
    return h  # [num_entities, hidden_dim]

# 字段语义向量生成
def encode_field_semantics(field_name, description, sample_values):
    text = f"[CLS] {field_name} [SEP] {description} [SEP] {' '.join(sample_values[:5])} [SEP]"
    return bert_model(text).pooler_output  # [768]

# 相似度计算与阈值判定
def align_entities(emb_struct, emb_semantic, threshold=0.85):
    fused = 0.6 * F.normalize(emb_struct) + 0.4 * F.normalize(emb_semantic)
    sim_matrix = torch.mm(fused, fused.t())  # 余弦相似度
    pairs = (sim_matrix > threshold).nonzero(as_tuple=False)
    return filter_mutual_best(pairs, sim_matrix)

# Debezium MySQL Connector 配置片段
connector.class: io.debezium.connector.mysql.MySqlConnector
database.hostname: meeting-db-prod
database.user: cdc_user
table.include.list: meeting_db.meetings, meeting_db.participants, meeting_db.recordings
transforms: unwrap, addTopicPrefix
transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
transforms.addTopicPrefix.type: org.apache.kafka.connect.transforms.InsertField$Value
transforms.addTopicPrefix.topic.field: _source_system
transforms.addTopicPrefix.static.field: source_system
transforms.addTopicPrefix.static.value: "zoom"

# 统一 Schema 片段
type Meeting @key(fields: "id") {
  id: ID!
  subject: String!
  startTime: DateTime!
  endTime: DateTime!
  organizer: User @provides(fields: "id displayName")
  participants: [Participant!]!
  recordings: [Recording!]!
  minutes: Minutes
  decisions: [Decision!]!
  actionItems: [ActionItem!]!
  confidentialityLevel: ConfidentialityLevel!
  sourceSystems: [SourceSystem!]!  # 标识数据来源系统
}

extend type User @key(fields: "id") {
  id: ID! @external
  displayName: String @external
  email: String @external
  department: Department @external
}

stateDiagram-v2
    [*] --> Draft : 创建草稿
    Draft --> Scheduled : 发送邀请
    Draft --> Cancelled : 取消草稿
    Scheduled --> Notified : 触发提醒 (T-15min)
    Notified --> InProgress : 主持人/参会人加入
    Scheduled --> InProgress : 直接加入 (即时会议)
    InProgress --> Paused : 全员离开/主持人锁定
    Paused --> InProgress : 重新加入
    InProgress --> Ending : 主持人结束/超时自动结束
    Ending --> Processing : 录制转码/ASR/NLP管线触发
    Processing --> Completed : 产出物入库/索引建立
    Processing --> Failed : 管线异常 (需人工干预/重试)
    Completed --> Archived : 归档策略触发 (冷存储)
    Cancelled --> [*]
    Failed --> [*] : 告警闭环后
    Archived --> [*]

-- 会议室资源冲突检测：新会议 [new_start, new_end) 与现有实例重叠
SELECT id FROM meeting_instances
WHERE room_id = $1
  AND status NOT IN ('Cancelled', 'Declined')
  AND tsrange(start_time, end_time) && tsrange($2, $3)  -- 利用 GiST 索引加速范围重叠查询
LIMIT 1;

@prefix iot: <http://example.org/iot#> .
@prefix brick: <https://brickschema.org/schema/Brick#> .  # 复用 Brick 极简本体

meeting:PhysicalRoom a owl:Class ;
    rdfs:subClassOf brick:Room ;
    owl:equivalentClass [
        a owl:Class ;
        owl:intersectionOf (
            brick:Room
            [ a owl:Restriction ; owl:onProperty brick:hasCapability ; owl:someValuesFrom meeting:VCCapability ]
        )
    ] .

meeting:VCCapability a owl:Class ;
    rdfs:label "视频会议能力集"@zh ;
    owl:oneOf ( meeting:TeamsRooms meeting:ZoomRooms meeting:Polycom meeting:CustomSIP ) .

# 设备实例示例
room:3A-201 a meeting:PhysicalRoom ;
    brick:hasName "3号楼A区201会议室" ;
    brick:hasCapacity "12"^^xsd:int ;
    brick:hasEquipment device:vc-bar-001, device:screen-001, device:mic-array-001 ;
    meeting:timezone "Asia/Shanghai" ;
    meeting:networkSegment "10.10.20.0/24" .

device:vc-bar-001 a brick:VideoConferenceBar ;
    brick:hasFirmwareVersion "v3.4.1" ;
    brick:managedBy iot:MDM-System-A ;
    meeting:supportedProtocols ( "H.323" "SIP" "Teams Native" "Zoom Native" ) .

原始录制 (MP4) 
   │
   ├─► 音频轨 ──► VAD + Speaker Diarization (PyAnnote) ──► 发言人分离片段
   │                │
   │                └─► ASR (Whisper-Large-v3 / Paraformer) ──► 带时间戳/发言人标签的转写文本
   │
   ├─► 视频流 (关键帧 1fps) ──► 视觉编码器 (InternVideo2 / CLIP-ViT) ──► 视觉向量序列
   │                │
   │                └─► 屏幕共享检测 (OCR + LayoutLMv3) ──► PPT/文档/代码结构化文本
   │
   └─► 白板/协作文档事件流 (WebSocket Replay) ──► 笔迹向量化 / 文档版本快照
   
   ▼
多模态融合模块 (Cross-Modal Aligner)
   │
   ├─► 文本模态: LLM (Qwen2-72B / GPT-4o) 抽取实体/关系/事件 (Schema-constrained Decoding)
   ├─► 视觉模态: MLLM (Qwen-VL / LLaVA-NeXT) 理解图表/架构图/白板草图 → 结构化描述
   └─► 对齐模块: 基于时间戳将视觉实体 (如"PPT第5页架构图") 链接至文本实体 (如"核心架构决策")
   
   ▼
知识图谱入库 (Neo4j / TuGraph) + 向量索引 (Milvus)

# 检索流程伪代码
def multimodal_rag_query(question: str):
    # 1. 问题分解与路由
    intent = llm_classify(question)  # {type: "visual_retrieval", target: "architecture_diagram", speaker: "李明", time_range: "last_week"}
    
    # 2. 图谱精准定位 (结构化检索)
    cypher = """
    MATCH (m:Meeting)-[:HAS_ARTIFACT]->(a:WhiteboardSnapshot)
    WHERE m.startTime > $start AND m.startTime < $end
      AND (m)-[:HAS_PARTICIPANT]->(:Person {name: $speaker})
      AND a.contentType = 'architecture_diagram'
    RETURN a.vectorId, a.timestamp, a.description
    """
    candidates = graph_db.run(cypher, params)
    
    # 3. 向量精排 (语义匹配)
    query_vec = clip_encode_text(question)
    top_k = milvus.search(query_vec, candidates.vectorIds, top_k=3)
    
    # 4. 多模态答案生成
    context = build_context(top_k)  # 包含图片 Base64、时间戳、周围转写文本
    prompt = f"""基于以下会议片段回答：
    [视频片段 {top_k[0].timestamp}] 图像描述: {top_k[0].description}
    [转写上下文] {top_k[0].surrounding_transcript}
    问题: {question}
    要求: 直接引用图像内容，标注时间戳。"""
    return vlm_generate(prompt, images=[top_k[0].image_bytes])

# meeting_pipeline.yaml - 单一事实来源
pipeline_id: "meeting_kg_daily_build"
schedule: "0 3 * * *"  # 每日 03:00
source_systems:
  - system_id: "zoom"
    connector: "zoom_api_v2"
    incremental_key: "updated_at"
    entities: ["meetings", "participants", "recordings", "transcripts"]
  - system_id: "tencent_meeting"
    connector: "tencent_openapi"
    entities: ["meetings", "records", "attendees"]

transform_layers:
  - layer: "raw"          # 原始落盘 (Iceberg/Hudi)
    partition_by: "dt=source_system"
  - layer: "standardized" # 标准化映射 (DBT Models)
    models:
      - "stg_meetings"
      - "stg_participants"
      - "dim_meeting_standard"  # 核心维度表
  - layer: "entity_resolution" # 实体消歧 (Python UDF / Spark Job)
    blocking_keys: ["email_hash", "phone_hash", "display_name_ngram"]
    model_path: "s3://models/er_model_v3.pt"
  - layer: "kg_construction"   # 图构建
    steps:
      - "build_vertices"
      - "build_edges"
      - "compute_pagerank"      # 会议/人影响力计算
      - "export_to_graphdb"     # 批量导入 Neo4j (LOAD CSV / Bolt)

data_quality_gates:  # 每层强制门禁
  - layer: "standardized"
    rules:
      - "not_null(meeting_id)"
      - "referential_integrity(organizer_id -> dim_user.user_id)"
      - "freshness(max(updated_at) > now() - 24h)"
  - layer: "kg_construction"
    rules:
      - "vertex_count(Person) > 10000"
      - "edge_count(HAS_ACTION_ITEM) / vertex_count(Meeting) > 0.8"  # 业务健康度

alerting:
  on_failure: "pagerduty:critical"
  on_quality_drop: "slack:#data-alerts"

sequenceDiagram
    participant OEM as 主机厂
    participant Supplier as 供应商
    participant TTP as 可信第三方/区块链锚定

    Note over OEM, Supplier: 阶段 1: 隐私求交 (PSI) - 确定共同参会人/会议实例
    OEM->>TTP: Encrypt(Hash(Attendee_IDs)) 
    Supplier->>TTP: Encrypt(Hash(Attendee_IDs))
    TTP-->>OEM: Intersection_Set (加密)
    TTP-->>Supplier: Intersection_Set (加密)

    Note over OEM, Supplier: 阶段 2: 联邦嵌入训练 - 学习跨域实体表示
    loop Local Epochs
        OEM->>OEM: Train Local GNN on Local Meeting Graph
        Supplier->>Supplier: Train Local GNN on Local Meeting Graph
    end
    OEM->>TTP: Upload Local Model Delta (DP Noise Added)
    Supplier->>TTP: Upload Local Model Delta (DP Noise Added)
    TTP->>OEM: Global Model (Aggregated)
    TTP->>Supplier: Global Model (Aggregated)

    Note over OEM, Supplier: 阶段 3: 语义映射推理 - 无明文交互
    OEM->>Supplier: Encrypted Query: "Find Decisions about 'Battery Safety'"
    Supplier-->>OEM: Encrypted Result: [Decision_IDs, Confidence]

# ChaosMesh 实验定义
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: meeting-kg-graphdb-pod-kill
spec:
  action: pod-failure
  mode: one
  selector:
    namespaces: [prod]
    labelSelectors:
      app: neo4j-core
  duration: "30s"
  scheduler:
    cron: "@once"  # 手动触发
---
# 验证指标 (自动化断言)
assertions:
  - metric: "graphql_gateway_error_rate"
    threshold: "< 0.01"  # 熔断降级生效，错误率 < 1%
  - metric: "meeting_detail_page_p99_latency"
    threshold: "< 3000"  # 降级模式下 < 3s
  - metric: "neo4j_cluster_recovery_time"
    threshold: "< 60"    # 集群自愈 < 60s

{
  "type": "record",
  "name": "StandardMeetingEvent",
  "namespace": "com.enterprise.meeting.avro",
  "doc": "标准化会议事件，用于 Kafka Topic: meeting.events.standardized",
  "fields": [
    {"name": "event_id", "type": "string", "doc": "事件唯一ID (UUID v7)"},
    {"name": "event_type", "type": {"type": "enum", "name": "MeetingEventType", "symbols": ["CREATED", "UPDATED", "STARTED", "ENDED", "CANCELLED", "ARTIFACT_GENERATED"]}},
    {"name": "timestamp", "type": {"type": "long", "logicalType": "timestamp-micros"}, "doc": "事件发生时间 UTC"},
    {"name": "source_system", "type": "string", "doc": "来源系统标识"},
    {"name": "meeting", "type": {
      "type": "record", "name": "MeetingCore", "fields": [
        {"name": "meeting_id", "type": "string"},
        {"name": "subject", "type": ["null", "string"], "default": null},
        {"name": "meeting_type", "type": {"type": "enum", "name": "MeetingType", "symbols": ["INSTANT", "SCHEDULED", "RECURRING", "WEBINAR"]}},
        {"name": "start_time", "type": ["null", {"type": "long", "logicalType": "timestamp-micros"}], "default": null},
        {"name": "end_time", "type": ["null", {"type": "long", "logicalType": "timestamp-micros"}], "default": null},
        {"name": "timezone", "type": ["null", "string"], "default": "UTC"},
        {"name": "organizer_id", "type": "string"},
        {"name": "confidentiality", "type": {"type": "enum", "name": "Confidentiality", "symbols": ["PUBLIC", "INTERNAL", "CONFIDENTIAL", "TOP_SECRET"]}},
        {"name": "tags", "type": {"type": "array", "items": "string"}, "default": []}
      ]
    }},
    {"name": "participants", "type": {"type": "array", "items": {
      "type": "record", "name": "ParticipantInfo", "fields": [
        {"name": "user_id", "type": "string"},
        {"name": "display_name", "type": "string"},
        {"name": "role", "type": {"type": "enum", "name": "ParticipantRole", "symbols": ["HOST", "CO_HOST", "PRESENTER", "ATTENDEE", "INTERPRETER", "BOT"]}},
        {"name": "join_time", "type": ["null", {"type": "long", "logicalType": "timestamp-micros"}], "default": null},
        {"name": "leave_time", "type": ["null", {"type": "long", "logicalType": "timestamp-micros"}], "default": null},
        {"name": "is_external", "type": "boolean", "default": false}
      ]
    }}},
    {"name": "artifacts", "type": ["null", {"type": "array", "items": {
      "type": "record", "name": "ArtifactInfo", "fields": [
        {"name": "artifact_id", "type": "string"},
        {"name": "artifact_type", "type": {"type": "enum", "name": "ArtifactType", "symbols": ["RECORDING", "TRANSCRIPT", "MINUTES", "DECISION", "ACTION_ITEM", "WHITEBOARD", "CHAT_LOG", "POLL_RESULT"]}},
        {"name": "storage_url", "type": "string"},
        {"name": "mime_type", "type": "string"},
        {"name": "size_bytes", "type": "long"},
        {"name": "processing_status", "type": {"type": "enum", "name": "ProcessingStatus", "symbols": ["PENDING", "PROCESSING", "COMPLETED", "FAILED"]}}
      ]
    }}, "default": null}
  ],
  "connect.name": "com.enterprise.meeting.avro.StandardMeetingEvent"
}