diff --git a/domain/adapters/providers/alist.py b/domain/adapters/providers/alist.py index fdee4de..78175a1 100644 --- a/domain/adapters/providers/alist.py +++ b/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 diff --git a/domain/adapters/providers/foxel.py b/domain/adapters/providers/foxel.py index 65a54ae..cc89fa0 100644 --- a/domain/adapters/providers/foxel.py +++ b/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) diff --git a/domain/adapters/providers/ftp.py b/domain/adapters/providers/ftp.py index c2030db..8363e2c 100644 --- a/domain/adapters/providers/ftp.py +++ b/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() diff --git a/domain/adapters/providers/local.py b/domain/adapters/providers/local.py index 12ddd4f..4f4d5e2 100644 --- a/domain/adapters/providers/local.py +++ b/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() diff --git a/domain/adapters/providers/quark.py b/domain/adapters/providers/quark.py index 2af3d27..1b2ad0c 100644 --- a/domain/adapters/providers/quark.py +++ b/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"\n{i+1}\n{etags[i]}\n\n" for i in range(len(etags))] + body_xml = "\n\n" + "".join(parts_xml) + "" + 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") diff --git a/domain/adapters/providers/sftp.py b/domain/adapters/providers/sftp.py index 715925d..aa887b0 100644 --- a/domain/adapters/providers/sftp.py +++ b/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: diff --git a/domain/adapters/providers/telegram.py b/domain/adapters/providers/telegram.py index 61e2f68..28d243d 100644 --- a/domain/adapters/providers/telegram.py +++ b/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: diff --git a/domain/virtual_fs/routes.py b/domain/virtual_fs/routes.py index 83e70ad..5a0b44b 100644 --- a/domain/virtual_fs/routes.py +++ b/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)