tiku/tiku-server/app/services/knowledge.py

"""知识点树解析 + 编码生成模块"""

import json
import re
from pathlib import Path

from app.database import get_connection, sync_knowledge_points


def parse_knowledge_tree(tree_path: str | Path) -> dict:
    """解析 5 层 Markdown 知识点树，返回 code → {path, leaf_name, chapter_code, chapter_name} 映射

    层级: # (章) → ## (节) → ### (小节) → #### (项) → - (叶子)
    编码: 每层两位数字，各层独立，用 . 连接
    示例: 01.01.03.02.01
    """
    tree_path = Path(tree_path)
    if not tree_path.exists():
        print(f"⚠ 知识点树文件不存在: {tree_path}")
        return {}

    with open(tree_path, "r", encoding="utf-8") as f:
        content = f.read()

    lines = content.split("\n")
    knowledge_map: dict[str, dict] = {}

    # 层级计数器，每层独立（5 层编码）
    counters = [0, 0, 0, 0, 0]  # 章, 节, 小节, 项, 叶子
    chapter_code = ""
    chapter_name = ""
    current_path_parts = ["", "", "", "", ""]

    # 标题行: # → ## → ### → #### (1-4 级)
    heading_re = re.compile(r"^(#{1,4})\s+(.+)$")
    # 叶子行: - xxx（先 strip 后匹配）
    leaf_re = re.compile(r"^-\s+(.+)$")

    for line in lines:
        stripped = line.strip()
        if not stripped:
            continue

        match = heading_re.match(stripped)
        leaf_match = leaf_re.match(stripped)

        if match:
            level = len(match.group(1))
            title = match.group(2).strip()
            idx = level - 1  # 0-based

            if idx < 4:
                counters[idx] += 1
                # 重置下层计数器（包括叶子层）
                for i in range(idx + 1, 5):
                    counters[i] = 0

                current_path_parts[idx] = title
                for i in range(idx + 1, 5):
                    current_path_parts[i] = ""

                if idx == 0:
                    chapter_code = f"{counters[0]:02d}"
                    chapter_name = title

        elif leaf_match:
            leaf_name = leaf_match.group(1).strip()
            counters[4] += 1  # 叶子层独立计数

            code_parts = [f"{counters[i]:02d}" for i in range(5)]
            code = ".".join(code_parts)

            current_path_parts[4] = leaf_name
            path_parts = [p for p in current_path_parts if p]
            path = " > ".join(path_parts)

            knowledge_map[code] = {
                "path": path,
                "leaf_name": leaf_name,
                "chapter_code": chapter_code,
                "chapter_name": chapter_name,
            }

    return knowledge_map


def get_chapter_from_code(code: str, knowledge_map: dict) -> str:
    """从编码反推章节名，如 '01.01.03.02.01' → '01-集合与常用逻辑用语'"""
    # 取前两位作为章编码
    chapter_code = ".".join(code.split(".")[:2])  # 前两层编码
    for c, info in knowledge_map.items():
        if c.startswith(chapter_code[:2] + "."):
            ch_code = info.get("chapter_code", "")
            ch_name = info.get("chapter_name", "")
            if ch_code and ch_name:
                return f"{ch_code}-{ch_name}"
    return ""


def find_code_by_name(name: str, knowledge_map: dict) -> str | None:
    """根据知识点名称查找编码"""
    name = name.strip()
    for code, info in knowledge_map.items():
        if info["leaf_name"] == name:
            return code
    return None


def find_code_by_name_fuzzy(name: str, knowledge_map: dict) -> str | None:
    """模糊匹配知识点名称"""
    name = name.strip()
    # 精确匹配
    for code, info in knowledge_map.items():
        if info["leaf_name"] == name:
            return code
    # 包含匹配
    for code, info in knowledge_map.items():
        if name in info["leaf_name"] or info["leaf_name"] in name:
            return code
    return None


def build_knowledge(config):
    """全流程：解析知识点树 → 存 JSON → 同步数据库"""
    from app.config import Config
    cfg = config if isinstance(config, Config) else config

    print("解析知识点树...")
    knowledge_map = parse_knowledge_tree(cfg.knowledge_tree_path)

    if not knowledge_map:
        print("❌ 知识点树解析失败")
        return

    # 存 JSON
    import json
    cfg.knowledge_map_path.parent.mkdir(parents=True, exist_ok=True)
    with open(cfg.knowledge_map_path, "w", encoding="utf-8") as f:
        json.dump(knowledge_map, f, ensure_ascii=False, indent=2)

    print(f"✓ knowledge_map.json 已生成 ({len(knowledge_map)} 个叶子)")

    # 同步数据库（先确保数据库已初始化）
    from app.database import init_db
    init_db(cfg.db_path)
    conn = get_connection(cfg.db_path)
    try:
        sync_knowledge_points(conn, knowledge_map)
        print(f"✓ 知识点已同步到数据库")
    finally:
        conn.close()


def load_knowledge_map(config) -> dict:
    """加载已有的 knowledge_map.json"""
    import json
    map_path = config.knowledge_map_path
    if map_path.exists():
        with open(map_path, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}