sincos.cc-website/website/main.py

# -*- 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",
}


@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")