File size: 3,232 Bytes
f6e10bb 79a5b4c f6e10bb 1a859c0 f6e10bb | 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 | import os
import random
import asyncio
import nest_asyncio
import uvicorn
import aiohttp
from fastapi import FastAPI
from contextlib import asynccontextmanager
# Lifespan context manager for startup/shutdown events
@asynccontextmanager
async def lifespan(app: FastAPI):
asyncio.create_task(start_periodic_task())
yield
app = FastAPI(
title="UltXtraHF",
version="1.0.2",
contact={
"name": "🌀ʊʄ⊕ք🌀",
"url": "https://github.com/ufoptg/UltroidBackup/",
},
docs_url=None,
redoc_url=None,
lifespan=lifespan
)
@app.get("/")
def root():
return {"message": "Welcome! Please use the /status endpoint."}
@app.get("/status")
def status():
return {"message": "running"}
# Get the Hugging Face token from the environment
HUGGING_TOKEN = os.environ.get("HF2")
# Define an async HTTP GET helper using aiohttp
async def async_searcher(api_url, headers, re_json=True):
async with aiohttp.ClientSession() as session:
async with session.get(api_url, headers=headers, timeout=10) as response:
if re_json:
return await response.json()
return await response.text()
# Asynchronously call the Hugging Face API endpoint with authorization
async def hfv(api_url, timeout=10):
try:
headers = {
"Authorization": f"Bearer {HUGGING_TOKEN}",
"Content-Type": "application/json",
}
response = await async_searcher(api_url, headers=headers, re_json=True)
stat = response.get("message", "ded")
return stat
except Exception as e:
print("Error occurred in hfv:", e)
return False
# Asynchronously call the second API (without special headers)
async def hfv2(api_url, timeout=10):
try:
async with aiohttp.ClientSession() as session:
async with session.get(api_url, timeout=timeout) as response:
resp_text = await response.text()
print("second api called successfully, RESPONSE:")
return resp_text
except Exception as e:
print("Error occurred in hfv2:", e)
return False
# Periodically call the Hugging Face API endpoint
async def periodic_hfv(api_url, interval_range):
while True:
result = await hfv(api_url)
print("Periodic API call result:", result)
interval = random.randint(*interval_range)
await asyncio.sleep(interval)
# Periodically call the second API every 5 minutes (300 seconds)
async def periodic_second_api(api_url, interval):
while True:
await hfv2(api_url)
await asyncio.sleep(interval)
# Start the periodic tasks for both APIs
async def start_periodic_task():
# Start periodic task for Hugging Face API: interval between 2 to 4 hours
hf_api_url = "https://akborana4-apimaster.hf.space/status"
asyncio.create_task(periodic_hfv(hf_api_url, (2 * 3600, 4 * 3600)))
# Start periodic task for second API: fixed interval every 5 minutes (300 seconds)
second_api_url = "https://filetolink-rn17.onrender.com"
asyncio.create_task(periodic_second_api(second_api_url, 300))
if __name__ == "__main__":
nest_asyncio.apply()
uvicorn.run(app, host="0.0.0.0", port=7860)
|