"""Nancode Portal 后端 — SQLite 持久化链接 + SSO Cookie 认证""" import os, sqlite3 from contextlib import contextmanager from fastapi import FastAPI, Request, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from jose import jwt, JWTError # ── 配置 ───────────────────────────────── DB_PATH = os.path.join(os.path.dirname(__file__), "portal.db") SSO_SECRET_KEY = "your-secret-key-change-in-production" # 与 SSO 中心一致 ALGORITHM = "HS256" # ── 数据库 ─────────────────────────────── @contextmanager def get_db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row try: yield conn finally: conn.close() def init_db(): with get_db() as conn: conn.execute("""CREATE TABLE IF NOT EXISTS links ( id TEXT PRIMARY KEY, name TEXT NOT NULL, url TEXT NOT NULL, desc TEXT DEFAULT '', icon TEXT DEFAULT '', category TEXT DEFAULT '', sort_order INTEGER DEFAULT 0 )""") # 兼容迁移：旧库补充 category 列 cols = [r[1] for r in conn.execute("PRAGMA table_info(links)").fetchall()] if "category" not in cols: conn.execute("ALTER TABLE links ADD COLUMN category TEXT DEFAULT ''") conn.execute("""CREATE TABLE IF NOT EXISTS categories ( id TEXT PRIMARY KEY, name TEXT NOT NULL UNIQUE, sort_order INTEGER DEFAULT 0 )""") # 首次启动：如果表为空则写入默认数据 count = conn.execute("SELECT COUNT(*) FROM links").fetchone()[0] if count == 0: defaults = [ ('1','个人博客','http://129.211.173.87:8888','Next.js技术博客，个人技术文章与生活记录','book-open',1), ('2','相册查看器','http://129.211.173.87:8892','图片相册浏览与管理工具','image',2), ('3','快乐游戏','http://129.211.173.87:8895','Flask小游戏集合，休闲益智','gamepad-2',3), ('4','网游引擎','http://129.211.173.87:8900','网络游戏引擎服务平台','server',4), ('5','有声书播放器','http://129.211.173.87:8899','在线有声书播放与TTS服务','headphones',5), ('6','WanAndroid','http://129.211.173.87:8893','Material Design风格的Web应用','smartphone',6), ('7','FumaMX CRM','http://129.211.173.87:10000','客户关系管理系统','briefcase',7), ('8','AI聊天服务','http://129.211.173.87:8890','AI对话机器人服务','bot',8), ('9','新媒体平台','http://129.211.173.87:18888','CMS内容管理系统与网站后台','layout-dashboard',9), ('10','微信小程序桥接','http://129.211.173.87:5000','微信-Nanobot消息桥接服务','message-circle',10), ('11','音频录制工具','http://129.211.173.87:8901','在线音频录制与处理工具','mic',11), ('12','粒子实验室','http://129.211.173.87:8902','粒子特效与图形渲染实验','sparkles',12), ('13','WebGPU演示','http://129.211.173.87:8902','WebGPU与PlayCanvas 3D演示','globe',13), ('14','Hermes网关','http://129.211.173.87:8894','AI网关与代理服务','hexagon',14), ] conn.executemany("INSERT INTO links (id,name,url,desc,icon,sort_order) VALUES (?,?,?,?,?,?)", defaults) conn.commit() init_db() # ── SSO 认证 ───────────────────────────── def require_sso(request: Request) -> str: """从 Cookie 验签，返回 username；失败抛 401""" token = request.cookies.get("access_token") if not token: raise HTTPException(401, "未登录") try: payload = jwt.decode(token, SSO_SECRET_KEY, algorithms=[ALGORITHM]) username = payload.get("sub") if not username: raise HTTPException(401, "无效令牌") return username except JWTError: raise HTTPException(401, "令牌已过期或无效") # ── FastAPI ────────────────────────────── app = FastAPI(title="Nancode Portal API") app.add_middleware( CORSMiddleware, allow_origins=[ "http://129.211.173.87", "http://129.211.173.87:8887", "http://129.211.173.87:8888", "http://129.211.173.87:9000", ], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class LinkCreate(BaseModel): name: str url: str desc: str = "" icon: str = "" category: str = "" class CategoryCreate(BaseModel): name: str @app.get("/api/links") def list_links(): with get_db() as conn: rows = conn.execute("SELECT id,name,url,desc,icon,category FROM links ORDER BY sort_order, id").fetchall() return [dict(r) for r in rows] @app.get("/api/categories") def list_categories(): with get_db() as conn: rows = conn.execute("SELECT id,name FROM categories ORDER BY sort_order, id").fetchall() return [dict(r) for r in rows] @app.post("/api/categories") def add_category(cat: CategoryCreate, username: str = Depends(require_sso)): import time, random name = cat.name.strip() if not name: raise HTTPException(400, "分类名称不能为空") cid = f"c{int(time.time()):x}{random.randint(0,0xffffff):06x}" with get_db() as conn: try: max_sort = conn.execute("SELECT COALESCE(MAX(sort_order),0) FROM categories").fetchone()[0] conn.execute("INSERT INTO categories (id,name,sort_order) VALUES (?,?,?)", (cid, name, max_sort + 1)) conn.commit() except sqlite3.IntegrityError: raise HTTPException(400, "分类已存在") return {"id": cid, "name": name} @app.post("/api/links") def add_link(link: LinkCreate, username: str = Depends(require_sso)): import time, random lid = f"{int(time.time()):x}{random.randint(0,0xffffff):06x}" with get_db() as conn: max_sort = conn.execute("SELECT COALESCE(MAX(sort_order),0) FROM links").fetchone()[0] conn.execute("INSERT INTO links (id,name,url,desc,icon,category,sort_order) VALUES (?,?,?,?,?,?,?)", (lid, link.name, link.url, link.desc, link.icon, link.category, max_sort + 1)) conn.commit() return {"id": lid, "name": link.name, "url": link.url, "desc": link.desc, "icon": link.icon, "category": link.category} @app.delete("/api/links/{link_id}") def delete_link(link_id: str, username: str = Depends(require_sso)): with get_db() as conn: cur = conn.execute("DELETE FROM links WHERE id=?", (link_id,)) conn.commit() if cur.rowcount == 0: raise HTTPException(404, "链接不存在") return {"success": True} @app.put("/api/links/{link_id}") def update_link(link_id: str, link: LinkCreate, username: str = Depends(require_sso)): with get_db() as conn: row = conn.execute("SELECT * FROM links WHERE id=?", (link_id,)).fetchone() if not row: raise HTTPException(404, "链接不存在") # 图标不传则保留原值，避免改名称时图标被重新生成 icon = link.icon if link.icon else row["icon"] # 分类始终按提交值更新（空字符串 = 未归类） conn.execute("UPDATE links SET name=?, url=?, desc=?, icon=?, category=? WHERE id=?", (link.name, link.url, link.desc, icon, link.category, link_id)) conn.commit() return {"id": link_id, "name": link.name, "url": link.url, "desc": link.desc, "icon": icon, "category": link.category} if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8887)