341 lines
14 KiB
Python
341 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
从源代码和 DDL 中提取 API → ODS → DWD 数据流映射，生成 Markdown 文档。
|
||
用法: python scripts/ops/gen_dataflow_doc.py
|
||
输出: $FULL_DATAFLOW_DOC_ROOT/dataflow_api_ods_dwd.md（由 .env 配置）
|
||
"""
|
||
import re
|
||
import ast
|
||
import sys
|
||
import os
|
||
from pathlib import Path
|
||
from collections import OrderedDict
|
||
|
||
ROOT = Path(__file__).resolve().parents[2]
|
||
ETL = ROOT / "apps" / "etl" / "pipelines" / "feiqiu"
|
||
DB = ROOT / "db" / "etl_feiqiu" / "schemas"
|
||
from _env_paths import get_output_path as _get_path
|
||
OUT = _get_path("FULL_DATAFLOW_DOC_ROOT") / "dataflow_api_ods_dwd.md"
|
||
|
||
|
||
# ── 1. 从 DDL 解析表结构 ──────────────────────────────────────────
|
||
def parse_ddl_tables(sql_path: Path, schema: str) -> dict[str, list[dict]]:
|
||
"""解析 CREATE TABLE 语句，返回 {schema.table: [{col, type}, ...]}"""
|
||
text = sql_path.read_text(encoding="utf-8")
|
||
tables: dict[str, list[dict]] = {}
|
||
# 匹配 CREATE TABLE IF NOT EXISTS table_name (...)
|
||
pattern = re.compile(
|
||
r"CREATE\s+TABLE\s+IF\s+NOT\s+EXISTS\s+"
|
||
r"(?:(\w+)\.)?(\w+)\s*\((.*?)\)\s*;",
|
||
re.DOTALL | re.IGNORECASE,
|
||
)
|
||
for m in pattern.finditer(text):
|
||
s = m.group(1) or schema
|
||
tname = m.group(2)
|
||
body = m.group(3)
|
||
cols = []
|
||
for line in body.split("\n"):
|
||
line = line.strip().rstrip(",")
|
||
if not line or line.upper().startswith("PRIMARY") or line.startswith("--"):
|
||
continue
|
||
# 跳过约束行
|
||
if re.match(r"^(CONSTRAINT|UNIQUE|CHECK|FOREIGN|EXCLUDE)\b", line, re.I):
|
||
continue
|
||
parts = line.split()
|
||
if len(parts) >= 2:
|
||
col_name = parts[0].strip('"')
|
||
col_type = parts[1]
|
||
# 合并类型修饰符
|
||
if len(parts) > 2 and parts[2].startswith("("):
|
||
col_type += parts[2]
|
||
cols.append({"col": col_name, "type": col_type})
|
||
full = f"{s}.{tname}"
|
||
tables[full] = cols
|
||
return tables
|
||
|
||
|
||
# ── 2. 从 Python 源码解析 TABLE_MAP ──────────────────────────────
|
||
def parse_table_map(py_path: Path) -> dict[str, str]:
|
||
"""解析 TABLE_MAP: dict[str, str] = {...}"""
|
||
text = py_path.read_text(encoding="utf-8")
|
||
# 找到 TABLE_MAP 字典
|
||
match = re.search(
|
||
r"TABLE_MAP\s*(?::\s*dict\[.*?\])?\s*=\s*\{(.*?)\}",
|
||
text,
|
||
re.DOTALL,
|
||
)
|
||
if not match:
|
||
return {}
|
||
body = match.group(1)
|
||
result = {}
|
||
for m in re.finditer(r'"([^"]+)"\s*:\s*"([^"]+)"', body):
|
||
result[m.group(1)] = m.group(2)
|
||
return result
|
||
|
||
|
||
# ── 3. 从 Python 源码解析 FACT_MAPPINGS ──────────────────────────
|
||
def parse_fact_mappings(py_path: Path) -> dict[str, list[tuple]]:
|
||
"""解析 FACT_MAPPINGS 字典，返回 {dwd_table: [(dwd_col, ods_expr, cast), ...]}"""
|
||
text = py_path.read_text(encoding="utf-8")
|
||
# 找到 FACT_MAPPINGS 块
|
||
start = text.find("FACT_MAPPINGS")
|
||
if start < 0:
|
||
return {}
|
||
# 找到第一个 { 后的内容
|
||
brace_start = text.find("{", start)
|
||
if brace_start < 0:
|
||
return {}
|
||
|
||
# 手动匹配大括号
|
||
depth = 0
|
||
end = brace_start
|
||
for i in range(brace_start, len(text)):
|
||
if text[i] == "{":
|
||
depth += 1
|
||
elif text[i] == "}":
|
||
depth -= 1
|
||
if depth == 0:
|
||
end = i + 1
|
||
break
|
||
|
||
block = text[brace_start:end]
|
||
result = {}
|
||
# 匹配每个表的映射列表
|
||
table_pattern = re.compile(r'"([^"]+)"\s*:\s*\[', re.DOTALL)
|
||
for tm in table_pattern.finditer(block):
|
||
table_name = tm.group(1)
|
||
list_start = tm.end()
|
||
# 找到对应的 ]
|
||
bracket_depth = 1
|
||
list_end = list_start
|
||
for i in range(list_start, len(block)):
|
||
if block[i] == "[":
|
||
bracket_depth += 1
|
||
elif block[i] == "]":
|
||
bracket_depth -= 1
|
||
if bracket_depth == 0:
|
||
list_end = i
|
||
break
|
||
list_body = block[list_start:list_end]
|
||
# 匹配 (dwd_col, ods_expr, cast|None)
|
||
tuples = []
|
||
tuple_pattern = re.compile(
|
||
r'\(\s*"([^"]+)"\s*,\s*"([^"]+)"\s*,\s*(?:"([^"]+)"|None)\s*\)'
|
||
)
|
||
for tp in tuple_pattern.finditer(list_body):
|
||
tuples.append((tp.group(1), tp.group(2), tp.group(3)))
|
||
result[table_name] = tuples
|
||
return result
|
||
|
||
|
||
# ── 4. 从 Python 源码解析 ODS_TASK_SPECS ─────────────────────────
|
||
def parse_ods_specs(py_path: Path) -> list[dict]:
|
||
"""解析 ODS_TASK_SPECS，提取 code, table_name, endpoint, list_key, description"""
|
||
text = py_path.read_text(encoding="utf-8")
|
||
specs = []
|
||
# 匹配每个 OdsTaskSpec(...)
|
||
pattern = re.compile(r"OdsTaskSpec\s*\((.*?)\)\s*,", re.DOTALL)
|
||
for m in pattern.finditer(text):
|
||
body = m.group(1)
|
||
spec = {}
|
||
for key in ("code", "table_name", "endpoint", "list_key", "description"):
|
||
km = re.search(rf'{key}\s*=\s*"([^"]*)"', body)
|
||
if km:
|
||
spec[key] = km.group(1)
|
||
if "code" in spec:
|
||
specs.append(spec)
|
||
return specs
|
||
|
||
|
||
# ── 5. 生成文档 ──────────────────────────────────────────────────
|
||
def generate_doc():
|
||
ods_ddl = parse_ddl_tables(DB / "ods.sql", "ods")
|
||
dwd_ddl = parse_ddl_tables(DB / "dwd.sql", "dwd")
|
||
|
||
dwd_task_py = ETL / "tasks" / "dwd" / "dwd_load_task.py"
|
||
table_map = parse_table_map(dwd_task_py)
|
||
fact_mappings = parse_fact_mappings(dwd_task_py)
|
||
|
||
ods_specs = parse_ods_specs(ETL / "tasks" / "ods" / "ods_tasks.py")
|
||
# ODS 表 → API 端点映射
|
||
ods_to_api: dict[str, dict] = {}
|
||
for spec in ods_specs:
|
||
tn = spec.get("table_name", "")
|
||
ods_to_api[tn] = spec
|
||
|
||
lines = []
|
||
lines.append("# API → ODS → DWD 数据流对比文档")
|
||
lines.append("")
|
||
lines.append("> 自动生成于 `scripts/ops/gen_dataflow_doc.py`，基于 DDL 和 ETL 源码解析。")
|
||
lines.append("")
|
||
lines.append("## 概览")
|
||
lines.append("")
|
||
lines.append(f"- ODS 表数量: {len(ods_ddl)}")
|
||
lines.append(f"- DWD 表数量: {len(dwd_ddl)}")
|
||
lines.append(f"- TABLE_MAP 映射条目: {len(table_map)}")
|
||
lines.append(f"- ODS 任务数量: {len(ods_specs)}")
|
||
lines.append("")
|
||
|
||
# ── 按 ODS 表分组 ──
|
||
# 先建立 ODS 表 → DWD 表列表的反向映射
|
||
ods_to_dwd: dict[str, list[str]] = {}
|
||
for dwd_t, ods_t in table_map.items():
|
||
ods_to_dwd.setdefault(ods_t, []).append(dwd_t)
|
||
|
||
# 收集所有涉及的 ODS 表（去重、排序）
|
||
all_ods = sorted(set(list(ods_to_dwd.keys()) + [s.get("table_name", "") for s in ods_specs]))
|
||
|
||
lines.append("## 目录")
|
||
lines.append("")
|
||
for i, ods_t in enumerate(all_ods, 1):
|
||
anchor = ods_t.replace(".", "").replace("_", "-")
|
||
short = ods_t.split(".")[-1] if "." in ods_t else ods_t
|
||
lines.append(f"{i}. [{short}](#{anchor})")
|
||
lines.append("")
|
||
lines.append("---")
|
||
lines.append("")
|
||
|
||
# ── 逐表详情 ──
|
||
for ods_t in all_ods:
|
||
short = ods_t.split(".")[-1] if "." in ods_t else ods_t
|
||
lines.append(f"## {short}")
|
||
lines.append("")
|
||
|
||
# API 信息
|
||
api_info = ods_to_api.get(ods_t, {})
|
||
if api_info:
|
||
lines.append("### API 端点")
|
||
lines.append("")
|
||
lines.append(f"- 任务编码: `{api_info.get('code', 'N/A')}`")
|
||
lines.append(f"- 端点: `{api_info.get('endpoint', 'N/A')}`")
|
||
lk = api_info.get("list_key")
|
||
if lk:
|
||
lines.append(f"- 数据路径: `data.{lk}`")
|
||
desc = api_info.get("description", "")
|
||
if desc:
|
||
lines.append(f"- 说明: {desc}")
|
||
lines.append("")
|
||
|
||
# ODS 表字段
|
||
ods_cols = ods_ddl.get(ods_t, [])
|
||
if ods_cols:
|
||
lines.append(f"### ODS 表: `{ods_t}` ({len(ods_cols)} 列)")
|
||
lines.append("")
|
||
lines.append("| # | 列名 | 类型 |")
|
||
lines.append("|---|------|------|")
|
||
for idx, c in enumerate(ods_cols, 1):
|
||
lines.append(f"| {idx} | `{c['col']}` | {c['type']} |")
|
||
lines.append("")
|
||
|
||
# DWD 表
|
||
dwd_tables = ods_to_dwd.get(ods_t, [])
|
||
if dwd_tables:
|
||
for dwd_t in sorted(dwd_tables):
|
||
dwd_cols = dwd_ddl.get(dwd_t, [])
|
||
is_dim = "dim_" in dwd_t
|
||
is_ex = dwd_t.endswith("_ex")
|
||
table_type = "维度" if is_dim else "事实"
|
||
if is_ex:
|
||
table_type += "（扩展）"
|
||
|
||
mappings = fact_mappings.get(dwd_t, [])
|
||
|
||
lines.append(f"### DWD 表: `{dwd_t}` — {table_type} ({len(dwd_cols)} 列)")
|
||
lines.append("")
|
||
|
||
# 字段对比表
|
||
lines.append("| # | DWD 列名 | DWD 类型 | ODS 来源表达式 | 转换 | 备注 |")
|
||
lines.append("|---|----------|----------|----------------|------|------|")
|
||
|
||
# 建立映射查找
|
||
mapping_dict = {m[0]: (m[1], m[2]) for m in mappings}
|
||
|
||
for idx, c in enumerate(dwd_cols, 1):
|
||
col_name = c["col"]
|
||
col_type = c["type"]
|
||
|
||
# SCD2 列
|
||
scd2_cols = {"scd2_start_time", "scd2_end_time", "scd2_is_current", "scd2_version"}
|
||
if col_name.lower().replace("scd2_", "scd2_") in scd2_cols or col_name.lower() in scd2_cols:
|
||
lines.append(f"| {idx} | `{col_name}` | {col_type} | — | — | DWD 慢变元数据 |")
|
||
continue
|
||
|
||
if col_name in mapping_dict:
|
||
ods_expr, cast = mapping_dict[col_name]
|
||
cast_str = f"CAST → {cast}" if cast else "直接映射"
|
||
# 判断是否为 JSONB 提取
|
||
note = ""
|
||
if "->>" in ods_expr:
|
||
note = "JSONB 提取"
|
||
elif "CASE" in ods_expr.upper():
|
||
note = "派生计算"
|
||
elif ods_expr != col_name:
|
||
note = "字段重命名"
|
||
lines.append(f"| {idx} | `{col_name}` | {col_type} | `{ods_expr}` | {cast_str} | {note} |")
|
||
else:
|
||
# 同名直传
|
||
ods_col_names = {oc["col"].lower() for oc in ods_cols}
|
||
if col_name.lower() in ods_col_names:
|
||
lines.append(f"| {idx} | `{col_name}` | {col_type} | `{col_name}` | 直接映射 | 同名直传 |")
|
||
else:
|
||
lines.append(f"| {idx} | `{col_name}` | {col_type} | — | — | 未在 FACT_MAPPINGS 中显式映射 |")
|
||
|
||
lines.append("")
|
||
else:
|
||
lines.append(f"*该 ODS 表暂无 DWD 映射（仅用于 DWS 或其他下游）*")
|
||
lines.append("")
|
||
|
||
lines.append("---")
|
||
lines.append("")
|
||
|
||
# ── 附录：ETL 元数据列说明 ──
|
||
lines.append("## 附录：ETL 元数据列")
|
||
lines.append("")
|
||
lines.append("所有 ODS 表均包含以下 ETL 元数据列，不映射到 DWD：")
|
||
lines.append("")
|
||
lines.append("| 列名 | 类型 | 说明 |")
|
||
lines.append("|------|------|------|")
|
||
lines.append("| `content_hash` | TEXT | 记录内容哈希，用于去重和变更检测 |")
|
||
lines.append("| `source_file` | TEXT | 原始导出文件名，用于数据追溯 |")
|
||
lines.append("| `source_endpoint` | TEXT | 采集来源接口/文件路径 |")
|
||
lines.append("| `fetched_at` | TIMESTAMPTZ | 采集/入库时间戳 |")
|
||
lines.append("| `payload` | JSONB | 完整原始 JSON 记录快照 |")
|
||
lines.append("")
|
||
|
||
lines.append("## 附录：DWD 维度表 SCD2 列")
|
||
lines.append("")
|
||
lines.append("所有 DWD 维度表（`dim_*`）均包含以下 SCD2 慢变维度列：")
|
||
lines.append("")
|
||
lines.append("| 列名 | 类型 | 说明 |")
|
||
lines.append("|------|------|------|")
|
||
lines.append("| `scd2_start_time` | TIMESTAMPTZ | 版本生效起点 |")
|
||
lines.append("| `scd2_end_time` | TIMESTAMPTZ | 版本失效时间（9999-12-31 = 当前） |")
|
||
lines.append("| `scd2_is_current` | INT | 当前版本标记（1=当前，0=历史） |")
|
||
lines.append("| `scd2_version` | INT | 版本号（自增） |")
|
||
lines.append("")
|
||
|
||
lines.append("## 附录：DWD 事实表增量策略")
|
||
lines.append("")
|
||
lines.append("事实表按时间窗口增量写入，优先使用以下业务时间列进行过滤（按优先级排序）：")
|
||
lines.append("")
|
||
lines.append("1. `pay_time` — 支付时间")
|
||
lines.append("2. `create_time` — 创建时间")
|
||
lines.append("3. `update_time` — 更新时间")
|
||
lines.append("4. `occur_time` — 发生时间")
|
||
lines.append("5. `settle_time` — 结算时间")
|
||
lines.append("6. `start_use_time` — 开始使用时间")
|
||
lines.append("7. `fetched_at` — 入库时间（兜底）")
|
||
lines.append("")
|
||
|
||
# 写入文件
|
||
OUT.parent.mkdir(parents=True, exist_ok=True)
|
||
OUT.write_text("\n".join(lines), encoding="utf-8")
|
||
print(f"文档已生成: {OUT}")
|
||
print(f" ODS 表: {len(ods_ddl)}, DWD 表: {len(dwd_ddl)}")
|
||
print(f" TABLE_MAP: {len(table_map)} 条, FACT_MAPPINGS: {len(fact_mappings)} 条")
|
||
print(f" ODS 任务: {len(ods_specs)} 个")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
generate_doc()
|