copywriting_production/tests/mytest/t_markitdown_parse_url_task.py

import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))

from src.flow_task.crawl_asin_flow import parse_url_to_markdown_task
from prefect import flow

def test_parse_url_task():
    """测试parse_url_to_markdown_task函数"""
    # 测试URL列表（来自t_markitdown_xlsx.py中的示例）
    test_urls = [
        "http://s3.vs1.lan/public/amazone/copywriting_production/product/202508/1P镊子压刀.xlsx",
        "http://s3.vs1.lan/public/amazone/copywriting_production/product/202508/3P一体不锈钢迷你园艺铲.xlsx",
        "http://s3.vs1.lan/public/amazone/copywriting_production/product/202508/磁吸固定夹.xlsx",
        "http://s3.vs1.lan/public/amazone/copywriting_production/product/202508/锯齿固定夹.xlsx",
        "http://s3.vs1.lan/public/amazone/copywriting_production/product/202508/魔术贴金属扣.xlsx",
        "http://s3.vs1.lan/public/amazone/copywriting_production/product/202508/黑白轧带.xlsx"
    ]
    
    # 选择第一个URL进行测试
    test_url = test_urls[0]
    print(f"正在测试URL: {test_url}")
    
    try:
        # 调用task
        result = parse_url_to_markdown_task.with_options(refresh_cache=True)(test_url)
        
        if result:
            print("解析成功！内容如下：")
            print(result)
            print("\n" + "="*50 + "\n")
        else:
            print("解析失败，返回结果为空")
            
    except Exception as e:
        print(f"测试过程中发生错误: {e}")
        import traceback
        traceback.print_exc()

@flow(name="测试URL解析任务")
def test_flow():
    """测试flow"""
    test_parse_url_task()

if __name__ == "__main__":
    print("开始测试parse_url_to_markdown_task...")
    test_parse_url_task()
    print("测试完成")