File size: 3,588 Bytes
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 |
import os
import asyncio
import json
import logging
import httpx
from pikpakapi import PikPakApi
from typing import Union
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
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)
):
# 验证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.get("/files")
async def get_files(kw: str = ""):
return await THUNDERX_CLIENT.file_list()
@app.get("/userinfo")
async def userinfo():
return THUNDERX_CLIENT.get_user_info()
|