Files
whoami/xinquan/app.py
T

617 lines
25 KiB
Python
Raw Blame History

This file contains invisible Unicode characters
This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""xinquandashboard —— 密码保护的图文页面应用(零依赖,标准库 only)。
路由(Caddy 通过 handle_path 剥掉 /xinquandashboard 前缀后转发至此):
GET / 单页 UI(HTML+JS 内联)
GET /api/pages 页面列表
POST /api/pages 新建页面 body: {"title","content"}
GET /api/pages/<id> 单页详情
PUT /api/pages/<id> 编辑页面 body: {"title","content"}
DELETE /api/pages/<id> 删除页面
POST /api/upload 上传图片 body: {"data":"<base64>","name":"a.png","mime":"image/png"}
GET /media/<file> 图片静态读取
持久化:/data/pages.json(页面元数据)+ /data/uploads/(图片文件)。
鉴权:由 Caddy basic_auth 在网关卡负责,本应用不鉴权。
"""
import base64
import hashlib
import hmac
import json
import os
import re
import threading
import time
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
DATA_DIR = os.environ.get("DATA_DIR", "/data")
PAGES_FILE = os.path.join(DATA_DIR, "pages.json")
UPLOADS_DIR = os.path.join(DATA_DIR, "uploads")
PORT = int(os.environ.get("PORT", "8000"))
_lock = threading.Lock()
ALLOWED_IMG_EXT = {".png": "image/png", ".jpg": "image/jpeg",
".jpeg": "image/jpeg", ".gif": "image/gif",
".webp": "image/webp", ".svg": "image/svg+xml"}
# ---- 应用内表单登录(替代 Caddy basic_auth,兼容微信内置浏览器)----
AUTH_USER = os.environ.get("XQ_USER", "xqzk")
AUTH_PASS = os.environ.get("XQ_PASS", "66668888")
AUTH_SECRET = os.environ.get("XQ_SECRET", "xinquan-session-secret-change-me")
SESSION_DAYS = 7
RE_AUTH = re.compile(r"__AUTH__")
def _sign_token(user, exp):
payload = f"{user}|{exp}".encode()
sig = hmac.new(AUTH_SECRET.encode(), payload, hashlib.sha256).hexdigest()
return base64.urlsafe_b64encode(payload + b"|" + sig.encode()).decode()
def _verify_token(token):
try:
raw = base64.urlsafe_b64decode(token.encode()).decode()
user, exp, sig = raw.split("|")
if int(exp) < int(time.time()):
return None
expect = hmac.new(AUTH_SECRET.encode(), f"{user}|{exp}".encode(),
hashlib.sha256).hexdigest()
return user if hmac.compare_digest(expect, sig) else None
except Exception:
return None
def _ensure_dirs():
os.makedirs(DATA_DIR, exist_ok=True)
os.makedirs(UPLOADS_DIR, exist_ok=True)
if not os.path.exists(PAGES_FILE):
_write_pages({})
def _read_pages():
if not os.path.exists(PAGES_FILE):
return {}
try:
with open(PAGES_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _write_pages(pages):
tmp = PAGES_FILE + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(pages, f, ensure_ascii=False, indent=2)
os.replace(tmp, PAGES_FILE)
def _summary(content, n=160):
text = re.sub(r"[\n\r]+", " ", content or "")
text = re.sub(r"[\s ]+", " ", text).strip()
return text[:n] + ("…" if len(text) > n else "")
# ------------------------- 页面 与 UI -------------------------
BOOTSTRAP = "https://cdn.jsdelivr.net/npm/marked@12/marked.min.js"
INDEX_HTML = r"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Xinquan Dashboard</title>
<script src="__MARKED__"></script>
<style>
:root{
--bg:#f4f6fb; --card:#fff; --border:#e2e6ef; --text:#1c2233;
--muted:#64718c; --brand:#6366f1; --brand2:#0ea5e9;
--grad:linear-gradient(120deg,#6366f1,#0ea5e9 65%,#22d3ee);
}
@media(prefers-color-scheme:dark){
:root{--bg:#0a0f1e; --card:#111829; --border:#232c42; --text:#eef2ff; --muted:#8b97b3;}
}
*{box-sizing:border-box;margin:0;padding:0}
body{font-family:-apple-system,BlinkMacSystemFont,"Segoe UI","PingFang SC","Microsoft YaHei",sans-serif;
background:var(--bg);color:var(--text);min-height:100vh}
.top{position:sticky;top:0;z-index:20;display:flex;align-items:center;gap:12px;
padding:12px 20px;background:color-mix(in srgb,var(--bg) 82%,transparent);
backdrop-filter:blur(12px);border-bottom:1px solid var(--border)}
.top .logo{width:34px;height:34px;border-radius:10px;background:var(--grad);display:grid;place-items:center;
color:#fff;font-weight:800;font-size:18px}
.top h1{font-size:18px;font-weight:700}
.top .actions{margin-left:auto;display:flex;gap:8px}
.wrap{max-width:860px;margin:0 auto;padding:22px 18px 60px}
button,.btn{border:none;cursor:pointer;padding:8px 14px;border-radius:10px;font-size:14px;
background:var(--card);border:1px solid var(--border);color:var(--text);transition:.16s}
button.primary,.btn.primary{background:var(--grad);color:#fff;font-weight:600;border-color:transparent}
button:hover,.btn:hover{opacity:.9}
.grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(240px,1fr));gap:14px}
.tile{background:var(--card);border:1px solid var(--border);border-radius:16px;padding:16px;
text-decoration:none;color:var(--text);display:block;transition:.16s;box-shadow:0 2px 10px rgba(20,30,70,.05)}
.tile:hover{transform:translateY(-2px);box-shadow:0 10px 24px rgba(20,30,70,.1)}
.tile h3{font-size:16px;margin-bottom:6px}
.tile p{font-size:13px;color:var(--muted);line-height:1.5;
display:-webkit-box;-webkit-line-clamp:2;-webkit-box-orient:vertical;overflow:hidden}
.tile time{display:block;margin-top:10px;font-size:12px;color:var(--muted)}
.empty{text-align:center;color:var(--muted);padding:60px 0;line-height:2}
.login{max-width:380px;margin:10vh auto 0;background:var(--card);border:1px solid var(--border);
border-radius:20px;padding:32px 30px;box-shadow:0 10px 30px rgba(20,30,70,.08)}
.login .lg{width:52px;height:52px;border-radius:14px;background:var(--grad);display:grid;place-items:center;
color:#fff;font-weight:800;font-size:24px;margin:0 auto 16px}
.login h2{text-align:center;font-size:20px;margin-bottom:22px}
.login input{width:100%;padding:11px 12px;margin-bottom:12px;border-radius:10px;
border:1px solid var(--border);background:var(--bg);color:var(--text);font-size:15px;outline:none}
.login input:focus{border-color:var(--brand)}
.login button{width:100%;padding:11px;border:none;border-radius:10px;background:var(--grad);
color:#fff;font-size:15px;font-weight:600;cursor:pointer}
.login .err{color:#ef4444;font-size:13px;text-align:center;height:18px;margin-bottom:8px}
.page{padding-top:8px}
.page .title{font-size:26px;font-weight:800;letter-spacing:-.02em;margin-bottom:4px}
.page .meta{color:var(--muted);font-size:13px;margin-bottom:18px}
.article{background:var(--card);border:1px solid var(--border);border-radius:16px;padding:22px 26px;line-height:1.75}
.article img{max-width:100%;border-radius:10px;margin:8px 0}
.article pre{background:color-mix(in srgb,var(--text) 6%,transparent);padding:14px;border-radius:10px;overflow:auto}
.article code{background:color-mix(in srgb,var(--text) 8%,transparent);padding:2px 5px;border-radius:6px;font-size:14px}
.article pre code{background:none;padding:0}
.article blockquote{border-left:3px solid var(--brand);color:var(--muted);padding-left:14px;margin:10px 0}
.article table{border-collapse:collapse;margin:10px 0}
.article th,.article td{border:1px solid var(--border);padding:6px 10px}
.editor .field{margin-bottom:16px}
.editor label{display:block;font-weight:600;margin-bottom:6px}
.editor input[type=text]{width:100%;padding:10px 12px;border-radius:10px;border:1px solid var(--border);
background:var(--card);color:var(--text);font-size:15px}
.editor textarea{width:100%;min-height:60vh;padding:12px;border-radius:10px;border:1px solid var(--border);
background:var(--card);color:var(--text);font-size:14px;font-family:ui-monospace,Consolas,monospace;line-height:1.6}
.drop{display:flex;flex-wrap:wrap;gap:8px;align-items:center;margin-bottom:8px}
.bar-row{display:flex;gap:8px;flex-wrap:wrap}
.toast{position:fixed;left:50%;bottom:24px;transform:translateX(-50%);background:var(--text);
color:var(--bg);padding:10px 18px;border-radius:999px;font-size:14px;opacity:0;pointer-events:none;transition:.25s;z-index:50}
.toast.show{opacity:1}
.muted{color:var(--muted)}
</style>
</head>
<body>
<div class="top">
<div class="logo">XQ</div>
<h1>Xinquan Dashboard</h1>
<div class="actions" id="topActions"></div>
</div>
<div class="wrap" id="app"></div>
<div class="toast" id="toast"></div>
<script>
const BASE = (()=>{const p=location.pathname;const i=p.indexOf('/xinquandashboard');
return i>=0?p.slice(0,i+'/xinquandashboard'.length):'';})();
const $ = s => document.querySelector(s);
const AUTH = __AUTH__;
function api(path, opts){
return fetch(BASE+path, opts).then(async r=>{
if(!r.ok){let e='HTTP '+r.status;try{e=(await r.json()).error||e}catch(_){};throw new Error(e)}
const ct=r.headers.get('content-type')||'';
return ct.includes('json')?r.json():r.text();
});
}
function esc(s){const d=document.createElement('div');d.textContent=s;return d.innerHTML;}
function md(s){ try{return marked.parse(s||'');}catch(e){return esc(s);} }
const state={pages:[], mode:'list', current:null, content:''};
const umarked = typeof marked==='undefined';
function toast(msg){const t=$('#toast');t.textContent=msg;t.classList.add('show');clearTimeout(t._h);t._h=setTimeout(()=>t.classList.remove('show'),1800);}
function fmt(ts){ if(!ts)return ''; const s=String(ts);
const iso=(/Z$/i.test(s)||/([+-]\d{2}:\d{2})$/.test(s))?s:s+'Z';
try{return new Date(iso).toLocaleString('zh-CN');}catch(e){return ts;} }
async function loadPages(){
state.pages = await api('/api/pages');
state.mode='list';
renderList();
}
function renderList(){
$('#topActions').innerHTML = `<button class="primary" onclick="newPage()">+ 新建页面</button>
<button onclick="logout()">👤 退出</button>`;
const a=$('#app');
if(!state.pages.length){ a.innerHTML=`<div class="empty">还没有内容。<br><button class="primary" onclick="newPage()">创建第一个页面</button></div>`; return; }
a.innerHTML = `<div class="grid">${state.pages.map(p=>`
<a class="tile" onclick="openPage('${p.id}')">
<h3>${esc(p.title||'未命名')}</h3>
<p>${esc(p.summary||'')}</p>
<time>${fmt(p.updated)}</time>
</a>`).join('')}</div>`;
}
async function openPage(id){
try{
const p = await api('/api/pages/'+encodeURIComponent(id));
$('#topActions').innerHTML=`<button onclick="loadPages()">← 返回</button>
<button onclick="editPage('${p.id}')">✏️ 编辑</button>
<button onclick="delPage('${p.id}')">🗑</button>`;
$('#app').innerHTML=`<div class="page">
<h1 class="title">${esc(p.title)}</h1>
<div class="meta">更新于 ${fmt(p.updated)}</div>
<div class="article">${md(p.content)}</div></div>`;
state.mode='view';
}catch(e){ toast('打开失败: '+e.message); }
}
let editingId=null;
function newPage(){ editingId=null; beginEdit(null); }
function editPage(id){ editingId=id; beginEdit(id); }
function beginEdit(id){
(id? api('/api/pages/'+encodeURIComponent(id)):Promise.resolve({title:'',content:''})).then(p=>{
$('#topActions').innerHTML=`<button onclick="cancelEdit()">← 取消</button>
<button class="primary" onclick="savePage()">💾 保存</button>`;
$('#app').innerHTML=`<div class="editor">
<div class="field"><label>标题</label><input type="text" id="eTitle" value="${esc(p.title||'')}"></div>
<div class="field">
<label>正文(Markdown,支持图片)</label>
<div class="drop">
<button onclick="$('#file').click()">🖼 选择图片</button>
<span class="muted" style="font-size:13px">或把图片拖进来(自动插入引用)</span>
<input type="file" id="file" accept="image/*" multiple hidden>
</div>
<textarea id="eContent" ondrop="dropImg(event)" ondragover="event.preventDefault()">${esc(p.content||'')}</textarea>
</div>
</div>`;
state.mode='edit';
state.content=p.content||'';
$('#file').onchange=e=>{ for(const f of e.target.files) upImg(f); };
}).catch(e=>toast('e: '+e.message));
}
async function upImg(file){
if(!ALLOWED(file)) { toast('仅支持图片'); return; }
const b64 = await readB64(file);
try{
const r = await api('/api/upload',{method:'POST',headers:{'content-type':'application/json'},
body:JSON.stringify({data:b64,name:file.name,mime:file.type})});
const ta=$('#eContent'); if(!ta){ return; }
ta.value += (ta.value&&!ta.value.endsWith('\n')?'\n\n':'\n') + `\n![${esc(file.name.replace(/[\[\]]/g,'_'))}](${r.url})\n`;
ta.scrollTop=ta.scrollHeight;
}catch(e){ toast('上传失败: '+e.message); }
}
function ALLOWED(f){ return /^image\/(png|jpe?g|gif|webp|svg\+xml)$/.test(f.type); }
function readB64(file){ return new Promise((res,rej)=>{const r=new FileReader();r.onload=()=>{
const str=r.result;const m=String(str).match(/^data:[^;]+;base64,(.*)$/s);res(m?m[1]:str);};r.onerror=rej;r.readAsDataURL(file);});}
function dropImg(ev){
ev.preventDefault();
for(const f of ev.dataTransfer.files) if(ALLOWED(f)) upImg(f);
}
async function savePage(){
const title=($('#eTitle').value||'').trim(), content=$('#eContent').value;
if(!title){ toast('请填写标题'); return; }
try{
const body={title,content};
if(editingId) await api('/api/pages/'+encodeURIComponent(editingId),{method:'PUT',headers:{'content-type':'application/json'},body:JSON.stringify(body)});
else { const r=await api('/api/pages',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify(body)}); editingId=r.id; }
toast('已保存'); await loadPages();
}catch(e){ toast('保存失败: '+e.message); }
}
async function delPage(id){
if(!confirm('确定删除该页面?')) return;
try{ await api('/api/pages/'+encodeURIComponent(id),{method:'DELETE'}); toast('已删除'); loadPages(); }
catch(e){ toast('删除失败: '+e.message); }
}
function cancelEdit(){ loadPages(); }
async function login(u,p){
try{
const r = await fetch(BASE+'/login',{method:'POST',headers:{'content-type':'application/json'},body:JSON.stringify({username:u,password:p})});
if(!r.ok){ const j=await r.json().catch(()=>({})); return j.error||'登录失败'; }
return null;
}catch(e){ return '网络错误'; }
}
function renderLogin(){
$('#topActions').innerHTML='';
$('#app').innerHTML=`<div class="login">
<div class="lg">XQ</div>
<h2>Xinquan Dashboard</h2>
<div class="err" id="err"></div>
<input id="u" type="text" placeholder="用户名" autocomplete="username">
<input id="p" type="password" placeholder="密码" autocomplete="current-password">
<button id="go">登 录</button></div>`;
const doIt=async ()=>{
const err=$('#err'); err.textContent='';
const msg=await login($('#u').value.trim(), $('#p').value);
if(msg){ err.textContent=msg; return; }
$('#topActions').innerHTML='<button onclick="logout()">👤 退出登录</button>';
loadPages();
};
$('#go').onclick=doIt;
$('#p').onkeydown=e=>{ if(e.key==='Enter') doIt(); };
$('#u').focus();
}
async function logout(){
await fetch(BASE+'/logout',{method:'POST'});
location.reload();
}
if(!AUTH){ renderLogin(); }
else { onAuth(); }
function onAuth(){
$('#topActions').innerHTML='<button onclick="logout()">👤 退出登录</button>';
loadPages();
}
if(umarked){ document.addEventListener('DOMContentLoaded',()=>toast('Markdown 渲染库加载失败,回退为纯文本')); }
</script>
</body>
</html>
"""
# ------------------------- 服务器 -------------------------
_PREFIXES = ("/xinquandashboard",) # 即使未剥前缀也能自适应
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
# ---- 基础 ----
def _send(self, code, body: bytes, ct="application/json; charset=utf-8",
extra=None):
self.send_response(code)
self.send_header("Content-Type", ct)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
if extra:
for k, v in extra.items():
self.send_header(k, v)
self.end_headers()
self.wfile.write(body)
def _json(self, obj, code=200):
self._send(code, json.dumps(obj, ensure_ascii=False).encode("utf-8"))
def _read_body(self):
n = int(self.headers.get("Content-Length") or 0)
return self.rfile.read(n) if n else b""
def _read_json(self):
raw = self._read_body()
try:
return json.loads(raw.decode("utf-8"))
except Exception:
return None
# ---- 路由 ----
def _split(self):
path = self.path.split("?", 1)[0]
for pre in _PREFIXES:
if path.startswith(pre):
path = path[len(pre):] or "/"
break
if not path.startswith("/"):
path = "/" + path
return path
# ---- 会话 ----
def _user(self):
cookies = self.headers.get("Cookie") or ""
tok = None
for pair in cookies.split(";"):
if not pair:
continue
kv = pair.strip().split("=", 1)
if len(kv) == 2 and kv[0] == "xq_auth":
tok = kv[1]
return _verify_token(tok) if tok else None
def _secure_cookie(self):
return (self.headers.get("X-Forwarded-Proto", "") == "https")
def _unauthed(self):
"""返回 True 表示未登录(调用方应返回 401)。"""
return True
def _login_ok(self, user):
# 生成 7 天会话
exp = int(time.time()) + SESSION_DAYS * 86400
tok = _sign_token(user, exp)
cookie = f"xq_auth={tok}; Path=/; HttpOnly; SameSite=Lax; Max-Age={SESSION_DAYS*86400}"
if self._secure_cookie():
cookie += "; Secure"
self._send(200, b'{"ok":true}', extra={"Set-Cookie": cookie})
def _login_fail(self):
self._json({"error": "用户名或密码错误"}, 401)
def _auth_denied(self):
if self.path.startswith("/media"):
return self._send(401, b"unauthorized", "text/plain; charset=utf-8")
return self._json({"error": "unauthorized"}, 401)
# ---- 路由 ----
def _split(self):
path = self.path.split("?", 1)[0]
for pre in _PREFIXES:
if path.startswith(pre):
path = path[len(pre):] or "/"
break
if not path.startswith("/"):
path = "/" + path
return path
def do_GET(self):
path = self._split()
if path == "/health":
return self._json({"ok": True})
user = self._user()
if path.startswith("/api/"):
if not user:
return self._auth_denied()
if path.startswith("/api/pages/"):
pid = path[len("/api/pages/"):]
return self._get_page(pid) if pid else self._json({"error": "missing id"}, 400)
if path == "/api/pages":
return self._list_pages()
return self._json({"error": "not found"}, 404)
if path.startswith("/media/"):
if not user:
return self._auth_denied()
return self._media(path[len("/media/"):])
# 其余(/, /login, SPA 内部路由):总是渲染外壳;未登录则显示登录表单
return self._send(200, self._index_html(bool(user)),
"text/html; charset=utf-8")
def _index_html(self, authed=True):
flag = "true" if authed else "false"
html = INDEX_HTML.replace("__MARKED__", BOOTSTRAP)
# 注入登录态,避免未授权时闪出内容
result = RE_AUTH.sub(flag, html, count=1)
return result.encode("utf-8")
def do_POST(self):
path = self._split()
if path == "/login":
data = self._read_json()
if data and data.get("username") == AUTH_USER and \
data.get("password") == AUTH_PASS:
return self._login_ok(AUTH_USER)
return self._login_fail()
if path == "/logout":
return self._logout()
if not self._user():
return self._auth_denied()
if path == "/api/pages":
return self._create_page()
if path == "/api/upload":
return self._upload()
self._json({"error": "not found"}, 404)
def _logout(self):
cookie = "xq_auth=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0"
self._send(200, b'{"ok":true}', extra={"Set-Cookie": cookie})
def do_PUT(self):
if not self._user():
return self._auth_denied()
path = self._split()
if path.startswith("/api/pages/"):
pid = path[len("/api/pages/"):]
if pid:
return self._update_page(pid)
self._json({"error": "not found"}, 404)
def do_DELETE(self):
if not self._user():
return self._auth_denied()
path = self._split()
if path.startswith("/api/pages/"):
pid = path[len("/api/pages/"):]
if pid:
return self._delete_page(pid)
self._json({"error": "not found"}, 404)
def log_message(self, *a): # 静音默认日志
pass
# ---- 页面 API ----
def _list_pages(self):
with _lock:
pages = _read_pages()
out = [{"id": pid, "title": p.get("title", ""),
"summary": _summary(p.get("content", "")),
"updated": p.get("updated")}
for pid, p in pages.items()]
out.sort(key=lambda x: x["updated"] or "", reverse=True)
self._json(out)
def _get_page(self, pid):
with _lock:
pages = _read_pages()
p = pages.get(pid)
if p is None:
return self._json({"error": "page not found"}, 404)
self._json({"id": pid, **p})
def _create_page(self):
data = self._read_json()
if not data or not data.get("title", "").strip():
return self._json({"error": "title required"}, 400)
with _lock:
pages = _read_pages()
pid = uuid.uuid4().hex[:12]
now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
pages[pid] = {"title": data["title"].strip(),
"content": data.get("content", ""),
"created": now, "updated": now}
_write_pages(pages)
self._json({"id": pid}, 201)
def _update_page(self, pid):
data = self._read_json()
if not data:
return self._json({"error": "bad body"}, 400)
with _lock:
pages = _read_pages()
if pid not in pages:
return self._json({"error": "page not found"}, 404)
if data.get("title", "").strip():
pages[pid]["title"] = data["title"].strip()
if "content" in data:
pages[pid]["content"] = data["content"] or ""
pages[pid]["updated"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
_write_pages(pages)
self._json({"id": pid})
def _delete_page(self, pid):
with _lock:
pages = _read_pages()
if pid not in pages:
return self._json({"error": "page not found"}, 404)
del pages[pid]
_write_pages(pages)
self._json({"ok": True})
# ---- 图片 ----
def _upload(self):
data = self._read_json()
if not data or not data.get("data"):
return self._json({"error": "no image data"}, 400)
name = (data.get("name") or "img").lower()
ext = os.path.splitext(name)[1].replace(".", "")
if ext:
full = "." + ext
if full not in ALLOWED_IMG_EXT:
# 依据 mime 兜底
mime = data.get("mime", "")
full = next((e for e, m in ALLOWED_IMG_EXT.items() if m == mime), ".png")
else:
mime = data.get("mime", "")
full = next((e for e, m in ALLOWED_IMG_EXT.items() if m == mime), ".png")
try:
raw = base64.b64decode(data["data"])
except Exception:
return self._json({"error": "bad base64"}, 400)
if not raw or len(raw) > 20 * 1024 * 1024:
return self._json({"error": "image empty or too large (>20MB)"}, 413)
fname = uuid.uuid4().hex + full
with open(os.path.join(UPLOADS_DIR, fname), "wb") as f:
f.write(raw)
self._json({"url": f"media/{fname}", "name": name})
def _media(self, name):
if "/" in name or "\\" in name or not name:
return self._json({"error": "bad media"}, 400)
p = os.path.join(UPLOADS_DIR, name)
if not os.path.isfile(p):
return self._send(404, b"not found", "text/plain")
ext = os.path.splitext(name)[1].lower()
ct = ALLOWED_IMG_EXT.get(ext, "application/octet-stream")
with open(p, "rb") as f:
body = f.read()
self._send(200, body, ct, {"Cache-Control": "public, max-age=86400"})
def main():
_ensure_dirs()
srv = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
print(f"xinquandashboard running on 0.0.0.0:{PORT}", flush=True)
srv.serve_forever()
if __name__ == "__main__":
main()