"""将已采集元数据转换为待审阅 Markdown，仅允许输出到工作区外临时目录。

本程序不修改文档仓库。审阅临时结果后，使用 apply_patch 纳入交付。
"""
import argparse
import collections
import hashlib
import json
import pathlib
import re


def cell(value):
    if value is None:
        return "NULL"
    text = str(value).replace("&", "&amp;").replace("|", "\\|").replace("<", "&lt;").replace(">", "&gt;").replace("{", "&#123;").replace("}", "&#125;").replace("`", "&#96;").replace("\r", "").replace("\n", "<br />")
    # 数据库注释是原始文本；占位地址也不能被 GFM 自动提升为可解析链接。
    text = re.sub(r"\b(https?):(?=//)", r"\1&#58;", text, flags=re.I)
    return text or "（空）"


def table(headers, rows):
    return "\n".join(["| " + " | ".join(headers) + " |", "| " + " | ".join("---" for _ in headers) + " |"] + ["| " + " | ".join(cell(x) for x in row) + " |" for row in rows]) + "\n"


def header(title, slug, snapshot=None):
    text = f"---\ntitle: {title}\nslug: /database-design/catalog/{slug}\n---\n\n# {title}\n\n"
    if snapshot:
        digest = hashlib.sha256(snapshot.read_bytes()).hexdigest()
        data = json.loads(snapshot.read_text(encoding="utf-8"))
        text += f"采集时间：`{data.get('captured_at')}`；来源：只读结构元数据。原始临时快照：`{snapshot.name}`；SHA-256：`{digest}`。\n\n"
    return text


def mysql_pages(path):
    data = json.loads(path.read_text(encoding="utf-8"))
    stem = path.stem
    if data["status"] != "ok":
        return {stem + ".md": header(stem, stem, path) + "采集失败，未核验。\n"}
    tables = data["tables"]
    pages = {}
    index = header(data["database"] + "：MySQL 实测字典", stem, path)
    index += f"服务端版本：`{data['server_version']}`。共 {len(tables)} 个表/视图，{len(data['columns'])} 个字段，{len(data['foreign_keys'])} 条物理外键。\n\n"
    index += "本页记录数据库现状，不自动认定表的维护服务。业务归属与代码差异见对应微服务文档。默认值 NULL 指元数据未提供显式默认值，不替代可空性判断；索引以逐列顺序记录。\n\n"
    groups = {kind: collections.defaultdict(list) for kind in ("columns", "indexes", "constraints", "keys", "foreign_keys")}
    for kind, by_table in groups.items():
        for row in data[kind]:
            by_table[row["table_name"]].append(row)
    inventory = []
    for start in range(0, len(tables), max(len(tables), 1)):
        batch = tables
        content = ""
        for item in batch:
            name = item["table_name"]
            inventory.append((name, item["table_type"], item["table_comment"], f"[字段与约束](#table-{name})"))
            content += f'<span id="table-{name}"></span>\n\n## {name}\n\n类型：{cell(item["table_type"])}；引擎：{cell(item["engine"])}；排序规则：{cell(item["table_collation"])}；说明：{cell(item["table_comment"])}。\n\n'
            columns = []
            for col in groups["columns"][name]:
                default = col["column_default"]
                if default not in (None, "") and re.search(r"password|secret|token|credential", col["column_name"], re.I):
                    default = "[敏感字段默认值不公开]"
                columns.append((col["ordinal_position"], col["column_name"], col["column_type"], col["is_nullable"], default,
                                col["extra"], col["column_comment"], col["collation_name"], col["generation_expression"]))
            content += table(["序号", "列", "实际类型", "可空", "默认值", "附加属性", "数据库注释", "排序规则", "生成表达式"], columns)
            content += "\n索引：\n\n" + table(["索引", "唯一", "序号", "列", "前缀长度", "类型", "注释"], [
                (x["index_name"], "是" if x["non_unique"] == 0 else "否", x["seq_in_index"], x["column_name"], x["sub_part"], x["index_type"], x["index_comment"])
                for x in groups["indexes"][name]])
            content += "\n约束：\n\n" + table(["名称", "类型"], [(x["constraint_name"], x["constraint_type"]) for x in groups["constraints"][name]])
            fk = groups["foreign_keys"][name]
            if fk:
                content += "\n物理外键：\n\n" + table(["名称", "列", "目标库.表.列", "更新规则", "删除规则"], [
                    (x["constraint_name"], k["column_name"], f"{k['referenced_table_schema']}.{k['referenced_table_name']}.{k['referenced_column_name']}", x["update_rule"], x["delete_rule"])
                    for x in fk for k in groups["keys"][name] if k["constraint_name"] == x["constraint_name"]])
            else:
                content += "\n此次元数据未发现物理外键；这不代表没有业务逻辑关联。\n"
            content += "\n"
        pages[stem + ".md"] = content
    # 链接本身不经过 cell 转义，以保留 Markdown 导航。
    index += table(["对象", "类型", "数据库注释", "详情"], inventory)
    pages[stem + ".md"] = index + partition_summary(path) + pages.get(stem + ".md", "")
    return pages


def partition_summary(path):
    supplement = path.parent / "query-layout" / ("mysql-layout-" + path.stem.removeprefix("mysql-") + ".json")
    if not supplement.exists():
        return "\n分区元数据未采集，不能推断是否分区。\n\n"
    data = json.loads(supplement.read_text(encoding="utf-8"))
    content = "\n## 实测分区\n\n"
    content += f"补采时间：`{data.get('captured_at')}`；快照：`{supplement.name}`；SHA-256：`{hashlib.sha256(supplement.read_bytes()).hexdigest()}`。\n\n"
    if data.get("status") != "ok":
        return content + "分区采集失败，未核验。\n\n"
    groups = collections.defaultdict(list)
    for row in data["partitions"]:
        if row["partition_name"] is not None:
            groups[row["table_name"]].append(row)
    if not groups:
        return content + "此次目录未发现有声明分区的表。\n\n"
    rows = []
    for name, partitions in groups.items():
        first = partitions[0]
        descriptions = [x["partition_description"] for x in partitions]
        names = [x["partition_name"] for x in partitions]
        try:
            values = sorted(int(x) for x in descriptions)
            numeric = True
        except (TypeError, ValueError):
            numeric = False
        prefix = names[0][:1]
        if numeric and prefix in ("p", "P") and all(x["partition_name"] == prefix + str(x["partition_description"]) for x in partitions):
            ranges = []
            begin = end = values[0]
            for value in values[1:]:
                if value == end + 1:
                    end = value
                else:
                    ranges.append(str(begin) if begin == end else f"{begin}..{end}")
                    begin = end = value
            ranges.append(str(begin) if begin == end else f"{begin}..{end}")
            boundaries = f"{prefix}N VALUES IN (N)，N=" + ", ".join(ranges) + "；每段逐值连续，段间值不在此分区清单"
        elif first["partition_method"] == "KEY" and all(x is None for x in descriptions) and names == [f"p{i}" for i in range(len(names))]:
            boundaries = f"p0..p{len(names)-1}；KEY 分区无显式 VALUES 边界"
        else:
            boundaries = "; ".join(f"{x['partition_name']}={x['partition_description']}" for x in partitions)
        rows.append((name, len(partitions), first["partition_method"], first["partition_expression"], boundaries))
    content += table(["表", "分区数", "方法", "分区表达式", "分区名与边界（按实测压缩）"], rows)
    return content + "\n分区定义不等于索引；是否裁剪仍取决于实际查询条件及优化器。此次未执行 EXPLAIN 或业务查询。\n\n"


def td_page(path):
    data = json.loads(path.read_text(encoding="utf-8"))
    stem = path.stem
    content = header(data["database"] + "：TDengine 实测字典", stem, path)
    content += f"采集状态：`{data['status']}`。版本元数据：`{json.dumps(data.get('server_version'), ensure_ascii=False)}`。\n\n"
    content += "采集超级表/普通表定义及系统目录子表数量；不导出设备标签值、业务行或逐设备子表名称。子表继承对应超级表结构，动态命名规则见服务文档。零子表不代表超级表不存在。\n\n"
    counts = {x["stable_name"]: x["child_count"] for x in data.get("child_counts", [])}
    content += table(["对象", "类型", "子表数"], [(name, item["kind"], counts.get(name, 0)) for name, item in data.get("objects", {}).items()])
    for name, item in data.get("objects", {}).items():
        content += f"\n## {name}\n\n"
        if "columns" in item:
            rows = item["columns"]
            keys = list(rows[0]) if rows else []
            content += table(keys or ["结果"], [[row.get(key) for key in keys] for row in rows])
        else:
            content += "结构获取失败，未核验。\n"
    return {stem + ".md": content}


def es_pages(path):
    data = json.loads(path.read_text(encoding="utf-8"))
    index = header("Elasticsearch：业务索引、映射与查询参数", "elasticsearch", path)
    index += f"服务端版本：`{data.get('server_version')}`；采集范围为 open 索引 mapping 与 alias，未读取文档数据，未覆盖 closed 索引。\n\n"
    index += "业务索引按完整 mapping 内容分组，相同结构只列一次；月份仍逐一保留以便核对类型漂移。其他系统、监控和归属未明索引仅登记名称，不展开与本次业务无关的字段。字段索引、物理索引名、分片数含义不同，业务条件见[查询与分区指南](../query-performance.md)。\n\n"
    inventory = []
    def fields(properties, prefix=""):
        rows = []
        for key, spec in properties.items():
            full = prefix + key
            options = {k: v for k, v in spec.items() if k not in ("properties", "fields", "_meta")}
            rows.append((full, spec.get("type", "object" if "properties" in spec else "未声明"), json.dumps(options, ensure_ascii=False, sort_keys=True)))
            rows.extend(fields(spec.get("properties", {}), full + "."))
            rows.extend(fields(spec.get("fields", {}), full + "."))
        return rows
    groups = {}
    external = []
    for name, item in sorted(data.get("mappings", {}).items()):
        aliases = data.get("aliases", {}).get(name, {}).get("aliases", {})
        if name not in ("base_device", "apk_push_history") and not name.startswith("app_install_device"):
            external.append((name, ", ".join(aliases) or "无"))
            continue
        mappings = item.get("mappings", {})
        canonical = json.dumps(mappings, ensure_ascii=True, sort_keys=True)
        if canonical not in groups:
            groups[canonical] = {"id": "mapping-" + str(len(groups) + 1), "names": [], "mapping": mappings}
        group = groups[canonical]
        group["names"].append(name)
        inventory.append((name, len(fields(mappings.get("properties", {}))), ", ".join(aliases) or "无", f"[结构](#{group['id']})"))
    index += table(["业务索引", "字段路径数", "别名", "mapping 分组"], inventory)
    for group in groups.values():
        mappings = group["mapping"]
        content = f'\n<span id="{group["id"]}"></span>\n\n## {group["id"]}\n\n'
        content += "适用索引：" + "、".join(f"`{name}`" for name in group["names"]) + "。\n\n"
        for name in group["names"]:
            content += f'<span id="index-{name}"></span>\n\n'
        content += "映射顶层选项：`" + json.dumps({k: v for k, v in mappings.items() if k not in ("properties", "_meta")}, ensure_ascii=False, sort_keys=True).replace("`", "\\`") + "`。\n\n"
        content += table(["字段路径（含 multi-field）", "类型", "实际 mapping 参数"], fields(mappings.get("properties", {})))
        index += content
    settings_path = path.parent / "query-layout/es-query-settings.json"
    index += "\n## 分片、分析器与排序设置\n\n"
    if settings_path.exists():
        settings = json.loads(settings_path.read_text(encoding="utf-8"))
        index += f"补采时间：`{settings.get('captured_at')}`；快照：`{settings_path.name}`；SHA-256：`{hashlib.sha256(settings_path.read_bytes()).hexdigest()}`；状态：`{settings.get('status')}`。\n\n"
        index += table(["索引", "主分片", "副本", "mac_analyzer", "显式 index.sort"], [
            (name, spec.get("index.number_of_shards"), spec.get("index.number_of_replicas"),
             "pattern，分隔符冒号" if spec.get("index.analysis.analyzer.mac_analyzer.pattern") == ":" else "未返回",
             json.dumps({k: v for k, v in spec.items() if k.startswith("index.sort.")}, ensure_ascii=False))
            for name, spec in sorted(settings.get("settings", {}).items())])
        index += "\n本次返回的分配策略为 `index.routing.allocation.include._tier_preference=data_content`，是节点分配设置，不是文档 routing 字段。未返回 `index.sort.*` 不代表请求没有排序。未采集 max_result_window，不能把代码的分页阈值当作服务端实测配置。\n\n"
    else:
        index += "未采集 settings。\n\n"
    index += '<span id="external-scope"></span>\n\n## 其他开放索引：仅登记范围\n\n'
    index += table(["索引", "别名"], external)
    index += "\n`ee_default_alias` 成员既包含业务月份索引，也包含 `document_s1`，且未覆盖全部已存在月份。不能用这个别名直接代表完整业务历史。上方两张清单共同保留所有开放索引的别名成员关系；未导出别名过滤条件和写索引标记。\n"
    return {"elasticsearch.md": index}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", type=pathlib.Path, required=True)
    parser.add_argument("--output", type=pathlib.Path, required=True)
    parser.add_argument("--workspace", type=pathlib.Path, required=True)
    args = parser.parse_args()
    root, output = args.workspace.resolve(), args.output.resolve()
    if output == root or root in output.parents:
        parser.error("输出必须位于工作区之外的临时目录")
    pages = {}
    for path in sorted(args.input.glob("mysql-*.json")):
        pages.update(mysql_pages(path))
    td_content = header("TDengine：各库结构、标签与子表规模", "tdengine")
    td_content += "同页集中记录各库结构，不为日期粒度或设备子表新增文档。业务读写及时间/TAG 过滤见[查询指南](../query-performance.md)。\n\n"
    td_paths = sorted(args.input.glob("tdengine-*.json"))
    td_content += table(["数据库", "结构与标签"], [(path.stem.removeprefix("tdengine-"), f"[跳转](#td-{path.stem.removeprefix('tdengine-')})") for path in td_paths])
    for path in td_paths:
        original = next(iter(td_page(path).values()))
        body = original.split("---\n", 2)[2].strip()
        body = re.sub(r"^(#{1,5}) ", r"\1# ", body, flags=re.M)
        td_content += f'\n<span id="td-{path.stem.removeprefix("tdengine-")}"></span>\n\n{body}\n'
    pages["tdengine.md"] = td_content
    pages.update(es_pages(args.input / "elasticsearch.json"))
    for name, content in pages.items():
        target = output / name
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(content, encoding="utf-8")
    print(json.dumps({"pages": len(pages), "characters": sum(map(len, pages.values()))}))


if __name__ == "__main__":
    main()