data_analysis/desc_table.py

#!/usr/bin/env python
# coding=utf-8
"""
查看 ODPS 表结构

使用示例:
    python desc_table.py loghubods.opengid_base_data
    python desc_table.py videoods.wx_user_wechar_detail
"""
import argparse
from pathlib import Path

from lib.odps_module import ODPSClient

PROJECT_ROOT = Path(__file__).parent


def desc_table(table_name: str):
    """查看表结构，返回格式化的字符串"""
    # 解析 project.table 格式
    if '.' in table_name:
        project, table = table_name.split('.', 1)
    else:
        project = 'loghubods'
        table = table_name

    client = ODPSClient(project=project)
    t = client.odps.get_table(table)

    lines = []
    lines.append(f"表名: {project}.{table}")
    lines.append(f"注释: {t.comment or '(无)'}")
    lines.append(f"创建时间: {t.creation_time}")
    lines.append(f"最后修改: {t.last_data_modified_time}")
    lines.append("")
    lines.append("=" * 60)
    lines.append(f"{'字段名':<30} {'类型':<15} {'注释'}")
    lines.append("=" * 60)

    for col in t.table_schema.columns:
        comment = col.comment or ''
        lines.append(f"{col.name:<30} {col.type.name:<15} {comment}")

    if t.table_schema.partitions:
        lines.append("")
        lines.append("分区字段:")
        lines.append("-" * 60)
        for p in t.table_schema.partitions:
            comment = p.comment or ''
            lines.append(f"{p.name:<30} {p.type.name:<15} {comment}")

    return "\n".join(lines), project, table


def save_and_print(table_name: str):
    """查看表结构并保存到文件"""
    content, project, table = desc_table(table_name)

    # 输出目录: tables/project/table.txt
    output_dir = PROJECT_ROOT / "tables" / project
    output_dir.mkdir(parents=True, exist_ok=True)
    output_file = output_dir / f"{table}.txt"

    # 保存文件
    with open(output_file, 'w', encoding='utf-8') as f:
        f.write(content)

    # 打印到终端
    print(content)
    print()
    print(f"已保存到: {output_file}")


def main():
    parser = argparse.ArgumentParser(description='查看 ODPS 表结构')
    parser.add_argument('table', type=str, help='表名，如 loghubods.table_name')
    args = parser.parse_args()

    save_and_print(args.table)


if __name__ == "__main__":
    main()