first commit
This commit is contained in:
+320
@@ -0,0 +1,320 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
ResLib License 申请系统后端 (FastAPI)
|
||||
|
||||
存储：data/requests/<id>.txt（DokuWiki 式纯文本）
|
||||
公开：申请 / 状态查询 / 下载 license
|
||||
管理：密码登录 -> 审批 -> 生成 res_center.lic
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import secrets
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import FastAPI, Form, HTTPException, Request, Response
|
||||
from fastapi.responses import FileResponse, PlainTextResponse, RedirectResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from license import generate_lic, normalize_mac
|
||||
|
||||
BASE_DIR = Path(__file__).resolve().parent
|
||||
DATA_DIR = BASE_DIR / "data"
|
||||
REQ_DIR = DATA_DIR / "requests"
|
||||
LIC_DIR = DATA_DIR / "licenses"
|
||||
PKG_DIR = DATA_DIR / "packages"
|
||||
STATIC_DIR = BASE_DIR / "static"
|
||||
|
||||
|
||||
def load_admin_password() -> str:
|
||||
env = os.environ.get("RESLIB_ADMIN_PASSWORD", "").strip()
|
||||
if env:
|
||||
return env
|
||||
env_file = BASE_DIR / ".env"
|
||||
if env_file.is_file():
|
||||
for line in env_file.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
k, _, v = line.partition("=")
|
||||
if k.strip() == "RESLIB_ADMIN_PASSWORD":
|
||||
return v.strip()
|
||||
return "reslib-admin"
|
||||
|
||||
|
||||
ADMIN_PASSWORD = load_admin_password()
|
||||
SESSION_FILE = DATA_DIR / "session.txt"
|
||||
|
||||
STATUS_PENDING = "pending"
|
||||
STATUS_APPROVED = "approved"
|
||||
STATUS_DENIED = "denied"
|
||||
|
||||
for d in (REQ_DIR, LIC_DIR, PKG_DIR):
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
app = FastAPI(title="ResLib License System", docs_url=None, redoc_url=None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 工具
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def new_request_id() -> str:
|
||||
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
return f"{stamp}-{secrets.token_hex(3).upper()}"
|
||||
|
||||
|
||||
def request_path(req_id: str) -> Path:
|
||||
if not re.fullmatch(r"\d{8}-\d{6}-[0-9A-F]{6}", req_id):
|
||||
raise HTTPException(400, "无效的申请编号")
|
||||
return REQ_DIR / f"{req_id}.txt"
|
||||
|
||||
|
||||
def read_request(req_id: str) -> dict:
|
||||
path = request_path(req_id)
|
||||
if not path.is_file():
|
||||
raise HTTPException(404, "申请不存在")
|
||||
data = {}
|
||||
for line in path.read_text(encoding="utf-8").splitlines():
|
||||
if ":" in line:
|
||||
k, _, v = line.partition(":")
|
||||
data[k.strip()] = v.strip()
|
||||
return data
|
||||
|
||||
|
||||
def write_request(data: dict):
|
||||
path = request_path(data["id"])
|
||||
content = "\n".join(f"{k}: {data.get(k, '')}" for k in (
|
||||
"id", "created", "status", "mac", "company", "contact",
|
||||
"email", "creo", "seats", "notes", "reason",
|
||||
))
|
||||
path.write_text(content + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def save_lic(req_id: str, mac: str) -> Path:
|
||||
lic_path = LIC_DIR / f"{req_id}.lic"
|
||||
lic_path.write_text(generate_lic(mac) + "\n", encoding="ascii")
|
||||
return lic_path
|
||||
|
||||
|
||||
def new_session_token() -> str:
|
||||
token = secrets.token_hex(24)
|
||||
SESSION_FILE.write_text(token, encoding="utf-8")
|
||||
return token
|
||||
|
||||
|
||||
def get_session_token() -> str:
|
||||
if SESSION_FILE.is_file():
|
||||
return SESSION_FILE.read_text(encoding="utf-8").strip()
|
||||
return ""
|
||||
|
||||
|
||||
def is_admin(request: Request) -> bool:
|
||||
cookie = request.cookies.get("reslib_admin")
|
||||
return bool(cookie) and secrets.compare_digest(cookie, get_session_token())
|
||||
|
||||
|
||||
def require_admin(request: Request):
|
||||
if not is_admin(request):
|
||||
raise HTTPException(401, "未登录或会话已失效")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 公开 API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class ApplyForm(BaseModel):
|
||||
mac: str = Field(..., min_length=1)
|
||||
company: str = Field(..., min_length=1)
|
||||
contact: str = Field(..., min_length=1)
|
||||
email: str = Field(..., min_length=1)
|
||||
creo: str = ""
|
||||
seats: str = "1"
|
||||
notes: str = ""
|
||||
|
||||
|
||||
@app.post("/api/apply")
|
||||
def api_apply(form: ApplyForm):
|
||||
try:
|
||||
mac = normalize_mac(form.mac)
|
||||
except ValueError as e:
|
||||
raise HTTPException(400, str(e))
|
||||
if "@" not in form.email or "." not in form.email.split("@")[-1]:
|
||||
raise HTTPException(400, "邮箱格式不正确")
|
||||
|
||||
req_id = new_request_id()
|
||||
data = {
|
||||
"id": req_id,
|
||||
"created": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"status": STATUS_PENDING,
|
||||
"mac": mac,
|
||||
"company": form.company.strip(),
|
||||
"contact": form.contact.strip(),
|
||||
"email": form.email.strip(),
|
||||
"creo": form.creo.strip(),
|
||||
"seats": form.seats.strip(),
|
||||
"notes": form.notes.strip(),
|
||||
"reason": "",
|
||||
}
|
||||
write_request(data)
|
||||
return {"id": req_id, "status": STATUS_PENDING}
|
||||
|
||||
|
||||
@app.get("/api/status/{req_id}")
|
||||
def api_status(req_id: str):
|
||||
data = read_request(req_id)
|
||||
return {
|
||||
"id": data.get("id"),
|
||||
"status": data.get("status"),
|
||||
"created": data.get("created"),
|
||||
"reason": data.get("reason", ""),
|
||||
}
|
||||
|
||||
|
||||
@app.get("/api/license/{req_id}")
|
||||
def api_license(req_id: str):
|
||||
data = read_request(req_id)
|
||||
if data.get("status") != STATUS_APPROVED:
|
||||
raise HTTPException(403, "申请尚未通过审批")
|
||||
lic_path = LIC_DIR / f"{req_id}.lic"
|
||||
if not lic_path.is_file():
|
||||
save_lic(req_id, data["mac"])
|
||||
lic_path = LIC_DIR / f"{req_id}.lic"
|
||||
return FileResponse(
|
||||
lic_path, filename="res_center.lic",
|
||||
media_type="application/octet-stream",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 资源包（仅审核通过的用户可获取）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@app.get("/api/package/{req_id}")
|
||||
def api_package_list(req_id: str):
|
||||
data = read_request(req_id)
|
||||
if data.get("status") != STATUS_APPROVED:
|
||||
raise HTTPException(403, "资源包仅对审核通过的用户提供，请先完成 License 审批")
|
||||
files = sorted(p.name for p in PKG_DIR.glob("*") if p.is_file())
|
||||
if not files:
|
||||
return {"files": []}
|
||||
return {"files": files}
|
||||
|
||||
|
||||
@app.get("/api/package/download/{req_id}/{filename}")
|
||||
def api_package_download(req_id: str, filename: str):
|
||||
data = read_request(req_id)
|
||||
if data.get("status") != STATUS_APPROVED:
|
||||
raise HTTPException(403, "资源包仅对审核通过的用户提供")
|
||||
name = os.path.basename(filename)
|
||||
path = PKG_DIR / name
|
||||
if not path.is_file():
|
||||
raise HTTPException(404, "资源包不存在")
|
||||
return FileResponse(path, filename=name, media_type="application/octet-stream")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 管理 API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@app.post("/api/admin/login")
|
||||
def admin_login(response: Response, password: str = Form(...)):
|
||||
if not secrets.compare_digest(password, ADMIN_PASSWORD):
|
||||
raise HTTPException(401, "密码错误")
|
||||
token = new_session_token()
|
||||
response.set_cookie("reslib_admin", token, httponly=True, samesite="lax", max_age=86400 * 7)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@app.post("/api/admin/logout")
|
||||
def admin_logout(request: Request, response: Response):
|
||||
response.delete_cookie("reslib_admin")
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@app.get("/api/admin/list")
|
||||
def admin_list(request: Request):
|
||||
require_admin(request)
|
||||
items = []
|
||||
for path in sorted(REQ_DIR.glob("*.txt"), reverse=True):
|
||||
items.append(read_request(path.stem))
|
||||
return items
|
||||
|
||||
|
||||
@app.post("/api/admin/approve/{req_id}")
|
||||
def admin_approve(req_id: str, request: Request):
|
||||
require_admin(request)
|
||||
data = read_request(req_id)
|
||||
if data.get("status") == STATUS_APPROVED:
|
||||
return {"ok": True, "id": req_id}
|
||||
save_lic(req_id, data["mac"])
|
||||
data["status"] = STATUS_APPROVED
|
||||
data["reason"] = ""
|
||||
write_request(data)
|
||||
return {"ok": True, "id": req_id}
|
||||
|
||||
|
||||
@app.post("/api/admin/deny/{req_id}")
|
||||
def admin_deny(req_id: str, request: Request, reason: str = Form("")):
|
||||
require_admin(request)
|
||||
data = read_request(req_id)
|
||||
data["status"] = STATUS_DENIED
|
||||
data["reason"] = reason.strip()[:200]
|
||||
write_request(data)
|
||||
return {"ok": True, "id": req_id}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 页面与静态资源
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@app.get("/")
|
||||
def index():
|
||||
return FileResponse(STATIC_DIR / "home.html")
|
||||
|
||||
|
||||
@app.get("/about/")
|
||||
def about_page():
|
||||
return FileResponse(STATIC_DIR / "about.html")
|
||||
|
||||
|
||||
@app.get("/reslib/")
|
||||
def reslib_home():
|
||||
return FileResponse(STATIC_DIR / "index.html")
|
||||
|
||||
|
||||
@app.get("/reslib/apply")
|
||||
def reslib_apply():
|
||||
return FileResponse(STATIC_DIR / "apply.html")
|
||||
|
||||
|
||||
@app.get("/reslib/status")
|
||||
def reslib_status():
|
||||
return FileResponse(STATIC_DIR / "status.html")
|
||||
|
||||
|
||||
@app.get("/reslib/package")
|
||||
def reslib_package():
|
||||
return FileResponse(STATIC_DIR / "package.html")
|
||||
|
||||
|
||||
@app.get("/admin")
|
||||
def admin_page(request: Request):
|
||||
if not is_admin(request):
|
||||
return RedirectResponse("/admin/login", status_code=302)
|
||||
return FileResponse(STATIC_DIR / "admin.html")
|
||||
|
||||
|
||||
@app.get("/admin/login")
|
||||
def admin_login_page():
|
||||
return FileResponse(STATIC_DIR / "admin_login.html")
|
||||
|
||||
|
||||
@app.get("/admin/logout")
|
||||
def admin_logout_page(request: Request):
|
||||
return RedirectResponse("/admin/login", status_code=302)
|
||||
|
||||
|
||||
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|
||||
app.mount("/blog", StaticFiles(directory=STATIC_DIR / "blog", html=True, check_dir=False), name="blog")
|
||||
Reference in New Issue
Block a user