1import asyncio
2import logging
3from contextlib import asynccontextmanager
4from datetime import datetime as dt
5from datetime import timedelta as td
6from typing import AsyncGenerator
7
8from apscheduler.schedulers.asyncio import AsyncIOScheduler
9from fastapi import FastAPI, Request, status
10from fastapi.encoders import jsonable_encoder
11from fastapi.exceptions import RequestValidationError
12from fastapi.middleware.cors import CORSMiddleware
13from fastapi.responses import HTMLResponse, JSONResponse
14from fastapi_pagination import add_pagination
15from starlette.staticfiles import StaticFiles
16from uvicorn import Config, Server
17
18from app.config.env import (
19 DEBUG,
20 DOCS,
21 HOME_PAGE_TEMPLATE,
22 UVICORN_HOST,
23 UVICORN_PORT,
24 UVICORN_SSL_CERTFILE,
25 UVICORN_SSL_KEYFILE,
26 UVICORN_UDS,
27 DASHBOARD_PATH,
28 WEBHOOK_ADDRESS,
29)
30from app.templates import render_template
31from . import __version__, telegram
32from .routes import api_router
33from .tasks import (
34 delete_expired_reminders,
35 nodes_startup,
36 record_realtime_bandwidth,
37 record_user_usages,
38 reset_user_data_usage,
39 review_users,
40 send_notifications,
41)
42
43logger = logging.getLogger(__name__)
44
45
46@asynccontextmanager
47async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
48 await nodes_startup()
49 yield
50 scheduler.shutdown()
51 logger.info("Sending pending notifications before shutdown...")
52 await send_notifications()
53
54
55app = FastAPI(
56 title="MarzneshinAPI",
57 description="Unified GUI Censorship Resistant Solution Powered by Xray",
58 version=__version__,
59 lifespan=lifespan,
60 docs_url="/docs" if DOCS else None,
61 redoc_url="/redoc" if DOCS else None,
62)
63
64app.include_router(api_router)
65add_pagination(app)
66
67
68@app.get("/", response_class=HTMLResponse)
69def home_page():
70 return render_template(HOME_PAGE_TEMPLATE)
71
72
73app.add_middleware(
74 CORSMiddleware,
75 allow_origins=["*"],
76 allow_credentials=True,
77 allow_methods=["*"],
78 allow_headers=["*"],
79)
80
81scheduler = AsyncIOScheduler(timezone="UTC")
82scheduler.add_job(record_user_usages, "interval", coalesce=True, seconds=30)
83scheduler.add_job(
84 review_users, "interval", seconds=30, coalesce=True, max_instances=1
85)
86scheduler.add_job(reset_user_data_usage, "interval", coalesce=True, hours=1)
87scheduler.add_job(
88 record_realtime_bandwidth,
89 "interval",
90 seconds=1,
91 coalesce=True,
92 max_instances=1,
93)
94
95if WEBHOOK_ADDRESS:
96 scheduler.add_job(
97 send_notifications, "interval", seconds=30, replace_existing=True
98 )
99 scheduler.add_job(
100 delete_expired_reminders,
101 "interval",
102 hours=2,
103 start_date=dt.utcnow() + td(minutes=1),
104 )
105
106
107@app.exception_handler(RequestValidationError)
108def validation_exception_handler(
109 request: Request, exc: RequestValidationError
110):
111 details = {}
112 for error in exc.errors():
113 details[error["loc"][-1]] = error.get("msg")
114 return JSONResponse(
115 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
116 content=jsonable_encoder({"detail": details}),
117 )
118
119
120async def main():
121 if not DEBUG:
122 app.mount(
123 DASHBOARD_PATH,
124 StaticFiles(directory="dashboard/dist", html=True),
125 name="dashboard",
126 )
127 app.mount(
128 "/static/",
129 StaticFiles(directory="dashboard/dist/static"),
130 name="static",
131 )
132 app.mount(
133 "/locales/",
134 StaticFiles(directory="dashboard/dist/locales"),
135 name="locales",
136 )
137 scheduler.start()
138 asyncio.create_task(telegram.start_bot())
139 cfg = Config(
140 app=app,
141 host=UVICORN_HOST,
142 port=UVICORN_PORT,
143 uds=(None if DEBUG else UVICORN_UDS),
144 ssl_certfile=UVICORN_SSL_CERTFILE,
145 ssl_keyfile=UVICORN_SSL_KEYFILE,
146 workers=1,
147 reload=DEBUG,
148 log_level=logging.DEBUG if DEBUG else logging.INFO,
149 )
150 server = Server(cfg)
151 await server.serve()