fix(cell-storage): 储存证 ACL 保存避免 collation 混用 503

ACL 表与 sys_user JOIN/条件比较统一 utf8mb4_unicode_ci，并补建表 COLLATE 与 CONVERT 修复脚本（#97）。

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/core/db_collation.py

@@ -9,7 +9,19 @@ def unicode_ci(col):
 
 
 def eq_unicode(left, right):
-    """跨表字符串等值比较（JOIN / WHERE）时统一为 utf8mb4_unicode_ci。"""
+    """跨表字符串等值比较（JOIN / WHERE）时统一为 utf8mb4_unicode_ci。
+
+    列与列：两侧 collate。列与字面量：仅列侧 collate，避免连接 collation
+    （常见 utf8mb4_general_ci）与列 collation（utf8mb4_unicode_ci）混用。
+    """
+    left_is_str = isinstance(left, str)
+    right_is_str = isinstance(right, str)
+    if left_is_str and right_is_str:
+        return left == right
+    if left_is_str:
+        return left == unicode_ci(right)
+    if right_is_str:
+        return unicode_ci(left) == right
     return unicode_ci(left) == unicode_ci(right)
 
 

backend/app/crud/sample_storage_cert_qc_acl.py

@@ -6,6 +6,7 @@ from typing import Any, Dict, List, Optional, Set
 
 from sqlalchemy.orm import Session
 
+from app.core.db_collation import eq_unicode, unicode_ci
 from app.core.utils import generate_uuid
 from app.crud.child_row_sync import sync_child_rows
 from app.models.models import BizStorageCertQcAcl, BizStorageCertQcAclUser, SysUser
@@ -22,8 +23,8 @@ def is_acl_strict(db: Session, tenant_no: str) -> bool:
     row = (
         db.query(BizStorageCertQcAcl.id)
         .filter(
-            BizStorageCertQcAcl.tenant_no == tenant_no,
-            BizStorageCertQcAcl.configured_flag == "1",
+            eq_unicode(BizStorageCertQcAcl.tenant_no, tenant_no),
+            eq_unicode(BizStorageCertQcAcl.configured_flag, "1"),
             BizStorageCertQcAcl.del_flag == 0,
         )
         .first()
@@ -48,8 +49,8 @@ def resolve_editable_field_keys(
     acls = (
         db.query(BizStorageCertQcAcl)
         .filter(
-            BizStorageCertQcAcl.tenant_no == tenant_no,
-            BizStorageCertQcAcl.configured_flag == "1",
+            eq_unicode(BizStorageCertQcAcl.tenant_no, tenant_no),
+            eq_unicode(BizStorageCertQcAcl.configured_flag, "1"),
             BizStorageCertQcAcl.del_flag == 0,
         )
         .all()
@@ -60,9 +61,9 @@ def resolve_editable_field_keys(
     membership = (
         db.query(BizStorageCertQcAclUser.acl_id)
         .filter(
-            BizStorageCertQcAclUser.tenant_no == tenant_no,
-            BizStorageCertQcAclUser.acl_id.in_(acl_ids),
-            BizStorageCertQcAclUser.user_id == uid,
+            eq_unicode(BizStorageCertQcAclUser.tenant_no, tenant_no),
+            unicode_ci(BizStorageCertQcAclUser.acl_id).in_(acl_ids),
+            eq_unicode(BizStorageCertQcAclUser.user_id, uid),
             BizStorageCertQcAclUser.del_flag == 0,
         )
         .all()
@@ -115,8 +116,8 @@ def _validate_user_ids(db: Session, tenant_no: str, user_ids: Any) -> List[str]:
     rows = (
         db.query(SysUser.id)
         .filter(
-            SysUser.tenant_no == tenant_no,
-            SysUser.id.in_(out),
+            eq_unicode(SysUser.tenant_no, tenant_no),
+            unicode_ci(SysUser.id).in_(out),
             SysUser.del_flag == 0,
         )
         .all()
@@ -133,8 +134,8 @@ def get_acl_detail(db: Session, tenant_no: str, scope: str) -> Dict[str, Any]:
     row = (
         db.query(BizStorageCertQcAcl)
         .filter(
-            BizStorageCertQcAcl.tenant_no == tenant_no,
-            BizStorageCertQcAcl.scope == scope_key,
+            eq_unicode(BizStorageCertQcAcl.tenant_no, tenant_no),
+            eq_unicode(BizStorageCertQcAcl.scope, scope_key),
             BizStorageCertQcAcl.del_flag == 0,
         )
         .first()
@@ -152,11 +153,11 @@ def get_acl_detail(db: Session, tenant_no: str, scope: str) -> Dict[str, Any]:
         db.query(BizStorageCertQcAclUser, SysUser)
         .outerjoin(
             SysUser,
-            (SysUser.id == BizStorageCertQcAclUser.user_id) & (SysUser.del_flag == 0),
+            eq_unicode(SysUser.id, BizStorageCertQcAclUser.user_id) & (SysUser.del_flag == 0),
         )
         .filter(
-            BizStorageCertQcAclUser.tenant_no == tenant_no,
-            BizStorageCertQcAclUser.acl_id == row.id,
+            eq_unicode(BizStorageCertQcAclUser.tenant_no, tenant_no),
+            eq_unicode(BizStorageCertQcAclUser.acl_id, row.id),
             BizStorageCertQcAclUser.del_flag == 0,
         )
         .order_by(BizStorageCertQcAclUser.order_num.asc(), BizStorageCertQcAclUser.create_time.asc())
@@ -202,8 +203,8 @@ def upsert_acl(
     row = (
         db.query(BizStorageCertQcAcl)
         .filter(
-            BizStorageCertQcAcl.tenant_no == tenant_no,
-            BizStorageCertQcAcl.scope == scope_key,
+            eq_unicode(BizStorageCertQcAcl.tenant_no, tenant_no),
+            eq_unicode(BizStorageCertQcAcl.scope, scope_key),
             BizStorageCertQcAcl.del_flag == 0,
         )
         .first()

backend/tests/test_sample_storage_cert_qc_acl.py

@@ -1,6 +1,11 @@
 """储存证质检字段 ACL：合法字段与 scope。"""
 import unittest
+from unittest.mock import MagicMock
 
+from sqlalchemy.dialects import mysql
+
+from app.crud import sample_storage_cert_qc_acl as acl_svc
+from app.models.models import BizStorageCertQcAcl, BizStorageCertQcAclUser, SysUser
 from app.services.sample_storage_cert_qc import (
     ACL_SCOPE_PRODUCTION,
     ACL_SCOPE_QC,
@@ -31,5 +36,66 @@ class TestAclFieldKeys(unittest.TestCase):
         self.assertEqual(ACL_SCOPE_PRODUCTION, "production")
 
 
+class TestAclUserJoinCollation(unittest.TestCase):
+    def test_detail_user_join_uses_unicode_collate(self):
+        """JOIN sys_user 时统一 utf8mb4_unicode_ci，避免 1267 mix of collations。"""
+        acl = MagicMock()
+        acl.id = "acl-1"
+        acl.field_keys = "[]"
+        acl.configured_flag = "1"
+
+        captured = {}
+
+        class _Query:
+            def __init__(self, *args, **kwargs):
+                pass
+
+            def filter(self, *args, **kwargs):
+                captured.setdefault("filters", []).extend(args)
+                return self
+
+            def first(self):
+                return acl
+
+            def outerjoin(self, *args, **kwargs):
+                captured["join_on"] = args[1] if len(args) > 1 else None
+                return self
+
+            def order_by(self, *args, **kwargs):
+                return self
+
+            def all(self):
+                return []
+
+        db = MagicMock()
+        db.query.side_effect = lambda *args, **kwargs: _Query()
+
+        detail = acl_svc.get_acl_detail(db, "t1", "qc")
+        self.assertEqual(detail["scope"], "qc")
+        join_on = captured.get("join_on")
+        self.assertIsNotNone(join_on)
+        sql = str(
+            join_on.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})
+        ).lower()
+        self.assertIn("utf8mb4_unicode_ci", sql)
+        filter_sql = " ".join(
+            str(f.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})).lower()
+            for f in captured.get("filters", [])
+        )
+        self.assertIn("utf8mb4_unicode_ci", filter_sql)
+        self.assertIn("sys_user", SysUser.__tablename__)
+        self.assertIn("biz_storage_cert_qc_acl_user", BizStorageCertQcAclUser.__tablename__)
+        self.assertEqual(BizStorageCertQcAcl.__tablename__, "biz_storage_cert_qc_acl")
+
+    def test_eq_unicode_literal_only_collates_column(self):
+        from app.core.db_collation import eq_unicode
+
+        expr = eq_unicode(BizStorageCertQcAcl.tenant_no, "t1")
+        sql = str(
+            expr.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True})
+        ).lower()
+        self.assertIn("utf8mb4_unicode_ci", sql)
+        self.assertEqual(sql.count("utf8mb4_unicode_ci"), 1)
+
 if __name__ == "__main__":
     unittest.main()

docs/sql/migrations/2026-11-11-storage-cert-qc-acl-collation-fix.sql

@@ -0,0 +1,14 @@
+-- Issue #97：将储存证质检 ACL 表 collation 对齐库默认 utf8mb4_unicode_ci。
+-- 修复与 sys_user 等表 JOIN 时 Illegal mix of collations。可重复执行。
+
+ALTER TABLE biz_storage_cert_qc_acl
+  CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
+
+ALTER TABLE biz_storage_cert_qc_acl_user
+  CONVERT TO CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
+
+SELECT table_name, table_collation
+FROM information_schema.tables
+WHERE table_schema = DATABASE()
+  AND table_name IN ('biz_storage_cert_qc_acl', 'biz_storage_cert_qc_acl_user')
+ORDER BY table_name;

docs/sql/migrations/2026-11-11-storage-cert-qc-field-acl.sql

@@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS biz_storage_cert_qc_acl (
   del_flag SMALLINT NOT NULL DEFAULT 0,
   PRIMARY KEY (id),
   KEY ix_tenant_scope (tenant_no, scope, del_flag)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
 
 CREATE TABLE IF NOT EXISTS biz_storage_cert_qc_acl_user (
   id VARCHAR(64) NOT NULL,
@@ -35,7 +35,7 @@ CREATE TABLE IF NOT EXISTS biz_storage_cert_qc_acl_user (
   PRIMARY KEY (id),
   KEY ix_tenant_acl (tenant_no, acl_id, del_flag),
   KEY ix_tenant_user (tenant_no, user_id, del_flag)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
 
 INSERT INTO sys_menu (
     id, menu_name, parent_id, menu_type, path, component, perms,