# -*- coding: utf-8 -*- """WS8771 folder/json reconcile helper. 현재 저장 폴더의 실제 미디어 파일과 parent folder.json 기록을 대조한다. - JSON에는 있는데 실제 파일이 없는 항목을 찾아낸다. - 실제 파일은 있는데 JSON에는 없는 파일을 찾아낸다. - (1), (2), (3) 같은 Windows 중복 번호 파일/기록을 표시한다. - 적용(apply=True) 시 JSON 백업을 만든 뒤 실제 없는 항목만 제거하고 인덱스를 재작성한다. 실제 미디어 파일을 삭제하거나, JSON에 미등록 파일을 자동 추가하지 않는다. """ from __future__ import annotations import json import os import re import shutil from datetime import datetime from typing import Any, Dict, Iterable, List, Tuple MEDIA_EXTS = {".mp4", ".mov", ".webm", ".m4v", ".jpg", ".jpeg", ".png", ".webp", ".avif"} DUP_SUFFIX_RE = re.compile(r"\s\((\d+)\)(?=\.[^.]+$)", re.IGNORECASE) def now_stamp() -> str: return datetime.now().strftime("%Y%m%d_%H%M%S") def now_text() -> str: return datetime.now().strftime("%Y-%m-%d %H:%M:%S") def safe_path_part(value: Any, fallback: str = "instagram") -> str: s = str(value or fallback or "instagram") for ch in '\\/:*?"<>|\x00': s = s.replace(ch, "_") s = "_".join(s.split()).strip("._ ")[:120] return s or str(fallback or "instagram") def read_json(path: str) -> Dict[str, Any]: with open(path, "r", encoding="utf-8") as f: data = json.load(f) if not isinstance(data, dict): raise ValueError("folder_json_root_must_be_object") return data def write_json_atomic(path: str, data: Dict[str, Any]) -> None: os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) os.replace(tmp, path) def infer_media_folder(json_path: str, doc: Dict[str, Any], media_folder: str = "") -> str: if media_folder: return os.path.abspath(media_folder) folder_name = str(doc.get("folder_name") or "").strip() json_dir = os.path.dirname(os.path.abspath(json_path)) if folder_name: candidate = os.path.join(json_dir, safe_path_part(folder_name, folder_name)) if os.path.isdir(candidate): return os.path.abspath(candidate) # fallback: old layout where json may live inside the media folder return os.path.abspath(json_dir) def candidate_json_paths(media_folder: str = "", json_path: str = "") -> List[str]: out: List[str] = [] if json_path: out.append(os.path.abspath(json_path)) if media_folder: folder = os.path.abspath(media_folder) folder_name = os.path.basename(folder.rstrip("\\/")) or "instagram" parent = os.path.dirname(folder.rstrip("\\/")) or folder out.append(os.path.join(parent, safe_path_part(folder_name, "instagram") + ".json")) out.append(os.path.join(folder, safe_path_part(folder_name, "instagram") + ".json")) # keep order, remove duplicates seen = set() uniq = [] for p in out: k = os.path.normcase(os.path.abspath(p)) if k not in seen: seen.add(k) uniq.append(os.path.abspath(p)) return uniq def choose_existing_json(media_folder: str = "", json_path: str = "") -> str: paths = candidate_json_paths(media_folder, json_path) for p in paths: if os.path.exists(p): return p return paths[0] if paths else "" def is_media_file(name: str) -> bool: return os.path.splitext(str(name or ""))[1].lower() in MEDIA_EXTS def has_duplicate_suffix(name: str) -> bool: return bool(DUP_SUFFIX_RE.search(str(name or ""))) def item_file_name(item: Dict[str, Any]) -> str: name = str(item.get("file_name") or "").strip() if name: return os.path.basename(name) path = str(item.get("file_path") or "").strip() return os.path.basename(path) if path else "" def item_abs_path(item: Dict[str, Any], media_folder: str) -> str: path = str(item.get("file_path") or "").strip() if path and os.path.isabs(path): return os.path.abspath(path) name = item_file_name(item) return os.path.abspath(os.path.join(media_folder, name)) if name else "" def existing_media_files(media_folder: str) -> Dict[str, str]: out: Dict[str, str] = {} if not media_folder or not os.path.isdir(media_folder): return out for name in os.listdir(media_folder): path = os.path.join(media_folder, name) if os.path.isfile(path) and is_media_file(name): out[os.path.normcase(name)] = os.path.abspath(path) return out def rebuild_indexes(doc: Dict[str, Any]) -> Dict[str, Any]: items = doc.get("items") if isinstance(doc.get("items"), dict) else {} by_shortcode: Dict[str, List[str]] = {} by_shortcode_asset: Dict[str, List[str]] = {} by_orig: Dict[str, str] = {} for key, item in items.items(): if not isinstance(item, dict): continue name = item_file_name(item) if not name: continue shortcode = str(item.get("shortcode") or "").strip() asset_kind = str(item.get("asset_kind") or "").strip().lower() try: media_index = int(item.get("media_index") or 0) except Exception: media_index = 0 if shortcode: by_shortcode.setdefault(shortcode, []) if name not in by_shortcode[shortcode]: by_shortcode[shortcode].append(name) if asset_kind: sc_asset_key = f"{shortcode}::{asset_kind}::{media_index}" by_shortcode_asset.setdefault(sc_asset_key, []) if name not in by_shortcode_asset[sc_asset_key]: by_shortcode_asset[sc_asset_key].append(name) orig = str(item.get("instagram_original_filename") or "").strip() if orig: by_orig[orig] = name doc["items_by_shortcode"] = by_shortcode doc["items_by_shortcode_asset"] = by_shortcode_asset doc["items_by_instagram_original_filename"] = by_orig doc["updated_at"] = now_text() return doc def _trim_entries(entries: List[Dict[str, Any]], limit: int) -> List[Dict[str, Any]]: try: n = max(1, int(limit or 80)) except Exception: n = 80 return entries[:n] def reconcile_folder_json(media_folder: str = "", json_path: str = "", apply: bool = False, max_list: int = 80) -> Dict[str, Any]: json_path = choose_existing_json(media_folder, json_path) if not json_path: return {"ok": False, "error": "folder_json_path_empty"} if not os.path.exists(json_path): return {"ok": False, "error": "folder_json_not_found", "json_path": json_path} doc = read_json(json_path) media_folder = infer_media_folder(json_path, doc, media_folder) items = doc.get("items") if isinstance(doc.get("items"), dict) else {} actual = existing_media_files(media_folder) actual_names = set(actual.keys()) json_file_names: Dict[str, str] = {} missing_entries: List[Dict[str, Any]] = [] duplicate_suffix_entries: List[Dict[str, Any]] = [] kept_items: Dict[str, Any] = {} for key, item in items.items(): if not isinstance(item, dict): missing_entries.append({"key": str(key), "reason": "bad_item", "file_name": ""}) continue name = item_file_name(item) norm_name = os.path.normcase(name) if name: json_file_names[norm_name] = str(key) abs_path = item_abs_path(item, media_folder) exists = bool(abs_path and os.path.exists(abs_path)) if has_duplicate_suffix(name): duplicate_suffix_entries.append({"key": str(key), "file_name": name, "exists": exists}) if not name or not exists: missing_entries.append({ "key": str(key), "file_name": name, "expected_path": abs_path, "reason": "file_missing" if name else "file_name_empty", "duplicate_suffix": has_duplicate_suffix(name), }) else: kept_items[str(key)] = item extra_files = [] for norm_name, path in sorted(actual.items(), key=lambda kv: kv[0]): if norm_name not in json_file_names: extra_files.append({"file_name": os.path.basename(path), "file_path": path, "duplicate_suffix": has_duplicate_suffix(path)}) result: Dict[str, Any] = { "ok": True, "applied": False, "json_path": os.path.abspath(json_path), "media_folder": os.path.abspath(media_folder), "item_count": len(items), "actual_file_count": len(actual), "missing_json_count": len(missing_entries), "extra_file_count": len(extra_files), "duplicate_suffix_json_count": len(duplicate_suffix_entries), "missing_json_entries": _trim_entries(missing_entries, max_list), "extra_files": _trim_entries(extra_files, max_list), "duplicate_suffix_json_entries": _trim_entries(duplicate_suffix_entries, max_list), } if apply and missing_entries: backup = os.path.abspath(json_path) + ".bak_" + now_stamp() shutil.copy2(json_path, backup) new_doc = dict(doc) new_doc["items"] = kept_items new_doc["folder_reconcile"] = { "applied_at": now_text(), "removed_missing_json_count": len(missing_entries), "actual_file_count": len(actual), "extra_file_count": len(extra_files), "duplicate_suffix_json_count": len(duplicate_suffix_entries), "backup_path": backup, } rebuild_indexes(new_doc) write_json_atomic(json_path, new_doc) result.update({ "applied": True, "backup_path": backup, "removed_missing_json_count": len(missing_entries), "item_count_after": len(kept_items), }) return result def format_summary(result: Dict[str, Any]) -> str: if not result.get("ok"): return f"폴더정리 실패: {result.get('error', '')}\njson: {result.get('json_path', '')}" lines = [ "폴더정리 결과", f"json: {result.get('json_path', '')}", f"folder: {result.get('media_folder', '')}", f"JSON 항목: {result.get('item_count', 0)}", f"실제 미디어 파일: {result.get('actual_file_count', 0)}", f"JSON에는 있는데 실제 없음: {result.get('missing_json_count', 0)}", f"실제 있는데 JSON 없음: {result.get('extra_file_count', 0)}", f"(1)(2)(3)류 JSON 기록: {result.get('duplicate_suffix_json_count', 0)}", ] if result.get("applied"): lines.append(f"적용: 실제 없는 JSON 항목 {result.get('removed_missing_json_count', 0)}개 제거") lines.append(f"백업: {result.get('backup_path', '')}") else: lines.append("적용: 안 함") missing = result.get("missing_json_entries") or [] if missing: lines.append("") lines.append("실제 없는 JSON 항목 일부:") for it in missing[:10]: lines.append(f"- {it.get('file_name') or it.get('key')} ({it.get('reason','')})") extra = result.get("extra_files") or [] if extra: lines.append("") lines.append("JSON에 없는 실제 파일 일부:") for it in extra[:10]: lines.append(f"- {it.get('file_name')}") return "\n".join(lines) if __name__ == "__main__": import argparse ap = argparse.ArgumentParser(description="WS8771 folder/json reconcile") ap.add_argument("--folder", default="") ap.add_argument("--json", default="") ap.add_argument("--apply", action="store_true") args = ap.parse_args() res = reconcile_folder_json(args.folder, args.json, apply=bool(args.apply)) print(format_summary(res))