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
 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(
 33    db_user: SubUserDep,
 34    request: Request,
 35    db: DBDep,
 36    user_agent: str = Header(default=""),
 37):
 38    """
 39    Subscription link, V2ray and Clash supported
 40    """
 41    accept_header = request.headers.get("Accept", "")
 42
 43    user: UserResponse = UserResponse.model_validate(db_user)
 44
 45    if "text/html" in accept_header:
 46        links = generate_subscription(user=user, config_format="v2ray").split()
 47        return HTMLResponse(
 48            render_template(SUBSCRIPTION_PAGE_TEMPLATE, {"user": user, "links": links})
 49        )
 50
 51    response_headers = {
 52        "content-disposition": f'attachment; filename="{user.username}"',
 53        "profile-web-page-url": str(request.url),
 54        "support-url": SUB_SUPPORT_URL,
 55        "profile-title": encode_title(SUB_PROFILE_TITLE),
 56        "profile-update-interval": SUB_UPDATE_INTERVAL,
 57        "subscription-userinfo": "; ".join(
 58            f"{key}={val}"
 59            for key, val in get_subscription_user_info(user).items()
 60            if val is not None
 61        ),
 62    }
 63
 64    crud.update_user_sub(db, db_user, user_agent)
 65
 66    if re.match("^([Cc]lash-verge|[Cc]lash-?[Mm]eta)", user_agent):
 67        conf = generate_subscription(user=user, config_format="clash-meta")
 68        return Response(content=conf, media_type="text/yaml", headers=response_headers)
 69    elif re.match("^([Cc]lash|[Ss]tash)", user_agent):
 70        conf = generate_subscription(user=user, config_format="clash")
 71        return Response(content=conf, media_type="text/yaml", headers=response_headers)
 72    elif re.match("^(SFA|SFI|SFM|SFT)", user_agent):
 73        conf = generate_subscription(user=user, config_format="sing-box")
 74        return Response(
 75            content=conf, media_type="application/json", headers=response_headers
 76        )
 77    else:
 78        conf = generate_subscription(user=user, config_format="v2ray", as_base64=True)
 79        return Response(content=conf, media_type="text/plain", headers=response_headers)
 80
 81
 82@router.get("/{username}/{key}/info", response_model=UserResponse)
 83def user_subscription_info(db_user: SubUserDep):
 84    return db_user
 85
 86
 87@router.get("/{username}/{key}/usage")
 88def user_get_usage(
 89    db_user: SubUserDep, db: DBDep, start_date: StartDateDep, end_date: EndDateDep
 90):
 91    usages = crud.get_user_usages(db, db_user, start_date, end_date)
 92
 93    return {"usages": usages, "username": db_user.username}
 94
 95
 96@router.get("/{username}/{key}/{client_type}")
 97def user_subscription_with_client_type(
 98    db_user: SubUserDep,
 99    request: Request,
100    client_type: str = Path(regex="sing-box|clash-meta|clash|xray|v2ray"),
101):
102    """
103    Subscription by client type; v2ray, xray, sing-box, clash and clash-meta formats supported
104    """
105
106    user: UserResponse = UserResponse.model_validate(db_user)
107
108    response_headers = {
109        "content-disposition": f'attachment; filename="{user.username}"',
110        "profile-web-page-url": str(request.url),
111        "support-url": SUB_SUPPORT_URL,
112        "profile-title": encode_title(SUB_PROFILE_TITLE),
113        "profile-update-interval": SUB_UPDATE_INTERVAL,
114        "subscription-userinfo": "; ".join(
115            f"{key}={val}"
116            for key, val in get_subscription_user_info(user).items()
117            if val is not None
118        ),
119    }
120
121    if client_type == "clash-meta":
122        conf = generate_subscription(user=user, config_format="clash-meta")
123        return Response(content=conf, media_type="text/yaml", headers=response_headers)
124
125    elif client_type == "sing-box":
126        conf = generate_subscription(user=user, config_format="sing-box")
127        return Response(
128            content=conf, media_type="application/json", headers=response_headers
129        )
130    elif client_type == "clash":
131        conf = generate_subscription(user=user, config_format="clash")
132        return Response(content=conf, media_type="text/yaml", headers=response_headers)
133
134    elif client_type == "v2ray":
135        conf = generate_subscription(user=user, config_format="v2ray", as_base64=True)
136        return Response(content=conf, media_type="text/plain", headers=response_headers)
137    elif client_type == "xray":
138        return Response(
139            content=generate_subscription(user=user, config_format="xray"),
140            headers=response_headers,
141        )
142    else:
143        raise HTTPException(status_code=400, detail="Invalid subscription type")