support text direct downloads

Author: monlor

apps/base/views.py

@@ -4,13 +4,14 @@
 import os
 import uuid
 from datetime import timedelta
-from urllib.parse import unquote
+from urllib.parse import quote, unquote
 
 from typing import Optional, Tuple, Union
 
 from fastapi import APIRouter, Form, Request, UploadFile, File, Depends, HTTPException
 from pydantic import BaseModel, ValidationError
 from starlette import status
+from starlette.responses import Response
 from tortoise.expressions import Case, F, Q, When
 
 from apps.admin.dependencies import share_required_login
@@ -334,6 +335,17 @@ async def get_code_file(code: str, ip: str = Depends(ip_limit["error"])):
     assert isinstance(file_code, FileCodes)
     if not await consume_file_usage(file_code):
         return APIResponse(code=404, detail="文件已过期")
+    if file_code.text is not None:
+        filename = f"{file_code.prefix or 'Text'}{file_code.suffix or '.txt'}"
+        return Response(
+            content=file_code.text,
+            media_type="text/plain",
+            headers={
+                "Content-Disposition": (
+                    f"attachment; filename*=UTF-8''{quote(filename, safe='')}"
+                )
+            },
+        )
     return await file_storage.get_file_response(file_code)
 
 

tests/test_issue_482_share_usage.py

@@ -71,6 +71,7 @@ async def _run_scenario(self):
             await self._assert_count_consumption_is_atomic()
             await self._assert_time_expiration_and_usage_are_atomic()
             await self._assert_download_url_is_single_use()
+            await self._assert_text_share_downloads_as_txt()
             await self._assert_limited_files_do_not_expose_direct_urls()
             await self._assert_metadata_counts_all_attempts()
         finally:
@@ -156,6 +157,30 @@ async def _assert_download_url_is_single_use(self):
         self.assertEqual(record.expired_count, 0)
         self.assertEqual(record.used_count, 1)
 
+    async def _assert_text_share_downloads_as_txt(self):
+        record = await FileCodes.create(
+            code="text-download",
+            prefix="TAG",
+            text="hello, 文本直链",
+            expired_count=-1,
+            expired_at=None,
+        )
+
+        with patch.dict(views.storages, {"local": FakeStorage}):
+            response = await views.get_code_file(
+                code=record.code, ip="127.0.0.1"
+            )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.body.decode("utf-8"), record.text)
+        self.assertEqual(response.media_type, "text/plain")
+        self.assertEqual(
+            response.headers["content-disposition"],
+            "attachment; filename*=UTF-8''TAG.txt",
+        )
+        await record.refresh_from_db()
+        self.assertEqual(record.used_count, 1)
+
     async def _assert_limited_files_do_not_expose_direct_urls(self):
         record = await FileCodes.create(
             code="proxy-only",