Как разобрать файлы на Mac за 10–15 минут и удалить дубли фото, видео и документов + готовый скрипт

TMP_SCRIPT="$(mktemp /tmp/mac_file_organizer.XXXXXX)"cat > "$TMP_SCRIPT" <<'PY'from __future__ import annotationsimport csv, datetime as dt, hashlib, json, os, re, shutil, subprocess, sys, unicodedata, zipfilefrom collections import Counter, defaultdictfrom pathlib import PathPHOTO={".jpg",".jpeg",".png",".heic",".heif",".webp",".gif",".tif",".tiff",".bmp",".dng",".raw",".cr2",".cr3",".nef",".arw"}VIDEO={".mov",".mp4",".m4v",".avi",".mkv",".webm",".mts",".m2ts",".3gp"}AUDIO={".mp3",".m4a",".aac",".wav",".flac",".ogg",".aiff"}DOCS={".pdf",".doc",".docx",".rtf",".txt",".odt",".pages"}SHEETS={".xls",".xlsx",".xlsm",".csv",".tsv",".ods",".numbers"}PRESENTATIONS={".ppt",".pptx",".key",".odp"}DESIGN={".psd",".ai",".eps",".svg",".sketch",".fig"}ARCHIVES={".zip",".rar",".7z",".tar",".gz"}CODE={".py",".js",".ts",".tsx",".jsx",".php",".html",".htm",".css",".sql",".json",".xml",".yaml",".yml",".sh",".command"}MEDIA=PHOTO|VIDEO|AUDIOWORK=DOCS|SHEETS|PRESENTATIONS|DESIGN|ARCHIVESSKIP_NAMES={".git","node_modules",".Trash","__pycache__","_РАЗОБРАНО"}SKIP_PREFIXES=("_FILE_AUDIT_","_FILE_QUARANTINE_","_ORGANIZER_PLAN_","_ORGANIZER_RUN_")PROJECT_MARKERS={".git","package.json","pyproject.toml","requirements.txt","Cargo.toml","go.mod","composer.json"}GENERIC={"разное","старое","копия","фото с телефона","новая папка","backup","old","temp","photos","pictures","documents","downloads"}PRIVATE={"паспорт","права","свидетельство","личное"}MONTHS={"январ":1,"феврал":2,"март":3,"апрел":4,"май":5,"мая":5,"июн":6,"июл":7,"август":8,"сентябр":9,"октябр":10,"ноябр":11,"декабр":12}def stamp(): return dt.datetime.now().strftime("%Y%m%d_%H%M%S")def norm(s): return unicodedata.normalize("NFC",str(s))def clean(s):    s=re.sub(r"\s+"," ",norm(s).replace("/"," - ").replace(":"," - ")).strip(" ._-")    return (s or "Без названия")[:120]def human(n):    x=float(n)    for u in ("Б","КБ","МБ","ГБ","ТБ"):        if x<1024 or u=="ТБ": return f"{int(x)} {u}" if u=="Б" else f"{x:.1f} {u}"        x/=1024def choose_folder():    r=subprocess.run(["osascript","-e",'POSIX path of (choose folder with prompt "Выберите папку, которую нужно разобрать")'],                     capture_output=True,text=True)    if not r.stdout.strip(): print("Папка не выбрана."); raise SystemExit    return Path(r.stdout.strip()).expanduser().resolve()def is_project_dir(p):    try: return bool({x.name for x in p.iterdir()} & PROJECT_MARKERS)    except Exception: return Falsedef iter_files(root):    for dp,dirs,files in os.walk(root):        cur=Path(dp)        if cur!=root and is_project_dir(cur):            dirs[:]=[]            continue        dirs[:]=[d for d in dirs if d not in SKIP_NAMES and not any(d.startswith(p) for p in SKIP_PREFIXES)]        for name in files:            p=cur/name            try:                if not p.is_symlink(): yield p            except Exception: passdef sha256_file(p,block=1024*1024):    h=hashlib.sha256()    try:        with p.open("rb") as f:            while True:                b=f.read(block)                if not b: break                h.update(b)        return h.hexdigest()    except Exception: return Nonedef category(p):    e=p.suffix.lower()    for exts,name in ((PHOTO,"Фото"),(VIDEO,"Видео"),(AUDIO,"Аудио"),(DOCS,"Документы"),                      (SHEETS,"Таблицы"),(PRESENTATIONS,"Презентации"),(DESIGN,"Дизайн"),                      (ARCHIVES,"Архивы"),(CODE,"Код")):        if e in exts: return name    return "Прочее"def parse_date(text):    s=norm(text).lower()    patterns=[        (r"(?<!\d)(20\d{2}|19\d{2})[-_. /](0?[1-9]|1[0-2])[-_. /](0?[1-9]|[12]\d|3[01])(?!\d)","ymd"),        (r"(?<!\d)(20\d{2}|19\d{2})(0[1-9]|1[0-2])([0-2]\d|3[01])(?!\d)","ymd"),        (r"(?<!\d)(0?[1-9]|[12]\d|3[01])[-_. /](0?[1-9]|1[0-2])[-_. /](20\d{2}|19\d{2})(?!\d)","dmy"),    ]    for rx,order in patterns:        m=re.search(rx,s)        if not m: continue        a,b,c=map(int,m.groups())        try: return (dt.datetime(a,b,c),100) if order=="ymd" else (dt.datetime(c,b,a),100)        except Exception: pass    for word,month in MONTHS.items():        if word in s:            y=re.search(r"(?<!\d)(20\d{2}|19\d{2})(?!\d)",s)            if y: return dt.datetime(int(y.group(1)),month,1),85    y=re.search(r"(?<!\d)(20\d{2}|19\d{2})(?!\d)",s)    return (dt.datetime(int(y.group(1)),1,1),60) if y else (None,0)def exif_dates(paths):    exe=shutil.which("exiftool")    if not exe: return {}    out={}    print("ExifTool найден — читаю даты фото и видео...")    for start in range(0,len(paths),200):        chunk=paths[start:start+200]        try:            r=subprocess.run([exe,"-j","-DateTimeOriginal","-MediaCreateDate","-CreateDate","-TrackCreateDate",*[str(p) for p in chunk]],                             capture_output=True,text=True,timeout=120)            for item in json.loads(r.stdout or "[]"):                src=item.get("SourceFile")                if not src: continue                for key in ("DateTimeOriginal","MediaCreateDate","CreateDate","TrackCreateDate"):                    m=re.match(r"(\d{4}):(\d{2}):(\d{2})",str(item.get(key,"")))                    if m:                        out[os.path.abspath(src)]=dt.datetime(*map(int,m.groups()))                        break        except Exception: pass    return outdef spotlight_date(p):    try:        r=subprocess.run(["mdls","-raw","-name","kMDItemContentCreationDate",str(p)],capture_output=True,text=True,timeout=5)        m=re.search(r"(20\d{2}|19\d{2})-(\d{2})-(\d{2})",r.stdout)        return dt.datetime(*map(int,m.groups())) if m else None    except Exception: return Nonedef filesystem_date(p):    try:        st=p.stat()        return dt.datetime.fromtimestamp(getattr(st,"st_birthtime",None) or st.st_mtime)    except Exception: return dt.datetime.now()def infer_date(p,root,exif):    ap=os.path.abspath(str(p))    if ap in exif: return exif[ap],100,"EXIF / QuickTime"    value,conf=parse_date(p.name)    if value and conf>=85: return value,conf,"имя файла"    best=(value,conf,"имя файла" if value else "")    for part in reversed(p.relative_to(root).parts[:-1]):        v,c=parse_date(part)        if v and c>best[1]: best=(v,c,f"папка: {part}")    if best[0] and best[1]>=85: return best    if p.suffix.lower() in MEDIA:        v=spotlight_date(p)        if v: return v,75,"Spotlight"    return best if best[0] else (filesystem_date(p),30,"дата файла macOS")def extract_text(p,limit=50000):    e=p.suffix.lower()    try:        if e in {".txt",".csv",".tsv",".html",".htm",".rtf"}:            return p.read_text(encoding="utf-8",errors="ignore")[:limit]        if e==".docx":            with zipfile.ZipFile(p) as z: raw=z.read("word/document.xml")            return re.sub(r"\s+"," ",re.sub(r"<[^>]+>"," ",raw.decode("utf-8","ignore")))[:limit]        if e in {".xlsx",".xlsm"}:            chunks=[]            with zipfile.ZipFile(p) as z:                for name in ("xl/sharedStrings.xml","xl/workbook.xml"):                    if name in z.namelist():                        chunks.append(re.sub(r"<[^>]+>"," ",z.read(name).decode("utf-8","ignore")))            return re.sub(r"\s+"," "," ".join(chunks))[:limit]        if e==".pdf":            r=subprocess.run(["mdls","-raw","-name","kMDItemTextContent",str(p)],capture_output=True,text=True,timeout=5)            return "" if r.stdout.strip()=="(null)" else r.stdout[:limit]    except Exception: pass    return ""def document_type(ctx,e):    ctx=ctx.lower()    rules=[        (("приложение","дополнительное соглашение","доп. соглашение"),"Приложения"),        (("договор","contract"),"Договоры"),        (("акт выполн","акт №"),"Акты"),        (("счет на оплат","счёт на оплат","invoice"),"Счета"),        (("коммерческое предложение","коммерческ","кп "),"Коммерческие предложения"),        (("бриф","brief"),"Брифы"),        (("техническое задание","тз "),"ТЗ"),        (("отчет","отчёт","report"),"Отчёты"),        (("реквизит",),"Реквизиты"),        (("seo","семантик","метрика","директ","поисковые запрос"),"SEO и реклама"),    ]    for words,name in rules:        if any(w in ctx for w in words): return name    if e in SHEETS: return "Таблицы"    if e in PRESENTATIONS: return "Презентации"    if e in DESIGN: return "Дизайн"    if e in ARCHIVES: return "Архивы"    return "Документы"def unique_path(dst,reserved=None):    reserved=reserved if reserved is not None else set()    if not dst.exists() and str(dst) not in reserved:        reserved.add(str(dst)); return dst    for i in range(2,10000):        c=dst.with_name(f"{dst.stem}__{i}{dst.suffix}")        if not c.exists() and str(c) not in reserved:            reserved.add(str(c)); return c    raise RuntimeError("Не удалось подобрать уникальное имя")def write_csv(path,fields,rows):    with path.open("w",encoding="utf-8-sig",newline="") as f:        w=csv.DictWriter(f,fieldnames=fields); w.writeheader(); w.writerows(rows)def audit(root):    print(f"\nСканирую:\n{root}")    files=list(iter_files(root)); cats=Counter(); sizes=defaultdict(list); total=0    for i,p in enumerate(files,1):        try: size=p.stat().st_size        except Exception: size=0        total+=size; cats[category(p)]+=1        if size>0: sizes[size].append(p)        if i%1000==0: print(f"Проверено {i}/{len(files)}")    print("\nИщу точные дубли...")    groups=[]    for size,paths in sizes.items():        if len(paths)<2: continue        hashes=defaultdict(list)        for p in paths:            h=sha256_file(p)            if h: hashes[h].append(p)        groups += [{"hash":h,"size":size,"paths":ps} for h,ps in hashes.items() if len(ps)>1]    dup_files=sum(len(g["paths"])-1 for g in groups)    dup_bytes=sum(g["size"]*(len(g["paths"])-1) for g in groups)    out=root/f"_FILE_AUDIT_{stamp()}"; out.mkdir()    (out/"summary.json").write_text(json.dumps({        "files":len(files),"size":human(total),"categories":dict(cats),        "duplicate_groups":len(groups),"duplicate_files":dup_files,"duplicate_size":human(dup_bytes)    },ensure_ascii=False,indent=2),encoding="utf-8")    print("\n"+"="*48+"\nАУДИТ ГОТОВ\n"+"="*48)    print(f"\nФайлов: {len(files)}\nРазмер:  {human(total)}")    for name,count in cats.most_common(): print(f"{name:<14} {count}")    print(f"\nГрупп дублей: {len(groups)}\nЛишних копий: {dup_files}\nОбъём дублей: {human(dup_bytes)}")    return groupsdef keeper_score(p,root):    rel=p.relative_to(root); text=str(rel).lower()    return len(rel.parts)*10 + (1000 if "разобран" in text else 0) - (500 if re.search(r"(копия|copy|\(\d+\))",p.name,re.I) else 0) + (100 if re.search(r"(19|20)\d{2}",text) else 0)def quarantine(root,groups):    n=sum(len(g["paths"])-1 for g in groups); b=sum(g["size"]*(len(g["paths"])-1) for g in groups)    print(f"\nБудет перемещено: {n} файлов / {human(b)}\nФайлы НЕ будут удалены.")    if input("\nВведите КАРАНТИН: ").strip()!="КАРАНТИН":        print("Отменено."); return    q=root/f"_FILE_QUARANTINE_{stamp()}"; q.mkdir()    undo=[]; moved=skipped=0    for g in groups:        paths=sorted(g["paths"],key=lambda p:keeper_score(p,root),reverse=True)        expected=g["hash"]; keeper=paths[0]        if any((not p.exists()) or sha256_file(p)!=expected for p in paths):            skipped+=len(paths)-1; continue        for p in paths[1:]:            dst=q/p.relative_to(root); dst.parent.mkdir(parents=True,exist_ok=True)            try:                shutil.move(str(p),str(dst))                undo.append({"original":str(p),"quarantine":str(dst),"keeper":str(keeper),"sha256":expected}); moved+=1            except Exception: skipped+=1    write_csv(q/"undo.csv",["original","quarantine","keeper","sha256"],undo)    print(f"\nПеремещено: {moved}\nПропущено: {skipped}\nКарантин:\n{q}")def latest(root,prefix):    xs=sorted((p for p in root.iterdir() if p.is_dir() and p.name.startswith(prefix)),reverse=True)    return xs[0] if xs else Nonedef restore_quarantine(root):    q=latest(root,"_FILE_QUARANTINE_")    if not q or not (q/"undo.csv").exists(): print("Карантин не найден."); return    rows=list(csv.DictReader((q/"undo.csv").open(encoding="utf-8-sig")))    ok=skip=0    for r in reversed(rows):        src,dst=Path(r["quarantine"]),Path(r["original"])        if not src.exists() or dst.exists(): skip+=1; continue        try:            dst.parent.mkdir(parents=True,exist_ok=True); shutil.move(str(src),str(dst)); ok+=1        except Exception: skip+=1    print(f"\nВозвращено: {ok}\nПропущено: {skip}")def ask_rules():    print("\nМожно добавить проекты/клиентов через запятую. Если не нужно — Enter.")    projects=[clean(x.strip()) for x in input("Проекты: ").split(",") if x.strip()]    generic=set(GENERIC)    print("Игнорируются: разное, старое, копия, backup, old, temp, новая папка")    generic.update(x.strip().lower() for x in input("Добавить ещё или Enter: ").split(",") if x.strip())    return projects,genericdef detect_project(rel,text,projects,generic):    ctx=(" ".join(rel.parts)+" "+text[:5000]).lower()    for project in projects:        if project.lower() in ctx: return project    if len(rel.parts)>1 and not rel.parts[0].startswith("_"):        first=clean(rel.parts[0])        if first.lower() not in generic: return first    return "Без проекта"def build_plan(root,projects,generic):    print("\nСтрою будущую структуру...")    files=list(iter_files(root)); exif=exif_dates([p for p in files if p.suffix.lower() in MEDIA])    reserved=set(); plan=[]; summary=Counter(); confidence=Counter()    for i,p in enumerate(files,1):        rel=p.relative_to(root); e=p.suffix.lower()        date,conf,source=infer_date(p,root,exif); year=str(date.year); month=f"{date.year:04d}-{date.month:02d}"        if e in PHOTO|VIDEO:            dst=root/"_РАЗОБРАНО"/"02_ФОТО_И_ВИДЕО"/year/month            if len(rel.parts)>1:                first=clean(rel.parts[0])                if first.lower() not in generic and not first.startswith("_"): dst/=first            dst/=p.name; group="Фото/видео"        elif e in AUDIO:            dst=root/"_РАЗОБРАНО"/"03_АУДИО"/year/p.name; group="Аудио"        elif e in WORK:            text=extract_text(p); ctx=(" ".join(rel.parts)+" "+text)            dtype=document_type(ctx,e)            if any(w in ctx.lower() for w in PRIVATE):                dst=root/"_РАЗОБРАНО"/"04_ЛИЧНОЕ"/dtype/year/p.name; group="Личное"            else:                dst=root/"_РАЗОБРАНО"/"01_РАБОТА"/detect_project(rel,text,projects,generic)/dtype/year/p.name; group="Работа"        elif e in CODE:            continue        else:            dst=root/"_РАЗОБРАНО"/"90_ПРОЧЕЕ"/year/p.name; group="Прочее"        dst=unique_path(dst,reserved)        try: size=p.stat().st_size        except Exception: size=0        plan.append({"source":str(p),"destination":str(dst),"group":group,"date":date.isoformat(sep=" ",timespec="seconds"),                     "date_source":source,"confidence":conf,"size_bytes":size})        summary[group]+=1        confidence["Высокая" if conf>=85 else "Средняя" if conf>=60 else "Низкая"]+=1        if i%500==0: print(f"Обработано {i}/{len(files)}")    out=root/f"_ORGANIZER_PLAN_{stamp()}"; out.mkdir()    write_csv(out/"plan.csv",["source","destination","group","date","date_source","confidence","size_bytes"],plan)    print("\n"+"="*48+"\nПЛАН ГОТОВ\n"+"="*48)    print(f"\nФайлов: {len(plan)}")    for k,v in summary.items(): print(f"{k:<16} {v}")    print("\nУверенность даты:")    for k in ("Высокая","Средняя","Низкая"): print(f"{k:<16} {confidence[k]}")    print(f"\nПолный план:\n{out/'plan.csv'}")    return plandef apply_plan(root,plan):    if input("\nВведите РАЗОБРАТЬ для начала: ").strip()!="РАЗОБРАТЬ":        print("Ничего не перемещено."); return    run=root/f"_ORGANIZER_RUN_{stamp()}"; run.mkdir()    undo=[]; moved=skipped=0    for i,r in enumerate(plan,1):        src,dst=Path(r["source"]),Path(r["destination"])        if not src.exists(): skipped+=1; continue        dst.parent.mkdir(parents=True,exist_ok=True)        if dst.exists(): dst=unique_path(dst)        try:            shutil.move(str(src),str(dst)); undo.append({"original":str(src),"current":str(dst)}); moved+=1        except Exception: skipped+=1        if i%500==0: print(f"Перемещено {i}/{len(plan)}")    write_csv(run/"undo.csv",["original","current"],undo)    print(f"\nПеремещено: {moved}\nПропущено: {skipped}\nНовая структура:\n{root/'_РАЗОБРАНО'}")    subprocess.run(["open",str(root/"_РАЗОБРАНО")],check=False)def undo_sort(root):    run=latest(root,"_ORGANIZER_RUN_")    if not run or not (run/"undo.csv").exists(): print("Сортировка не найдена."); return    rows=list(csv.DictReader((run/"undo.csv").open(encoding="utf-8-sig")))    ok=skip=0    for r in reversed(rows):        src,dst=Path(r["current"]),Path(r["original"])        if not src.exists() or dst.exists(): skip+=1; continue        try:            dst.parent.mkdir(parents=True,exist_ok=True); shutil.move(str(src),str(dst)); ok+=1        except Exception: skip+=1    print(f"\nВозвращено: {ok}\nПропущено: {skip}")def main():    if sys.platform!="darwin": print("Этот вариант рассчитан на macOS."); return    root=choose_folder()    print(f"\nВыбрана папка:\n{root}\n\n1 — проверить и разобрать\n2 — откатить последнюю сортировку\n3 — вернуть последний карантин\n4 — выйти")    choice=input("\nВведите 1, 2, 3 или 4: ").strip()    if choice=="2": undo_sort(root); return    if choice=="3": restore_quarantine(root); return    if choice!="1": return    groups=audit(root)    print("\n1 — закончить\n2 — убрать точные дубли в карантин\n3 — перейти к сортировке")    choice=input("\nВведите 1, 2 или 3: ").strip()    if choice=="1": return    if choice=="2":        quarantine(root,groups)        if input("\nПродолжить сортировку? Введите ДА или Enter: ").strip()!="ДА": return    projects,generic=ask_rules()    plan=build_plan(root,projects,generic)    print("\n1 — только посмотреть план\n2 — разложить файлы")    if input("\nВведите 1 или 2: ").strip()=="2": apply_plan(root,plan)    else: print("\nНичего не перемещено.")if __name__=="__main__":    main()PYif ! command -v python3 >/dev/null 2>&1; then    echo    echo "Python 3 не найден. Установите Python 3 и повторите запуск."    rm -f "$TMP_SCRIPT"    exit 1fipython3 "$TMP_SCRIPT"STATUS=$?rm -f "$TMP_SCRIPT"if [ "$STATUS" -ne 0 ]; then    echo    echo "Скрипт завершился с ошибкой: $STATUS"fi