1import re
2
3from fastapi import APIRouter
4from fastapi import Header, HTTPException, Path, Request, Response
5from fastapi.responses import HTMLResponse
6
7from app.db import crud
8from app.dependencies import DBDep, SubUserDep, StartDateDep, EndDateDep
9from app.models.user import UserResponse
10from app.templates import render_template
11from app.utils.share import encode_title, generate_subscription, generate_v2ray_links
12from config import (
13 SUB_PROFILE_TITLE,
14 SUB_SUPPORT_URL,
15 SUB_UPDATE_INTERVAL,
16 SUBSCRIPTION_PAGE_TEMPLATE
17)
18
19router = APIRouter(prefix="/sub", tags=['Subscription'])
20
21
22def get_subscription_user_info(user: UserResponse) -> dict:
23 return {
24 "upload": 0,
25 "download": user.used_traffic,
26 "total": user.data_limit,
27 "expire": user.expire,
28 }
29
30
31@router.get("/{username}/{key}")
32def user_subscription(db_user: SubUserDep,
33 request: Request,
34 db: DBDep,
35 user_agent: str = Header(default="")):
36 """
37 Subscription link, V2ray and Clash supported
38 """
39 accept_header = request.headers.get("Accept", "")
40
41 user: UserResponse = UserResponse.model_validate(db_user)
42
43 if "text/html" in accept_header:
44 links = generate_v2ray_links(user.inbounds,
45 user.key,
46 user.model_dump(
47 exclude={"subscription_url", "services", "inbounds", "links"}
48 ))
49 return HTMLResponse(
50 render_template(
51 SUBSCRIPTION_PAGE_TEMPLATE,
52 {"user": user, "links": links}
53 )
54 )
55
56 response_headers = {
57 "content-disposition": f'attachment; filename="{user.username}"',
58 "profile-web-page-url": str(request.url),
59 "support-url": SUB_SUPPORT_URL,
60 "profile-title": encode_title(SUB_PROFILE_TITLE),
61 "profile-update-interval": SUB_UPDATE_INTERVAL,
62 "subscription-userinfo": "; ".join(
63 f"{key}={val}"
64 for key, val in get_subscription_user_info(user).items()
65 if val is not None
66 )
67 }
68
69 crud.update_user_sub(db, db_user, user_agent)
70
71 if re.match('^([Cc]lash-verge|[Cc]lash-?[Mm]eta)', user_agent):
72 conf = generate_subscription(user=user, config_format="clash-meta", as_base64=False)
73 return Response(content=conf, media_type="text/yaml", headers=response_headers)
74
75 elif re.match('^([Cc]lash|[Ss]tash)', user_agent):
76 conf = generate_subscription(user=user, config_format="clash", as_base64=False)
77 return Response(content=conf, media_type="text/yaml", headers=response_headers)
78
79 elif re.match('^(SFA|SFI|SFM|SFT)', user_agent):
80 conf = generate_subscription(user=user, config_format="sing-box", as_base64=False)
81 return Response(content=conf, media_type="application/json", headers=response_headers)
82
83 elif re.match('^(SS|SSR|SSD|SSS|Outline|Shadowsocks|SSconf)', user_agent):
84 conf = generate_subscription(user=user, config_format="outline", as_base64=False)
85 return Response(content=conf, media_type="application/json", headers=response_headers)
86
87 else:
88 conf = generate_subscription(user=user, config_format="v2ray", as_base64=True)
89 return Response(content=conf, media_type="text/plain", headers=response_headers)
90
91
92@router.get("/{username}/{key}/info", response_model=UserResponse)
93def user_subscription_info(db_user: SubUserDep):
94 return db_user
95
96
97@router.get("/{username}/{key}/usage")
98def user_get_usage(db_user: SubUserDep,
99 db: DBDep,
100 start_date: StartDateDep,
101 end_date: EndDateDep):
102 usages = crud.get_user_usages(db, db_user, start_date, end_date)
103
104 return {"usages": usages, "username": db_user.username}
105
106
107@router.get("/{username}/{key}/{client_type}")
108def user_subscription_with_client_type(
109 db_user: SubUserDep,
110 request: Request,
111 client_type: str = Path(..., regex="sing-box|clash-meta|clash|outline|v2ray")):
112 """
113 Subscription link, v2ray, clash, sing-box, outline and clash-meta supported
114 """
115
116 user: UserResponse = UserResponse.model_validate(db_user)
117
118 response_headers = {
119 "content-disposition": f'attachment; filename="{user.username}"',
120 "profile-web-page-url": str(request.url),
121 "support-url": SUB_SUPPORT_URL,
122 "profile-title": encode_title(SUB_PROFILE_TITLE),
123 "profile-update-interval": SUB_UPDATE_INTERVAL,
124 "subscription-userinfo": "; ".join(
125 f"{key}={val}"
126 for key, val in get_subscription_user_info(user).items()
127 if val is not None
128 )
129 }
130
131 if client_type == "clash-meta":
132 conf = generate_subscription(user=user, config_format="clash-meta", as_base64=False)
133 return Response(content=conf, media_type="text/yaml", headers=response_headers)
134
135 elif client_type == "sing-box":
136 conf = generate_subscription(user=user, config_format="sing-box", as_base64=False)
137 return Response(content=conf, media_type="application/json", headers=response_headers)
138
139 elif client_type == "clash":
140 conf = generate_subscription(user=user, config_format="clash", as_base64=False)
141 return Response(content=conf, media_type="text/yaml", headers=response_headers)
142
143 elif client_type == "v2ray":
144 conf = generate_subscription(user=user, config_format="v2ray", as_base64=True)
145 return Response(content=conf, media_type="text/plain", headers=response_headers)
146
147 elif client_type == "outline":
148 conf = generate_subscription(user=user, config_format="outline", as_base64=False)
149 return Response(content=conf, media_type="application/json", headers=response_headers)
150
151 else:
152 raise HTTPException(status_code=400, detail="Invalid subscription type")