debug_directory.py

#!/usr/bin/env python3
"""
调试目录遍历问题
"""

from pathlib import Path

from sqllineage.config import SQLLineageConfig


def debug_directory_traversal():
    """调试目录遍历"""
    print("🔍 调试目录遍历问题...")

    # 获取配置的目录
    root = Path(SQLLineageConfig.DIRECTORY)
    print(f"根目录: {root}")
    print(f"根目录存在: {root.exists()}")
    print(f"根目录是目录: {root.is_dir()}")

    if not root.exists():
        print("❌ 根目录不存在!")
        return

    # 测试根目录内容
    try:
        children = list(root.iterdir())
        print(f"根目录子项数量: {len(children)}")
        for child in children:
            print(f"  - {child.name} (is_dir: {child.is_dir()})")
    except Exception as e:
        print(f"❌ 读取根目录失败: {e}")
        return

    # 查找 tpcds 目录
    tpcds_path = root / "tpcds"
    print(f"\ntpcds 路径: {tpcds_path}")
    print(f"tpcds 存在: {tpcds_path.exists()}")
    print(f"tpcds 是目录: {tpcds_path.is_dir()}")

    if tpcds_path.exists() and tpcds_path.is_dir():
        try:
            tpcds_children = list(tpcds_path.iterdir())
            print(f"tpcds 子项数量: {len(tpcds_children)}")
            sql_files = [f for f in tpcds_children if f.name.endswith(".sql")]
            print(f"SQL 文件数量: {len(sql_files)}")
            print("前10个SQL文件:")
            for i, file in enumerate(sorted(sql_files)[:10]):
                print(f"  {i + 1}. {file.name}")
        except Exception as e:
            print(f"❌ 读取 tpcds 目录失败: {e}")

    # 测试 build_tree 函数
    print("\n🧪 测试 build_tree 函数...")

    def build_tree(path):
        try:
            children = []
            for p in sorted(path.iterdir(), key=lambda _: (not _.is_dir(), _.name)):
                if p.is_dir():
                    children.append({"id": str(p), "name": p.name, "is_dir": True, "children": build_tree(p)})
                else:
                    children.append({"id": str(p), "name": p.name, "is_dir": False})
            return children
        except PermissionError as e:
            print(f"❌ 权限错误: {e}")
            return []
        except Exception as e:
            print(f"❌ 其他错误: {e}")
            return []

    result = build_tree(root)
    print(f"build_tree 结果长度: {len(result)}")

    # 查找 tpcds 在结果中
    for child in result:
        if child["name"] == "tpcds" and child["is_dir"]:
            print("✅ 找到 tpcds 目录")
            print(f"tpcds children 数量: {len(child.get('children', []))}")
            if child.get("children"):
                print("tpcds 前5个文件:")
                for i, file in enumerate(child["children"][:5]):
                    print(f"  {i + 1}. {file['name']} (is_dir: {file['is_dir']})")
            break
    else:
        print("❌ 未找到 tpcds 目录")


if __name__ == "__main__":
    debug_directory_traversal()