667 lines
21 KiB
Python
667 lines
21 KiB
Python
import json
|
|
import os
|
|
import platform
|
|
import shutil
|
|
import subprocess
|
|
import unicodedata
|
|
from typing import Dict, List, Optional
|
|
from .config import ENABLE_JSON_SIDECAR, METADATA_EXT
|
|
from .logger import get_logger
|
|
|
|
# Supported extensions
|
|
IMAGE_EXTS = (".png", ".jpg", ".jpeg", ".webp", ".gif")
|
|
VIDEO_EXTS = (".mp4", ".mov", ".webm", ".mkv")
|
|
AUDIO_EXTS = (".wav", ".mp3", ".flac", ".ogg", ".m4a", ".aac")
|
|
MODEL3D_EXTS = (".obj", ".fbx", ".glb", ".gltf", ".stl")
|
|
WINDOWS_RATING_INDEX = 14 # fallback
|
|
WINDOWS_TAGS_INDEX = 21 # fallback
|
|
STAR_CHARS = ("\u2605", "\u2606", "\u22c6", "\u272d", "\u272e", "\u2730")
|
|
RATING_KEYS = (
|
|
"rating",
|
|
"note",
|
|
"notation",
|
|
"evaluation",
|
|
"evaluations",
|
|
"etoile",
|
|
"etoiles",
|
|
"star",
|
|
"stars",
|
|
)
|
|
TAG_KEYS = ("tag", "mot-cle", "mot cle", "mots cle", "keywords", "keyword")
|
|
|
|
_WIN32COM: Optional[object] = None
|
|
_WIN32COM_CHECKED = False
|
|
_CACHED_RATING_INDEX: Optional[int] = None
|
|
_CACHED_TAGS_INDEX: Optional[int] = None
|
|
_EXIFTOOL_PATH: Optional[str] = None
|
|
_EXIFTOOL_CHECKED = False
|
|
_META_CACHE: Dict[str, tuple[Optional[float], int, dict]] = {}
|
|
_CACHE_EPOCH = 0
|
|
|
|
log = get_logger(__name__)
|
|
|
|
|
|
def _env_timeout(name: str, default: float, min_s: float = 0.5, max_s: float = 60.0) -> float:
|
|
"""
|
|
Parse a timeout value from env and clamp it to a safe range.
|
|
Keeps external metadata tools from hanging the server.
|
|
"""
|
|
try:
|
|
raw = os.environ.get(name, None)
|
|
val = float(default) if raw is None else float(raw)
|
|
except Exception:
|
|
val = float(default)
|
|
try:
|
|
return max(float(min_s), min(float(val), float(max_s)))
|
|
except Exception:
|
|
return float(default)
|
|
|
|
|
|
def _force_sidecar_enabled() -> bool:
|
|
val = str(os.environ.get("MJR_FORCE_SIDECAR", "0")).lower()
|
|
return val in ("1", "true", "yes", "on")
|
|
|
|
|
|
def _sidecar_allowed_for(path: str) -> bool:
|
|
return ENABLE_JSON_SIDECAR or _force_sidecar_enabled()
|
|
|
|
|
|
def _normalize_label(text) -> str:
|
|
try:
|
|
normalized = unicodedata.normalize("NFKD", str(text or ""))
|
|
normalized = "".join(ch for ch in normalized if not unicodedata.combining(ch))
|
|
return normalized.lower().strip()
|
|
except Exception:
|
|
return str(text or "").lower().strip()
|
|
|
|
|
|
def _get_mtime_safe(path: str) -> Optional[float]:
|
|
try:
|
|
return os.path.getmtime(path)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def safe_import_win32com():
|
|
global _WIN32COM, _WIN32COM_CHECKED
|
|
if _WIN32COM_CHECKED:
|
|
return _WIN32COM
|
|
_WIN32COM_CHECKED = True
|
|
if platform.system().lower() == "windows":
|
|
try:
|
|
import win32com.client # type: ignore
|
|
|
|
_WIN32COM = win32com.client
|
|
except ImportError:
|
|
log.warning("Windows logic disabled (pywin32 not found)")
|
|
_WIN32COM = None
|
|
else:
|
|
_WIN32COM = None
|
|
return _WIN32COM
|
|
|
|
|
|
def _get_exiftool_path() -> Optional[str]:
|
|
global _EXIFTOOL_PATH, _EXIFTOOL_CHECKED
|
|
if _EXIFTOOL_CHECKED:
|
|
return _EXIFTOOL_PATH
|
|
_EXIFTOOL_CHECKED = True
|
|
_EXIFTOOL_PATH = shutil.which("exiftool") or shutil.which("exiftool.exe")
|
|
return _EXIFTOOL_PATH
|
|
|
|
|
|
def _parse_rating_value(raw) -> int:
|
|
if raw in (None, ""):
|
|
return 0
|
|
s_raw = str(raw).strip()
|
|
if not s_raw:
|
|
return 0
|
|
|
|
star_count = sum(s_raw.count(ch) for ch in STAR_CHARS)
|
|
if star_count:
|
|
return max(0, min(star_count, 5))
|
|
|
|
try:
|
|
numeric = float(s_raw.replace(",", ".").split()[0])
|
|
except Exception:
|
|
return 0
|
|
|
|
if numeric <= 0:
|
|
return 0
|
|
if numeric > 5:
|
|
if numeric <= 25:
|
|
return 2
|
|
if numeric <= 50:
|
|
return 3
|
|
if numeric <= 75:
|
|
return 4
|
|
return 5
|
|
return int(round(min(numeric, 5)))
|
|
|
|
|
|
def rating_to_windows_percent(stars: int) -> int:
|
|
"""
|
|
Map 0..5 stars to Windows percent scale expected by SharedUserRating.
|
|
"""
|
|
try:
|
|
r = int(stars)
|
|
except Exception:
|
|
r = 0
|
|
mapping = {0: 0, 1: 1, 2: 25, 3: 50, 4: 75, 5: 99}
|
|
return max(0, min(99, mapping.get(r, 0 if r < 0 else r)))
|
|
|
|
|
|
def windows_percent_to_stars(percent: int) -> int:
|
|
"""
|
|
Convert Windows percent (SharedUserRating) to 0..5 stars using thresholds.
|
|
"""
|
|
try:
|
|
p = int(percent)
|
|
except Exception:
|
|
return 0
|
|
if p <= 0:
|
|
return 0
|
|
if p <= 12:
|
|
return 1
|
|
if p <= 37:
|
|
return 2
|
|
if p <= 62:
|
|
return 3
|
|
if p <= 87:
|
|
return 4
|
|
return 5
|
|
|
|
|
|
def tags_to_windows_category(tags: list) -> str:
|
|
"""
|
|
Join tags for Microsoft:Category; use '; ' which Explorer parses.
|
|
"""
|
|
if not isinstance(tags, list):
|
|
return ""
|
|
cleaned = [str(t).strip() for t in tags if str(t).strip()]
|
|
return "; ".join(cleaned)
|
|
|
|
|
|
def windows_category_to_tags(cat: str) -> list:
|
|
"""
|
|
Split Microsoft:Category back into a list; preserve order, drop empties.
|
|
"""
|
|
if not cat:
|
|
return []
|
|
parts = [p.strip() for p in str(cat).split(";") if p.strip()]
|
|
return parts
|
|
|
|
|
|
def get_windows_metadata(file_path: str) -> dict:
|
|
"""Fetch file metadata via the Windows Shell API."""
|
|
try:
|
|
win32com = safe_import_win32com()
|
|
if not win32com:
|
|
return {"rating": 0, "tags": []}
|
|
|
|
shell = win32com.Dispatch("Shell.Application")
|
|
folder = shell.Namespace(os.path.dirname(file_path))
|
|
file = folder.ParseName(os.path.basename(file_path))
|
|
|
|
rating_idx, tags_idx = _resolve_shell_indices(folder)
|
|
|
|
rating_raw = folder.GetDetailsOf(file, rating_idx)
|
|
tags_raw = folder.GetDetailsOf(file, tags_idx)
|
|
|
|
rating = 0
|
|
# Windows may expose either stars or SharedUserRating percent; normalize to stars
|
|
try:
|
|
numeric = float(str(rating_raw).strip())
|
|
if 0 <= numeric <= 100 and numeric > 5:
|
|
rating = windows_percent_to_stars(int(round(numeric)))
|
|
else:
|
|
rating = _parse_rating_value(numeric)
|
|
except Exception:
|
|
rating = _parse_rating_value(rating_raw)
|
|
|
|
tags = []
|
|
if tags_raw:
|
|
tag_text = str(tags_raw)
|
|
for sep in [",", ";"]:
|
|
if sep in tag_text:
|
|
tags = [t.strip() for t in tag_text.split(sep) if t.strip()]
|
|
break
|
|
else:
|
|
t = tag_text.strip()
|
|
if t:
|
|
tags = [t]
|
|
|
|
return {
|
|
"rating": rating,
|
|
"tags": tags,
|
|
}
|
|
except Exception as e:
|
|
log.warning("Error reading metadata for %s: %s", file_path, e)
|
|
return {"rating": 0, "tags": []}
|
|
|
|
|
|
# Canonical ExifTool reader for rating/tags (avoid duplicate implementations)
|
|
def get_exif_metadata(file_path: str) -> dict:
|
|
"""
|
|
Fast read via exiftool when available.
|
|
Returns {"rating": int, "tags": List[str]} or {} on fallback.
|
|
"""
|
|
exe = _get_exiftool_path()
|
|
if not exe:
|
|
return {}
|
|
try:
|
|
timeout_s = _env_timeout("MJR_EXIFTOOL_READ_TIMEOUT", 2.0, min_s=0.5, max_s=60.0)
|
|
cmd = [
|
|
exe,
|
|
"-n",
|
|
"-json",
|
|
"-Rating",
|
|
"-RatingPercent",
|
|
"-Microsoft:SharedUserRating",
|
|
"-Microsoft:Category",
|
|
"-Subject",
|
|
"-Keywords",
|
|
"-XPKeywords",
|
|
file_path,
|
|
]
|
|
out = subprocess.check_output(cmd, timeout=timeout_s)
|
|
data = json.loads(out)[0]
|
|
is_video = file_path.lower().endswith((".mp4", ".mov", ".m4v", ".webm", ".mkv"))
|
|
|
|
rating_raw = data.get("Rating")
|
|
if rating_raw is None:
|
|
rating_raw = data.get("RatingPercent")
|
|
if rating_raw is None:
|
|
rating_raw = data.get("Microsoft:SharedUserRating")
|
|
rating = 0
|
|
if rating_raw is not None:
|
|
try:
|
|
num = float(rating_raw)
|
|
except Exception:
|
|
num = None
|
|
if num is not None and 0 <= num <= 5:
|
|
rating = int(round(num))
|
|
elif num is not None and num > 5:
|
|
rating = windows_percent_to_stars(int(round(num)))
|
|
else:
|
|
rating = _parse_rating_value(rating_raw)
|
|
|
|
tags: List[str] = []
|
|
if is_video:
|
|
cat = data.get("Microsoft:Category")
|
|
if cat:
|
|
tags = windows_category_to_tags(cat)
|
|
if not tags:
|
|
for key in ("Subject", "Keywords", "XPKeywords"):
|
|
val = data.get(key)
|
|
if isinstance(val, list):
|
|
tags.extend([str(v).strip() for v in val if str(v).strip()])
|
|
elif isinstance(val, str):
|
|
tags.extend([t.strip() for t in val.split(",") if t.strip()])
|
|
|
|
if tags:
|
|
tags = sorted(set(tags))
|
|
|
|
return {"rating": rating, "tags": tags}
|
|
except subprocess.TimeoutExpired:
|
|
return {}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
# Self-check for rating normalization; kept lightweight
|
|
def _test_rating_normalization():
|
|
assert windows_percent_to_stars(5) == 1 # percent-like but low
|
|
assert windows_percent_to_stars(99) == 5
|
|
assert windows_percent_to_stars(75) == 4
|
|
assert windows_percent_to_stars(1) == 1
|
|
assert _parse_rating_value(5) == 5
|
|
assert _parse_rating_value(0) == 0
|
|
|
|
_test_rating_normalization()
|
|
|
|
|
|
# Canonical ExifTool writer for rating/tags (video-aware); do not duplicate
|
|
def set_exif_metadata(file_path: str, rating: int, tags: list) -> bool:
|
|
"""
|
|
Write Rating/Keywords via exiftool when available. Returns True on success.
|
|
"""
|
|
exe = _get_exiftool_path()
|
|
if not exe:
|
|
return False
|
|
|
|
global _CACHE_EPOCH
|
|
_CACHE_EPOCH += 1
|
|
epoch = _CACHE_EPOCH
|
|
original_mtime = _get_mtime_safe(file_path)
|
|
|
|
def _collect_preserve_args_for_video() -> list:
|
|
if not exe:
|
|
return []
|
|
try:
|
|
timeout_s = _env_timeout("MJR_EXIFTOOL_PRESERVE_TIMEOUT", 3.0, min_s=0.5, max_s=60.0)
|
|
cmd = [
|
|
exe,
|
|
"-j",
|
|
"-Comment",
|
|
"-UserComment",
|
|
"-Description",
|
|
"-UserData",
|
|
"-XMP:Comment",
|
|
"-XMP:Description",
|
|
"-ItemList:UserComment",
|
|
file_path,
|
|
]
|
|
out = subprocess.check_output(cmd, stderr=subprocess.DEVNULL, timeout=timeout_s)
|
|
data_list = json.loads(out) if out else []
|
|
if not data_list:
|
|
return []
|
|
entry = data_list[0]
|
|
preserve_args = []
|
|
for key in ("Comment", "UserComment", "Description", "UserData", "XMP:Comment", "XMP:Description", "ItemList:UserComment"):
|
|
val = entry.get(key)
|
|
if isinstance(val, list):
|
|
val = " ".join(str(v) for v in val if str(v).strip())
|
|
if isinstance(val, str):
|
|
val = val.strip()
|
|
if val:
|
|
preserve_args.append(f"-{key}={val}")
|
|
return preserve_args
|
|
except subprocess.TimeoutExpired:
|
|
return []
|
|
except Exception:
|
|
return []
|
|
|
|
try:
|
|
write_timeout_s = _env_timeout("MJR_EXIFTOOL_WRITE_TIMEOUT", 3.0, min_s=0.5, max_s=120.0)
|
|
r = max(0, min(int(rating), 5))
|
|
win_r = rating_to_windows_percent(r)
|
|
|
|
is_video = file_path.lower().endswith((".mp4", ".mov", ".m4v", ".webm", ".mkv"))
|
|
|
|
preserve_args = _collect_preserve_args_for_video() if is_video else []
|
|
|
|
cmd = [exe, "-overwrite_original", *preserve_args]
|
|
cmd += [
|
|
f"-xmp:rating={r}",
|
|
f"-rating={r}",
|
|
f"-ratingpercent={win_r}",
|
|
"-Subject=",
|
|
"-Keywords=",
|
|
"-XPKeywords=",
|
|
]
|
|
if is_video:
|
|
cmd.append(f"-Microsoft:SharedUserRating={win_r}")
|
|
cat = tags_to_windows_category(tags)
|
|
cmd.append(f"-Microsoft:Category={cat}")
|
|
|
|
clean_tags = [str(t).strip() for t in tags if str(t).strip()]
|
|
# Always write Category for video, even if empty, to clear stale values
|
|
if is_video and not clean_tags:
|
|
cmd.append(f"-Microsoft:Category=")
|
|
if clean_tags:
|
|
for t in clean_tags:
|
|
cmd.append(f"-xmp:subject+={t}")
|
|
cmd.append(f"-iptc:keywords+={t}")
|
|
# XPKeywords expects a semicolon-separated string on Windows
|
|
cmd.append(f"-xpkeywords={'; '.join(clean_tags)}")
|
|
|
|
cmd.append(file_path)
|
|
subprocess.run(cmd, check=True, capture_output=True, timeout=write_timeout_s)
|
|
|
|
# Restore mtime to avoid reordering the grid after rating/tag updates
|
|
if original_mtime is not None:
|
|
try:
|
|
os.utime(file_path, (original_mtime, original_mtime))
|
|
except Exception:
|
|
pass
|
|
|
|
# Update cache with current epoch
|
|
_META_CACHE[file_path] = (
|
|
original_mtime if original_mtime is not None else _get_mtime_safe(file_path),
|
|
epoch,
|
|
{"rating": rating, "tags": tags},
|
|
)
|
|
return True
|
|
except subprocess.TimeoutExpired:
|
|
return False
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def set_windows_metadata(file_path: str, rating: int, tags: list) -> bool:
|
|
"""Write metadata via the Windows Shell API."""
|
|
try:
|
|
win32com = safe_import_win32com()
|
|
if not win32com:
|
|
return False
|
|
|
|
shell = win32com.Dispatch("Shell.Application")
|
|
folder = shell.Namespace(os.path.dirname(file_path))
|
|
file = folder.ParseName(os.path.basename(file_path))
|
|
|
|
rating_idx, tags_idx = _resolve_shell_indices(folder)
|
|
|
|
# Rating (scale 1-99); Windows 5-star uses 1-5, but accepting broader range
|
|
if rating is not None:
|
|
try:
|
|
r_int = int(rating)
|
|
except Exception:
|
|
r_int = 0
|
|
# Accept 0-5 stars or already-percent; normalize to Windows percent scale
|
|
if 0 <= r_int <= 5:
|
|
rating_val = rating_to_windows_percent(r_int)
|
|
else:
|
|
rating_val = max(0, min(99, r_int))
|
|
folder.GetDetailsOf(file, rating_idx) # force index load
|
|
folder.SetDetailsOf(file, rating_idx, str(rating_val))
|
|
|
|
if tags:
|
|
folder.GetDetailsOf(file, tags_idx)
|
|
folder.SetDetailsOf(file, tags_idx, "; ".join(tags))
|
|
elif tags == []:
|
|
folder.GetDetailsOf(file, tags_idx)
|
|
folder.SetDetailsOf(file, tags_idx, "")
|
|
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _resolve_shell_indices(folder) -> (int, int):
|
|
"""
|
|
Tente de localiser les colonnes "Rating"/"Tags" en fonction de la locale.
|
|
Retourne un tuple (rating_idx, tags_idx) avec fallback sur les constantes.
|
|
"""
|
|
global _CACHED_RATING_INDEX, _CACHED_TAGS_INDEX
|
|
if _CACHED_RATING_INDEX is not None and _CACHED_TAGS_INDEX is not None:
|
|
return _CACHED_RATING_INDEX, _CACHED_TAGS_INDEX
|
|
|
|
rating_idx = WINDOWS_RATING_INDEX
|
|
tags_idx = WINDOWS_TAGS_INDEX
|
|
|
|
try:
|
|
for i in range(0, 256):
|
|
col_name = _normalize_label(folder.GetDetailsOf(None, i))
|
|
if not col_name:
|
|
continue
|
|
if _CACHED_RATING_INDEX is None and any(key in col_name for key in RATING_KEYS):
|
|
_CACHED_RATING_INDEX = i
|
|
if _CACHED_TAGS_INDEX is None and any(key in col_name for key in TAG_KEYS):
|
|
_CACHED_TAGS_INDEX = i
|
|
if _CACHED_RATING_INDEX is not None and _CACHED_TAGS_INDEX is not None:
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
if _CACHED_RATING_INDEX is None:
|
|
_CACHED_RATING_INDEX = rating_idx
|
|
if _CACHED_TAGS_INDEX is None:
|
|
_CACHED_TAGS_INDEX = tags_idx
|
|
|
|
return _CACHED_RATING_INDEX, _CACHED_TAGS_INDEX
|
|
|
|
|
|
def apply_windows_metadata(file_path: str, rating, tags: list) -> dict:
|
|
"""
|
|
Apply rating/tags via Windows API while handling defaults.
|
|
Retourne {"rating": int, "tags": List[str]}.
|
|
"""
|
|
try:
|
|
rating_int = int(rating) if rating is not None else 0
|
|
except Exception:
|
|
rating_int = 0
|
|
tags_list: List[str] = []
|
|
if isinstance(tags, list):
|
|
tags_list = [str(t) for t in tags if str(t).strip()]
|
|
set_windows_metadata(file_path, rating_int, tags_list)
|
|
return {"rating": rating_int, "tags": tags_list}
|
|
|
|
|
|
# Canonical metadata updater (Windows + ExifTool + optional sidecar); keep single definition
|
|
def update_metadata_with_windows(file_path: str, updates: dict) -> dict:
|
|
"""
|
|
Update rating/tags via Windows + sidecar (if enabled).
|
|
Returns final metadata.
|
|
"""
|
|
global _CACHE_EPOCH
|
|
_CACHE_EPOCH += 1
|
|
epoch = _CACHE_EPOCH
|
|
|
|
# Load current values (prefer system metadata, fallback to sidecar)
|
|
sys_meta = get_system_metadata(file_path) or {}
|
|
sidecar = load_metadata(file_path) or {}
|
|
current_rating = sys_meta.get("rating", sidecar.get("rating", 0))
|
|
current_tags = sys_meta.get("tags", sidecar.get("tags", []))
|
|
|
|
has_rating = "rating" in updates
|
|
has_tags = "tags" in updates
|
|
|
|
if not has_rating and not has_tags:
|
|
return {"rating": current_rating, "tags": current_tags}
|
|
|
|
rating = updates.get("rating", current_rating if has_rating else current_rating)
|
|
tags = updates.get("tags", current_tags if has_tags else current_tags)
|
|
|
|
# Only write tags when provided; otherwise preserve existing tags in system metadata
|
|
target_tags = tags if has_tags else current_tags
|
|
|
|
ok_win = False
|
|
ok_exif = False
|
|
if has_rating or has_tags:
|
|
ok_win = set_windows_metadata(file_path, rating if has_rating else current_rating, target_tags if has_tags else current_tags)
|
|
ok_exif = set_exif_metadata(file_path, rating if has_rating else current_rating, target_tags if has_tags else current_tags)
|
|
|
|
if _sidecar_allowed_for(file_path):
|
|
save_metadata(file_path, {"rating": rating if has_rating else current_rating, "tags": target_tags})
|
|
|
|
effective_rating = (rating if has_rating else current_rating) if (ok_win or ok_exif) else current_rating
|
|
effective_tags = target_tags
|
|
_META_CACHE[file_path] = (_get_mtime_safe(file_path), epoch, {"rating": effective_rating, "tags": effective_tags})
|
|
return {"rating": effective_rating, "tags": effective_tags}
|
|
|
|
|
|
def apply_system_metadata(file_path: str, rating, tags: list) -> dict:
|
|
"""
|
|
Try exiftool first (if installed), otherwise Windows Shell.
|
|
"""
|
|
global _CACHE_EPOCH
|
|
_CACHE_EPOCH += 1
|
|
epoch = _CACHE_EPOCH
|
|
try:
|
|
rating_int = int(rating) if rating is not None else 0
|
|
except Exception:
|
|
rating_int = 0
|
|
tags_list: List[str] = []
|
|
if isinstance(tags, list):
|
|
tags_list = [str(t) for t in tags if str(t).strip()]
|
|
|
|
# exiftool (meilleure compat Windows + autres OS)
|
|
if set_exif_metadata(file_path, rating_int, tags_list):
|
|
mtime = _get_mtime_safe(file_path)
|
|
_META_CACHE[file_path] = (mtime, epoch, {"rating": rating_int, "tags": tags_list})
|
|
return {"rating": rating_int, "tags": tags_list}
|
|
|
|
# fallback shell API
|
|
meta = apply_windows_metadata(file_path, rating_int, tags_list)
|
|
mtime = _get_mtime_safe(file_path)
|
|
_META_CACHE[file_path] = (mtime, epoch, meta)
|
|
return meta
|
|
|
|
|
|
def get_system_metadata(file_path: str) -> dict:
|
|
"""
|
|
Prefer ExifTool metadata (Rating/Keywords), otherwise Windows API.
|
|
"""
|
|
mtime = _get_mtime_safe(file_path)
|
|
cached = _META_CACHE.get(file_path)
|
|
if cached and cached[0] == mtime and cached[1] == _CACHE_EPOCH:
|
|
return dict(cached[2])
|
|
|
|
meta = get_exif_metadata(file_path)
|
|
if meta.get("rating") or meta.get("tags"):
|
|
_META_CACHE[file_path] = (mtime, _CACHE_EPOCH, meta)
|
|
return meta
|
|
|
|
meta = get_windows_metadata(file_path)
|
|
if (meta.get("rating") or meta.get("tags")) and not (meta.get("rating") == 0 and not meta.get("tags")):
|
|
_META_CACHE[file_path] = (mtime, _CACHE_EPOCH, meta)
|
|
return meta
|
|
|
|
# Sidecar fallback only when enabled/forced
|
|
if _sidecar_allowed_for(file_path):
|
|
sidecar = load_metadata(file_path)
|
|
if sidecar:
|
|
r = sidecar.get("rating", 0)
|
|
t = sidecar.get("tags", [])
|
|
meta = {"rating": r if isinstance(r, int) else 0, "tags": t if isinstance(t, list) else []}
|
|
_META_CACHE[file_path] = (mtime, _CACHE_EPOCH, meta)
|
|
return meta
|
|
|
|
_META_CACHE[file_path] = (mtime, _CACHE_EPOCH, meta)
|
|
return meta
|
|
|
|
|
|
def classify_ext(lower_name: str) -> str:
|
|
"""Retourne image / video / audio / model3d / other."""
|
|
if lower_name.endswith(IMAGE_EXTS):
|
|
return "image"
|
|
if lower_name.endswith(VIDEO_EXTS):
|
|
return "video"
|
|
if lower_name.endswith(AUDIO_EXTS):
|
|
return "audio"
|
|
if lower_name.endswith(MODEL3D_EXTS):
|
|
return "model3d"
|
|
return "other"
|
|
|
|
def ensure_dir(path):
|
|
if not os.path.exists(path):
|
|
os.makedirs(path)
|
|
|
|
def is_image(fname: str):
|
|
f = fname.lower()
|
|
return f.endswith(IMAGE_EXTS)
|
|
|
|
def metadata_path(image_path: str):
|
|
if not _sidecar_allowed_for(image_path):
|
|
return None
|
|
return image_path + METADATA_EXT
|
|
|
|
def load_metadata(image_path):
|
|
meta_file = metadata_path(image_path)
|
|
if not meta_file:
|
|
return {}
|
|
if not os.path.exists(meta_file):
|
|
return {}
|
|
try:
|
|
with open(meta_file, "r", encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except Exception:
|
|
return {}
|
|
|
|
def save_metadata(image_path, meta: dict):
|
|
meta_file = metadata_path(image_path)
|
|
if not meta_file:
|
|
return
|
|
with open(meta_file, "w", encoding="utf-8") as f:
|
|
json.dump(meta, f, indent=2, ensure_ascii=False)
|