94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
import requests
|
|
import juneberry
|
|
from juneberry.colors import Color, R, G, B
|
|
from uvicorn import run
|
|
from fastapi import FastAPI
|
|
from starlette.requests import Request
|
|
from fastapi.responses import PlainTextResponse, StreamingResponse, RedirectResponse
|
|
|
|
app = FastAPI()
|
|
logger = juneberry.Logger(theme=juneberry.Theme(
|
|
Color.Custom.from_rgb((R(189), G(224), B(254))),
|
|
Color.Custom.from_rgb((R(205), G(180), B(219))),
|
|
Color.Custom.from_rgb((R(162), G(210), B(255))),
|
|
Color.Custom.from_rgb((R(255), G(200), B(221))),
|
|
Color.Custom.from_rgb((R(255), G(175), B(204))),
|
|
Color.Custom.from_rgb((R(255), G(254), B(214))),
|
|
Color.Custom.from_rgb((R(255), G(254), B(214)))
|
|
))
|
|
|
|
|
|
@app.get("/")
|
|
@app.head("/")
|
|
async def root(request: Request):
|
|
logger.info(f'Root accessed from {request.client.host}')
|
|
return PlainTextResponse(
|
|
f"Simple linkbox.to direct link grabber. \n"
|
|
f"Paste full link to LinkBox shared file to get direct download link, that you can share to friends. \n"
|
|
f"Add /s to end of this link, to create proxied streamed link that can be accessible from curl or wget "
|
|
f"(slower stream, shorter link). \n"
|
|
f"Add /l to end of this link to get direct link on LinkBox cdn (faster stream, longer link) \n"
|
|
f"Paste file id (symbols after last slash in original link) to directly download file. You can also add /s or "
|
|
f"/l to it. \n"
|
|
f"Use {request.url._url}[original share link]/d to download file directly."
|
|
)
|
|
|
|
|
|
@app.get("/{linkbox:path}")
|
|
async def return_hui(linkbox: str, request: Request):
|
|
mode = None
|
|
for catch in ['d', 'l', 's']:
|
|
if '/' + catch == linkbox[-2:]:
|
|
mode = catch
|
|
linkbox = linkbox[:-2]
|
|
|
|
if linkbox[-1] == '/':
|
|
linkbox = linkbox[:-1]
|
|
|
|
if (is_link := 'linkbox.to' in linkbox or 'lbx.to' in linkbox) and \
|
|
(resp := request.url._url.replace(request.url.path, '/' + linkbox.split('/')[-1])) and \
|
|
mode != 'd':
|
|
logger.info(f"{request.client.host} - Url for https://lbx.to/f/{linkbox} - {resp} (default full link)")
|
|
return PlainTextResponse(resp)
|
|
elif is_link:
|
|
logger.info(f"{request.client.host} - Url for https://lbx.to/f/{linkbox} - {resp} (d mode)")
|
|
return RedirectResponse(resp)
|
|
|
|
try:
|
|
itemId = requests.get(
|
|
f'https://linkbox.to/api/file/share_out_list?shareToken={linkbox}'
|
|
).json()["data"]["itemId"]
|
|
iteminfo = requests.get(f'https://linkbox.to/api/file/detail?itemId={itemId}').json()['data']['itemInfo']
|
|
url = iteminfo['url'].replace('wht.nuplink', 'wdl.nuplink') + f'&filename={iteminfo["name"]}'
|
|
except Exception as e:
|
|
logger.error(f"{request.client.host} - Error {e} while processing {linkbox}")
|
|
return PlainTextResponse(f'Error: {e}. Check file id or url.', status_code=404)
|
|
|
|
if mode == 's':
|
|
def iterate_file():
|
|
session = requests.Session()
|
|
r = session.get(url, stream=True)
|
|
r.raise_for_status()
|
|
|
|
for chunk in r.iter_content(1024 * 1024):
|
|
yield chunk
|
|
|
|
logger.info(f"{request.client.host} - Streaming {iteminfo['name']} from {url}")
|
|
return StreamingResponse(iterate_file(), headers=requests.head(url).headers)
|
|
|
|
if mode == 'l':
|
|
logger.info(f"{request.client.host} - Url for https://lbx.to/f/{linkbox} - {url} (l mode)")
|
|
return PlainTextResponse(url)
|
|
|
|
logger.info(f"{request.client.host} - Redirecting https://lbx.to/f/{linkbox} to {url} (default mode)")
|
|
return RedirectResponse(url)
|
|
|
|
|
|
@app.on_event('startup')
|
|
def on_startup():
|
|
logger.info('Started!')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
run(app=app, host="0.0.0.0", port=1488, log_level='critical')
|