admin
678cbf98f8
- 首页：主力项目 Sequence-Diagram 大图展示 + 前沿研究三卡 + 工具矩阵（去彩色渐变块） - 新增工具页：串口调试助手（Android）与湿空气焓湿图（Android），含实机截图 - 下架 Engineer Craft / Five Phases / UniTouch（保留页面与路由） - 全站品牌 Easy Data Collection → Easy Design Collection（EDC 总品牌） - 资源中心更名「内部研究」，仅保留 footer 入口 - 导航下拉更新；产品线统计更新为 6
418 lines
12 KiB
Python
418 lines
12 KiB
Python
# -*- 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("/research/")
|
||
def research_page():
|
||
return FileResponse(STATIC_DIR / "research.html")
|
||
|
||
|
||
@app.get("/research/neural/")
|
||
def research_neural():
|
||
return FileResponse(STATIC_DIR / "neural.html")
|
||
|
||
|
||
@app.get("/research/reslib/")
|
||
def research_reslib():
|
||
return FileResponse(STATIC_DIR / "index.html")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 工具开发
|
||
# ---------------------------------------------------------------------------
|
||
|
||
TOOL_PAGES = {
|
||
"sequence-diagram": "tool-sequence-diagram.html",
|
||
"uni-viewer": "tool-uni-viewer.html",
|
||
"engineer-craft": "tool-engineer-craft.html",
|
||
"five-phases-observer": "tool-five-phases.html",
|
||
"unitouch": "tool-unitouch.html",
|
||
"edc-lite-erp": "tool-edc-lite-erp.html",
|
||
"physics-ai-fluent": "tool-physics-ai-fluent.html",
|
||
"serial-terminal": "tool-serial-terminal.html",
|
||
"enthalpy-diagram": "tool-enthalpy-diagram.html",
|
||
}
|
||
|
||
|
||
@app.get("/tools/")
|
||
def tools_page():
|
||
return FileResponse(STATIC_DIR / "tools.html")
|
||
|
||
|
||
@app.get("/tools/{slug}/")
|
||
def tool_page(slug: str):
|
||
filename = TOOL_PAGES.get(slug)
|
||
if not filename:
|
||
raise HTTPException(404, "页面不存在")
|
||
return FileResponse(STATIC_DIR / filename)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# License（申请 / 查询下载 / 资源包）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.get("/license/")
|
||
def license_page():
|
||
return FileResponse(STATIC_DIR / "license.html")
|
||
|
||
|
||
@app.get("/license/apply")
|
||
def license_apply():
|
||
return FileResponse(STATIC_DIR / "apply.html")
|
||
|
||
|
||
@app.get("/license/status")
|
||
def license_status():
|
||
return FileResponse(STATIC_DIR / "status.html")
|
||
|
||
|
||
@app.get("/license/package")
|
||
def license_package():
|
||
return FileResponse(STATIC_DIR / "package.html")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 资源中心
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.get("/resources/")
|
||
def resources_page():
|
||
return FileResponse(STATIC_DIR / "resources.html")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 兼容旧路径
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@app.get("/neural/")
|
||
def neural_page():
|
||
return FileResponse(STATIC_DIR / "neural.html")
|
||
|
||
|
||
@app.get("/links.html")
|
||
def links_page():
|
||
return FileResponse(STATIC_DIR / "resources.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")
|
||
app.mount("/apps", StaticFiles(directory=STATIC_DIR / "apps", html=True, check_dir=False), name="apps")
|