Repository: opendatalab/labelU

Issue:
图片连接包含井号#时，图片无法展示
图片名称为`a#1.jpg`，上传图片无法展示。

![Image](https://github.com/user-attachments/assets/2acc9ce0-1543-41d6-abde-320644f10640)

图片链接返回
```
{"msg":"Attachment file not found","err_code":51001}
```

Discussion:
@gary-Shen: OK，我们修一下。@gary-Shen: 你使用的什么版本？@gary-Shen: 1.2.7 已修复。

Solution patch:
diff --git a/labelu/internal/application/service/attachment.py b/labelu/internal/application/service/attachment.py
index cc9cb24..8314151 100644
--- a/labelu/internal/application/service/attachment.py
+++ b/labelu/internal/application/service/attachment.py
@@ -38,7 +38,7 @@ async def create(
     #  filename = str(uuid.uuid4())[0:8] + "-" + path_filename[-1] NOTE: If you want keep filename safe, you can use uuid as filename 
     filename = path_filename[-1]
     sanitized = re.sub(r'%', '_pct_', filename)
-    sanitized = re.sub(r'[\\/*?:"<>|]', '_', sanitized)
+    sanitized = re.sub(r'[\\/*?:"<>|#]', '_', sanitized)
     path = "/".join(path_filename[:-1])
     attachment_relative_base_dir = Path(settings.UPLOAD_DIR).joinpath(
         str(task_id), path
diff --git a/labelu/internal/application/service/pre_annotation.py b/labelu/internal/application/service/pre_annotation.py
index 6400056..ed4fa82 100644
--- a/labelu/internal/application/service/pre_annotation.py
+++ b/labelu/internal/application/service/pre_annotation.py
@@ -210,9 +210,23 @@ async def delete_pre_annotation_file(
     db: Session, task_id: int, file_id: int, current_user: User
 ) -> CommonDataResp:
     with db.begin():
+        task = crud_task.get(db=db, task_id=task_id, lock=True)
+        if not task:
+            logger.error("cannot find task:{}", task_id)
+            raise LabelUException(
+                code=ErrorCode.CODE_50002_TASK_NOT_FOUND,
+                status_code=status.HTTP_404_NOT_FOUND,
+            )
+        
         attachments = crud_attachment.get_by_ids(
             db=db, attachment_ids=[file_id]
         )
+
+        if len(attachments) == 0:
+            raise LabelUException(
+                code=ErrorCode.CODE_51001_TASK_ATTACHMENT_NOT_FOUND,
+                status_code=status.HTTP_404_NOT_FOUND,
+            )
         
         for attachment in attachments:
             file_full_path = Path(settings.MEDIA_ROOT).joinpath(attachment.path)
diff --git a/labelu/internal/application/service/task.py b/labelu/internal/application/service/task.py
index ac966c6..629ad94 100644
--- a/labelu/internal/application/service/task.py
+++ b/labelu/internal/application/service/task.py
@@ -184,7 +184,7 @@ async def add_collaborator(db: Session, task_id: int, user_id: int, current_user
     if not user:
         logger.error("cannot find user:{}", user_id)
         raise LabelUException(
-            code=ErrorCode.CODE_50001_USER_NOT_FOUND,
+            code=ErrorCode.CODE_40002_USER_NOT_FOUND,
             status_code=status.HTTP_404_NOT_FOUND,
         )
         


Test patch:
diff --git a/labelu/tests/internal/adapter/routers/test_attachment.py b/labelu/tests/internal/adapter/routers/test_attachment.py
index 0f7cdef..4fb3bc6 100644
--- a/labelu/tests/internal/adapter/routers/test_attachment.py
+++ b/labelu/tests/internal/adapter/routers/test_attachment.py
@@ -1,4 +1,5 @@
 from pathlib import Path
+import re
 
 from sqlalchemy.orm import Session
 from fastapi.testclient import TestClient
@@ -317,3 +318,150 @@ class TestClassTaskAttachmentRouter:
         # check
         assert r.status_code == 403
         assert r.json()["err_code"] == 30001
+
+    def test_partial_content(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # prepare data
+        data = {
+            "name": "partial content task",
+            "description": "task description",
+            "tips": "task tips",
+        }
+        task = client.post(
+            f"{settings.API_V1_STR}/tasks", headers=testuser_token_headers, json=data
+        )
+        task_id = task.json()["data"]["id"]
+        
+        empty_task_upload(task_id, "test.txt")
+
+        # 上传测试文件
+        with Path("labelu/tests/data/test.txt").open(mode="rb") as f:
+            attachment = client.post(
+                f"{settings.API_V1_STR}/tasks/{task_id}/attachments",
+                headers=testuser_token_headers,
+                files={"file": ("test.txt", f, "text/plain")},
+            )
+        url = attachment.json()["data"]["url"]
+        
+        # 转换为 partial URL
+        partial_url = url.replace("/attachment/", "/partial/")
+        
+        # 测试 Range 请求
+        r1 = client.get(
+            partial_url,
+            headers={**testuser_token_headers, "Range": "bytes=0-9"},
+        )
+        
+        # 检查
+        assert r1.status_code == 206  # 部分内容响应码
+        assert r1.headers.get("Content-Length") == "10"
+        assert r1.headers.get("Content-Range").startswith("bytes 0-9/")
+        
+        # 测试没有 Range 的请求
+        r2 = client.get(
+            partial_url,
+            headers=testuser_token_headers,
+        )
+        
+        # 检查
+        assert r2.status_code == 200
+        assert "Accept-Ranges" in r2.headers
+        
+        # 测试无效的 Range 请求
+        r3 = client.get(
+            partial_url,
+            headers={**testuser_token_headers, "Range": "invalid"},
+        )
+        
+        # 检查 - 应该回退到完整响应
+        assert r3.status_code == 200
+        
+        # 测试超出文件范围的请求
+        with Path("labelu/tests/data/test.txt").open(mode="rb") as f:
+            file_size = len(f.read())
+            
+        r4 = client.get(
+            partial_url,
+            headers={**testuser_token_headers, "Range": f"bytes={file_size + 10}-{file_size + 20}"},
+        )
+        
+        # 检查 - 应该回退到完整响应
+        assert r4.status_code == 200
+
+    def test_partial_content_file_not_found(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # 测试不存在的文件
+        r = client.get(
+            f"{settings.API_V1_STR}/partial/non-existent/file.txt",
+            headers=testuser_token_headers,
+        )
+        
+        # 检查
+        assert r.status_code == 404
+        assert r.json()["err_code"] == 30003
+
+    def test_upload_file_with_hash_symbol(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # prepare data
+        data = {
+            "name": "hash symbol test",
+            "description": "test for files with # in name",
+            "tips": "test tips",
+        }
+        task = client.post(
+            f"{settings.API_V1_STR}/tasks", headers=testuser_token_headers, json=data
+        )
+        task_id = task.json()["data"]["id"]
+        
+        empty_task_upload(task_id, "test.txt")
+
+        # 创建一个带有井号的临时文件名
+        original_filename = "test#file#with#hash.txt"
+        
+        # 打开测试文件并上传（使用带有井号的文件名）
+        with Path("labelu/tests/data/test.txt").open(mode="rb") as f:
+            response = client.post(
+                f"{settings.API_V1_STR}/tasks/{task_id}/attachments",
+                headers=testuser_token_headers,
+                files={"file": (original_filename, f, "text/plain")},
+            )
+        
+        # 检查
+        json = response.json()
+        assert response.status_code == 201
+        
+        # 验证文件名中的井号被替换为下划线
+        assert "#" not in json["data"]["filename"]
+        assert "test_file_with_hash.txt" == json["data"]["filename"]
+        
+        # 验证文件实际存在于文件系统
+        parts = json["data"]["url"].split("/")[-3:]
+        assert Path(f"{settings.MEDIA_ROOT}").joinpath("/".join(parts)).exists()
+
+
+class TestAttachmentUtils:
+    def test_filename_sanitization(self):
+        """测试文件名清理功能，验证井号(#)是否被正确处理"""
+        # 模拟 attachment.py 中的文件名清理逻辑
+        original_filename = "test#file#with#hash.txt"
+        sanitized = re.sub(r'%', '_pct_', original_filename)
+        sanitized = re.sub(r'[\\/*?:"<>|#]', '_', sanitized)
+        
+        # 验证井号被替换为下划线
+        assert "#" not in sanitized
+        assert "test_file_with_hash.txt" == sanitized
+        
+        # 测试其他特殊字符也能被正确处理
+        complex_filename = r"file\with/special*chars?:and\"<>|#symbols.txt"
+        sanitized = re.sub(r'%', '_pct_', complex_filename)
+        sanitized = re.sub(r'[\\/*?:"<>|#]', '_', sanitized)
+        
+        # 检查正确结果（拆分为多个断言便于排除问题）
+        assert sanitized.startswith("file_with_special_chars_")
+        assert sanitized.endswith("_symbols.txt")
+        assert "and" in sanitized
+        # 直接检查整个转义序列被正确处理
+        assert "\\" not in sanitized
diff --git a/labelu/tests/internal/adapter/routers/test_pre_annotation.py b/labelu/tests/internal/adapter/routers/test_pre_annotation.py
index f6c293e..e35d2aa 100644
--- a/labelu/tests/internal/adapter/routers/test_pre_annotation.py
+++ b/labelu/tests/internal/adapter/routers/test_pre_annotation.py
@@ -487,3 +487,211 @@ class TestClassTaskPreAnnotationRouter:
 
         # check
         assert r.status_code == 200
+
+    def test_pre_annotation_files_list(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # prepare data
+        current_user = crud_user.get_user_by_username(
+            db=db, username="test@example.com"
+        )
+        task = crud_task.create(
+            db=db,
+            task=Task(
+                name="list files task",
+                description="description",
+                tips="tips",
+                created_by=current_user.id,
+                updated_by=current_user.id,
+            ),
+        )
+        
+        # 上传预注释文件
+        empty_task_upload(task.id, "test.jsonl")
+        with Path("labelu/tests/data/test.jsonl").open(mode="rb") as f:
+            jsonl1 = client.post(
+                f"{settings.API_V1_STR}/tasks/{task.id}/attachments",
+                headers=testuser_token_headers,
+                files={"file": f},
+            )
+        
+        # 为文件创建预注释
+        client.post(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations",
+            headers=testuser_token_headers,
+            json=[{"file_id": jsonl1.json()["data"]["id"]}],
+        )
+
+        # 测试 page 列表
+        r1 = client.get(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations/files",
+            headers=testuser_token_headers,
+            params={"page": 0, "size": 10},
+        )
+
+        # 检查
+        json1 = r1.json()
+        assert r1.status_code == 200
+        assert len(json1["data"]) == 1
+        assert json1["meta_data"]["total"] == 1
+        
+        # 测试 before 列表
+        r2 = client.get(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations/files",
+            headers=testuser_token_headers,
+            params={"before": json1["data"][0]["id"], "size": 10},
+        )
+        
+        # 检查
+        json2 = r2.json()
+        assert r2.status_code == 200
+        assert len(json2["data"]) == 0
+        assert json2["meta_data"]["total"] == 0
+        
+        # 测试 after 列表
+        r3 = client.get(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations/files",
+            headers=testuser_token_headers,
+            params={"after": json1["data"][0]["id"], "size": 10},
+        )
+        
+        # 检查
+        json3 = r3.json()
+        assert r3.status_code == 200
+        assert len(json3["data"]) == 0  # 没有更多的文件了
+        
+        # 测试排序
+        r4 = client.get(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations/files",
+            headers=testuser_token_headers,
+            params={"page": 0, "size": 10, "sort": "created_at:desc"},
+        )
+        
+        # 检查
+        json4 = r4.json()
+        assert r4.status_code == 200
+        assert len(json4["data"]) == 1
+        # 检查降序排序（最新的文件应该在前面）
+        # assert json4["data"][0]["id"] > json4["data"][1]["id"]
+        
+    def test_pre_annotation_files_list_with_errors(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # 测试不存在的任务
+        r = client.get(
+            f"{settings.API_V1_STR}/tasks/9999/pre_annotations/files",
+            headers=testuser_token_headers,
+            params={"page": 0, "size": 10},
+        )
+        
+        # 检查
+        assert r.status_code == 200
+        
+        # 测试无效的排序规则
+        task = crud_task.create(
+            db=db,
+            task=Task(
+                name="invalid sort task",
+                description="description",
+                tips="tips",
+                created_by=0,
+                updated_by=0,
+            ),
+        )
+        
+        r2 = client.get(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations/files",
+            headers=testuser_token_headers,
+            params={"page": 0, "size": 10, "sort": "invalid:desc"},
+        )
+        
+        # 检查 - 应该返回验证错误
+        assert r2.status_code == 422
+
+    def test_delete_pre_annotation_file(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # prepare data
+        current_user = crud_user.get_user_by_username(
+            db=db, username="test@example.com"
+        )
+        task = crud_task.create(
+            db=db,
+            task=Task(
+                name="delete file task",
+                description="description",
+                tips="tips",
+                created_by=current_user.id,
+                updated_by=current_user.id,
+            ),
+        )
+        
+        # 上传预注释文件
+        empty_task_upload(task.id, "test.jsonl")
+        with Path("labelu/tests/data/test.jsonl").open(mode="rb") as f:
+            jsonl = client.post(
+                f"{settings.API_V1_STR}/tasks/{task.id}/attachments",
+                headers=testuser_token_headers,
+                files={"file": f},
+            )
+        
+        file_id = jsonl.json()["data"]["id"]
+        
+        # 为文件创建预注释
+        pre_resp = client.post(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations",
+            headers=testuser_token_headers,
+            json=[{"file_id": file_id}],
+        )
+        
+        # 删除预注释文件
+        r = client.delete(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations/files/{file_id}",
+            headers=testuser_token_headers,
+        )
+        
+        # 检查
+        json = r.json()
+        assert r.status_code == 200
+        assert json["data"]["ok"] == True
+        
+        # 确认文件已被删除
+        files_resp = client.get(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations/files",
+            headers=testuser_token_headers,
+            params={"page": 0, "size": 10},
+        )
+        
+        assert len(files_resp.json()["data"]) == 0
+        
+    def test_delete_pre_annotation_file_errors(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # 测试不存在的任务
+        r1 = client.delete(
+            f"{settings.API_V1_STR}/tasks/9999/pre_annotations/files/1",
+            headers=testuser_token_headers,
+        )
+        
+        # 检查
+        assert r1.status_code == 404
+        
+        # 测试不存在的文件
+        task = crud_task.create(
+            db=db,
+            task=Task(
+                name="not found file task",
+                description="description",
+                tips="tips",
+                created_by=0,
+                updated_by=0,
+            ),
+        )
+        
+        r2 = client.delete(
+            f"{settings.API_V1_STR}/tasks/{task.id}/pre_annotations/files/9999",
+            headers=testuser_token_headers,
+        )
+        
+        # 检查
+        assert r2.status_code == 404
diff --git a/labelu/tests/internal/adapter/routers/test_task.py b/labelu/tests/internal/adapter/routers/test_task.py
index ca96bc4..e946b3d 100644
--- a/labelu/tests/internal/adapter/routers/test_task.py
+++ b/labelu/tests/internal/adapter/routers/test_task.py
@@ -3,6 +3,7 @@ from fastapi.testclient import TestClient
 
 from labelu.internal.common.config import settings
 from labelu.internal.domain.models.task import Task
+from labelu.internal.domain.models.user import User
 from labelu.internal.adapter.persistence import crud_user
 from labelu.internal.adapter.persistence import crud_task
 
@@ -346,3 +347,250 @@ class TestClassTaskRouter:
         json = r.json()
         assert r.status_code == 403
         assert json["err_code"] == 30001
+
+    def test_task_get_collaborators(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # prepare data
+        data = {
+            "name": "task name",
+            "description": "task description",
+            "tips": "task tips",
+        }
+        task_response = client.post(
+            f"{settings.API_V1_STR}/tasks",
+            headers=testuser_token_headers,
+            json=data,
+        )
+        task_id = task_response.json()["data"]["id"]
+
+        # run
+        r = client.get(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators",
+            headers=testuser_token_headers,
+        )
+
+        # check
+        json = r.json()
+        assert r.status_code == 200
+        # 创建者可能不会自动成为协作者，根据实际业务逻辑修改断言
+        # assert len(json["data"]) >= 1
+
+    def test_task_add_collaborator(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # prepare data
+        current_user = crud_user.get_user_by_username(
+            db=db, username="test@example.com"
+        )
+        data = {
+            "name": "collaborative task",
+            "description": "task description",
+            "tips": "task tips",
+        }
+        task_response = client.post(
+            f"{settings.API_V1_STR}/tasks",
+            headers=testuser_token_headers,
+            json=data,
+        )
+        task_id = task_response.json()["data"]["id"]
+
+        # run - 尝试添加当前用户为协作者（即使它已经是协作者）
+        r = client.post(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators",
+            headers=testuser_token_headers,
+            json={"user_id": current_user.id},
+        )
+
+        # check
+        json = r.json()
+        assert r.status_code == 201
+        assert json["data"]["id"] == current_user.id
+
+    def test_task_batch_add_collaborators(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # prepare data
+        current_user = crud_user.get_user_by_username(
+            db=db, username="test@example.com"
+        )
+        # 获取或创建测试用户
+        new_user = crud_user.get_user_by_username(db=db, username="collaborator@example.com")
+        if not new_user:
+            new_user = crud_user.create(
+                db=db,
+                user=User(
+                    username="collaborator@example.com",
+                    hashed_password="hashed_password",
+                ),
+            )
+        
+        data = {
+            "name": "batch collaborative task",
+            "description": "task description",
+            "tips": "task tips",
+        }
+        task_response = client.post(
+            f"{settings.API_V1_STR}/tasks",
+            headers=testuser_token_headers,
+            json=data,
+        )
+        task_id = task_response.json()["data"]["id"]
+
+        # run
+        r = client.post(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators/batch_add",
+            headers=testuser_token_headers,
+            json={"user_ids": [current_user.id, new_user.id]},
+        )
+
+        # check
+        json = r.json()
+        assert r.status_code == 201
+        # 期望只有一个协作者
+        assert len(json["data"]) == 1
+        user_ids = [user["id"] for user in json["data"]]
+        assert new_user.id in user_ids
+
+    def test_task_batch_remove_collaborators(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # prepare data
+        current_user = crud_user.get_user_by_username(
+            db=db, username="test@example.com"
+        )
+        # 获取或创建测试用户
+        new_user = crud_user.get_user_by_username(db=db, username="to_remove@example.com")
+        if not new_user:
+            new_user = crud_user.create(
+                db=db,
+                user=User(
+                    username="to_remove@example.com",
+                    hashed_password="hashed_password",
+                ),
+            )
+        
+        data = {
+            "name": "remove collaborative task",
+            "description": "task description",
+            "tips": "task tips",
+        }
+        task_response = client.post(
+            f"{settings.API_V1_STR}/tasks",
+            headers=testuser_token_headers,
+            json=data,
+        )
+        task_id = task_response.json()["data"]["id"]
+
+        # 先添加协作者
+        client.post(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators/batch_add",
+            headers=testuser_token_headers,
+            json={"user_ids": [new_user.id]},
+        )
+
+        # run - 移除协作者
+        r = client.post(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators/batch_remove",
+            headers=testuser_token_headers,
+            json={"user_ids": [new_user.id]},
+        )
+
+        # check
+        json = r.json()
+        assert r.status_code == 201
+        assert json["data"]["ok"] == True
+
+        # 验证协作者已被移除
+        collaborators = client.get(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators",
+            headers=testuser_token_headers,
+        )
+        assert new_user.id not in [c["id"] for c in collaborators.json()["data"]]
+
+    def test_task_remove_collaborator(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # prepare data
+        # 获取或创建测试用户
+        new_user = crud_user.get_user_by_username(db=db, username="single_remove@example.com")
+        if not new_user:
+            new_user = crud_user.create(
+                db=db,
+                user=User(
+                    username="single_remove@example.com",
+                    hashed_password="hashed_password",
+                ),
+            )
+        
+        data = {
+            "name": "single remove task",
+            "description": "task description",
+            "tips": "task tips",
+        }
+        task_response = client.post(
+            f"{settings.API_V1_STR}/tasks",
+            headers=testuser_token_headers,
+            json=data,
+        )
+        task_id = task_response.json()["data"]["id"]
+
+        # 先添加协作者
+        client.post(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators",
+            headers=testuser_token_headers,
+            json={"user_id": new_user.id},
+        )
+
+        # run - 移除单个协作者
+        r = client.delete(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators/{new_user.id}",
+            headers=testuser_token_headers,
+        )
+
+        # check
+        json = r.json()
+        assert r.status_code == 200
+        assert json["data"]["ok"] == True
+
+        # 验证协作者已被移除
+        collaborators = client.get(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators",
+            headers=testuser_token_headers,
+        )
+        assert new_user.id not in [c["id"] for c in collaborators.json()["data"]]
+
+    def test_task_not_found_for_collaborators(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # 测试不存在的任务
+        r1 = client.get(
+            f"{settings.API_V1_STR}/tasks/99999/collaborators",
+            headers=testuser_token_headers,
+        )
+        assert r1.status_code == 404
+        
+    def test_user_not_found_for_collaborators(
+        self, client: TestClient, testuser_token_headers: dict, db: Session
+    ) -> None:
+        # 测试添加不存在的用户
+        
+        data = {
+            "name": "error test task",
+            "description": "task description",
+            "tips": "task tips",
+        }
+        task_response = client.post(
+            f"{settings.API_V1_STR}/tasks",
+            headers=testuser_token_headers,
+            json=data,
+        )
+        task_id = task_response.json()["data"]["id"]
+        
+        r2 = client.post(
+            f"{settings.API_V1_STR}/tasks/{task_id}/collaborators",
+            headers=testuser_token_headers,
+            json={"user_id": 99999},  # 不存在的用户ID
+        )
+        # 验证状态码是4xx，表示客户端错误
+        assert r2.status_code == 404