File size: 4,516 Bytes
cdd5c14 059e444 cdd5c14 152d722 acc27be 152d722 3e73bd3 acc27be 152d722 cdd5c14 79c0611 cdd5c14 acc27be 152d722 059e444 152d722 059e444 79c0611 acc27be 152d722 cdd5c14 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
import os
import asyncio
import json
import logging
import httpx
from pikpakapi import PikPakApi
from typing import Union, Any, Dict, List, Optional
from fastapi import (
FastAPI,
Depends,
Request,
Body,
Response,
HTTPException,
status,
Request,
)
from fastapi.responses import StreamingResponse, HTMLResponse, JSONResponse
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.templating import Jinja2Templates
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Extra
class PostRequest(BaseModel):
class Config:
extra = Extra.allow
class FileRequest(BaseModel):
size: int = 100
parent_id: str | None = ""
next_page_token: str | None = ""
additional_filters: Dict | None = {}
class Config:
extra = Extra.allow
class OfflineRequest(BaseModel):
file_url: str = ""
parent_id: str | None = ""
name: str | None = ""
class Config:
extra = Extra.allow
security = HTTPBearer()
# SECRET_TOKEN = "SECRET_TOKEN"
SECRET_TOKEN = os.getenv("SECRET_TOKEN")
if SECRET_TOKEN is None:
raise ValueError("请在环境变量中设置SECRET_TOKEN，确保安全!")
THUNDERX_USERNAME = os.getenv("THUNDERX_USERNAME")
if THUNDERX_USERNAME is None:
raise ValueError("请在环境变量中设置THUNDERX_USERNAME，用户名【邮箱】用来登陆!")
THUNDERX_PASSWORD = os.getenv("THUNDERX_PASSWORD")
if THUNDERX_PASSWORD is None:
raise ValueError("请在环境变量中设置THUNDERX_PASSWORD，密码用来登陆!")
PROXY_URL = os.getenv("PROXY_URL")
async def verify_token(
request: Request, credentials: HTTPAuthorizationCredentials = Depends(security)
):
excluded_paths = ["/"] # 需要排除的路径列表
if request.url.path in excluded_paths:
return # 直接跳过验证
# 验证Bearer格式
if credentials.scheme != "Bearer":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication scheme",
)
# 验证令牌内容
if credentials.credentials != SECRET_TOKEN:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired token"
)
app = FastAPI(dependencies=[Depends(verify_token)])
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
templates = Jinja2Templates(
directory="templates", variable_start_string="{[", variable_end_string="]}"
)
THUNDERX_CLIENT = None
async def log_token(THUNDERX_CLIENT, extra_data):
logging.info(f"Token: {THUNDERX_CLIENT.encoded_token}, Extra Data: {extra_data}")
@app.on_event("startup")
async def init_client():
global THUNDERX_CLIENT
if not os.path.exists("thunderx.txt"):
THUNDERX_CLIENT = PikPakApi(
username=THUNDERX_USERNAME,
password=THUNDERX_PASSWORD,
httpx_client_args=None,
token_refresh_callback=log_token,
token_refresh_callback_kwargs={"extra_data": "test"},
)
await THUNDERX_CLIENT.login()
await THUNDERX_CLIENT.refresh_access_token()
with open("thunderx.json", "w") as f:
f.write(json.dumps(THUNDERX_CLIENT.to_dict(), indent=4))
else:
with open("thunderx.txt", "r") as f:
data = json.load(f)
THUNDERX_CLIENT = PikPakApi.from_dict(data)
# await client.refresh_access_token()
print(json.dumps(THUNDERX_CLIENT.get_user_info(), indent=4))
print(
json.dumps(
await THUNDERX_CLIENT.events(),
indent=4,
)
)
@app.get("/", response_class=HTMLResponse)
async def home(request: Request):
return templates.TemplateResponse("index.html", {"request": request})
@app.post("/files")
async def get_files(item: FileRequest):
return await THUNDERX_CLIENT.file_list(
item.size, item.parent_id, item.next_page_token, item.additional_filters
)
@app.get("/files/{file_id}")
async def get_file_info(file_id: str):
return await THUNDERX_CLIENT.get_download_url(file_id)
@app.post("/offline")
async def offline(item: OfflineRequest):
return await THUNDERX_CLIENT.offline_download(
item.file_url, item.parent_id, item.name
)
@app.get("/userinfo")
async def userinfo():
return THUNDERX_CLIENT.get_user_info()
|