feat: implement write_upload_file method for various adapters to handle file uploads

domain/adapters/providers/alist.py

@@ -381,6 +381,31 @@ class AListApiAdapterBase:
             except Exception:
                 pass
 
+    async def write_upload_file(self, root: str, rel: str, file_obj, filename: str | None, file_size: int | None = None, content_type: str | None = None):
+        full_path = _join_fs_path(root, rel)
+        token = await self._ensure_token()
+        headers = {
+            "Authorization": token,
+            "File-Path": quote(full_path, safe="/"),
+        }
+        name = filename or Path(rel).name or "file"
+        mime = content_type or "application/octet-stream"
+        files = {"file": (name, file_obj, mime)}
+        async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client:
+            resp = await client.put(self.base_url + "/api/fs/form", headers=headers, files=files)
+            resp.raise_for_status()
+            payload = resp.json()
+        if not isinstance(payload, dict):
+            raise HTTPException(502, detail=f"{self.product_name} upload: invalid response")
+        code = payload.get("code")
+        if code not in (0, 200):
+            msg = payload.get("message") or payload.get("msg") or ""
+            raise HTTPException(502, detail=f"{self.product_name} upload failed: {msg}")
+        data = payload.get("data")
+        if isinstance(data, dict) and file_size is not None and "size" not in data:
+            data["size"] = file_size
+        return data
+
     async def write_file_stream(self, root: str, rel: str, data_iter: AsyncIterator[bytes]):
         full_path = _join_fs_path(root, rel)
         suffix = Path(rel).suffix

domain/adapters/providers/foxel.py

@@ -250,6 +250,30 @@ class FoxelAdapter:
             return True
         raise HTTPException(502, detail="Foxel 写入失败")
 
+    async def write_upload_file(self, root: str, rel: str, file_obj, filename: str | None, file_size: int | None = None, content_type: str | None = None):
+        rel = (rel or "").lstrip("/")
+        full_path = _join_fs_path(root, rel)
+        url = self.base_url + self._file_path(full_path)
+        name = filename or Path(rel).name or "file"
+        mime = content_type or "application/octet-stream"
+        for attempt in range(2):
+            try:
+                if callable(getattr(file_obj, "seek", None)):
+                    file_obj.seek(0)
+            except Exception:
+                pass
+            token = await self._ensure_token()
+            headers = {"Authorization": f"Bearer {token}"}
+            files = {"file": (name, file_obj, mime)}
+            async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client:
+                resp = await client.post(url, headers=headers, files=files)
+            if resp.status_code == 401 and attempt == 0:
+                self._token = None
+                continue
+            resp.raise_for_status()
+            return {"size": file_size or 0}
+        raise HTTPException(502, detail="Foxel 上传失败")
+
     async def write_file_stream(self, root: str, rel: str, data_iter: AsyncIterator[bytes]):
         rel = (rel or "").lstrip("/")
         full_path = _join_fs_path(root, rel)

domain/adapters/providers/ftp.py

@@ -238,6 +238,39 @@ class FTPAdapter:
 
         await asyncio.to_thread(_do_write)
 
+    async def write_upload_file(self, root: str, rel: str, file_obj, filename: str | None, file_size: int | None = None, content_type: str | None = None):
+        path = _join_remote(root, rel)
+
+        def _ensure_dirs(ftp: FTP, dir_path: str):
+            parts = [p for p in dir_path.strip("/").split("/") if p]
+            cur = "/"
+            for p in parts:
+                cur = _join_remote(cur, p)
+                try:
+                    ftp.mkd(cur)
+                except Exception:
+                    pass
+
+        def _do_upload():
+            ftp = self._connect()
+            try:
+                parent = "/" if "/" not in path.strip("/") else path.rsplit("/", 1)[0]
+                _ensure_dirs(ftp, parent)
+                try:
+                    if callable(getattr(file_obj, "seek", None)):
+                        file_obj.seek(0)
+                except Exception:
+                    pass
+                ftp.storbinary("STOR " + path, file_obj)
+            finally:
+                try:
+                    ftp.quit()
+                except Exception:
+                    pass
+
+        await asyncio.to_thread(_do_upload)
+        return {"size": file_size or 0}
+
     async def write_file_stream(self, root: str, rel: str, data_iter: AsyncIterator[bytes]):
         # KISS: 聚合后一次性写入
         buf = bytearray()

domain/adapters/providers/local.py

@@ -114,6 +114,32 @@ class LocalAdapter:
         if not pre_exists:
             await asyncio.to_thread(_apply_mode, fp, DEFAULT_FILE_MODE)
 
+    async def write_upload_file(self, root: str, rel: str, file_obj, filename: str | None, file_size: int | None = None, content_type: str | None = None):
+        fp = _safe_join(root, rel)
+        pre_exists = fp.exists()
+        await asyncio.to_thread(os.makedirs, fp.parent, mode=DEFAULT_DIR_MODE, exist_ok=True)
+
+        def _copy():
+            try:
+                if callable(getattr(file_obj, "seek", None)):
+                    file_obj.seek(0)
+            except Exception:
+                pass
+            with open(fp, "wb") as f:
+                shutil.copyfileobj(file_obj, f)
+
+        await asyncio.to_thread(_copy)
+        if not pre_exists:
+            await asyncio.to_thread(_apply_mode, fp, DEFAULT_FILE_MODE)
+
+        size = file_size
+        if size is None:
+            try:
+                size = fp.stat().st_size
+            except Exception:
+                size = 0
+        return {"size": int(size or 0)}
+
     async def write_file_stream(self, root: str, rel: str, data_iter: AsyncIterator[bytes]):
         fp = _safe_join(root, rel)
         pre_exists = fp.exists()

domain/adapters/providers/quark.py

@@ -453,6 +453,159 @@ class QuarkAdapter:
             yield data
         return await self.write_file_stream(root, rel, gen())
 
+    async def write_upload_file(self, root: str, rel: str, file_obj, filename: str | None, file_size: int | None = None, content_type: str | None = None):
+        if not rel or rel.endswith("/"):
+            raise HTTPException(400, detail="Invalid file path")
+
+        parent = rel.rsplit("/", 1)[0] if "/" in rel else ""
+        name = filename or rel.rsplit("/", 1)[-1]
+        base_fid = root or self.root_fid
+        parent_fid = await self._resolve_dir_fid_from(base_fid, parent)
+
+        md5 = hashlib.md5()
+        sha1 = hashlib.sha1()
+        total = 0
+        try:
+            if callable(getattr(file_obj, "seek", None)):
+                file_obj.seek(0)
+        except Exception:
+            pass
+        while True:
+            chunk = file_obj.read(1024 * 1024)
+            if not chunk:
+                break
+            total += len(chunk)
+            md5.update(chunk)
+            sha1.update(chunk)
+
+        md5_hex = md5.hexdigest()
+        sha1_hex = sha1.hexdigest()
+
+        # 预上传，拿到上传信息
+        pre_resp = await self._upload_pre(name, total, parent_fid)
+        pre_data = pre_resp.get("data", {})
+
+        # hash 秒传
+        hash_body = {"md5": md5_hex, "sha1": sha1_hex, "task_id": pre_data.get("task_id")}
+        hash_resp = await self._request("POST", "/file/update/hash", json=hash_body)
+        if (hash_resp.get("data") or {}).get("finish") is True:
+            self._invalidate_children_cache(parent_fid)
+            return {"size": total}
+
+        # 分片上传
+        part_size = int((pre_resp.get("metadata") or {}).get("part_size") or 0)
+        if part_size <= 0:
+            raise HTTPException(502, detail="Invalid part_size from Quark")
+
+        bucket = pre_data.get("bucket")
+        obj_key = pre_data.get("obj_key")
+        upload_id = pre_data.get("upload_id")
+        upload_url = pre_data.get("upload_url")
+        if not (bucket and obj_key and upload_id and upload_url):
+            raise HTTPException(502, detail="Upload pre missing fields")
+
+        try:
+            upload_host = upload_url.split("://", 1)[1]
+        except Exception:
+            upload_host = upload_url
+        base_url = f"https://{bucket}.{upload_host}/{obj_key}"
+
+        try:
+            if callable(getattr(file_obj, "seek", None)):
+                file_obj.seek(0)
+        except Exception:
+            pass
+
+        etags: List[str] = []
+        oss_ua = "aliyun-sdk-js/6.6.1 Chrome 98.0.4758.80 on Windows 10 64-bit"
+        async with httpx.AsyncClient(timeout=None, follow_redirects=True) as client:
+            part_number = 1
+            left = total
+            while left > 0:
+                sz = min(part_size, left)
+                data_bytes = file_obj.read(sz)
+                if len(data_bytes) != sz:
+                    raise IOError("Failed to read part bytes")
+                now_str = time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime())
+                auth_meta = (
+                    "PUT\n\n"
+                    f"{self._guess_mime(name)}\n"
+                    f"{now_str}\n"
+                    f"x-oss-date:{now_str}\n"
+                    f"x-oss-user-agent:{oss_ua}\n"
+                    f"/{bucket}/{obj_key}?partNumber={part_number}&uploadId={upload_id}"
+                )
+                auth_req_body = {"auth_info": pre_data.get("auth_info"), "auth_meta": auth_meta, "task_id": pre_data.get("task_id")}
+                auth_resp = await self._request("POST", "/file/upload/auth", json=auth_req_body)
+                auth_key = (auth_resp.get("data") or {}).get("auth_key")
+                if not auth_key:
+                    raise HTTPException(502, detail="upload/auth missing auth_key")
+
+                put_headers = {
+                    "Authorization": auth_key,
+                    "Content-Type": self._guess_mime(name),
+                    "Referer": REFERER + "/",
+                    "x-oss-date": now_str,
+                    "x-oss-user-agent": oss_ua,
+                }
+                put_url = f"{base_url}?partNumber={part_number}&uploadId={upload_id}"
+                put_resp = await client.put(put_url, headers=put_headers, content=data_bytes)
+                if put_resp.status_code != 200:
+                    raise HTTPException(502, detail=f"Upload part failed status={put_resp.status_code} text={put_resp.text}")
+                etag = put_resp.headers.get("Etag", "")
+                etags.append(etag)
+                left -= sz
+                part_number += 1
+
+        parts_xml = [f"<Part>\n<PartNumber>{i+1}</PartNumber>\n<ETag>{etags[i]}</ETag>\n</Part>\n" for i in range(len(etags))]
+        body_xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n<CompleteMultipartUpload>\n" + "".join(parts_xml) + "</CompleteMultipartUpload>"
+        content_md5 = base64.b64encode(hashlib.md5(body_xml.encode("utf-8")).digest()).decode("ascii")
+        callback = pre_data.get("callback") or {}
+        try:
+            import json as _json
+            callback_b64 = base64.b64encode(_json.dumps(callback).encode("utf-8")).decode("ascii")
+        except Exception:
+            callback_b64 = ""
+
+        now_str = time.strftime("%a, %d %b %Y %H:%M:%S GMT", time.gmtime())
+        auth_meta_commit = (
+            "POST\n"
+            f"{content_md5}\n"
+            "application/xml\n"
+            f"{now_str}\n"
+            f"x-oss-callback:{callback_b64}\n"
+            f"x-oss-date:{now_str}\n"
+            f"x-oss-user-agent:{oss_ua}\n"
+            f"/{bucket}/{obj_key}?uploadId={upload_id}"
+        )
+        auth_commit_resp = await self._request("POST", "/file/upload/auth", json={"auth_info": pre_data.get("auth_info"), "auth_meta": auth_meta_commit, "task_id": pre_data.get("task_id")})
+        auth_key_commit = (auth_commit_resp.get("data") or {}).get("auth_key")
+        if not auth_key_commit:
+            raise HTTPException(502, detail="upload/auth(commit) missing auth_key")
+
+        async with httpx.AsyncClient(timeout=None, follow_redirects=True) as client:
+            commit_headers = {
+                "Authorization": auth_key_commit,
+                "Content-MD5": content_md5,
+                "Content-Type": "application/xml",
+                "Referer": REFERER + "/",
+                "x-oss-callback": callback_b64,
+                "x-oss-date": now_str,
+                "x-oss-user-agent": oss_ua,
+            }
+            commit_url = f"{base_url}?uploadId={upload_id}"
+            r = await client.post(commit_url, headers=commit_headers, content=body_xml.encode("utf-8"))
+            if r.status_code != 200:
+                raise HTTPException(502, detail=f"Upload commit failed status={r.status_code} text={r.text}")
+
+        await self._request("POST", "/file/upload/finish", json={"obj_key": obj_key, "task_id": pre_data.get("task_id")})
+        try:
+            await asyncio.sleep(1.0)
+        except Exception:
+            pass
+        self._invalidate_children_cache(parent_fid)
+        return {"size": total}
+
     async def write_file_stream(self, root: str, rel: str, data_iter: AsyncIterator[bytes]):
         if not rel or rel.endswith("/"):
             raise HTTPException(400, detail="Invalid file path")

domain/adapters/providers/sftp.py

@@ -157,6 +157,41 @@ class SFTPAdapter:
 
         await asyncio.to_thread(_do_write)
 
+    async def write_upload_file(self, root: str, rel: str, file_obj, filename: str | None, file_size: int | None = None, content_type: str | None = None):
+        path = _join_remote(root, rel)
+
+        def _ensure_dirs(sftp: paramiko.SFTPClient, dir_path: str):
+            parts = [p for p in dir_path.strip("/").split("/") if p]
+            cur = "/"
+            for p in parts:
+                cur = _join_remote(cur, p)
+                try:
+                    sftp.mkdir(cur)
+                except IOError:
+                    pass
+
+        def _do_upload():
+            sftp = self._connect()
+            try:
+                parent = "/" if "/" not in path.strip("/") else path.rsplit("/", 1)[0]
+                _ensure_dirs(sftp, parent)
+                try:
+                    if callable(getattr(file_obj, "seek", None)):
+                        file_obj.seek(0)
+                except Exception:
+                    pass
+                with sftp.open(path, "wb") as f:
+                    import shutil
+                    shutil.copyfileobj(file_obj, f)
+            finally:
+                try:
+                    sftp.close()
+                except Exception:
+                    pass
+
+        await asyncio.to_thread(_do_upload)
+        return {"size": file_size or 0}
+
     async def write_file_stream(self, root: str, rel: str, data_iter: AsyncIterator[bytes]):
         buf = bytearray()
         async for chunk in data_iter:

domain/adapters/providers/telegram.py

@@ -21,6 +21,30 @@ def _get_session_lock(session_string: str) -> asyncio.Lock:
         _SESSION_LOCKS[session_string] = lock
     return lock
 
+
+class _NamedFile:
+    def __init__(self, file_obj, name: str):
+        self._file = file_obj
+        self.name = name
+
+    def read(self, *args, **kwargs):
+        return self._file.read(*args, **kwargs)
+
+    def seek(self, *args, **kwargs):
+        return self._file.seek(*args, **kwargs)
+
+    def tell(self):
+        return self._file.tell()
+
+    def seekable(self):
+        return self._file.seekable()
+
+    def close(self):
+        return self._file.close()
+
+    def __getattr__(self, name):
+        return getattr(self._file, name)
+
 # 适配器类型标识
 ADAPTER_TYPE = "telegram"
 
@@ -278,13 +302,45 @@ class TelegramAdapter:
             if client.is_connected():
                 await client.disconnect()
 
+    async def write_upload_file(self, root: str, rel: str, file_obj, filename: str | None, file_size: int | None = None, content_type: str | None = None):
+        client = self._get_client()
+        name = filename or os.path.basename(rel) or "file"
+        file_like = _NamedFile(file_obj, name)
+
+        try:
+            await client.connect()
+            sent = await client.send_file(
+                self.chat_id,
+                file_like,
+                caption=file_like.name,
+                file_size=file_size,
+                mime_type=content_type,
+            )
+            message = sent[0] if isinstance(sent, list) and sent else sent
+            actual_rel = rel
+            size = file_size or 0
+            if message:
+                stored_name = file_like.name
+                file_meta = getattr(message, "file", None)
+                if file_meta and getattr(file_meta, "name", None):
+                    stored_name = file_meta.name
+                if getattr(message, "id", None) is not None:
+                    actual_rel = f"{message.id}_{stored_name}"
+                if file_meta and getattr(file_meta, "size", None):
+                    size = int(file_meta.size)
+            return {"rel": actual_rel, "size": size}
+        finally:
+            if client.is_connected():
+                await client.disconnect()
+
     async def write_file_stream(self, root: str, rel: str, data_iter: AsyncIterator[bytes]):
         """以流式方式上传文件"""
         client = self._get_client()
         filename = os.path.basename(rel) or "file"
         import tempfile
-        temp_dir = tempfile.gettempdir()
-        temp_path = os.path.join(temp_dir, filename)
+        suffix = os.path.splitext(filename)[1]
+        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tf:
+            temp_path = tf.name
         
         total_size = 0
         try:

domain/virtual_fs/routes.py

@@ -6,6 +6,7 @@ from fastapi import HTTPException, UploadFile
 from fastapi.responses import Response
 
 from domain.config import ConfigService
+from domain.tasks import TaskService
 from .thumbnail import (
     get_or_create_thumb,
     is_image_filename,
@@ -216,7 +217,7 @@ class VirtualFSRouteMixin(VirtualFSTempLinkMixin):
         full_path = cls._normalize_path(full_path)
         if full_path.endswith("/"):
             raise HTTPException(400, detail="Path must be a file")
-        adapter, _m, root, rel = await cls.resolve_adapter_and_rel(full_path)
+        adapter, adapter_model, root, rel = await cls.resolve_adapter_and_rel(full_path)
         exists_func = getattr(adapter, "exists", None)
         if not overwrite and callable(exists_func):
             try:
@@ -227,6 +228,21 @@ class VirtualFSRouteMixin(VirtualFSTempLinkMixin):
             except Exception:
                 pass
 
+        upload_func = getattr(adapter, "write_upload_file", None)
+        if callable(upload_func):
+            try:
+                await file.seek(0)
+            except Exception:
+                pass
+            size_hint = getattr(file, "size", None)
+            if not isinstance(size_hint, int):
+                size_hint = None
+            filename = file.filename or (rel.rsplit("/", 1)[-1] if rel else "file")
+            result = await upload_func(root, rel, file.file, filename, size_hint, file.content_type)
+            final_path, size = cls._normalize_written_result(full_path, adapter_model, result, size_hint or 0)
+            await TaskService.trigger_tasks("file_written", final_path)
+            return {"uploaded": True, "path": final_path, "size": size, "overwrite": overwrite}
+
         async def gen():
             while True:
                 chunk = await file.read(chunk_size)