tree.py

import logging
from genericpath import exists
import json
import os
import uuid
import sys
import re

id_set = set()

logger = logging.getLogger(__name__)


def load_json(p):
    with open(p, 'r') as f:
        return json.loads(f.read())


def dump_json(p, j, exist_ok=False, override=False):
    if os.path.exists(p):
        if exist_ok:
            if not override:
                return
        else:
            logger.error(f"{p} already exist")
            sys.exit(0)

    with open(p, 'w+') as f:
        f.write(json.dumps(j, indent=2, ensure_ascii=False))


def ensure_config(path):
    config_path = os.path.join(path, "config.json")
    if not os.path.exists(config_path):
        node = {"keywords": []}
        dump_json(config_path, node, exist_ok=True, override=False)
        return node
    else:
        return load_json(config_path)

def parse_no_name(d):
    p = r'(\d+)\.(.*)'
    m = re.search(p, d)

    try:
        no = int(m.group(1))
        dir_name = m.group(2)
    except:
        sys.exit(0)

    return no, dir_name


def check_export(base, cfg):
    flag = False
    exports = []
    for export in cfg.get('export', []):
        ecfg_path = os.path.join(base, export)
        if os.path.exists(ecfg_path):
            exports.append(export)
        else:
            flag = True
    if flag:
        cfg["export"] = exports
    return flag


def gen_tree(data_path):
    root = {}

    def gen_node_id():
        # return ''.join(str(uuid.uuid5(uuid.NAMESPACE_URL, 'skill_tree')).split('-'))
        return "oceanbase-" + uuid.uuid4().hex

    def list_dir(p):
        v = os.listdir(p)
        v.sort()
        for no_name in v:
            no_dir = os.path.join(p, no_name)
            if os.path.isdir(no_dir):
                yield no_dir, no_name

    def ensure_id_helper(node):
        flag = False

        if (node.get('node_id') is None) or node.get('node_id') in id_set:
            node['node_id'] = gen_node_id()
            flag = True

        id_set.add(node['node_id'])

        if 'children' in node:
            for c in node["children"]:
                flag = flag or ensure_id_helper(list(c.values())[0])

        return flag

    def ensure_node_id(cfg):
        return ensure_id_helper(cfg)

    def ensure_title_helper(node, cfg_path, title=""):
        flag = False

        if node.get('title') is None:
            if cfg_path:
                node['title'] = re.sub("^[0-9]{1,3}\.", "", os.path.split(os.path.dirname(cfg_path))[-1])
            else:
                node['title'] = title
            flag = True

        if 'children' in node:
            for c in node["children"]:
                flag = flag or ensure_title_helper(list(c.values())[0], None, list(c.keys())[0])

        return flag

    def ensure_title(cfg, cfg_path):
        return ensure_title_helper(cfg, cfg_path)

    def make_node(name, node_id, keywords, children=None):
        node = {}
        node_children = children or []
        node[name] = {
            'node_id': node_id,
            'keywords': keywords,
            'children': node_children
        }
        return node, node_children

    # 根节点
    cfg = ensure_config(data_path)
    cfg_path = os.path.join(data_path, 'config.json')
    if ensure_node_id(cfg):
        dump_json(cfg_path, cfg, exist_ok=True, override=True)

    if ensure_title(cfg, cfg_path):
        cfg["title"] = "C"
        dump_json(cfg_path, cfg, exist_ok=True, override=True)
    tree_node = {
        "node_id": cfg['node_id'],
        "keywords": cfg['keywords'],
        "children": []
    }
    root[cfg['tree_name']] = tree_node

    # 难度节点
    for level_no_dir, level_no_name in list_dir(data_path):
        print(level_no_dir)
        no, level_name = parse_no_name(level_no_name)
        level_path = os.path.join(level_no_dir, 'config.json')
        level_cfg = ensure_config(level_no_dir)
        if ensure_node_id(level_cfg) or check_export(level_no_dir, level_cfg):
            dump_json(level_path, level_cfg, exist_ok=True, override=True)
        if ensure_title(level_cfg, level_path):
            dump_json(level_path, level_cfg, exist_ok=True, override=True)

        level_node, level_node_children = make_node(
            level_name, level_cfg['node_id'], level_cfg['keywords'])
        tree_node['children'].append(level_node)

        # 章节点
        for chapter_no_dir, chapter_no_name in list_dir(level_no_dir):
            no, chapter_name = parse_no_name(chapter_no_name)
            chapter_path = os.path.join(chapter_no_dir, 'config.json')
            chapter_cfg = ensure_config(chapter_no_dir)
            if ensure_node_id(chapter_cfg) or check_export(chapter_no_dir, chapter_cfg):
                dump_json(chapter_path, chapter_cfg, exist_ok=True, override=True)
            if ensure_title(chapter_cfg, chapter_path):
                dump_json(chapter_path, chapter_cfg, exist_ok=True, override=True)

            chapter_node, chapter_node_children = make_node(
                chapter_name, chapter_cfg['node_id'], chapter_cfg['keywords'])
            level_node_children.append(chapter_node)

            # 知识点
            for section_no_dir, section_no_name in list_dir(chapter_no_dir):
                no, section_name = parse_no_name(section_no_name)
                sec_path = os.path.join(section_no_dir, 'config.json')
                sec_cfg = ensure_config(section_no_dir)
                flag = ensure_node_id(sec_cfg) or check_export(section_no_dir, sec_cfg)

                section_node, section_node_children = make_node(
                    section_name, sec_cfg['node_id'], sec_cfg['keywords'], sec_cfg.get('children', []))
                chapter_node_children.append(section_node)

                # 确保习题分配了习题ID

                for export in sec_cfg.get("export", []):
                    ecfg_path = os.path.join(section_no_dir, export)
                    ecfg = load_json(ecfg_path)

                    if (ecfg.get('exercise_id') is None) or (ecfg.get('exercise_id') in id_set):
                        ecfg['exercise_id'] = uuid.uuid4().hex
                        dump_json(ecfg_path, ecfg, exist_ok=True, override=True)

                    id_set.add(ecfg['exercise_id'])

                if flag:
                    dump_json(sec_path, sec_cfg, exist_ok=True, override=True)

                if ensure_title(sec_cfg, sec_path):
                    dump_json(sec_path, sec_cfg, exist_ok=True, override=True)

                    # 保存技能树骨架
    tree_path = os.path.join(data_path, 'tree.json')
    dump_json(tree_path, root, exist_ok=True, override=True)