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

1""" 

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

3 

4Category: data 

5""" 

6 

7 

8def run(action: str, file_path: str = "", query: str = "", output_path: str = "", 

9 delimiter: str = ",", encoding: str = "utf-8") -> str: 

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

11 import csv, os 

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

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

14 try: 

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

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

17 if reader.fieldnames is None: 

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

19 headers = list(reader.fieldnames) 

20 rows = list(reader) 

21 if action == "headers": 

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

23 if action == "read": 

24 preview = rows[:20] 

25 lines = [delimiter.join(headers)] 

26 for row in preview: 

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

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

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

30 if action == "stats": 

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

32 for h in headers: 

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

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

35 try: 

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

37 if nums: 

38 res.append(f" {h}: 非空={ne}, 数值={len(nums)}, min={min(nums):.2f}, max={max(nums):.2f}, avg={sum(nums)/len(nums):.2f}") 

39 else: 

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

41 except ValueError: 

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

43 return "\n".join(res) 

44 if action == "filter": 

45 if not query: return "[csv-toolkit] filter 需要 query='col op value'" 

46 parts = query.split(maxsplit=2) 

47 if len(parts)<3: return "[csv-toolkit] query格式: 'col op value'" 

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

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

50 filtered=[] 

51 for row in rows: 

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

53 try: 

54 if op==">": m = float(cell)>float(val) 

55 elif op=="<": m = float(cell)<float(val) 

56 elif op==">=": m = float(cell)>=float(val) 

57 elif op=="<=": m = float(cell)<=float(val) 

58 elif op=="==": m = str(cell).strip()==val.strip() 

59 elif op=="!=": m = str(cell).strip()!=val.strip() 

60 elif op=="contains": m = val.strip().lower() in str(cell).lower() 

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

62 except ValueError: m=False 

63 if m: filtered.append(row) 

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

65 if output_path and filtered: 

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

67 w=csv.DictWriter(f,fieldnames=headers); w.writeheader(); w.writerows(filtered) 

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

69 elif filtered: 

70 lines=[delimiter.join(headers)] 

71 for row in filtered[:10]: lines.append(delimiter.join(str(row.get(h,"")) for h in headers)) 

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

73 return s 

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

75 except Exception as e: 

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

77 

78 

79__all__ = ["run"]