DB Write sanitizer

database.py

@@ -40,7 +40,7 @@
 
 from tz_helper import UTC_NOW_SQL
 
-from sanitization import sanitize_db_field
+from sanitization import sanitize_db_field, sanitize_string_for_db
 
 from config import (
     TASK_STATUS_PENDING,
@@ -761,7 +761,7 @@ def persist_chromaprint(server_id, provider_track_id, fingerprint):
             "VALUES (%s, %s, %s, now()) "
             "ON CONFLICT (server_id, provider_track_id) DO UPDATE SET "
             "fingerprint = EXCLUDED.fingerprint, updated_at = now()",
-            (str(server_id), str(provider_track_id), blob),
+            (str(server_id), sanitize_string_for_db(str(provider_track_id)), blob),
         )
         conn.commit()
     except Exception:
@@ -787,7 +787,7 @@ def get_chromaprint(server_id, provider_track_id):
         cur.execute(
             "SELECT fingerprint FROM chromaprint "
             "WHERE server_id = %s AND provider_track_id = %s",
-            (str(server_id), str(provider_track_id)),
+            (str(server_id), sanitize_string_for_db(str(provider_track_id))),
         )
         row = cur.fetchone()
         return bytes(row[0]) if row and row[0] is not None else None
@@ -1554,6 +1554,7 @@ def init_db():
             _seed_registry_from_legacy_config(cur)
             _drop_unconfigured_servers(cur)
             _migrate_artist_mapping_to_server_map(cur)
+            _scrub_control_chars_from_map_ids(cur)
             _migrate_playlist_server_column(cur)
             removed_media_keys = purge_media_keys_from_app_config(cur)
             if removed_media_keys:
@@ -1631,6 +1632,57 @@ def _migrate_file_path_to_track_server_map(cur):
         )
 
 
+_MAP_ID_SCRUB_MARKER = 'map_id_c0_scrub_v1'
+
+
+def _scrub_control_chars_from_map_ids(cur):
+    cur.execute("SELECT 1 FROM app_config WHERE key = %s", (_MAP_ID_SCRUB_MARKER,))
+    if cur.fetchone():
+        return
+    controls = ''.join(
+        chr(c) for c in (*range(0x01, 0x09), 0x0B, 0x0C, *range(0x0E, 0x20))
+    )
+    klass = '[' + controls + ']'
+    cur.execute(
+        "DELETE FROM track_server_map t WHERE t.provider_track_id ~ %s AND ("
+        "regexp_replace(t.provider_track_id, %s, '', 'g') = '' "
+        "OR EXISTS (SELECT 1 FROM track_server_map c WHERE c.server_id = t.server_id "
+        "AND c.provider_track_id = regexp_replace(t.provider_track_id, %s, '', 'g')) "
+        "OR t.provider_track_id > (SELECT MIN(d.provider_track_id) FROM track_server_map d "
+        "WHERE d.server_id = t.server_id AND d.provider_track_id ~ %s "
+        "AND regexp_replace(d.provider_track_id, %s, '', 'g') = "
+        "regexp_replace(t.provider_track_id, %s, '', 'g')))",
+        (klass, klass, klass, klass, klass, klass),
+    )
+    dropped = cur.rowcount
+    cur.execute(
+        "UPDATE track_server_map SET provider_track_id = "
+        "regexp_replace(provider_track_id, %s, '', 'g'), updated_at = now() "
+        "WHERE provider_track_id ~ %s",
+        (klass, klass),
+    )
+    rewritten = cur.rowcount
+    cur.execute(
+        "DELETE FROM artist_server_map WHERE artist_name ~ %s OR provider_artist_id ~ %s",
+        (klass, klass),
+    )
+    artist_rows = cur.rowcount
+    cur.execute("DELETE FROM chromaprint WHERE provider_track_id ~ %s", (klass,))
+    chroma_rows = cur.rowcount
+    if dropped or rewritten or artist_rows or chroma_rows:
+        logger.warning(
+            "Scrubbed control characters from legacy map rows: %d track map ids "
+            "rewritten, %d colliding/empty track rows dropped, %d artist rows and "
+            "%d chromaprint rows removed (they re-populate on the next sweep/analysis)",
+            rewritten, dropped, artist_rows, chroma_rows,
+        )
+    cur.execute(
+        "INSERT INTO app_config (key, value) VALUES (%s, %s) "
+        "ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value",
+        (_MAP_ID_SCRUB_MARKER, 'done'),
+    )
+
+
 def _ensure_track_server_map_key(cur):
     """Ensure track_server_map carries the (server_id, provider_track_id) unique
     index and the relaxed PRIMARY KEY the N:1 upserts arbitrate on. Dedupes any

sanitization.py

@@ -15,6 +15,7 @@
 Main Features:
 * ``sanitize_db_field`` removes non-printable characters and truncates to a max length.
 * ``sanitize_string_for_db`` / ``sanitize_json_for_db`` strip NUL and control chars from text and nested JSON.
+* ``sanitize_string_for_db_loud`` is the same transform with a per-value warning when it changed something.
 * ``sanitize_for_json`` converts numpy int/float/bool/array types to native Python.
 """
 
@@ -26,6 +27,8 @@
 
 logger = logging.getLogger(__name__)
 
+_DB_CONTROL_CHARS = re.compile(r'[\x00-\x08\x0B-\x0C\x0E-\x1F]')
+
 
 def sanitize_string_for_db(value: Optional[str]) -> Optional[str]:
     if value is None:
@@ -34,11 +37,18 @@ def sanitize_string_for_db(value: Optional[str]) -> Optional[str]:
     if not isinstance(value, str):
         value = str(value)
 
-    value = value.replace('\x00', '')
+    return _DB_CONTROL_CHARS.sub('', value)
 
-    value = re.sub(r'[\x01-\x08\x0B-\x0C\x0E-\x1F]', '', value)
 
-    return value
+def sanitize_string_for_db_loud(value, field):
+    text = value if isinstance(value, str) else str(value)
+    cleaned = _DB_CONTROL_CHARS.sub('', text)
+    if cleaned != text:
+        logger.warning(
+            "Sanitized control characters out of %s %r before database write",
+            field, text,
+        )
+    return cleaned
 
 
 def sanitize_db_field(s, max_length=1000, field_name="field"):

tasks/analysis/helper.py

@@ -44,6 +44,7 @@
 from database import get_db, save_task_status
 from psycopg2 import OperationalError
 from psycopg2 import sql as pgsql
+from sanitization import sanitize_string_for_db
 
 from error import error_manager
 from error.error_dictionary import ERR_DB_CONNECTION
@@ -144,15 +145,17 @@ def report(message, progress, **kwargs):
 
 
 def _str_ids(ids):
-    return [str(i) for i in ids]
+    return [sanitize_string_for_db(str(i)) for i in ids]
 
 
 def attach_catalog_item_ids(tracks, server_id=None):
     if not tracks:
         return tracks
     from tasks.mediaserver import context, registry
 
-    provider_ids = [str(t.get('Id') or t.get('id')) for t in tracks]
+    provider_ids = [
+        sanitize_string_for_db(str(t.get('Id') or t.get('id'))) for t in tracks
+    ]
     active_server_id = server_id or context.active_server_id()
     mapped = registry.reverse_translate_ids(provider_ids, active_server_id)
     for item, provider_id in zip(tracks, provider_ids):

tasks/analysis/song.py

@@ -48,6 +48,8 @@
 )
 from psycopg2 import OperationalError
 
+from sanitization import sanitize_string_for_db
+
 from error import error_manager
 from error.error_dictionary import (
     ERR_DB_QUERY,
@@ -588,11 +590,13 @@ def analyze_track(file_path, mood_labels_list, model_paths, onnx_sessions=None,
 
 
 def catalog_item_id(item):
-    return str(item.get('_catalog_item_id') or item.get('Id') or item.get('id'))
+    return sanitize_string_for_db(
+        str(item.get('_catalog_item_id') or item.get('Id') or item.get('id'))
+    )
 
 
 def provider_item_id(item):
-    return str(item.get('Id') or item.get('id'))
+    return sanitize_string_for_db(str(item.get('Id') or item.get('id')))
 
 
 def ensure_musicnn_sessions(onnx_sessions, model_paths, session_recycler, album_name):

tasks/duplicate_repair.py

@@ -74,6 +74,7 @@
 
 import config
 from database import connect_raw
+from sanitization import sanitize_string_for_db
 from tasks import provider_probe
 from tasks import simhash
 from tasks.chromaprint import chromaprints_agree
@@ -169,7 +170,7 @@ def _server_durations(server):
             server['server_type'], server['creds'], apply_filter=True
         )
     return {
-        str(track['id']): track['duration']
+        sanitize_string_for_db(str(track['id'])): track['duration']
         for track in tracks
         if track.get('id') is not None and track.get('duration') is not None
     }
@@ -182,7 +183,10 @@ def _group_duration(provider_ids, durations):
     DURATION_TOLERANCE_SECONDS; the stamped value is the smallest, deterministic
     and within tolerance of the survivor's true length.
     """
-    values = [durations.get(provider_id) for provider_id in provider_ids]
+    values = [
+        durations.get(sanitize_string_for_db(provider_id))
+        for provider_id in provider_ids
+    ]
     if any(value is None for value in values):
         return None
     if (max(values) - min(values)) > config.DURATION_TOLERANCE_SECONDS:

tasks/fingerprint_canonicalize.py

@@ -72,6 +72,7 @@
 
 import config
 from database import connect_raw
+from sanitization import sanitize_string_for_db
 from tasks import simhash
 from tasks.mediaserver import registry
 from tasks.provider_migration_tasks import (
@@ -175,7 +176,7 @@ def _fetch_provider_durations(source_id, conn):
                 server['server_type'], server['creds'], apply_filter=False
             )
         durations = {
-            str(track['id']): track['duration']
+            sanitize_string_for_db(str(track['id'])): track['duration']
             for track in tracks
             if track.get('id') is not None and track.get('duration') is not None
         }
@@ -206,9 +207,10 @@ def _durations_for_rows(cur, ids, rows, provider_durations, source_id):
         for item_id, duration in cur.fetchall():
             durations[str(item_id)] = float(duration)
     unresolved = [i for i in wanted if i not in durations]
-    direct = [i for i in unresolved if i in provider_durations]
-    for item_id in direct:
-        durations[item_id] = provider_durations[item_id]
+    for item_id in unresolved:
+        value = provider_durations.get(sanitize_string_for_db(item_id))
+        if value is not None:
+            durations[item_id] = value
     unresolved = [i for i in unresolved if i not in durations]
     if unresolved and provider_durations:
         for begin in range(0, len(unresolved), _CHUNK_ROWS):
@@ -219,7 +221,7 @@ def _durations_for_rows(cur, ids, rows, provider_durations, source_id):
                 (source_id, chunk),
             )
             for item_id, provider_id in cur.fetchall():
-                value = provider_durations.get(str(provider_id))
+                value = provider_durations.get(sanitize_string_for_db(str(provider_id)))
                 if value is not None:
                     durations.setdefault(str(item_id), value)
     return durations

tasks/mediaserver/emby.py

@@ -549,8 +549,8 @@ def download_track(temp_dir, item):
                     f.write(chunk)
         logger.info(f"Downloaded '{item.get('Name', track_id)}' to '{local_filename}'")
         return local_filename
-    except Exception:
-        logger.exception(f"Failed to download track {item.get('Name', 'Unknown')}")
+    except Exception as exc:
+        logger.warning(f"Failed to download track {item.get('Name', 'Unknown')}: {exc}")
         return None
 
 

tasks/mediaserver/jellyfin.py

@@ -302,8 +302,8 @@ def download_track(temp_dir, item):
                     f.write(chunk)
         logger.info(f"Downloaded '{item.get('Name', track_id)}' to '{local_filename}'")
         return local_filename
-    except Exception:
-        logger.exception(f"Failed to download track {item.get('Name', 'Unknown')}")
+    except Exception as exc:
+        logger.warning(f"Failed to download track {item.get('Name', 'Unknown')}: {exc}")
         return None
 
 

tasks/mediaserver/lyrion.py

@@ -400,8 +400,8 @@ def download_track(temp_dir, item):
                         f.write(chunk)
         logger.info(f"Downloaded '{item.get('title', 'Unknown')}' to '{local_filename}'")
         return local_filename
-    except Exception:
-        logger.exception(f"Failed to download Lyrion track {item.get('title', 'Unknown')}")
+    except Exception as exc:
+        logger.warning(f"Failed to download Lyrion track {item.get('title', 'Unknown')}: {exc}")
     return None
 
 

tasks/mediaserver/plex.py

@@ -293,8 +293,8 @@ def download_track(temp_dir, item):
                     f.write(chunk)
         logger.info(f"Downloaded '{item.get('Name', 'Unknown')}' to '{local_filename}'")
         return local_filename
-    except Exception:
-        logger.exception(f"Failed to download track {item.get('Name', 'Unknown')}")
+    except Exception as exc:
+        logger.warning(f"Failed to download track {item.get('Name', 'Unknown')}: {exc}")
         return None