yunmei
098125164d
架构核对发现 2 处差距，本次补齐： - Kafka 上行支持 SSL/SASL_SSL 双向 mTLS（8.2 服务间 mTLS 边缘网关↔总线） - 网关日志支持结构化 JSON 输出（NFR 可维护：统一日志规范） - 配置示例与 README 验收口径同步更新
124 lines
5.4 KiB
Python
124 lines
5.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""数据总线命名与分区模板化 —— 换行业只改配置，内核零改动（PRD 5.2 / Issue #4）。
|
||
|
||
模板（template）是行业复制的唯一粒度，所有外部资源命名均由
|
||
「模板名 + 点位维度」推导：
|
||
|
||
- Kafka topic ：`{topic_prefix}.{device_id}.points`（与 edge-gateway 上行一致）；
|
||
分区 = hash(device_id) % num_partitions，保证单设备分区内有序；
|
||
- TDengine ：超级表 `{tpl}_points`、每测点子表 `{tpl}_pt_{point_id}`；
|
||
- PostgreSQL ：schema `tpl_{tpl}`，表名 `{schema}.{table}`；
|
||
- MinIO ：桶 `{template}-artifacts`，对象键 `features/{model_id}/{...}`。
|
||
|
||
命名清洗规则：
|
||
- Kafka topic / MinIO 桶：小写 + 保留 `-` / `_` / `.` 之外替换为 `_`；
|
||
- SQL 标识符（TDengine / PostgreSQL）：额外把 `-` 替换为 `_`，避免引号转义。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import re
|
||
from typing import List, Optional
|
||
|
||
# 保留字符集（Kafka topic / MinIO 桶名均允许小写字母、数字、- _ .）
|
||
_KEEP = re.compile(r"[^a-z0-9_.-]+")
|
||
# SQL 标识符额外排除 `-`（TDengine / PostgreSQL 不加引号时不允许）
|
||
_SQL_KEEP = re.compile(r"[^a-z0-9_.]+")
|
||
|
||
|
||
def sanitize(name: str) -> str:
|
||
"""通用命名清洗：小写 + 非安全字符替换为 `_`。"""
|
||
s = (name or "").strip().lower()
|
||
s = _KEEP.sub("_", s)
|
||
s = s.strip("._") or "tpl"
|
||
return s
|
||
|
||
|
||
def sanitize_sql(name: str) -> str:
|
||
"""SQL 标识符清洗：`-` 与其余非安全字符替换为 `_`。"""
|
||
s = sanitize(name)
|
||
s = _SQL_KEEP.sub("_", s)
|
||
return s.strip("._") or "tpl"
|
||
|
||
|
||
def _sql_str(value: str) -> str:
|
||
"""SQL 字符串字面量转义（单引号加倍）。"""
|
||
return "'" + str(value).replace("'", "''") + "'"
|
||
|
||
|
||
class TemplateNaming:
|
||
"""按模板推导全部外部资源命名（Kafka / TDengine / PostgreSQL / MinIO）。"""
|
||
|
||
def __init__(
|
||
self,
|
||
template: str,
|
||
topic_prefix: Optional[str] = None,
|
||
num_partitions: int = 12,
|
||
bucket_suffix: str = "artifacts",
|
||
):
|
||
self.template = sanitize(template)
|
||
# SQL 标识符用下划线形态（避免 `-` 需加引号）
|
||
self.tpl_sql = sanitize_sql(self.template)
|
||
self.topic_prefix = sanitize(topic_prefix) if topic_prefix else self.template
|
||
self.num_partitions = max(1, int(num_partitions))
|
||
self.bucket_suffix = sanitize_sql(bucket_suffix)
|
||
|
||
# ------------------------------------------------------------------
|
||
# Kafka：topic 命名 / 分区策略（子任务 #28）
|
||
# ------------------------------------------------------------------
|
||
def topic(self, device_id: str) -> str:
|
||
"""上行 topic：`{topic_prefix}.{device_id}.points`（对齐 edge-gateway）。"""
|
||
return f"{self.topic_prefix}.{sanitize(device_id)}.points"
|
||
|
||
def partition(self, device_id: str, num_partitions: Optional[int] = None) -> int:
|
||
"""分区策略：按 device_id 一致性哈希 → 单设备分区内严格有序。"""
|
||
n = num_partitions or self.num_partitions
|
||
digest = hashlib.md5(sanitize(device_id).encode("utf-8")).hexdigest()
|
||
return int(digest[:8], 16) % n
|
||
|
||
def partitions(self, device_ids: List[str], num_partitions: Optional[int] = None) -> dict:
|
||
"""设备 → 分区映射（模板配置台预览用）。"""
|
||
n = num_partitions or self.num_partitions
|
||
return {d: self.partition(d, n) for d in device_ids}
|
||
|
||
# ------------------------------------------------------------------
|
||
# TDengine：超级表 + 每测点子表（子任务 #29）
|
||
# ------------------------------------------------------------------
|
||
def stable(self) -> str:
|
||
"""时序超级表：`{tpl}_points`。"""
|
||
return f"{self.tpl_sql}_points"
|
||
|
||
def subtable(self, point_id: str) -> str:
|
||
"""测点子表：`{tpl}_pt_{point_id}`（点位维度，字典驱动自动生成）。"""
|
||
return f"{self.tpl_sql}_pt_{sanitize_sql(point_id)}"
|
||
|
||
# ------------------------------------------------------------------
|
||
# PostgreSQL：schema 与表（子任务 #30）
|
||
# ------------------------------------------------------------------
|
||
def pg_schema(self) -> str:
|
||
"""关系 schema：`tpl_{tpl}`。"""
|
||
return f"tpl_{self.tpl_sql}"
|
||
|
||
def pg_table(self, table: str) -> str:
|
||
"""`{schema}.{table}` 限定名。"""
|
||
return f"{self.pg_schema()}.{sanitize_sql(table)}"
|
||
|
||
# ------------------------------------------------------------------
|
||
# MinIO：对象桶与对象键（子任务 #31）
|
||
# ------------------------------------------------------------------
|
||
def bucket(self) -> str:
|
||
"""对象存储桶：`{template}-artifacts`（S3 桶名允许 `-`）。"""
|
||
return f"{self.template}-{self.bucket_suffix}"
|
||
|
||
def snapshot_key(self, model_id: str, date: str, seq: int) -> str:
|
||
"""特征快照对象键：`features/{model_id}/{date}/{seq:06d}.jsonl`。"""
|
||
return f"features/{sanitize_sql(model_id)}/{date}/{int(seq):06d}.jsonl"
|
||
|
||
def model_artifact_key(self, model_id: str, version: str) -> str:
|
||
"""模型 artifact 对象键：`models/{model_id}/{version}/model.bin`。"""
|
||
return f"models/{sanitize_sql(model_id)}/{sanitize_sql(version)}/model.bin"
|
||
|
||
# ------------------------------------------------------------------
|
||
def _sql_str(self, value: str) -> str:
|
||
return _sql_str(value)
|