Product-Analyzer/data_processing.py at main · AppleCook/Product-Analyzer · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import argparse
from parse_tsv import parse_tsv_file, get_product_sample

def process_data(input_file, limit=None, sample=False):
    """
    处理TSV数据文件

    参数:
        input_file (str): 输入TSV文件路径
        limit (int, optional): 处理的最大产品数量
        sample (bool): 是否只返回样本数据

    返回:
        list: 产品数据列表
    """
    # 解析TSV文件
    products = parse_tsv_file(input_file)

    # 限制数量
    if limit and isinstance(limit, int) and limit > 0:
        products = products[:limit]

    # 返回样本或全部数据
    if sample:
        return get_product_sample(products)
    return products

def main():
    # 解析命令行参数
    parser = argparse.ArgumentParser(description='处理产品数据')
    parser.add_argument('--input', '-i', required=True, help='输入TSV文件路径')
    parser.add_argument('--limit', '-l', type=int, help='处理的最大产品数量')
    parser.add_argument('--sample', '-s', action='store_true', help='是否只返回样本数据')
    parser.add_argument('--output', '-o', help='输出统计信息的文件路径')

    args = parser.parse_args()

    # 处理数据
    products = process_data(args.input, args.limit, args.sample)

    # 输出基本统计信息
    print(f"共处理 {len(products)} 个产品")

    # 计算一些基本统计信息
    if products:
        # 查找最高收入
        max_revenue_product = max(products, key=lambda x: str(x.get('text-base (3)', '0')))
        print(f"收入最高的产品: {max_revenue_product.get('text-base', '未知')} - {max_revenue_product.get('text-base (3)', '未知')}")

        # 统计收入分布
        revenue_counts = {}
        for p in products:
            revenue = p.get('text-base (3)', '未知')
            if revenue != '未知':
                # 提取收入范围（如10M-50M）
                if 'M' in revenue:
                    value = float(revenue.replace('M', ''))
                    if value < 10:
                        range_key = "< 10M"
                    elif value < 50:
                        range_key = "10M-50M"
                    elif value < 100:
                        range_key = "50M-100M"
                    else:
                        range_key = "> 100M"

                    revenue_counts[range_key] = revenue_counts.get(range_key, 0) + 1

        print("\n收入分布统计:")
        for range_key, count in revenue_counts.items():
            print(f"  {range_key}: {count} 个产品")

    # 保存统计信息
    if args.output:
        with open(args.output, 'w', encoding='utf-8') as f:
            f.write(f"# 产品数据统计\n\n")
            f.write(f"共分析 {len(products)} 个产品\n\n")

            if products:
                f.write("## 收入分布\n\n")
                for range_key, count in revenue_counts.items():
                    f.write(f"- {range_key}: {count} 个产品\n")

                f.write("\n## 产品样本\n\n")
                sample = get_product_sample(products, 3)
                for i, product in enumerate(sample):
                    f.write(f"### {product.get('text-base', '未知产品')}\n\n")
                    f.write(f"- 描述: {product.get('text-sm', '无描述')}\n")
                    f.write(f"- 收入: {product.get('text-base (3)', '未知')}\n")
                    f.write(f"- 排名: {product.get('text-center', '未知')}\n\n")

            print(f"统计信息已保存到: {args.output}")

    return products

if __name__ == "__main__":
    # 确保logs目录存在
    if not os.path.exists('logs'):
        os.makedirs('logs')

    main()