four-level-schedule/parse-service/test_local.py

#!/usr/bin/env python3
"""
本地测试 parse-service
先直接用 Python 测试解析，再启动 FastAPI 服务
"""
import sys
import os

# 添加当前目录到路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from core.router import ParserFactory
from utils.logger import log


async def test_parse_file(file_path):
    """测试解析单个文件"""
    log.info(f"开始测试解析文件: {file_path}")

    if not os.path.exists(file_path):
        log.error(f"文件不存在: {file_path}")
        return None

    factory = ParserFactory()
    result = await factory.parse(file_path)

    log.info(f"解析完成! 文件类型: {result.file_type}")
    log.info(f"内容长度: {len(result.content)}")
    log.info(f"元数据: {result.metadata}")

    print("\n" + "="*80)
    print("解析结果:")
    print("="*80)
    print(result.content[:1000] + "..." if len(result.content) > 1000 else result.content)
    print("="*80)

    # 生成性能报告
    report = factory.generate_performance_report()
    print("\n" + report)

    return result


if __name__ == "__main__":
    import asyncio

    # 先尝试用 duomotai 里的示例文件
    example_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "duomotai", "examples")

    if len(sys.argv) > 1:
        test_file = sys.argv[1]
    else:
        # 找一个示例文件
        test_file = None
        if os.path.exists(example_dir):
            for f in os.listdir(example_dir):
                if f.endswith(('.pdf', '.docx', '.xlsx', '.png', '.jpg')):
                    test_file = os.path.join(example_dir, f)
                    break

        if not test_file:
            print("用法: python test_local.py <文件路径>")
            print("或在 duomotai/examples 目录下放一个测试文件")
            sys.exit(1)

    print(f"测试文件: {test_file}")
    asyncio.run(test_parse_file(test_file))