fix url_encode aggressively encoding brackets

Author: chapmanjacobd

library/utils/arggroups.py

@@ -665,10 +665,32 @@ def sql_fs_post(args, table_prefix="m.") -> None:
             args.include[0] = shell_utils.resolve_absolute_path(args.include[0])
 
     if not args.no_url_encode_search:
+        from library.utils.path_utils import safe_unquote
         from library.utils.web import url_encode
 
-        args.include = [url_encode(s) if s.startswith("http") else s for s in args.include]
-        args.exclude = [url_encode(s) if s.startswith("http") else s for s in args.exclude]
+        new_include = []
+        for s in args.include:
+            if s.startswith("http"):
+                variants = {s, url_encode(s), safe_unquote(s)}
+                if len(variants) > 1:
+                    new_include.append(list(variants))
+                else:
+                    new_include.append(s)
+            else:
+                new_include.append(s)
+        args.include = new_include
+
+        new_exclude = []
+        for s in args.exclude:
+            if s.startswith("http"):
+                variants = {s, url_encode(s), safe_unquote(s)}
+                if len(variants) > 1:
+                    new_exclude.append(list(variants))
+                else:
+                    new_exclude.append(s)
+            else:
+                new_exclude.append(s)
+        args.exclude = new_exclude
 
     parse_args_limit(args)
 

library/utils/filter_engine.py

@@ -213,27 +213,51 @@ def construct_search_bindings(include, exclude, columns, exact=False, flexible_s
     sql = []
     bindings = {}
 
-    incl = ":" + param_key + "include{0}"
-    includes = "(" + " OR ".join([f"{col} LIKE {incl}" for col in columns]) + ")"
-    includes_sql_parts = []
-    for idx, inc in enumerate(include):
-        includes_sql_parts.append(includes.format(idx))
+    def get_include_sql(idx_str, inc):
+        incl = ":" + param_key + f"include{idx_str}"
         if exact:
-            bindings[f"{param_key}include{idx}"] = inc
+            bindings[f"{param_key}include{idx_str}"] = inc
+        else:
+            bindings[f"{param_key}include{idx_str}"] = "%" + inc.replace(" ", "%").replace("%%", " ") + "%"
+        return "(" + " OR ".join([f"{col} LIKE {incl}" for col in columns]) + ")"
+
+    include_sql_parts = []
+    for idx, inc in enumerate(include):
+        if isinstance(inc, list):
+            group_parts = [get_include_sql(f"{idx}_{sub_idx}", sub_inc) for sub_idx, sub_inc in enumerate(inc)]
+            include_sql_parts.append("(" + " OR ".join(group_parts) + ")")
         else:
-            bindings[f"{param_key}include{idx}"] = "%" + inc.replace(" ", "%").replace("%%", " ") + "%"
-    join_op = " OR " if flexible_search else " AND "
-    if len(includes_sql_parts) > 0:
-        sql.append("AND (" + join_op.join(includes_sql_parts) + ")")
+            include_sql_parts.append(get_include_sql(idx, inc))
+
+    if flexible_search:
+        if include_sql_parts:
+            sql.append("AND (" + " OR ".join(include_sql_parts) + ")")
+    else:
+        for part in include_sql_parts:
+            sql.append("AND " + part)
 
     excl = ":" + param_key + "exclude{0}"
     excludes = "AND (" + " AND ".join([f"COALESCE({col},'') NOT LIKE {excl}" for col in columns]) + ")"
     for idx, exc in enumerate(exclude):
-        sql.append(excludes.format(idx))
-        if exact:
-            bindings[f"{param_key}exclude{idx}"] = exc
+        if isinstance(exc, list):
+            # For exclusion, OR inside the group means if ANY variant matches, it's excluded
+            # This is equivalent to AND NOT variant1 AND NOT variant2
+            for sub_idx, sub_exc in enumerate(exc):
+                sub_idx_str = f"{idx}_{sub_idx}"
+                bindings[f"{param_key}exclude{sub_idx_str}"] = (
+                    sub_exc if exact else "%" + sub_exc.replace(" ", "%").replace("%%", " ") + "%"
+                )
+                sql.append(
+                    "AND ("
+                    + " AND ".join([f"COALESCE({col},'') NOT LIKE :{param_key}exclude{sub_idx_str}" for col in columns])
+                    + ")"
+                )
         else:
-            bindings[f"{param_key}exclude{idx}"] = "%" + exc.replace(" ", "%").replace("%%", " ") + "%"
+            sql.append(excludes.format(idx))
+            if exact:
+                bindings[f"{param_key}exclude{idx}"] = exc
+            else:
+                bindings[f"{param_key}exclude{idx}"] = "%" + exc.replace(" ", "%").replace("%%", " ") + "%"
 
     return sql, bindings
 

library/utils/web.py

@@ -838,7 +838,7 @@ def safe_quote(url):
 
     def selective_quote(component, restricted_chars):
         try:
-            quoted = quote(component, safe="/:[]@!$&'()*+,;=", errors="strict")
+            quoted = quote(component, errors="strict")
         except UnicodeDecodeError:
             return component
         return "".join(quote(char, safe="%") if char in restricted_chars else char for char in quoted)

tests/playback/test_links_open.py

@@ -19,15 +19,40 @@ def test_links_open(mock_souffle):
         {"path": "https://site2", "hostname": "site2", "category": "p1"},
     ]
 
+
 @mock.patch("library.playback.links_open.make_souffle")
 @mock.patch("library.playback.links_open.play")
 def test_links_open_brackets(mock_play, mock_souffle, temp_db):
     db_path = temp_db()
     url = "https://example.com/test?page[]=107"
+    encoded_url = "https://example.com/test?page%5B%5D=107"
+
+    # 1. Add the link RAW
     lb(["links-add", "--no-extract", db_path, url])
+
+    # 2. Search for it with RAW URL (should find it because it's in include as-is)
     lb(["links-open", db_path, "-s", url])
-    
-    assert mock_souffle.called
     media = mock_souffle.call_args[0][1]
     assert len(media) == 1
     assert media[0]["path"] == url
+
+    # 3. Search for it with ENCODED URL (should find it because it's added as group [encoded, raw])
+    lb(["links-open", db_path, "-s", encoded_url])
+    media = mock_souffle.call_args[0][1]
+    assert len(media) == 1
+    assert media[0]["path"] == url
+
+    # 4. Clear and add it ENCODED
+    import sqlite3
+
+    conn = sqlite3.connect(db_path)
+    conn.execute("DELETE FROM media")
+    conn.commit()
+    conn.close()
+    lb(["links-add", "--no-extract", db_path, encoded_url])
+
+    # 5. Search for it with RAW URL (should find it because it matches the encoded version which is added to include)
+    lb(["links-open", db_path, "-s", url])
+    media = mock_souffle.call_args[0][1]
+    assert len(media) == 1
+    assert media[0]["path"] == encoded_url

tests/playback/test_play_actions.py

@@ -20,11 +20,11 @@
     ("--downloaded-before '1 day'", 0, ""),
     ("--limit 1", 1, "corrupt.mp4"),
     ("-L 1", 1, "corrupt.mp4"),
-    ("--online-media-only", 1, "https://test/?tags[]="),
-    ("--no-fts -s https://test/?tags[]=", 1, "https://test/?tags[]="),
-    ("--no-fts -s https://test/?tags%5B%5D=", 1, "https://test/?tags[]="),
-    ("--no-url-encode-search --no-fts -s https://test/?tags[]=", 1, "https://test/?tags[]="),
-    ("--no-url-encode-search --no-fts -s https://test/?tags%5B%5D=", 0, ""),
+    ("--online-media-only", 1, "https://test/?tags%5B%5D="),
+    ("--no-fts -s https://test/?tags%5B%5D=", 1, "https://test/?tags%5B%5D="),
+    ("--no-fts -s https://test/?tags[]=", 1, "https://test/?tags%5B%5D="),
+    ("--no-url-encode-search --no-fts -s https://test/?tags%5B%5D=", 1, "https://test/?tags%5B%5D="),
+    ("--no-url-encode-search --no-fts -s https://test/?tags[]=", 0, ""),
     ("--offset 1", 4, "test.mp4"),
     ("-s tests -s 'tests AND data' -E 2 -s test -E 3", 4, "corrupt.mp4"),
     ("--created-within '30 years'", 5, "corrupt.mp4"),
@@ -83,7 +83,7 @@
     ("-O duration", 5, "test.gif"),
     ("-O locale_duration", 5, "test.gif"),
     ("-O locale_size", 5, "test_frame.gif"),
-    ("-O reverse_path_path", 5, "https://test/?tags[]="),
+    ("-O reverse_path_path", 5, "https://test/?tags%5B%5D="),
     ("-O size", 5, "test_frame.gif"),
     ("-w time_deleted=0", 5, "corrupt.mp4"),
 ]

tests/utils/__init__.py

@@ -70,7 +70,7 @@ def get_default_args(*funcs):
 v_db = p("tests/data/video.db")
 if not Path(v_db).exists():
     lb(["fs-add", v_db, "--scan-subtitles", p("tests/data/"), "-E", "Youtube"])
-    lb(["links-db", v_db, "--insert-only", "https://test/?tags[]="])
+    lb(["links-db", v_db, "--insert-only", "https://test/?tags%5B%5D="])
 
 
 tube_db = p("tests/data/tube.db")