Coverage for agentos/marketplace/skills/csv-toolkit/csv-toolkit.py: 2%

1""" 

2csv-toolkit — CSV 处理工具集：读取、过滤、统计、导出。 

3 

4Category: data 

5""" 

6 

7 

8def run( 

9 action: str, 

10 file_path: str = "", 

11 query: str = "", 

12 output_path: str = "", 

13 delimiter: str = ",", 

14 encoding: str = "utf-8", 

15) -> str: 

16 """CSV 文件操作工具。action: headers/read/stats/filter。filter 时 query 格式 'col op value'。""" 

17 import csv 

18 import os 

19 

20 if not file_path or not os.path.isfile(file_path): 

21 return f"[csv-toolkit] 文件不存在: {file_path}" 

22 try: 

23 with open(file_path, encoding=encoding, newline="") as f: 

24 reader = csv.DictReader(f, delimiter=delimiter) 

25 if reader.fieldnames is None: 

26 return "[csv-toolkit] 无法解析表头" 

27 headers = list(reader.fieldnames) 

28 rows = list(reader) 

29 if action == "headers": 

30 return f"表头({len(headers)}列): {', '.join(headers)}\n行数: {len(rows)}" 

31 if action == "read": 

32 preview = rows[:20] 

33 lines = [delimiter.join(headers)] 

34 for row in preview: 

35 lines.append(delimiter.join(str(row.get(h, "")) for h in headers)) 

36 tail = f"\n... (共{len(rows)}行,显示前{len(preview)}行)" if len(rows) > 20 else "" 

37 return "\n".join(lines) + tail 

38 if action == "stats": 

39 res = [f"文件: {file_path}", f"行数: {len(rows)}", f"列数: {len(headers)}"] 

40 for h in headers: 

41 vals = [row.get(h, "") for row in rows] 

42 ne = sum(1 for v in vals if v.strip()) 

43 try: 

44 nums = [float(v) for v in vals if v.strip()] 

45 if nums: 

46 res.append( 

47 f" {h}: 非空={ne}, 数值={len(nums)}, min={min(nums):.2f}, max={max(nums):.2f}, avg={sum(nums)/len(nums):.2f}" # noqa: E501 

48 ) 

49 else: 

50 res.append(f" {h}: 非空={ne}") 

51 except ValueError: 

52 res.append(f" {h}: 非空={ne}") 

53 return "\n".join(res) 

54 if action == "filter": 

55 if not query: 

56 return "[csv-toolkit] filter 需要 query='col op value'" 

57 parts = query.split(maxsplit=2) 

58 if len(parts) < 3: 

59 return "[csv-toolkit] query格式: 'col op value'" 

60 col, op, val = parts[0], parts[1], parts[2] 

61 if col not in headers: 

62 return f"[csv-toolkit] 列'{col}'不存在,可用: {', '.join(headers)}" 

63 filtered = [] 

64 for row in rows: 

65 cell = row.get(col, "") 

66 try: 

67 if op == ">": 

68 m = float(cell) > float(val) 

69 elif op == "<": 

70 m = float(cell) < float(val) 

71 elif op == ">=": 

72 m = float(cell) >= float(val) 

73 elif op == "<=": 

74 m = float(cell) <= float(val) 

75 elif op == "==": 

76 m = str(cell).strip() == val.strip() 

77 elif op == "!=": 

78 m = str(cell).strip() != val.strip() 

79 elif op == "contains": 

80 m = val.strip().lower() in str(cell).lower() 

81 else: 

82 return f"[csv-toolkit] 不支持操作符: {op}" 

83 except ValueError: 

84 m = False 

85 if m: 

86 filtered.append(row) 

87 s = f"过滤: {col} {op} {val}\n匹配: {len(filtered)}/{len(rows)}" 

88 if output_path and filtered: 

89 with open(output_path, "w", encoding=encoding, newline="") as f: 

90 w = csv.DictWriter(f, fieldnames=headers) 

91 w.writeheader() 

92 w.writerows(filtered) 

93 s += f"\n已写入: {output_path}" 

94 elif filtered: 

95 lines = [delimiter.join(headers)] 

96 for row in filtered[:10]: 

97 lines.append(delimiter.join(str(row.get(h, "")) for h in headers)) 

98 s += "\n" + ("\n".join(lines)) 

99 return s 

100 return f"[csv-toolkit] 未知操作: {action}, 支持: headers/read/stats/filter" 

101 except Exception as e: 

102 return f"[csv-toolkit] 失败: {e}" 

103 

104 

105__all__ = ["run"]