import
os
import
json
import
numpy
as
np
from
pathlib
import
Path
from
PIL
import
Image
from
typing
import
List, Dict
from
dataclasses
import
dataclass, asdict
@dataclass
class
MediaRecord
:
filepath: str
filename: str
width: int
height: int
tags: List[str]
scene: str
faces: int
text_content: str
class
MediaIndexer
:
"""Local media analysis — no cloud dependencies."""
def
__init__
(self):
self.scene_categories = [
"indoor"
,
"outdoor"
,
"nature"
,
"urban"
,
"beach"
,
"mountain"
,
"forest"
,
"garden"
,
"office"
,
"kitchen"
,
"restaurant"
,
"street"
,
"concert"
,
"sports"
,
"celebration"
,
"portrait"
,
]
self.object_tags = [
"person"
,
"car"
,
"bicycle"
,
"dog"
,
"cat"
,
"laptop"
,
"cell phone"
,
"book"
,
"plant"
,
"food"
,
"drink"
,
"furniture"
,
"instrument"
,
]
self._init_models()
def
_init_models
(self):
"""Lazy-load models to save memory when not indexing."""
self._models_loaded = False
def
_ensure_models
(self):
"""Load vision models on first use."""
if
self._models_loaded:
return
from
transformers
import
CLIPProcessor, CLIPModel
self.clip_model = CLIPModel.from_pretrained(
"openai/clip-vit-base-patch32"
)
self.clip_processor = CLIPProcessor.from_pretrained(
"openai/clip-vit-base-patch32"
)
import
face_recognition
self.face_detector = face_recognition
print(
"✓ Vision models loaded"
)
self._models_loaded = True
def
_classify_scene
(self, image: Image.Image) -> str:
"""Classify the scene using CLIP zero-shot classification."""
self._ensure_models()
categories = [f
"a photo of a {c} scene"
for
c
in
self.scene_categories]
inputs = self.clip_processor(
text=categories, images=image,
return_tensors=
"pt"
, padding=True
)
outputs = self.clip_model(**inputs)
scores = outputs.logits_per_image.softmax(dim=1)[0]
best_idx = scores.argmax().item()
return
self.scene_categories[best_idx]
def
_detect_faces
(self, image_path: str) -> int:
"""Count number of faces in image."""
self._ensure_models()
try
:
image = self.face_detector.load_image_file(image_path)
locations = self.face_detector.face_locations(image, model=
"hog"
)
return
len(locations)
except
:
return
0
def
_generate_tags
(self, image: Image.Image) -> List[str]:
"""Zero-shot object detection via CLIP."""
self._ensure_models()
prompts = [f
"a photo containing a {tag}"
for
tag
in
self.object_tags]
inputs = self.clip_processor(
text=prompts, images=image,
return_tensors=
"pt"
, padding=True
)
outputs = self.clip_model(**inputs)
scores = outputs.logits_per_image.softmax(dim=1)[0]
threshold = 0.15
matched = []
for
i, score
in
enumerate(scores):
if
score > threshold:
matched.append(self.object_tags[i])
return
matched
def
index_image
(self, filepath: str) -> MediaRecord:
"""Generate complete metadata record for a single image."""
filename = os.path.basename(filepath)
image = Image.open(filepath).convert(
"RGB"
)
width, height = image.size
tags = self._generate_tags(image)
scene = self._classify_scene(image)
faces = self._detect_faces(filepath)
return
MediaRecord(
filepath=str(filepath),
filename=filename,
width=width, height=height,
tags=tags, scene=scene,
faces=faces, text_content=
""
)
def
index_directory
(self, directory: str,
extensions: List[str] = None) -> List[MediaRecord]:
"""Walk a directory and index all images."""
if
extensions
is
None:
extensions = [
".jpg"
,
".jpeg"
,
".png"
,
".webp"
,
".bmp"
]
records = []
files = [f
for
f
in
Path(directory).rglob(
"*"
)
if
f.suffix.lower()
in
extensions]
for
i, filepath
in
enumerate(files):
try
:
record = self.index_image(str(filepath))
records.append(record)
if
(i + 1) % 20 == 0:
print(f
" {i+1}/{len(files)} images indexed...")
except
Exception
as
e:
print(f
" ✗ {filepath.name}: {e}")
return
records