Agent/examples/demand/export.py

import argparse
import json
from pathlib import Path
from typing import Any, List, Optional, Union


def build_lines_from_demand_items(input_path: Union[str, Path]) -> List[str]:
    """读取 demand items JSON，并格式化为逐行文本。"""
    path = Path(input_path)
    data: Any = json.loads(path.read_text(encoding="utf-8"))

    if not isinstance(data, list):
        raise ValueError(f"JSON 顶层必须是 list，实际是: {type(data).__name__}")

    lines: List[str] = []
    for idx, item in enumerate(data):
        if not isinstance(item, dict):
            raise ValueError(f"第 {idx + 1} 条记录不是对象: {type(item).__name__}")

        name = item.get("name", [])
        reason = str(item.get("reason", ""))
        suggestion = str(item.get("suggestion", ""))
        score = item.get("score", 0)
        ext_data = json.loads(item.get("ext_data", '{}'))
        type = ext_data.get('type')
        video_ids = ext_data.get('video_ids', [])

        lines.append(f"{name}｜{type}｜{reason}｜{suggestion}｜{score}｜{video_ids}")

    return lines


def export_lines(
    input_path: Union[str, Path],
    output_path: Optional[Union[str, Path]] = None,
) -> List[str]:
    """
    读取 JSON 并导出为逐行结果。
    - output_path 为空: 打印到标准输出
    - output_path 不为空: 写入文件（UTF-8）
    """
    lines = build_lines_from_demand_items(input_path)
    text = "\n".join(lines)

    if output_path is None:
        print(text)
    else:
        Path(output_path).write_text(text, encoding="utf-8")

    return lines


def main() -> None:
    account_name = 'R_50*泛知识*生活科普'
    input_path = f'/Users/shimeng/Desktop/py/Agent/examples/demand/result/{account_name}.json'
    output_path = f'/Users/shimeng/Desktop/py/Agent/examples/demand/result/{account_name}.txt'
    export_lines(input_path,
                 output_path)


if __name__ == "__main__":
    main()