|
|
@@ -64,7 +64,11 @@ class SQLGuard:
|
|
|
|
|
|
@staticmethod
|
|
|
def validate_product_efficiency_contract(
|
|
|
- sql: str, data_mode: str | None, *, bucketed: bool = True
|
|
|
+ sql: str,
|
|
|
+ data_mode: str | None,
|
|
|
+ *,
|
|
|
+ bucketed: bool = True,
|
|
|
+ version: str | None = None,
|
|
|
) -> None:
|
|
|
if data_mode not in {"offline", "realtime"}:
|
|
|
return
|
|
|
@@ -92,10 +96,25 @@ class SQLGuard:
|
|
|
)
|
|
|
|
|
|
if data_mode == "realtime":
|
|
|
+ video_tables = [
|
|
|
+ table
|
|
|
+ for table in statement.find_all(exp.Table)
|
|
|
+ if table.name.lower()
|
|
|
+ in {"video_action_log_per5min", "video_action_log_flow"}
|
|
|
+ ]
|
|
|
+ if (version or "").lower() == "all" and any(
|
|
|
+ table.name.lower() == "video_action_log_flow" for table in video_tables
|
|
|
+ ):
|
|
|
+ raise SQLValidationError(
|
|
|
+ "实时全版本产品效率视频行为必须优先使用 "
|
|
|
+ "loghubods.video_action_log_per5min"
|
|
|
+ )
|
|
|
+
|
|
|
invalid_sources: list[str] = []
|
|
|
for table in statement.find_all(exp.Table):
|
|
|
if table.name.lower() not in {
|
|
|
"useractive_log_per5min",
|
|
|
+ "video_action_log_per5min",
|
|
|
"user_share_log_per5min",
|
|
|
}:
|
|
|
continue
|
|
|
@@ -108,8 +127,23 @@ class SQLGuard:
|
|
|
f"实时表 {names} 必须使用 dt LIKE 'yyyyMMdd%' 查询当天累计数据"
|
|
|
)
|
|
|
|
|
|
+ for table in video_tables:
|
|
|
+ if table.name.lower() != "video_action_log_flow":
|
|
|
+ continue
|
|
|
+ select = table.find_ancestor(exp.Select)
|
|
|
+ if select is None or not SQLGuard._has_valid_flow_partitions(select, table):
|
|
|
+ raise SQLValidationError(
|
|
|
+ "实时 video_action_log_flow 必须使用精确分区 "
|
|
|
+ "year='yyyy'、month='MM'、dt='DD'；hh 如出现必须为 'HH'，"
|
|
|
+ "其中 dt 只能是两位日，不能写 yyyyMMdd"
|
|
|
+ )
|
|
|
+
|
|
|
for table in statement.find_all(exp.Table):
|
|
|
- if table.name.lower() not in {"video_action_log_applet", "video_action_log_flow"}:
|
|
|
+ if table.name.lower() not in {
|
|
|
+ "video_action_log_applet",
|
|
|
+ "video_action_log_flow",
|
|
|
+ "video_action_log_per5min",
|
|
|
+ }:
|
|
|
continue
|
|
|
select = table.find_ancestor(exp.Select)
|
|
|
if select is None:
|
|
|
@@ -202,3 +236,81 @@ class SQLGuard:
|
|
|
if isinstance(pattern, exp.Literal) and re.fullmatch(r"\d{8}%", str(pattern.this)):
|
|
|
return True
|
|
|
return False
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _has_valid_flow_partitions(select: exp.Select, table: exp.Table) -> bool:
|
|
|
+ validators = {
|
|
|
+ "year": lambda value: bool(re.fullmatch(r"\d{4}", value)),
|
|
|
+ "month": lambda value: bool(re.fullmatch(r"0[1-9]|1[0-2]", value)),
|
|
|
+ "dt": lambda value: bool(re.fullmatch(r"0[1-9]|[12]\d|3[01]", value)),
|
|
|
+ }
|
|
|
+ for name, validator in validators.items():
|
|
|
+ seen, values, unsupported = SQLGuard._partition_literal_filters(
|
|
|
+ select, table, name
|
|
|
+ )
|
|
|
+ if not seen or unsupported or len(values) != 1 or not validator(values[0]):
|
|
|
+ return False
|
|
|
+
|
|
|
+ seen_hh, hour_values, unsupported_hh = SQLGuard._partition_literal_filters(
|
|
|
+ select, table, "hh"
|
|
|
+ )
|
|
|
+ if not seen_hh:
|
|
|
+ return True
|
|
|
+ if unsupported_hh or not hour_values:
|
|
|
+ return False
|
|
|
+ return all(re.fullmatch(r"[01]\d|2[0-3]", value) for value in hour_values)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _partition_literal_filters(
|
|
|
+ select: exp.Select, table: exp.Table, column_name: str
|
|
|
+ ) -> tuple[bool, list[str], bool]:
|
|
|
+ sources = [
|
|
|
+ candidate
|
|
|
+ for candidate in select.find_all(exp.Table)
|
|
|
+ if candidate.find_ancestor(exp.Select) is select
|
|
|
+ ]
|
|
|
+ allow_unqualified = len(sources) == 1
|
|
|
+ qualifier = table.alias_or_name.lower()
|
|
|
+ predicates: list[exp.Expression] = [
|
|
|
+ where.this
|
|
|
+ for where in select.find_all(exp.Where)
|
|
|
+ if where.find_ancestor(exp.Select) is select
|
|
|
+ ]
|
|
|
+ predicates.extend(
|
|
|
+ join.args["on"]
|
|
|
+ for join in select.find_all(exp.Join)
|
|
|
+ if join.find_ancestor(exp.Select) is select and join.args.get("on") is not None
|
|
|
+ )
|
|
|
+
|
|
|
+ seen = False
|
|
|
+ values: list[str] = []
|
|
|
+ unsupported = False
|
|
|
+ for predicate in predicates:
|
|
|
+ for column in predicate.find_all(exp.Column):
|
|
|
+ if column.name.lower() != column_name:
|
|
|
+ continue
|
|
|
+ column_qualifier = column.table.lower() if column.table else ""
|
|
|
+ if column_qualifier != qualifier and not (
|
|
|
+ not column_qualifier and allow_unqualified
|
|
|
+ ):
|
|
|
+ continue
|
|
|
+ seen = True
|
|
|
+ parent = column.parent
|
|
|
+ if isinstance(parent, exp.EQ):
|
|
|
+ other = parent.expression if parent.this is column else parent.this
|
|
|
+ if isinstance(other, exp.Literal) and other.is_string:
|
|
|
+ values.append(str(other.this))
|
|
|
+ else:
|
|
|
+ unsupported = True
|
|
|
+ elif isinstance(parent, exp.In) and parent.this is column:
|
|
|
+ expressions = parent.expressions
|
|
|
+ if expressions and all(
|
|
|
+ isinstance(item, exp.Literal) and item.is_string
|
|
|
+ for item in expressions
|
|
|
+ ):
|
|
|
+ values.extend(str(item.this) for item in expressions)
|
|
|
+ else:
|
|
|
+ unsupported = True
|
|
|
+ else:
|
|
|
+ unsupported = True
|
|
|
+ return seen, values, unsupported
|
刘立冬