1import asyncio
2import atexit
3import logging
4import os
5from datetime import datetime as dt
6from datetime import timedelta as td
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
14from fastapi.responses import JSONResponse
15from starlette.staticfiles import StaticFiles
16from uvicorn import Server, Config
17
18import config
19from app.templates import render_template
20from config import DOCS, DEBUG, UVICORN_SSL_KEYFILE, UVICORN_SSL_CERTFILE, UVICORN_UDS, UVICORN_PORT, UVICORN_HOST
21from config import HOME_PAGE_TEMPLATE
22from . import __version__, telegram
23from .dashboard import build_dir, build, base_dir
24from .routes import api_router
25from .tasks import record_user_usages, review_users, reset_user_data_usage, \
26 record_realtime_bandwidth, send_notifications, delete_expired_reminders, nodes_startup
27
28logger = logging.getLogger(__name__)
29dashboard_path = "/dashboard/"
30
31app = FastAPI(
32 title="MarzneshinAPI",
33 description="Unified GUI Censorship Resistant Solution Powered by Xray",
34 version=__version__,
35 docs_url="/docs" if DOCS else None,
36 redoc_url="/redoc" if DOCS else None,
37)
38
39app.include_router(api_router)
40
41
42@app.get("/", response_class=HTMLResponse)
43def home_page():
44 return render_template(HOME_PAGE_TEMPLATE)
45
46
47app.add_middleware(
48 CORSMiddleware,
49 allow_origins=["*"],
50 allow_credentials=True,
51 allow_methods=["*"],
52 allow_headers=["*"],
53)
54
55scheduler = AsyncIOScheduler(timezone="UTC")
56scheduler.add_job(record_user_usages, 'interval', coalesce=True, seconds=30)
57scheduler.add_job(review_users, 'interval', seconds=600, coalesce=True, max_instances=1)
58scheduler.add_job(reset_user_data_usage, 'interval', coalesce=True, hours=1)
59scheduler.add_job(record_realtime_bandwidth, 'interval', seconds=1, coalesce=True, max_instances=1)
60
61if config.WEBHOOK_ADDRESS:
62 scheduler.add_job(send_notifications, "interval", seconds=30, replace_existing=True)
63 scheduler.add_job(delete_expired_reminders, "interval", hours=2, start_date=dt.utcnow() + td(minutes=1))
64
65
66@app.on_event("startup")
67async def on_start():
68 await nodes_startup()
69
70
71@app.on_event("shutdown")
72async def on_shutdown():
73 scheduler.shutdown()
74 logger.info("Sending pending notifications before shutdown...")
75 await send_notifications()
76
77
78@app.exception_handler(RequestValidationError)
79def validation_exception_handler(request: Request, exc: RequestValidationError):
80 details = {}
81 for error in exc.errors():
82 details[error["loc"][-1]] = error.get("msg")
83 return JSONResponse(
84 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
85 content=jsonable_encoder({"detail": details}),
86 )
87
88
89async def main():
90 if DEBUG:
91 proc = await asyncio.create_subprocess_exec(
92 'npm',
93 'run', 'dev', '--', '--host', '0.0.0.0', '--base',
94 dashboard_path, '--clearScreen', 'false',
95 env={**os.environ, 'VITE_BASE_API': config.VITE_BASE_API},
96 cwd=base_dir
97 )
98 atexit.register(proc.terminate)
99 else:
100 if not build_dir.is_dir():
101 build()
102
103 app.mount(
104 "/dashboard/",
105 StaticFiles(directory=build_dir, html=True),
106 name="dashboard"
107 )
108 scheduler.start()
109 asyncio.create_task(telegram.start_bot())
110 cfg = Config(
111 app=app,
112 host=UVICORN_HOST,
113 port=UVICORN_PORT,
114 uds=(None if DEBUG else UVICORN_UDS),
115 ssl_certfile=UVICORN_SSL_CERTFILE,
116 ssl_keyfile=UVICORN_SSL_KEYFILE,
117 workers=1,
118 reload=DEBUG,
119 log_level=logging.DEBUG if DEBUG else logging.INFO,
120 )
121 server = Server(cfg)
122 await server.serve()