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

1""" 

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

3 

4Category: utility 

5""" 

6 

7 

8def run( 

9 action: str, 

10 directory: str = "", 

11 pattern: str = "*", 

12 organize_by: str = "type", 

13 recursive: bool = False, 

14) -> str: 

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

16 import hashlib 

17 import os 

18 import shutil 

19 import time 

20 

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

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

23 

24 def _walk(): 

25 if recursive: 

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

27 for f in files: 

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

29 else: 

30 for f in os.listdir(directory): 

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

32 if os.path.isfile(fp): 

33 yield fp 

34 

35 try: 

36 if action == "stats": 

37 exts = {} 

38 sizes = [] 

39 total = 0 

40 for fp in _walk(): 

41 total += 1 

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

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

44 try: 

45 sizes.append(os.path.getsize(fp)) 

46 except Exception: 

47 pass 

48 total_size = sum(sizes) 

49 lines = [ 

50 f"目录: {directory}", 

51 f"文件总数: {total}", 

52 f"总大小: {_fmt_size(total_size)}", 

53 "", 

54 "按类型分布:", 

55 ] 

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

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

58 return "\n".join(lines) 

59 

60 if action == "organize": 

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

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

63 moved = 0 

64 for fp in _walk(): 

65 fname = os.path.basename(fp) 

66 if organize_by == "type": 

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

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

69 elif organize_by == "date": 

70 mt = os.path.getmtime(fp) 

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

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

73 elif organize_by == "size": 

74 sz = os.path.getsize(fp) 

75 if sz < 1024 * 1024: 

76 tier = "small" 

77 elif sz < 10 * 1024 * 1024: 

78 tier = "medium" 

79 elif sz < 100 * 1024 * 1024: 

80 tier = "large" 

81 else: 

82 tier = "xlarge" 

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

84 os.makedirs(subdir, exist_ok=True) 

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

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

87 shutil.move(fp, dst) 

88 moved += 1 

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

90 

91 if action == "dedupe": 

92 seen = {} 

93 dups = [] 

94 for fp in _walk(): 

95 try: 

96 sz = os.path.getsize(fp) 

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

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

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

100 if key in seen: 

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

102 else: 

103 seen[key] = fp 

104 except Exception: 

105 pass 

106 if not dups: 

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

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

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

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

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

112 lines.append("") 

113 return "\n".join(lines) 

114 

115 if action == "recent": 

116 files = [] 

117 for fp in _walk(): 

118 try: 

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

120 except Exception: 

121 pass 

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

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

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

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

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

127 return "\n".join(lines) 

128 

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

130 except Exception as e: 

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

132 

133 

134def _fmt_size(size): 

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

136 if size < 1024: 

137 return f"{size:.1f}{unit}" 

138 size /= 1024 

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

140 

141 

142__all__ = ["run"]