Coverage for agentos/marketplace/skills/file-organizer/file-organizer.py: 3%

1""" 

2file-organizer — 文件整理工具：按类型/日期归类、批量重命名、找重复文件。 

3 

4Category: utility 

5""" 

6 

7 

8def run(action: str, directory: str = "", pattern: str = "*", 

9 organize_by: str = "type", recursive: bool = False) -> str: 

10 """文件整理工具。action: organize/dedupe/stats/recent。organize_by: type/date/size。""" 

11 import os, shutil, hashlib, time 

12 

13 if not directory or not os.path.isdir(directory): 

14 return f"[file-organizer] 目录不存在: {directory}" 

15 

16 def _walk(): 

17 if recursive: 

18 for root, _, files in os.walk(directory): 

19 for f in files: 

20 yield os.path.join(root, f) 

21 else: 

22 for f in os.listdir(directory): 

23 fp = os.path.join(directory, f) 

24 if os.path.isfile(fp): yield fp 

25 

26 try: 

27 if action == "stats": 

28 exts = {} 

29 sizes = [] 

30 total = 0 

31 for fp in _walk(): 

32 total += 1 

33 ext = os.path.splitext(fp)[1].lower() or "(无后缀)" 

34 exts[ext] = exts.get(ext, 0) + 1 

35 try: sizes.append(os.path.getsize(fp)) 

36 except: pass 

37 total_size = sum(sizes) 

38 lines = [f"目录: {directory}", f"文件总数: {total}", f"总大小: {_fmt_size(total_size)}", "", 

39 "按类型分布:"] 

40 for ext, cnt in sorted(exts.items(), key=lambda x: -x[1]): 

41 lines.append(f" {ext}: {cnt} 个") 

42 return "\n".join(lines) 

43 

44 if action == "organize": 

45 if organize_by not in ("type", "date", "size"): 

46 return f"[file-organizer] organize_by 只支持: type/date/size" 

47 moved = 0 

48 for fp in _walk(): 

49 fname = os.path.basename(fp) 

50 if organize_by == "type": 

51 ext = os.path.splitext(fname)[1].lower().lstrip(".") or "other" 

52 subdir = os.path.join(directory, ext) 

53 elif organize_by == "date": 

54 mt = os.path.getmtime(fp) 

55 ts = time.strftime("%Y-%m", time.localtime(mt)) 

56 subdir = os.path.join(directory, ts) 

57 elif organize_by == "size": 

58 sz = os.path.getsize(fp) 

59 if sz < 1024*1024: tier = "small" 

60 elif sz < 10*1024*1024: tier = "medium" 

61 elif sz < 100*1024*1024: tier = "large" 

62 else: tier = "xlarge" 

63 subdir = os.path.join(directory, tier) 

64 os.makedirs(subdir, exist_ok=True) 

65 dst = os.path.join(subdir, fname) 

66 if fp != dst and not os.path.exists(dst): 

67 shutil.move(fp, dst) 

68 moved += 1 

69 return f"已整理 {moved} 个文件 (按 {organize_by}) → {directory}" 

70 

71 if action == "dedupe": 

72 seen = {} 

73 dups = [] 

74 for fp in _walk(): 

75 try: 

76 sz = os.path.getsize(fp) 

77 with open(fp, "rb") as f: 

78 h = hashlib.md5(f.read(8192)).hexdigest() 

79 key = f"{sz}_{h}" 

80 if key in seen: 

81 dups.append((fp, seen[key])) 

82 else: 

83 seen[key] = fp 

84 except: pass 

85 if not dups: 

86 return "[file-organizer] 未找到重复文件" 

87 lines = [f"找到 {len(dups)} 组疑似重复:"] 

88 for dup, orig in dups[:20]: 

89 lines.append(f" 重复: {dup}") 

90 lines.append(f" 原始: {orig}") 

91 lines.append("") 

92 return "\n".join(lines) 

93 

94 if action == "recent": 

95 files = [] 

96 for fp in _walk(): 

97 try: 

98 files.append((os.path.getmtime(fp), fp, os.path.getsize(fp))) 

99 except: pass 

100 files.sort(key=lambda x: -x[0]) 

101 lines = [f"最近修改的文件 (共{len(files)}个):"] 

102 for mt, fp, sz in files[:20]: 

103 ts = time.strftime("%Y-%m-%d %H:%M", time.localtime(mt)) 

104 lines.append(f" {ts} {_fmt_size(sz):>8} {fp}") 

105 return "\n".join(lines) 

106 

107 return f"[file-organizer] 未知操作: {action}, 支持: stats/organize/dedupe/recent" 

108 except Exception as e: 

109 return f"[file-organizer] 失败: {e}" 

110 

111 

112def _fmt_size(size): 

113 for unit in ("B", "KB", "MB", "GB"): 

114 if size < 1024: return f"{size:.1f}{unit}" 

115 size /= 1024 

116 return f"{size:.1f}TB" 

117 

118 

119__all__ = ["run"]