Hermes Agent commited on
Commit
fa9fa8f
·
1 Parent(s): 5abd90f

feat(ai_admin): enhance AI reconciliation engine, series deduplication, and artwork healing

Browse files
Thunder/bot/plugins/ai_admin.py CHANGED
@@ -30,7 +30,7 @@ from ai_admin.snapshot_store import SnapshotStore
30
  from ai_admin.telegram_router import TelegramAIAdminRouter
31
  from ai_admin.tool_registry import ToolRegistry
32
  from ai_admin.tool_runner import ToolRunner
33
- from ai_admin.tools_jellyfin import get_jellyfin_read_tool_definitions
34
  from ai_admin.tools_providers import get_provider_read_tool_definitions
35
 
36
  logger = logging.getLogger("Thunder.Plugin.AIAdmin")
@@ -110,6 +110,10 @@ def setup_ai_admin(client: Optional[Client] = None) -> Optional[TelegramAIAdminR
110
  for tool_def in get_jellyfin_read_tool_definitions(jf_client):
111
  registry.register(tool_def)
112
 
 
 
 
 
113
  prov_client = ProvidersClient(
114
  tmdb_api_key=os.environ.get("TMDB_API_KEY", ""),
115
  fanart_api_key=os.environ.get("FANART_API_KEY", ""),
 
30
  from ai_admin.telegram_router import TelegramAIAdminRouter
31
  from ai_admin.tool_registry import ToolRegistry
32
  from ai_admin.tool_runner import ToolRunner
33
+ from ai_admin.tools_jellyfin import get_jellyfin_active_mutation_definitions, get_jellyfin_read_tool_definitions
34
  from ai_admin.tools_providers import get_provider_read_tool_definitions
35
 
36
  logger = logging.getLogger("Thunder.Plugin.AIAdmin")
 
110
  for tool_def in get_jellyfin_read_tool_definitions(jf_client):
111
  registry.register(tool_def)
112
 
113
+ if config.mutations_enabled:
114
+ for tool_def in get_jellyfin_active_mutation_definitions(jf_client):
115
+ registry.register(tool_def)
116
+
117
  prov_client = ProvidersClient(
118
  tmdb_api_key=os.environ.get("TMDB_API_KEY", ""),
119
  fanart_api_key=os.environ.get("FANART_API_KEY", ""),
ai_admin/handler.py CHANGED
@@ -144,19 +144,23 @@ class AIAdminHandler:
144
  system_prompt = (
145
  "You are an intelligent, friendly, and expert Jellyfin Media Server AI Assistant.\n"
146
  "You have direct control to inspect and manage the Jellyfin library and query metadata/artwork providers:\n"
147
- "• Search Jellyfin library for movies, series, episodes (`jellyfin_search`, `jellyfin_get_item`, `jellyfin_get_media_info`).\n"
 
 
 
148
  "• Search artwork & posters across providers (`artwork_search` for multi-provider search, `tmdb_get_images`, `cinemeta_search`, `cinemeta_get_meta`, `tvmaze_search`, `fanart_get_images`).\n"
149
- "• Free providers like Cinemeta (IMDb catalog) and TVmaze require zero API keys and find authentic posters/thumbnails for movies and series.\n"
150
- "• Query TMDb (themoviedb) for official details and images (`tmdb_search`, `tmdb_get_details`, `tmdb_get_images`).\n"
151
  "• Search subtitles on OpenSubtitles (`opensubtitles_search`).\n\n"
152
  "Workflow Guidelines:\n"
153
- "1. When the user asks to find a movie/series and check or fix its poster/thumbnail:\n"
154
- " a) FIRST search Jellyfin (`jellyfin_search`) to locate the item, its ID, current metadata, and existing artwork.\n"
155
- " b) THEN search artwork via `artwork_search` or `cinemeta_search` / `tmdb_get_images` using the title or IMDb/TMDb ID.\n"
156
- " c) PRESENT the results conversationally with titles, years, current status, and direct preview links for the best authentic posters found.\n"
157
- " d) Offer to apply the selected poster.\n"
158
- "2. If the user is just chatting or asking a question that doesn't need tools, reply warmly, concisely, and conversationally.\n"
159
- "3. Format all responses using Telegram markdown (*bold*, `code`, [link](url)). Never output raw JSON unless specifically requested."
 
 
160
  )
161
 
162
  messages = [{"role": "system", "content": system_prompt}]
 
144
  system_prompt = (
145
  "You are an intelligent, friendly, and expert Jellyfin Media Server AI Assistant.\n"
146
  "You have direct control to inspect and manage the Jellyfin library and query metadata/artwork providers:\n"
147
+ "• Search Jellyfin library for movies, series, episodes (`jellyfin_search`, `jellyfin_get_item`, `jellyfin_get_media_info`, `jellyfin_list_library`).\n"
148
+ "• Ground-Truth Telegram Files: Access all files injected from Telegram with their exact raw file names and message IDs (`telegram_list_injected`). Telegram injected files are the 100% ground truth.\n"
149
+ "• Library Reconciliation & Audit: Compare Telegram ground truth with Jellyfin items to detect rogue un-injected items, duplicate series, and title mismatches (`library_audit_and_compare`).\n"
150
+ "• Automatic Fixing & Cleanup: Reconcile mismatched items (`library_fix_item`), remove rogue items / duplicate series (`library_cleanup_rogue_and_duplicates`), and apply artwork (`jellyfin_apply_artwork`).\n"
151
  "• Search artwork & posters across providers (`artwork_search` for multi-provider search, `tmdb_get_images`, `cinemeta_search`, `cinemeta_get_meta`, `tvmaze_search`, `fanart_get_images`).\n"
152
+ "• Free providers like Cinemeta (IMDb catalog) and TVmaze require zero API keys and find authentic posters/thumbnails for movies and series worldwide.\n"
 
153
  "• Search subtitles on OpenSubtitles (`opensubtitles_search`).\n\n"
154
  "Workflow Guidelines:\n"
155
+ "1. When the user asks to find a movie/series and check or fix its poster/thumbnail (e.g. Khalifa):\n"
156
+ " a) Search Jellyfin (`jellyfin_search`) or Telegram files (`telegram_list_injected`) to find the exact item ID and raw filename.\n"
157
+ " b) Search artwork via `artwork_search`, `cinemeta_search`, or `tmdb_get_images` using the title or IMDb ID (`tt22797990` for Khalifa).\n"
158
+ " c) If requested, fix the item directly (`library_fix_item` or `jellyfin_apply_artwork`) and present the poster options conversationally.\n"
159
+ "2. When the user asks about rogue movies, duplicate series, or mismatched items:\n"
160
+ " a) Run `library_audit_and_compare` to compare Telegram ground truth with Jellyfin.\n"
161
+ " b) Explain the findings clearly and perform cleanup or reconciliation as needed.\n"
162
+ "3. If the user is just chatting or asking a general question, reply warmly, concisely, and conversationally.\n"
163
+ "4. Format all responses using Telegram markdown (*bold*, `code`, [link](url)). Never output raw JSON unless specifically requested."
164
  )
165
 
166
  messages = [{"role": "system", "content": system_prompt}]
ai_admin/jellyfin_client.py CHANGED
@@ -98,11 +98,15 @@ class JellyfinClient:
98
  path: str,
99
  params: Optional[Dict[str, Any]] = None,
100
  json_body: Optional[Dict[str, Any]] = None,
 
 
101
  session: Optional[aiohttp.ClientSession] = None,
102
  ) -> Any:
103
  """Execute HTTP request against Jellyfin API with timeout and payload capping."""
104
  url = f"{self.base_url}{path}"
105
  headers = self._get_auth_headers()
 
 
106
 
107
  close_session = False
108
  if session is None:
@@ -110,19 +114,24 @@ class JellyfinClient:
110
  close_session = True
111
 
112
  try:
113
- async with session.request(
114
- method=method,
115
- url=url,
116
- params=params,
117
- json=json_body,
118
- headers=headers,
119
- allow_redirects=False,
120
- timeout=aiohttp.ClientTimeout(
121
  total=self.timeout_seconds,
122
  connect=min(3.0, self.timeout_seconds),
123
  sock_read=self.timeout_seconds,
124
  ),
125
- ) as resp:
 
 
 
 
 
 
126
  if resp.status in (301, 302, 303, 307, 308):
127
  raise JellyfinError(f"Unexpected redirect from Jellyfin server (HTTP {resp.status}).")
128
  if resp.status in (401, 403):
@@ -246,6 +255,7 @@ class JellyfinClient:
246
  "has_backdrop": len(backdrop_tags) > 0,
247
  "backdrop_count": len(backdrop_tags),
248
  "media_sources_count": len(data.get("MediaSources") or []),
 
249
  "path_basename": os.path.basename(str(data.get("Path") or "")),
250
  }
251
 
@@ -369,3 +379,121 @@ class JellyfinClient:
369
  "total_audio_tracks": len(audio_streams),
370
  "total_subtitle_tracks": len(subtitle_streams),
371
  }
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
98
  path: str,
99
  params: Optional[Dict[str, Any]] = None,
100
  json_body: Optional[Dict[str, Any]] = None,
101
+ raw_body: Optional[bytes] = None,
102
+ content_type: Optional[str] = None,
103
  session: Optional[aiohttp.ClientSession] = None,
104
  ) -> Any:
105
  """Execute HTTP request against Jellyfin API with timeout and payload capping."""
106
  url = f"{self.base_url}{path}"
107
  headers = self._get_auth_headers()
108
+ if content_type:
109
+ headers["Content-Type"] = content_type
110
 
111
  close_session = False
112
  if session is None:
 
114
  close_session = True
115
 
116
  try:
117
+ req_kwargs: Dict[str, Any] = {
118
+ "method": method,
119
+ "url": url,
120
+ "params": params,
121
+ "headers": headers,
122
+ "allow_redirects": False,
123
+ "timeout": aiohttp.ClientTimeout(
 
124
  total=self.timeout_seconds,
125
  connect=min(3.0, self.timeout_seconds),
126
  sock_read=self.timeout_seconds,
127
  ),
128
+ }
129
+ if json_body is not None:
130
+ req_kwargs["json"] = json_body
131
+ elif raw_body is not None:
132
+ req_kwargs["data"] = raw_body
133
+
134
+ async with session.request(**req_kwargs) as resp:
135
  if resp.status in (301, 302, 303, 307, 308):
136
  raise JellyfinError(f"Unexpected redirect from Jellyfin server (HTTP {resp.status}).")
137
  if resp.status in (401, 403):
 
255
  "has_backdrop": len(backdrop_tags) > 0,
256
  "backdrop_count": len(backdrop_tags),
257
  "media_sources_count": len(data.get("MediaSources") or []),
258
+ "path": data.get("Path"),
259
  "path_basename": os.path.basename(str(data.get("Path") or "")),
260
  }
261
 
 
379
  "total_audio_tracks": len(audio_streams),
380
  "total_subtitle_tracks": len(subtitle_streams),
381
  }
382
+
383
+ async def get_all_library_items(
384
+ self,
385
+ include_item_types: str = "Movie,Series",
386
+ limit: int = 1000,
387
+ session: Optional[aiohttp.ClientSession] = None,
388
+ ) -> List[Dict[str, Any]]:
389
+ """Fetch all media items in Jellyfin library with paths and provider IDs."""
390
+ bounded_limit = max(1, min(limit, 5000))
391
+ params = {
392
+ "Recursive": "true",
393
+ "IncludeItemTypes": include_item_types,
394
+ "Fields": "Path,ProviderIds,Overview,ProductionYear,ImageTags,BackdropImageTags,SeriesName,SeasonName",
395
+ "Limit": str(bounded_limit),
396
+ }
397
+ data = await self._request("GET", "/Items", params=params, session=session)
398
+ raw_items = data.get("Items") or []
399
+ results = []
400
+ for it in raw_items:
401
+ results.append({
402
+ "id": it.get("Id"),
403
+ "name": it.get("Name"),
404
+ "type": it.get("Type"),
405
+ "path": it.get("Path") or "",
406
+ "year": it.get("ProductionYear"),
407
+ "overview": it.get("Overview") or "",
408
+ "provider_ids": it.get("ProviderIds") or {},
409
+ "has_poster": bool((it.get("ImageTags") or {}).get("Primary")),
410
+ "has_backdrop": bool(it.get("BackdropImageTags")),
411
+ "series_name": it.get("SeriesName"),
412
+ })
413
+ return results
414
+
415
+ async def update_item_metadata(
416
+ self,
417
+ item_id: str,
418
+ fields: Dict[str, Any],
419
+ session: Optional[aiohttp.ClientSession] = None,
420
+ ) -> Dict[str, Any]:
421
+ """Update metadata fields (Name, ProductionYear, ProviderIds, Overview) for a Jellyfin item."""
422
+ if not validate_item_id(item_id):
423
+ raise ValueError(f"Invalid item_id format: '{item_id}'")
424
+
425
+ # Fetch current item representation first to preserve intact fields
426
+ current = await self._request("GET", f"/Items/{item_id}", session=session)
427
+ if not current:
428
+ raise JellyfinNotFoundError(f"Item '{item_id}' not found.")
429
+
430
+ # Merge allowed fields
431
+ for k, v in fields.items():
432
+ if k == "ProviderIds" and isinstance(v, dict):
433
+ current.setdefault("ProviderIds", {}).update(v)
434
+ else:
435
+ current[k] = v
436
+
437
+ await self._request("POST", f"/Items/{item_id}", json_body=current, session=session)
438
+ return {"id": item_id, "updated": True, "fields": list(fields.keys())}
439
+
440
+ async def set_item_image_bytes(
441
+ self,
442
+ item_id: str,
443
+ image_type: str,
444
+ image_bytes: bytes,
445
+ mime_type: str = "image/jpeg",
446
+ session: Optional[aiohttp.ClientSession] = None,
447
+ ) -> Dict[str, Any]:
448
+ """Set artwork for a Jellyfin item using raw image bytes."""
449
+ if not validate_item_id(item_id):
450
+ raise ValueError(f"Invalid item_id format: '{item_id}'")
451
+
452
+ clean_type = "Primary" if image_type.lower() in ("poster", "primary") else ("Backdrop" if image_type.lower() in ("backdrop", "background", "thumb") else "Logo")
453
+ await self._request(
454
+ "POST",
455
+ f"/Items/{item_id}/Images/{clean_type}",
456
+ raw_body=image_bytes,
457
+ content_type=mime_type,
458
+ session=session,
459
+ )
460
+ return {"id": item_id, "image_type": clean_type, "success": True}
461
+
462
+ async def delete_item(
463
+ self,
464
+ item_id: str,
465
+ session: Optional[aiohttp.ClientSession] = None,
466
+ ) -> Dict[str, Any]:
467
+ """Delete an item from Jellyfin."""
468
+ if not validate_item_id(item_id):
469
+ raise ValueError(f"Invalid item_id format: '{item_id}'")
470
+
471
+ await self._request("DELETE", f"/Items/{item_id}", session=session)
472
+ return {"id": item_id, "deleted": True}
473
+
474
+ async def refresh_item(
475
+ self,
476
+ item_id: str,
477
+ replace_all: bool = True,
478
+ session: Optional[aiohttp.ClientSession] = None,
479
+ ) -> Dict[str, Any]:
480
+ """Trigger metadata and image refresh for an item."""
481
+ if not validate_item_id(item_id):
482
+ raise ValueError(f"Invalid item_id format: '{item_id}'")
483
+
484
+ params = {
485
+ "MetadataRefreshMode": "FullRefresh",
486
+ "ImageRefreshMode": "FullRefresh",
487
+ "ReplaceAllMetadata": "true" if replace_all else "false",
488
+ "ReplaceAllImages": "true" if replace_all else "false",
489
+ }
490
+ await self._request("POST", f"/Items/{item_id}/Refresh", params=params, session=session)
491
+ return {"id": item_id, "refreshed": True}
492
+
493
+ async def refresh_library(
494
+ self,
495
+ session: Optional[aiohttp.ClientSession] = None,
496
+ ) -> Dict[str, Any]:
497
+ """Trigger full library scan in Jellyfin."""
498
+ await self._request("POST", "/Library/Refresh", session=session)
499
+ return {"refreshed": True}
ai_admin/reconciler.py ADDED
@@ -0,0 +1,414 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """Reconciliation engine comparing 100% ground-truth Telegram injected media against Jellyfin library."""
2
+
3
+ import aiohttp
4
+ import asyncio
5
+ import json
6
+ import logging
7
+ import os
8
+ import re
9
+ from typing import Any, Dict, List, Optional, Set, Tuple
10
+
11
+ from ai_admin.jellyfin_client import JellyfinClient
12
+ from ai_admin.net_guard import validate_outbound_url
13
+ from ai_admin.providers_client import ProvidersClient
14
+
15
+ logger = logging.getLogger("AIAdmin.Reconciler")
16
+
17
+ DEFAULT_CACHE_FILE = "/data/jellyfin/file_ids.json"
18
+
19
+
20
+ def load_telegram_injected_catalog(cache_path: str = DEFAULT_CACHE_FILE) -> List[Dict[str, Any]]:
21
+ """Load the ground-truth catalog of media injected from Telegram."""
22
+ if not os.path.exists(cache_path):
23
+ return []
24
+
25
+ try:
26
+ with open(cache_path, "r", encoding="utf-8") as f:
27
+ data = json.load(f)
28
+ except Exception as e:
29
+ logger.warning(f"[RECONCILER] Error reading {cache_path}: {e}")
30
+ return []
31
+
32
+ items = []
33
+ for key, entry in data.items():
34
+ if not isinstance(entry, dict):
35
+ continue
36
+
37
+ raw_name = entry.get("file_name") or entry.get("raw_name") or entry.get("title") or ""
38
+ strm_path = entry.get("strm_path") or ""
39
+
40
+ items.append({
41
+ "key": key,
42
+ "chat_id": entry.get("chat_id"),
43
+ "message_id": entry.get("message_id") or entry.get("msg_id"),
44
+ "raw_name": raw_name,
45
+ "title": entry.get("title") or raw_name,
46
+ "clean_title": entry.get("clean_title") or entry.get("title") or raw_name,
47
+ "year": entry.get("year"),
48
+ "is_tv": bool(entry.get("is_tv")),
49
+ "show_name": entry.get("show_name"),
50
+ "season": entry.get("season"),
51
+ "episode": entry.get("episode"),
52
+ "strm_path": strm_path,
53
+ "tmdb_id": entry.get("tmdb_id"),
54
+ "imdb_id": entry.get("imdb_id"),
55
+ "file_size": entry.get("file_size"),
56
+ "poster_url": entry.get("poster_url"),
57
+ "backdrop_url": entry.get("backdrop_url"),
58
+ })
59
+ return items
60
+
61
+
62
+ async def audit_and_compare(
63
+ jf_client: JellyfinClient,
64
+ cache_path: str = DEFAULT_CACHE_FILE,
65
+ ) -> Dict[str, Any]:
66
+ """Compare 100% ground-truth Telegram injected media with current Jellyfin library.
67
+
68
+ Identifies:
69
+ 1. Rogue / un-injected items in Jellyfin.
70
+ 2. Duplicate series or duplicate movies.
71
+ 3. Metadata mismatches (where Jellyfin title/year diverges from Telegram file).
72
+ 4. Missing artwork (items without posters or backdrops).
73
+ """
74
+ injected_files = load_telegram_injected_catalog(cache_path)
75
+ jellyfin_items = await jf_client.get_all_library_items(include_item_types="Movie,Series,Episode", limit=2000)
76
+
77
+ # Normalize strm paths from Telegram catalog
78
+ injected_by_strm: Dict[str, Dict[str, Any]] = {}
79
+ injected_by_name: Dict[str, List[Dict[str, Any]]] = {}
80
+
81
+ for inj in injected_files:
82
+ strm = os.path.normpath(str(inj.get("strm_path") or "")).lower()
83
+ if strm and strm.endswith(".strm"):
84
+ injected_by_strm[strm] = inj
85
+ base = os.path.splitext(os.path.basename(strm))[0].lower()
86
+ injected_by_strm[base] = inj
87
+
88
+ clean_name = re.sub(r"[^a-z0-9]+", " ", str(inj.get("clean_title") or inj.get("title") or "")).strip().lower()
89
+ if clean_name:
90
+ injected_by_name.setdefault(clean_name, []).append(inj)
91
+
92
+ rogue_items = []
93
+ duplicate_series = []
94
+ duplicate_movies = []
95
+ mismatches = []
96
+ missing_artwork = []
97
+
98
+ series_seen: Dict[str, List[Dict[str, Any]]] = {}
99
+ movie_seen: Dict[str, List[Dict[str, Any]]] = {}
100
+
101
+ for it in jellyfin_items:
102
+ itype = it.get("Type")
103
+ item_id = it.get("id")
104
+ name = it.get("name") or "Unknown"
105
+ path = os.path.normpath(str(it.get("path") or "")).lower()
106
+ prov_ids = it.get("provider_ids") or {}
107
+
108
+ # 1. Check duplicate Series
109
+ if itype == "Series":
110
+ norm_series = re.sub(r"[^a-z0-9]+", " ", name).strip().lower()
111
+ series_seen.setdefault(norm_series, []).append(it)
112
+ if not it.get("has_poster"):
113
+ missing_artwork.append({
114
+ "id": item_id,
115
+ "title": name,
116
+ "type": "Series",
117
+ "missing": "poster",
118
+ })
119
+ continue
120
+
121
+ # 2. Check Movies and Episodes
122
+ if itype == "Movie":
123
+ norm_movie = re.sub(r"[^a-z0-9]+", " ", name).strip().lower()
124
+ movie_seen.setdefault(norm_movie, []).append(it)
125
+
126
+ if not it.get("has_poster"):
127
+ missing_artwork.append({
128
+ "id": item_id,
129
+ "title": name,
130
+ "type": "Movie",
131
+ "missing": "poster",
132
+ })
133
+
134
+ # Match Jellyfin item to Telegram injection
135
+ matched_inj = injected_by_strm.get(path)
136
+ if not matched_inj and path:
137
+ base_fname = os.path.splitext(os.path.basename(path))[0].lower()
138
+ matched_inj = injected_by_strm.get(base_fname)
139
+
140
+ if not matched_inj:
141
+ # Check name match
142
+ norm_name = re.sub(r"[^a-z0-9]+", " ", name).strip().lower()
143
+ name_hits = injected_by_name.get(norm_name) or []
144
+ if name_hits:
145
+ matched_inj = name_hits[0]
146
+
147
+ if not matched_inj and path and path.endswith(".strm"):
148
+ # Rogue item in Jellyfin with no Telegram injection
149
+ rogue_items.append({
150
+ "id": item_id,
151
+ "title": name,
152
+ "type": itype,
153
+ "path": it.get("path"),
154
+ "reason": "No corresponding Telegram injected file found",
155
+ })
156
+ elif matched_inj:
157
+ # Check for title/TMDb mismatch
158
+ inj_raw = matched_inj.get("raw_name") or ""
159
+ inj_title = matched_inj.get("title") or ""
160
+ inj_tmdb = matched_inj.get("tmdb_id")
161
+ jf_tmdb = prov_ids.get("Tmdb")
162
+
163
+ norm_jf = re.sub(r"[^a-z0-9]+", "", name.lower())
164
+ norm_raw = re.sub(r"[^a-z0-9]+", "", inj_title.lower())
165
+
166
+ # Divergence check
167
+ if norm_raw and norm_jf and norm_raw not in norm_jf and norm_jf not in norm_raw:
168
+ mismatches.append({
169
+ "id": item_id,
170
+ "jellyfin_title": name,
171
+ "telegram_raw_name": inj_raw,
172
+ "expected_title": inj_title,
173
+ "expected_year": matched_inj.get("year"),
174
+ "jellyfin_tmdb": jf_tmdb,
175
+ "expected_tmdb": inj_tmdb,
176
+ "expected_imdb": matched_inj.get("imdb_id"),
177
+ })
178
+
179
+ # Collect duplicate Series
180
+ for s_name, s_list in series_seen.items():
181
+ if len(s_list) > 1:
182
+ duplicate_series.append({
183
+ "series_name": s_list[0].get("name"),
184
+ "count": len(s_list),
185
+ "items": [{"id": x.get("id"), "path": x.get("path")} for x in s_list],
186
+ })
187
+
188
+ # Collect duplicate Movies
189
+ for m_name, m_list in movie_seen.items():
190
+ if len(m_list) > 1:
191
+ duplicate_movies.append({
192
+ "movie_name": m_list[0].get("name"),
193
+ "count": len(m_list),
194
+ "items": [{"id": x.get("id"), "path": x.get("path")} for x in m_list],
195
+ })
196
+
197
+ return {
198
+ "total_telegram_injected": len(injected_files),
199
+ "total_jellyfin_items": len(jellyfin_items),
200
+ "rogue_items_count": len(rogue_items),
201
+ "rogue_items": rogue_items[:20],
202
+ "duplicate_series_count": len(duplicate_series),
203
+ "duplicate_series": duplicate_series[:10],
204
+ "duplicate_movies_count": len(duplicate_movies),
205
+ "duplicate_movies": duplicate_movies[:10],
206
+ "mismatches_count": len(mismatches),
207
+ "mismatches": mismatches[:20],
208
+ "missing_artwork_count": len(missing_artwork),
209
+ "missing_artwork": missing_artwork[:20],
210
+ }
211
+
212
+
213
+ async def apply_artwork_from_url(
214
+ jf_client: JellyfinClient,
215
+ item_id: str,
216
+ image_url: str,
217
+ image_type: str = "Primary",
218
+ ) -> Dict[str, Any]:
219
+ """Download image safely and upload directly to a Jellyfin item."""
220
+ is_safe, reason = validate_outbound_url(image_url)
221
+ if not is_safe:
222
+ raise ValueError(f"Artwork URL rejected by NetGuard: {reason}")
223
+
224
+ clean_type = "Primary" if image_type.lower() in ("poster", "primary") else "Backdrop"
225
+ mime = "image/jpeg"
226
+
227
+ async with aiohttp.ClientSession() as session:
228
+ async with session.get(
229
+ image_url,
230
+ timeout=aiohttp.ClientTimeout(total=15.0),
231
+ headers={"User-Agent": "Apex-ArtworkFetcher/1.0"},
232
+ ) as resp:
233
+ if resp.status != 200:
234
+ raise ValueError(f"Failed to download image (HTTP {resp.status}) from {image_url}")
235
+ content_type = resp.headers.get("Content-Type", "image/jpeg")
236
+ if "png" in content_type:
237
+ mime = "image/png"
238
+ img_bytes = await resp.read()
239
+
240
+ if len(img_bytes) < 100 or len(img_bytes) > 15 * 1024 * 1024:
241
+ raise ValueError(f"Invalid image payload size: {len(img_bytes)} bytes")
242
+
243
+ # Upload to Jellyfin
244
+ await jf_client.set_item_image_bytes(item_id, clean_type, img_bytes, mime_type=mime)
245
+ await jf_client.refresh_item(item_id, replace_all=False)
246
+ return {"id": item_id, "image_type": clean_type, "status": "applied", "url": image_url}
247
+
248
+
249
+ async def reconcile_item(
250
+ jf_client: JellyfinClient,
251
+ item_id: str,
252
+ title: Optional[str] = None,
253
+ year: Optional[int] = None,
254
+ tmdb_id: Optional[int] = None,
255
+ imdb_id: Optional[str] = None,
256
+ overview: Optional[str] = None,
257
+ poster_url: Optional[str] = None,
258
+ backdrop_url: Optional[str] = None,
259
+ cache_path: str = DEFAULT_CACHE_FILE,
260
+ raw_name: Optional[str] = None,
261
+ cache_key: Optional[str] = None,
262
+ ) -> Dict[str, Any]:
263
+ """Correct an item's metadata and artwork in Jellyfin and synchronize with FILE_ID_CACHE."""
264
+ current_name = None
265
+ current_path = None
266
+ try:
267
+ cur_item = await jf_client.get_item(item_id)
268
+ if cur_item:
269
+ current_name = cur_item.get("name")
270
+ current_path = os.path.normpath(str(cur_item.get("path") or "")).lower()
271
+ except Exception:
272
+ pass
273
+
274
+ fields: Dict[str, Any] = {}
275
+ if title:
276
+ fields["Name"] = title
277
+ if year:
278
+ fields["ProductionYear"] = int(year)
279
+ if overview:
280
+ fields["Overview"] = overview
281
+
282
+ prov_ids: Dict[str, str] = {}
283
+ if tmdb_id:
284
+ prov_ids["Tmdb"] = str(tmdb_id)
285
+ if imdb_id:
286
+ prov_ids["Imdb"] = str(imdb_id)
287
+ if prov_ids:
288
+ fields["ProviderIds"] = prov_ids
289
+
290
+ # 1. Update metadata in Jellyfin
291
+ if fields:
292
+ await jf_client.update_item_metadata(item_id, fields)
293
+
294
+ # 2. Apply artwork if provided
295
+ applied_art = []
296
+ if poster_url:
297
+ try:
298
+ res = await apply_artwork_from_url(jf_client, item_id, poster_url, image_type="Primary")
299
+ applied_art.append(res)
300
+ except Exception as e:
301
+ logger.warning(f"[RECONCILE] Failed to apply poster: {e}")
302
+
303
+ if backdrop_url:
304
+ try:
305
+ res = await apply_artwork_from_url(jf_client, item_id, backdrop_url, image_type="Backdrop")
306
+ applied_art.append(res)
307
+ except Exception as e:
308
+ logger.warning(f"[RECONCILE] Failed to apply backdrop: {e}")
309
+
310
+ # 3. Synchronize FILE_ID_CACHE on disk
311
+ if os.path.exists(cache_path):
312
+ try:
313
+ with open(cache_path, "r", encoding="utf-8") as f:
314
+ cache_data = json.load(f)
315
+
316
+ updated = False
317
+ for key, entry in cache_data.items():
318
+ if not isinstance(entry, dict):
319
+ continue
320
+ match = False
321
+ if cache_key and key == cache_key:
322
+ match = True
323
+ elif raw_name and raw_name.lower() in str(entry.get("raw_name") or entry.get("file_name") or "").lower():
324
+ match = True
325
+ elif current_path:
326
+ strm = os.path.normpath(str(entry.get("strm_path") or "")).lower()
327
+ if strm and (strm == current_path or os.path.basename(strm) == os.path.basename(current_path)):
328
+ match = True
329
+ if not match and current_name and (current_name.lower() in (entry.get("title") or "").lower() or (entry.get("title") or "").lower() in current_name.lower()):
330
+ match = True
331
+ if not match and title and title.lower() in (entry.get("title") or "").lower():
332
+ match = True
333
+ if not match and tmdb_id and entry.get("tmdb_id") == tmdb_id:
334
+ match = True
335
+
336
+ if match:
337
+ if title:
338
+ entry["title"] = title
339
+ entry["clean_title"] = title
340
+ if year:
341
+ entry["year"] = int(year)
342
+ if tmdb_id:
343
+ entry["tmdb_id"] = int(tmdb_id)
344
+ if imdb_id:
345
+ entry["imdb_id"] = str(imdb_id)
346
+ if poster_url:
347
+ entry["poster_url"] = poster_url
348
+ if backdrop_url:
349
+ entry["backdrop_url"] = backdrop_url
350
+ updated = True
351
+
352
+ if updated:
353
+ with open(cache_path, "w", encoding="utf-8") as f:
354
+ json.dump(cache_data, f, indent=2)
355
+ except Exception as e:
356
+ logger.warning(f"[RECONCILE] Error updating {cache_path}: {e}")
357
+
358
+ await jf_client.refresh_item(item_id, replace_all=True)
359
+ return {
360
+ "id": item_id,
361
+ "title": title,
362
+ "year": year,
363
+ "tmdb_id": tmdb_id,
364
+ "imdb_id": imdb_id,
365
+ "artwork": applied_art,
366
+ "status": "reconciled",
367
+ }
368
+
369
+
370
+ async def cleanup_rogue_and_duplicates(
371
+ jf_client: JellyfinClient,
372
+ remove_rogue: bool = True,
373
+ deduplicate_series: bool = True,
374
+ cache_path: str = DEFAULT_CACHE_FILE,
375
+ ) -> Dict[str, Any]:
376
+ """Clean up rogue un-injected items and merge duplicate series in Jellyfin."""
377
+ audit = await audit_and_compare(jf_client, cache_path=cache_path)
378
+
379
+ deleted_rogue = []
380
+ if remove_rogue:
381
+ for r in audit.get("rogue_items", []):
382
+ rid = r.get("id")
383
+ if rid:
384
+ try:
385
+ await jf_client.delete_item(rid)
386
+ deleted_rogue.append(r)
387
+ except Exception as e:
388
+ logger.warning(f"[CLEANUP] Failed to delete rogue item {rid}: {e}")
389
+
390
+ deleted_duplicate_series = []
391
+ if deduplicate_series:
392
+ for s in audit.get("duplicate_series", []):
393
+ items = s.get("items") or []
394
+ if len(items) > 1:
395
+ # Keep the first, delete the rest
396
+ for dup in items[1:]:
397
+ dup_id = dup.get("id")
398
+ if dup_id:
399
+ try:
400
+ await jf_client.delete_item(dup_id)
401
+ deleted_duplicate_series.append(dup)
402
+ except Exception as e:
403
+ logger.warning(f"[CLEANUP] Failed to delete duplicate series {dup_id}: {e}")
404
+
405
+ # Trigger library refresh
406
+ await jf_client.refresh_library()
407
+
408
+ return {
409
+ "deleted_rogue_count": len(deleted_rogue),
410
+ "deleted_rogue_items": deleted_rogue,
411
+ "deleted_duplicate_series_count": len(deleted_duplicate_series),
412
+ "deleted_duplicate_series": deleted_duplicate_series,
413
+ "status": "cleanup_complete",
414
+ }
ai_admin/tools_jellyfin.py CHANGED
@@ -5,6 +5,13 @@ from typing import Any, Dict, List, Optional
5
 
6
  from ai_admin.jellyfin_client import JellyfinClient
7
  from ai_admin.mutation_contracts import MUTATION_CONTRACTS
 
 
 
 
 
 
 
8
  from ai_admin.tool_registry import ToolDefinition, ToolRegistry
9
  from ai_admin.validators import validate_item_id
10
 
@@ -57,6 +64,105 @@ def make_jellyfin_get_media_info_handler(client: JellyfinClient):
57
  return handler
58
 
59
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
60
  # ==============================================================================
61
  # 2. Disabled Mutation Placeholders (Fail-closed)
62
  # ==============================================================================
@@ -157,6 +263,141 @@ def get_jellyfin_read_tool_definitions(client: JellyfinClient) -> List[ToolDefin
157
  handler=make_jellyfin_get_media_info_handler(client),
158
  validators=[lambda p: validate_item_id(p.get("item_id", ""))],
159
  ),
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
160
  ]
161
 
162
 
 
5
 
6
  from ai_admin.jellyfin_client import JellyfinClient
7
  from ai_admin.mutation_contracts import MUTATION_CONTRACTS
8
+ from ai_admin.reconciler import (
9
+ apply_artwork_from_url,
10
+ audit_and_compare,
11
+ cleanup_rogue_and_duplicates,
12
+ load_telegram_injected_catalog,
13
+ reconcile_item,
14
+ )
15
  from ai_admin.tool_registry import ToolDefinition, ToolRegistry
16
  from ai_admin.validators import validate_item_id
17
 
 
64
  return handler
65
 
66
 
67
+ def make_telegram_list_injected_handler():
68
+ async def handler(params: Dict[str, Any]) -> Dict[str, Any]:
69
+ query = (params.get("query") or "").strip().lower()
70
+ filter_type = params.get("filter_type")
71
+ items = load_telegram_injected_catalog()
72
+
73
+ if query:
74
+ items = [
75
+ it for it in items
76
+ if query in (it.get("raw_name") or "").lower()
77
+ or query in (it.get("title") or "").lower()
78
+ or query in (it.get("show_name") or "").lower()
79
+ ]
80
+
81
+ if filter_type == "movie":
82
+ items = [it for it in items if not it.get("is_tv")]
83
+ elif filter_type == "tv":
84
+ items = [it for it in items if it.get("is_tv")]
85
+
86
+ return {
87
+ "count": len(items),
88
+ "items": items[:params.get("limit", 50)],
89
+ }
90
+ return handler
91
+
92
+
93
+ def make_jellyfin_list_library_handler(client: JellyfinClient):
94
+ async def handler(params: Dict[str, Any]) -> Dict[str, Any]:
95
+ item_types = params.get("item_types", "Movie,Series")
96
+ limit = params.get("limit", 100)
97
+ items = await client.get_all_library_items(include_item_types=item_types, limit=limit)
98
+ return {
99
+ "count": len(items),
100
+ "items": items,
101
+ }
102
+ return handler
103
+
104
+
105
+ def make_library_audit_and_compare_handler(client: JellyfinClient):
106
+ async def handler(params: Dict[str, Any]) -> Dict[str, Any]:
107
+ audit = await audit_and_compare(jf_client=client)
108
+ return audit
109
+ return handler
110
+
111
+
112
+ def make_library_fix_item_handler(client: JellyfinClient):
113
+ async def handler(params: Dict[str, Any]) -> Dict[str, Any]:
114
+ item_id = params.get("item_id", "")
115
+ title = params.get("title")
116
+ year = params.get("year")
117
+ tmdb_id = params.get("tmdb_id")
118
+ imdb_id = params.get("imdb_id")
119
+ overview = params.get("overview")
120
+ poster_url = params.get("poster_url")
121
+ backdrop_url = params.get("backdrop_url")
122
+
123
+ res = await reconcile_item(
124
+ jf_client=client,
125
+ item_id=item_id,
126
+ title=title,
127
+ year=year,
128
+ tmdb_id=tmdb_id,
129
+ imdb_id=imdb_id,
130
+ overview=overview,
131
+ poster_url=poster_url,
132
+ backdrop_url=backdrop_url,
133
+ )
134
+ return res
135
+ return handler
136
+
137
+
138
+ def make_library_cleanup_rogue_and_duplicates_handler(client: JellyfinClient):
139
+ async def handler(params: Dict[str, Any]) -> Dict[str, Any]:
140
+ remove_rogue = params.get("remove_rogue", True)
141
+ deduplicate_series = params.get("deduplicate_series", True)
142
+ res = await cleanup_rogue_and_duplicates(
143
+ jf_client=client,
144
+ remove_rogue=remove_rogue,
145
+ deduplicate_series=deduplicate_series,
146
+ )
147
+ return res
148
+ return handler
149
+
150
+
151
+ def make_jellyfin_apply_artwork_handler(client: JellyfinClient):
152
+ async def handler(params: Dict[str, Any]) -> Dict[str, Any]:
153
+ item_id = params.get("item_id", "")
154
+ image_url = params.get("image_url", "")
155
+ image_type = params.get("image_type", "Primary")
156
+ res = await apply_artwork_from_url(
157
+ jf_client=client,
158
+ item_id=item_id,
159
+ image_url=image_url,
160
+ image_type=image_type,
161
+ )
162
+ return res
163
+ return handler
164
+
165
+
166
  # ==============================================================================
167
  # 2. Disabled Mutation Placeholders (Fail-closed)
168
  # ==============================================================================
 
263
  handler=make_jellyfin_get_media_info_handler(client),
264
  validators=[lambda p: validate_item_id(p.get("item_id", ""))],
265
  ),
266
+ ToolDefinition(
267
+ name="telegram_list_injected",
268
+ purpose="List raw media files injected from Telegram, with original file names, message IDs, and parsed details.",
269
+ json_schema={
270
+ "type": "object",
271
+ "properties": {
272
+ "query": {"type": "string", "maxLength": 100},
273
+ "filter_type": {"type": "string", "maxLength": 10},
274
+ "limit": {"type": "integer"},
275
+ },
276
+ "additionalProperties": False,
277
+ },
278
+ read_only=True,
279
+ mutates=False,
280
+ risk="low",
281
+ requires_confirmation="never",
282
+ permission="read",
283
+ timeout_seconds=10,
284
+ max_output_chars=4000,
285
+ handler=make_telegram_list_injected_handler(),
286
+ ),
287
+ ToolDefinition(
288
+ name="jellyfin_list_library",
289
+ purpose="List all media items currently in Jellyfin library, with item IDs, titles, and paths.",
290
+ json_schema={
291
+ "type": "object",
292
+ "properties": {
293
+ "item_types": {"type": "string", "maxLength": 50},
294
+ "limit": {"type": "integer"},
295
+ },
296
+ "additionalProperties": False,
297
+ },
298
+ read_only=True,
299
+ mutates=False,
300
+ risk="low",
301
+ requires_confirmation="never",
302
+ permission="read",
303
+ timeout_seconds=20,
304
+ max_output_chars=4000,
305
+ handler=make_jellyfin_list_library_handler(client),
306
+ ),
307
+ ToolDefinition(
308
+ name="library_audit_and_compare",
309
+ purpose="Compare Telegram injected files (the 100% ground truth) with Jellyfin items to detect rogue movies, duplicate series, and mismatches.",
310
+ json_schema={
311
+ "type": "object",
312
+ "properties": {},
313
+ "additionalProperties": False,
314
+ },
315
+ read_only=True,
316
+ mutates=False,
317
+ risk="low",
318
+ requires_confirmation="never",
319
+ permission="read",
320
+ timeout_seconds=30,
321
+ max_output_chars=4000,
322
+ handler=make_library_audit_and_compare_handler(client),
323
+ ),
324
+ ]
325
+
326
+
327
+ def get_jellyfin_active_mutation_definitions(client: JellyfinClient) -> List[ToolDefinition]:
328
+ """Tool definitions for active mutating operations when AI_MUTATIONS_ENABLED=true."""
329
+ return [
330
+ ToolDefinition(
331
+ name="library_fix_item",
332
+ purpose="Correct title, year, TMDb ID, IMDb ID, or authentic artwork for a Jellyfin item and update cache.",
333
+ json_schema={
334
+ "type": "object",
335
+ "properties": {
336
+ "item_id": {"type": "string", "maxLength": 64},
337
+ "title": {"type": "string", "maxLength": 100},
338
+ "year": {"type": "integer"},
339
+ "tmdb_id": {"type": "integer"},
340
+ "imdb_id": {"type": "string", "maxLength": 20},
341
+ "overview": {"type": "string", "maxLength": 1000},
342
+ "poster_url": {"type": "string", "maxLength": 300},
343
+ "backdrop_url": {"type": "string", "maxLength": 300},
344
+ },
345
+ "required": ["item_id"],
346
+ "additionalProperties": False,
347
+ },
348
+ read_only=False,
349
+ mutates=True,
350
+ risk="medium",
351
+ requires_confirmation="never",
352
+ permission="write",
353
+ timeout_seconds=30,
354
+ max_output_chars=4000,
355
+ handler=make_library_fix_item_handler(client),
356
+ validators=[lambda p: validate_item_id(p.get("item_id", ""))],
357
+ ),
358
+ ToolDefinition(
359
+ name="library_cleanup_rogue_and_duplicates",
360
+ purpose="Delete rogue un-injected items and merge/clean duplicate series in Jellyfin.",
361
+ json_schema={
362
+ "type": "object",
363
+ "properties": {
364
+ "remove_rogue": {"type": "boolean"},
365
+ "deduplicate_series": {"type": "boolean"},
366
+ },
367
+ "additionalProperties": False,
368
+ },
369
+ read_only=False,
370
+ mutates=True,
371
+ risk="medium",
372
+ requires_confirmation="never",
373
+ permission="write",
374
+ timeout_seconds=30,
375
+ max_output_chars=4000,
376
+ handler=make_library_cleanup_rogue_and_duplicates_handler(client),
377
+ ),
378
+ ToolDefinition(
379
+ name="jellyfin_apply_artwork",
380
+ purpose="Download and apply authentic poster or backdrop image from URL to a Jellyfin item.",
381
+ json_schema={
382
+ "type": "object",
383
+ "properties": {
384
+ "item_id": {"type": "string", "maxLength": 64},
385
+ "image_url": {"type": "string", "maxLength": 300},
386
+ "image_type": {"type": "string", "maxLength": 20},
387
+ },
388
+ "required": ["item_id", "image_url"],
389
+ "additionalProperties": False,
390
+ },
391
+ read_only=False,
392
+ mutates=True,
393
+ risk="medium",
394
+ requires_confirmation="never",
395
+ permission="write",
396
+ timeout_seconds=25,
397
+ max_output_chars=4000,
398
+ handler=make_jellyfin_apply_artwork_handler(client),
399
+ validators=[lambda p: validate_item_id(p.get("item_id", ""))],
400
+ ),
401
  ]
402
 
403
 
apex/tg_streamer.py CHANGED
@@ -230,7 +230,33 @@ def load_cache():
230
  cache_updated = True
231
  logger.info(f"[CACHE-HEAL] Corrected Generation Iron to real Iron Man 3 for '{raw_name}'")
232
 
233
- # 2. Clean release-size noise from title, clean_title, strm_base
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
234
  cur_title = data.get("title") or ""
235
  cur_clean = data.get("clean_title") or ""
236
  cur_base = data.get("strm_base") or ""
@@ -2938,19 +2964,46 @@ async def deduplicate_jellyfin_items():
2938
  payload = await response.json()
2939
  items = payload.get("Items") or []
2940
  groups = {}
 
 
 
 
 
 
 
 
 
 
2941
  for item in items:
 
2942
  path = os.path.normpath(str(item.get("Path") or ""))
 
 
 
 
 
 
 
 
 
 
2943
  if not path.lower().endswith(".strm"):
2944
  continue
2945
  if not (path.startswith(os.path.normpath(MOVIES_DIR)) or
2946
  path.startswith(os.path.normpath(SHOWS_DIR))):
2947
  continue
 
 
 
 
 
 
2948
  item_name = str(item.get("Name") or "").strip().lower().replace(" ", "_")
2949
  path_name = os.path.basename(path).lower().replace(" ", "_")
2950
  if item_name in {"unknown_media", "unknown"} or "unknown_media" in path_name:
2951
  groups.setdefault("unknown-media", []).append((item, path))
2952
  continue
2953
- provider_ids = item.get("ProviderIds") or {}
2954
  year_match = re.search(r"\((\d{4})\)", path)
2955
  year_suffix = year_match.group(1) if year_match else ""
2956
  if provider_ids.get("Tmdb"):
@@ -2979,7 +3032,17 @@ async def deduplicate_jellyfin_items():
2979
  + f":{year_suffix}"
2980
  )
2981
  groups.setdefault(identity, []).append((item, path))
2982
- duplicates = []
 
 
 
 
 
 
 
 
 
 
2983
  for identity, entries in groups.items():
2984
  if identity == "unknown-media":
2985
  duplicates.extend(entries)
@@ -2987,6 +3050,7 @@ async def deduplicate_jellyfin_items():
2987
  existing = [pair for pair in entries if os.path.exists(pair[1])]
2988
  keep = sorted(existing or entries, key=lambda pair: len(pair[1]))[0]
2989
  duplicates.extend(pair for pair in entries if pair is not keep)
 
2990
  for item, path in duplicates:
2991
  item_id = item.get("Id")
2992
  if not item_id:
 
230
  cache_updated = True
231
  logger.info(f"[CACHE-HEAL] Corrected Generation Iron to real Iron Man 3 for '{raw_name}'")
232
 
233
+ # 2. Heal Khalifa misidentification and assign authentic metadata & artwork
234
+ is_khalifa = (
235
+ "khalifa" in raw_lower
236
+ or "khalifa" in (data.get("tmdb_title") or "").lower()
237
+ or "khalifa" in (data.get("title") or "").lower()
238
+ or "khalifa" in (data.get("strm_base") or "").lower()
239
+ )
240
+ if is_khalifa:
241
+ data["title"] = "Khalifa"
242
+ data["clean_title"] = "Khalifa"
243
+ data["search_title"] = "khalifa"
244
+ data["tmdb_title"] = "Khalifa"
245
+ data["tmdb_id"] = 1036081
246
+ data["imdb_id"] = "tt22797990"
247
+ data["year"] = 2026
248
+ data["tmdb_year"] = 2026
249
+ data["strm_base"] = "Khalifa (2026)"
250
+ lang = data.get("language") or "Malayalam"
251
+ data["target_dir"] = os.path.join(MOVIES_DIR, lang, "Khalifa (2026)")
252
+ data["strm_path"] = os.path.join(data["target_dir"], "Khalifa (2026).strm")
253
+ data["poster_url"] = "https://images.metahub.space/poster/medium/tt22797990/img"
254
+ data["backdrop_url"] = "https://images.metahub.space/background/medium/tt22797990/img"
255
+ data["_force_art_refresh"] = True
256
+ cache_updated = True
257
+ logger.info(f"[CACHE-HEAL] Corrected Khalifa metadata and artwork for '{raw_name}'")
258
+
259
+ # 3. Clean release-size noise from title, clean_title, strm_base
260
  cur_title = data.get("title") or ""
261
  cur_clean = data.get("clean_title") or ""
262
  cur_base = data.get("strm_base") or ""
 
2964
  payload = await response.json()
2965
  items = payload.get("Items") or []
2966
  groups = {}
2967
+ series_groups = {}
2968
+ orphan_items = []
2969
+
2970
+ # Valid cached strm paths
2971
+ valid_cached_paths = {
2972
+ os.path.normpath(str(d.get("strm_path") or "")).lower()
2973
+ for d in FILE_ID_CACHE.values()
2974
+ if isinstance(d, dict) and d.get("strm_path")
2975
+ }
2976
+
2977
  for item in items:
2978
+ itype = item.get("Type")
2979
  path = os.path.normpath(str(item.get("Path") or ""))
2980
+ provider_ids = item.get("ProviderIds") or {}
2981
+
2982
+ # 1. Series deduplication
2983
+ if itype == "Series":
2984
+ s_name = str(item.get("Name") or "").strip().lower()
2985
+ tmdb_s = provider_ids.get("Tmdb")
2986
+ s_ident = f"series:tmdb:{tmdb_s}" if tmdb_s else f"series:name:{s_name}"
2987
+ series_groups.setdefault(s_ident, []).append((item, path))
2988
+ continue
2989
+
2990
  if not path.lower().endswith(".strm"):
2991
  continue
2992
  if not (path.startswith(os.path.normpath(MOVIES_DIR)) or
2993
  path.startswith(os.path.normpath(SHOWS_DIR))):
2994
  continue
2995
+
2996
+ # If .strm does not exist on disk, mark immediately as orphan for cleanup
2997
+ if not os.path.exists(path):
2998
+ orphan_items.append((item, path))
2999
+ continue
3000
+
3001
  item_name = str(item.get("Name") or "").strip().lower().replace(" ", "_")
3002
  path_name = os.path.basename(path).lower().replace(" ", "_")
3003
  if item_name in {"unknown_media", "unknown"} or "unknown_media" in path_name:
3004
  groups.setdefault("unknown-media", []).append((item, path))
3005
  continue
3006
+
3007
  year_match = re.search(r"\((\d{4})\)", path)
3008
  year_suffix = year_match.group(1) if year_match else ""
3009
  if provider_ids.get("Tmdb"):
 
3032
  + f":{year_suffix}"
3033
  )
3034
  groups.setdefault(identity, []).append((item, path))
3035
+
3036
+ duplicates = list(orphan_items)
3037
+
3038
+ # Process Series duplicates
3039
+ for s_ident, s_entries in series_groups.items():
3040
+ if len(s_entries) > 1:
3041
+ existing_s = [pair for pair in s_entries if os.path.exists(pair[1])]
3042
+ # Keep series with the deepest or existing folder path
3043
+ keep_s = sorted(existing_s or s_entries, key=lambda pair: len(pair[1]), reverse=True)[0]
3044
+ duplicates.extend(pair for pair in s_entries if pair is not keep_s)
3045
+
3046
  for identity, entries in groups.items():
3047
  if identity == "unknown-media":
3048
  duplicates.extend(entries)
 
3050
  existing = [pair for pair in entries if os.path.exists(pair[1])]
3051
  keep = sorted(existing or entries, key=lambda pair: len(pair[1]))[0]
3052
  duplicates.extend(pair for pair in entries if pair is not keep)
3053
+
3054
  for item, path in duplicates:
3055
  item_id = item.get("Id")
3056
  if not item_id:
tests/ai_admin/test_reconciler.py ADDED
@@ -0,0 +1,222 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ import json
2
+ import os
3
+ import pytest
4
+ from unittest.mock import AsyncMock, MagicMock, patch
5
+
6
+ from ai_admin.jellyfin_client import JellyfinClient
7
+ from ai_admin.reconciler import (
8
+ load_telegram_injected_catalog,
9
+ audit_and_compare,
10
+ reconcile_item,
11
+ cleanup_rogue_and_duplicates,
12
+ apply_artwork_from_url,
13
+ )
14
+
15
+
16
+ @pytest.fixture
17
+ def mock_jf_client():
18
+ client = MagicMock(spec=JellyfinClient)
19
+ client.get_all_library_items = AsyncMock()
20
+ client.get_item = AsyncMock(return_value={"id": "item-123", "name": "Old Title", "path": "/data/jellyfin/media/Movies/Old.strm"})
21
+ client.update_item_metadata = AsyncMock(return_value={"status": "ok"})
22
+ client.set_item_image_bytes = AsyncMock(return_value=True)
23
+ client.refresh_item = AsyncMock(return_value=True)
24
+ client.refresh_library = AsyncMock(return_value=True)
25
+ client.delete_item = AsyncMock(return_value=True)
26
+ return client
27
+
28
+
29
+ def test_load_telegram_injected_catalog_nonexistent(tmp_path):
30
+ res = load_telegram_injected_catalog(str(tmp_path / "nonexistent.json"))
31
+ assert res == []
32
+
33
+
34
+ def test_load_telegram_injected_catalog_valid(tmp_path):
35
+ cache_file = tmp_path / "file_ids.json"
36
+ cache_data = {
37
+ "-100123:456": {
38
+ "chat_id": -100123,
39
+ "message_id": 456,
40
+ "file_name": "Khalifa.2026.1080p.mkv",
41
+ "title": "Khalifa",
42
+ "clean_title": "Khalifa",
43
+ "year": 2026,
44
+ "is_tv": False,
45
+ "strm_path": "/data/jellyfin/media/Movies/Khalifa (2026)/Khalifa.strm",
46
+ "tmdb_id": 1036081,
47
+ "imdb_id": "tt22797990",
48
+ "file_size": 1500000000,
49
+ "poster_url": "https://images.metahub.space/poster/medium/tt22797990/img",
50
+ }
51
+ }
52
+ cache_file.write_text(json.dumps(cache_data))
53
+
54
+ catalog = load_telegram_injected_catalog(str(cache_file))
55
+ assert len(catalog) == 1
56
+ assert catalog[0]["title"] == "Khalifa"
57
+ assert catalog[0]["tmdb_id"] == 1036081
58
+ assert catalog[0]["strm_path"] == "/data/jellyfin/media/Movies/Khalifa (2026)/Khalifa.strm"
59
+
60
+
61
+ @pytest.mark.asyncio
62
+ async def test_audit_and_compare(mock_jf_client, tmp_path):
63
+ cache_file = tmp_path / "file_ids.json"
64
+ cache_data = {
65
+ "-1001:1": {
66
+ "chat_id": -1001,
67
+ "message_id": 1,
68
+ "file_name": "Inception.2010.mkv",
69
+ "title": "Inception",
70
+ "year": 2010,
71
+ "is_tv": False,
72
+ "strm_path": "/data/jellyfin/media/Movies/Inception (2010)/Inception.strm",
73
+ "tmdb_id": 27205,
74
+ }
75
+ }
76
+ cache_file.write_text(json.dumps(cache_data))
77
+
78
+ # Mock Jellyfin returning:
79
+ # 1. Matching movie (Inception) without poster
80
+ # 2. Rogue movie (Random.strm)
81
+ # 3. Duplicate series (Show A x2)
82
+ mock_jf_client.get_all_library_items.return_value = [
83
+ {
84
+ "id": "item-1",
85
+ "name": "Inception",
86
+ "Type": "Movie",
87
+ "path": "/data/jellyfin/media/Movies/Inception (2010)/Inception.strm",
88
+ "has_poster": False,
89
+ "provider_ids": {"Tmdb": "27205"},
90
+ },
91
+ {
92
+ "id": "item-2",
93
+ "name": "Rogue Movie",
94
+ "Type": "Movie",
95
+ "path": "/data/jellyfin/media/Movies/Rogue (2020)/Rogue.strm",
96
+ "has_poster": True,
97
+ "provider_ids": {},
98
+ },
99
+ {
100
+ "id": "series-1",
101
+ "name": "Dark",
102
+ "Type": "Series",
103
+ "path": "/data/jellyfin/media/TV Shows/Dark",
104
+ "has_poster": True,
105
+ },
106
+ {
107
+ "id": "series-2",
108
+ "name": "Dark",
109
+ "Type": "Series",
110
+ "path": "/data/jellyfin/media/TV Shows/Dark Duplicate",
111
+ "has_poster": True,
112
+ },
113
+ ]
114
+
115
+ audit = await audit_and_compare(mock_jf_client, cache_path=str(cache_file))
116
+
117
+ assert audit["total_telegram_injected"] == 1
118
+ assert audit["total_jellyfin_items"] == 4
119
+ assert audit["rogue_items_count"] == 1
120
+ assert audit["rogue_items"][0]["id"] == "item-2"
121
+
122
+ assert audit["duplicate_series_count"] == 1
123
+ assert audit["duplicate_series"][0]["series_name"] == "Dark"
124
+ assert audit["duplicate_series"][0]["count"] == 2
125
+
126
+ assert audit["missing_artwork_count"] == 1
127
+ assert audit["missing_artwork"][0]["id"] == "item-1"
128
+
129
+
130
+ @pytest.mark.asyncio
131
+ async def test_reconcile_item(mock_jf_client, tmp_path):
132
+ cache_file = tmp_path / "file_ids.json"
133
+ cache_data = {
134
+ "-1001:1": {
135
+ "title": "Old Title",
136
+ "year": 2020,
137
+ "tmdb_id": 100,
138
+ }
139
+ }
140
+ cache_file.write_text(json.dumps(cache_data))
141
+
142
+ res = await reconcile_item(
143
+ mock_jf_client,
144
+ item_id="item-123",
145
+ title="Khalifa",
146
+ year=2026,
147
+ tmdb_id=1036081,
148
+ imdb_id="tt22797990",
149
+ cache_path=str(cache_file),
150
+ )
151
+
152
+ assert res["status"] == "reconciled"
153
+ assert res["id"] == "item-123"
154
+ assert res["title"] == "Khalifa"
155
+ mock_jf_client.update_item_metadata.assert_called_once()
156
+ mock_jf_client.refresh_item.assert_called_once_with("item-123", replace_all=True)
157
+
158
+ # Check cache update
159
+ updated_cache = json.loads(cache_file.read_text())
160
+ entry = updated_cache["-1001:1"]
161
+ assert entry["title"] == "Khalifa"
162
+ assert entry["year"] == 2026
163
+ assert entry["tmdb_id"] == 1036081
164
+ assert entry["imdb_id"] == "tt22797990"
165
+
166
+
167
+ @pytest.mark.asyncio
168
+ async def test_cleanup_rogue_and_duplicates(mock_jf_client, tmp_path):
169
+ cache_file = tmp_path / "file_ids.json"
170
+ cache_file.write_text(json.dumps({}))
171
+
172
+ mock_jf_client.get_all_library_items.return_value = [
173
+ {
174
+ "id": "rogue-1",
175
+ "name": "Rogue",
176
+ "Type": "Movie",
177
+ "path": "/data/jellyfin/media/Movies/Rogue.strm",
178
+ "has_poster": True,
179
+ },
180
+ {
181
+ "id": "series-1",
182
+ "name": "Breaking Bad",
183
+ "Type": "Series",
184
+ "path": "/data/jellyfin/media/TV Shows/Breaking Bad",
185
+ "has_poster": True,
186
+ },
187
+ {
188
+ "id": "series-2",
189
+ "name": "Breaking Bad",
190
+ "Type": "Series",
191
+ "path": "/data/jellyfin/media/TV Shows/Breaking Bad (2)",
192
+ "has_poster": True,
193
+ },
194
+ ]
195
+
196
+ res = await cleanup_rogue_and_duplicates(
197
+ mock_jf_client,
198
+ remove_rogue=True,
199
+ deduplicate_series=True,
200
+ cache_path=str(cache_file),
201
+ )
202
+
203
+ assert res["status"] == "cleanup_complete"
204
+ assert res["deleted_rogue_count"] == 1
205
+ assert res["deleted_duplicate_series_count"] == 1
206
+
207
+ # Verified delete calls
208
+ deleted_ids = [call.args[0] for call in mock_jf_client.delete_item.call_args_list]
209
+ assert "rogue-1" in deleted_ids
210
+ assert "series-2" in deleted_ids
211
+ assert "series-1" not in deleted_ids
212
+ mock_jf_client.refresh_library.assert_called_once()
213
+
214
+
215
+ @pytest.mark.asyncio
216
+ async def test_apply_artwork_from_url_blocked_by_netguard(mock_jf_client):
217
+ with pytest.raises(ValueError, match="Artwork URL rejected by NetGuard"):
218
+ await apply_artwork_from_url(
219
+ mock_jf_client,
220
+ item_id="item-1",
221
+ image_url="http://192.168.1.1/exploit.jpg",
222
+ )
tests/ai_admin/test_stage6_e2e_integration.py CHANGED
@@ -120,14 +120,17 @@ def test_stage6_exact_tool_count_and_absence_of_mutations(monkeypatch, tmp_path)
120
  registry = router.handler.registry
121
  registered_tools = registry.list_tools()
122
 
123
- # Must contain exactly 13 read-only tools
124
- assert len(registered_tools) == 13, f"Expected 13 tools, got {len(registered_tools)}: {[t.name for t in registered_tools]}"
125
 
126
  expected_tool_names = {
127
  "jellyfin_search",
128
  "jellyfin_get_item",
129
  "jellyfin_list_missing_metadata",
130
  "jellyfin_get_media_info",
 
 
 
131
  "tmdb_search",
132
  "tmdb_get_details",
133
  "tmdb_get_images",
 
120
  registry = router.handler.registry
121
  registered_tools = registry.list_tools()
122
 
123
+ # Must contain exactly 16 read-only tools
124
+ assert len(registered_tools) == 16, f"Expected 16 tools, got {len(registered_tools)}: {[t.name for t in registered_tools]}"
125
 
126
  expected_tool_names = {
127
  "jellyfin_search",
128
  "jellyfin_get_item",
129
  "jellyfin_list_missing_metadata",
130
  "jellyfin_get_media_info",
131
+ "telegram_list_injected",
132
+ "jellyfin_list_library",
133
+ "library_audit_and_compare",
134
  "tmdb_search",
135
  "tmdb_get_details",
136
  "tmdb_get_images",