-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
71 lines (63 loc) · 2.82 KB
/
main.py
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
import threading
from fastapi import FastAPI, Request, Form
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from contextlib import asynccontextmanager
from db import Session
import db
import qb
from bangumi import DeleteBangumi, AddNewBangumi, coroutines
templates = Jinja2Templates(directory="templates")
# A signal method for signaling main thread main executed
import asyncio
from bangumi import BangumiMain
@asynccontextmanager
async def app_lifespan(app: FastAPI):
event_loop_bangumi = asyncio.new_event_loop()
bangumi_started = threading.Event()
def run_bangumi_main():
asyncio.set_event_loop(event_loop_bangumi)
event_loop_bangumi.call_soon_threadsafe(asyncio.create_task, BangumiMain(bangumi_started))
event_loop_bangumi.run_forever()
track_thread = threading.Thread(target=run_bangumi_main)
track_thread.start()
bangumi_started.wait()
yield
for coroutine in coroutines.values():
if not coroutine.done():
coroutine.cancel()
event_loop_bangumi.stop()
app = FastAPI(lifespan=app_lifespan)
@app.post("/delete_bangumi", response_class=RedirectResponse)
async def delete_bangumi(bangumi_id: int = Form(...)):
await DeleteBangumi(bangumi_id)
return RedirectResponse("/", status_code=303)
@app.post("/add_bangumi", response_class=RedirectResponse)
async def add_bangumi(name: str = Form(...), season: int = Form(...), rss: str = Form(...), regex: str = Form(...)):
await AddNewBangumi(name, season, rss, regex)
return RedirectResponse("/", status_code=303)
@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
with Session() as session:
bangumis = session.query(db.Bangumi).all()
bangumi_data = [b.to_dict() for b in bangumis]
for b in bangumi_data:
status = "Running"
if not (b['id'] in coroutines): status = 'Not Started'
elif coroutines[b['id']].done():
if coroutines[b['id']].cancelled(): status = 'Cancelled'
else: status = 'Done'
b['coroutine_status'] = status
return templates.TemplateResponse("index.html", {"request": request, "bangumis": bangumi_data})
@app.get("/track/{bangumi_id}", response_class=HTMLResponse)
async def track_bangumi(request: Request, bangumi_id: int):
with Session() as session:
bangumi = session.query(db.Bangumi).filter(db.Bangumi.id == bangumi_id).first()
bangumi_data = bangumi.to_dict()
for e in bangumi_data['torrents']:
status, rate = qb.GetTorrentStatus(e['hash'])
e['status'] = status
return templates.TemplateResponse("track.html", {"request": request, "bangumi": bangumi_data})
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)