bwstudio
75f32647f6
_resolve_icon 的本地缓存命中顺序改为先查 .ico/.png/.jpg/.gif 再查 .svg， 避免已生成的字母头像（.svg）遮盖后续补入的真实 favicon 缓存文件。
400 lines
14 KiB
Python
400 lines
14 KiB
Python
"""
|
|
前台导航页视图（无需登录）。
|
|
|
|
URL 路径（挂在根路径下，便于匿名访问）：
|
|
- GET /site 公开导航聚合首页（按分类展示所有启用的导航）
|
|
- GET /site/category/<name> 分类详情：展示指定分类下所有启用的导航
|
|
- GET /site/api/navs JSON 接口（供前端动态加载）
|
|
|
|
注意：
|
|
- 这些视图**不加** @authorize，避免 login_required 强制登录。
|
|
- 使用 url_prefix='/site' 避免与后台 / 路由冲突（后台 / 是登录后的工作台）。
|
|
"""
|
|
from collections import OrderedDict
|
|
import time
|
|
import os
|
|
import re
|
|
import threading
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
from flask import Blueprint, render_template, request, jsonify, current_app, send_file
|
|
|
|
from applications.models import Nav, NavCategory
|
|
from plugins.siteStats.view.stat import get_nav_visit_counts, my_top_navs
|
|
|
|
# 「我常访问」冷启动阈值：至少点过这么多个不同网址才展示个人区
|
|
MINE_MIN_ITEMS = 2
|
|
|
|
bp = Blueprint('public_nav', __name__, url_prefix='/site')
|
|
|
|
# 分类列表缓存（左侧侧栏 / 移动端抽屉共用，避免每个请求都查库）
|
|
_NAV_CATS_CACHE = {'ts': 0.0, 'data': []}
|
|
_NAV_CATS_TTL = 30.0
|
|
|
|
|
|
def get_nav_categories() -> "list[dict]":
|
|
"""返回前台导航用的分类列表（按后台 sort 排序，带稳定锚点）。
|
|
|
|
返回形如::
|
|
[{'name': '开发工具', 'icon': 'layui-icon-list',
|
|
'description': '', 'anchor': 'cat-1'}, ...]
|
|
|
|
锚点 ``anchor`` 基于「启用分类」的排序索引生成（从 1 开始），与首页
|
|
各 ``<section>`` 的 id 严格一致——即使某个分类下暂时没有导航（空分组
|
|
被过滤）也不会错位。
|
|
|
|
带 30s 进程内缓存：后台新增/调整分类后最多 30s 生效。
|
|
"""
|
|
now = time.time()
|
|
cached = _NAV_CATS_CACHE.get('data') or []
|
|
if cached and (now - _NAV_CATS_CACHE.get('ts', 0.0)) < _NAV_CATS_TTL:
|
|
return cached
|
|
try:
|
|
cats = (
|
|
NavCategory.query
|
|
.filter(NavCategory.enable == 1)
|
|
.order_by(NavCategory.sort.desc(), NavCategory.id.asc())
|
|
.all()
|
|
)
|
|
names = [c.name for c in cats]
|
|
meta = {c.name: (c.icon, c.description) for c in cats}
|
|
|
|
# 兜底：只在 Nav 表里出现、未登记到 NavCategory 的历史分类，
|
|
# 追加到末尾（与 _grouped_navs 的兜底顺序一致：按分类名升序），
|
|
# 保证侧栏锚点与首页 section 一一对应。
|
|
used = Nav.query.filter(Nav.enable == 1).with_entities(Nav.category).distinct().all()
|
|
names.extend(sorted(n for (n,) in used if n and n not in meta))
|
|
|
|
data = []
|
|
for idx, name in enumerate(names):
|
|
icon, desc = meta.get(name, (None, None))
|
|
data.append({
|
|
'name': name,
|
|
'icon': icon or 'layui-icon-list',
|
|
'description': desc or '',
|
|
'anchor': 'cat-%d' % (idx + 1),
|
|
})
|
|
except Exception: # 数据库尚未初始化（如迁移前）时不影响页面渲染
|
|
data = []
|
|
_NAV_CATS_CACHE['ts'] = now
|
|
_NAV_CATS_CACHE['data'] = data
|
|
return data
|
|
|
|
|
|
def get_cat_anchor_map() -> "dict[str, str]":
|
|
"""分类名 -> 锚点 id 的映射，供首页渲染 section id 使用。"""
|
|
return {c['name']: c['anchor'] for c in get_nav_categories()}
|
|
|
|
|
|
def clear_nav_categories_cache():
|
|
"""后台改动分类后调用：立即失效缓存。"""
|
|
_NAV_CATS_CACHE['ts'] = 0.0
|
|
_NAV_CATS_CACHE['data'] = []
|
|
|
|
|
|
@bp.get('/')
|
|
def index():
|
|
"""
|
|
公开首页：按 category 分组聚合，渲染模板
|
|
"""
|
|
return _render_index()
|
|
|
|
|
|
@bp.get('/category/<string:name>')
|
|
def category_detail(name: str):
|
|
"""
|
|
分类详情页：列出指定分类下所有启用的导航
|
|
"""
|
|
items = (
|
|
Nav.query
|
|
.filter(Nav.category == name, Nav.enable == 1)
|
|
.order_by(Nav.sort.asc(), Nav.id.asc())
|
|
.all()
|
|
)
|
|
counts = get_nav_visit_counts()
|
|
dict_items = []
|
|
for it in items:
|
|
d = it.to_dict()
|
|
d['visit_count'] = counts.get(it.id, 0)
|
|
dict_items.append(d)
|
|
return render_template(
|
|
'public/category.html',
|
|
category=name,
|
|
items=dict_items,
|
|
site_name='旺珂 · 导航',
|
|
)
|
|
|
|
|
|
@bp.get('/api/navs')
|
|
def api_navs():
|
|
"""
|
|
JSON 接口：返回按分类聚合的导航数据
|
|
可用于首页卡片动态加载、第三方嵌入等
|
|
"""
|
|
return jsonify({
|
|
'success': True,
|
|
'msg': '请求成功',
|
|
'data': _grouped_navs(),
|
|
})
|
|
|
|
|
|
def render_public_index():
|
|
"""给外部复用：渲染前台公开导航首页（无需登录）。"""
|
|
return _render_index()
|
|
|
|
|
|
def get_category_meta() -> "dict[str, dict]":
|
|
"""给外部复用：返回启用的分类元信息。"""
|
|
return _category_meta()
|
|
|
|
|
|
def _render_index():
|
|
groups = _grouped_navs()
|
|
total = sum(len(items) for items in groups.values())
|
|
mine = my_top_navs(8)
|
|
return render_template(
|
|
'public/index.html',
|
|
groups=groups,
|
|
total=total,
|
|
site_name='旺珂 · 导航',
|
|
cat_meta=_category_meta(),
|
|
cat_anchors=get_cat_anchor_map(),
|
|
top_navs=_top_navs(12),
|
|
my_navs=mine if len(mine) >= MINE_MIN_ITEMS else [],
|
|
)
|
|
|
|
|
|
def _top_navs(limit: int = 12) -> "list[dict]":
|
|
"""跨分类聚合：返回点击数 TOP N 的有效 nav 列表（按点击数降序）。
|
|
|
|
- 复用 get_nav_visit_counts() 的 60s 进程内缓存
|
|
- 0 点击时返回空列表（调用方负责隐藏整个分类块）
|
|
- 同一个 nav 不可能重复出现（按 nav.id 维度）
|
|
"""
|
|
counts = get_nav_visit_counts()
|
|
if not counts:
|
|
return []
|
|
# 取点击数最高的 N 个 nav_id（点击数相同的按 id asc 稳定排序）
|
|
top_ids = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[:limit]
|
|
nav_ids = [nid for nid, _ in top_ids]
|
|
rows = (
|
|
Nav.query
|
|
.filter(Nav.id.in_(nav_ids), Nav.enable == 1)
|
|
.all()
|
|
)
|
|
by_id = {n.id: n.to_dict() for n in rows}
|
|
out = []
|
|
for nid in nav_ids:
|
|
if nid in by_id:
|
|
d = by_id[nid]
|
|
d['visit_count'] = counts.get(nid, 0)
|
|
out.append(d)
|
|
return out
|
|
|
|
|
|
def _grouped_navs() -> "OrderedDict[str, list]":
|
|
"""按 category 分组，组内按 sort asc 排序；分组顺序优先按 NavCategory.sort 排列"""
|
|
# 先取启用的分类，按 sort 排好序
|
|
cats = (
|
|
NavCategory.query
|
|
.filter(NavCategory.enable == 1)
|
|
.order_by(NavCategory.sort.desc(), NavCategory.id.asc())
|
|
.all()
|
|
)
|
|
ordered_names = [c.name for c in cats]
|
|
cat_order = {name: idx for idx, name in enumerate(ordered_names)}
|
|
|
|
items = (
|
|
Nav.query
|
|
.filter(Nav.enable == 1)
|
|
.order_by(Nav.category.asc(), Nav.sort.asc(), Nav.id.asc())
|
|
.all()
|
|
)
|
|
# 一次性取所有 nav 的点击数（带缓存）
|
|
counts = get_nav_visit_counts()
|
|
groups = OrderedDict()
|
|
# 按 cat_order 初始化空分组，保证顺序
|
|
for name in ordered_names:
|
|
groups[name] = []
|
|
for item in items:
|
|
d = item.to_dict()
|
|
d['visit_count'] = counts.get(item.id, 0)
|
|
if item.category in groups:
|
|
groups[item.category].append(d)
|
|
else:
|
|
# 对于未在 NavCategory 登记的历史分类，放在最后
|
|
groups.setdefault(item.category, []).append(d)
|
|
# 过滤空分组（启用的分类下可能没有导航）
|
|
return OrderedDict((k, v) for k, v in groups.items() if v)
|
|
|
|
|
|
def _category_meta() -> "dict[str, dict]":
|
|
"""返回分类元信息 {name: {icon, description, ...}}，用于抽屉图标等"""
|
|
cats = NavCategory.query.filter(NavCategory.enable == 1).all()
|
|
return {c.name: {'icon': c.icon or 'layui-icon-list', 'description': c.description or ''} for c in cats}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 卡片图标本地缓存
|
|
# 首次请求某域名图标时，自动拉取并保存到 data/icon/，以后直接读本地文件，
|
|
# 不再重复拉取。拉取失败（含内网/不可达网址）时自动生成「站点名称首字母」SVG 兜底。
|
|
# ---------------------------------------------------------------------------
|
|
_ICON_DIR = None
|
|
_ICON_LOCKS = {}
|
|
_ICON_LOCKS_GUARD = threading.Lock()
|
|
_ICON_EXT_MIME = {
|
|
'.ico': 'image/x-icon',
|
|
'.png': 'image/png',
|
|
'.jpg': 'image/jpeg',
|
|
'.jpeg': 'image/jpeg',
|
|
'.gif': 'image/gif',
|
|
'.svg': 'image/svg+xml',
|
|
}
|
|
|
|
|
|
def _icon_dir():
|
|
global _ICON_DIR
|
|
if _ICON_DIR is None:
|
|
base = current_app.config.get('DATA_DIR')
|
|
if not base:
|
|
base = os.path.join(os.path.dirname(os.path.dirname(__file__)), '..', 'data')
|
|
_ICON_DIR = os.path.join(base, 'icon')
|
|
os.makedirs(_ICON_DIR, exist_ok=True)
|
|
return _ICON_DIR
|
|
|
|
|
|
def _safe_name(domain: str) -> str:
|
|
return re.sub(r'[^A-Za-z0-9.\-_]', '_', domain or 'unknown')
|
|
|
|
|
|
def _lock_for(key: str):
|
|
with _ICON_LOCKS_GUARD:
|
|
lock = _ICON_LOCKS.get(key)
|
|
if lock is None:
|
|
lock = threading.Lock()
|
|
_ICON_LOCKS[key] = lock
|
|
return lock
|
|
|
|
|
|
def _domain_hue(domain: str) -> int:
|
|
"""与 models.admin_nav.Nav.to_dict 中用于生成卡片字母头像的配色保持一致"""
|
|
s = 0
|
|
for ch in (domain or ''):
|
|
s = (s * 31 + ord(ch)) & 0xffffffff
|
|
return s % 360
|
|
|
|
|
|
def _letter_svg(letter: str, domain: str) -> bytes:
|
|
hue = _domain_hue(domain)
|
|
c1 = "hsl(%d,62%%,56%%)" % hue
|
|
c2 = "hsl(%d,72%%,46%%)" % ((hue + 28) % 360)
|
|
ch = (letter or (domain or '·')[:1] or '·').upper()[:1] or '·'
|
|
ch = re.sub(r'[<>&"]', '', ch) # 防止 SVG 注入
|
|
svg = (
|
|
'<svg xmlns="http://www.w3.org/2000/svg" width="64" height="64" viewBox="0 0 64 64">'
|
|
'<defs><linearGradient id="g" x1="0" y1="0" x2="1" y2="1">'
|
|
'<stop offset="0%%" stop-color="%s"/>'
|
|
'<stop offset="100%%" stop-color="%s"/>'
|
|
'</linearGradient></defs>'
|
|
'<rect width="64" height="64" rx="14" fill="url(#g)"/>'
|
|
'<text x="32" y="33" text-anchor="middle" dominant-baseline="central" '
|
|
'font-family="Arial, \'PingFang SC\', \'Microsoft YaHei\', sans-serif" '
|
|
'font-size="34" font-weight="700" fill="#ffffff">%s</text>'
|
|
'</svg>'
|
|
) % (c1, c2, ch)
|
|
return svg.encode('utf-8')
|
|
|
|
|
|
def _fetch_bytes(url: str, timeout: int = 5):
|
|
try:
|
|
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
data = resp.read()
|
|
ctype = resp.headers.get('Content-Type', '') or ''
|
|
return data, ctype
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _ext_from(ctype: str, data: bytes) -> str:
|
|
ct = (ctype or '').lower()
|
|
if 'png' in ct:
|
|
return '.png'
|
|
if 'jpeg' in ct or 'jpg' in ct:
|
|
return '.jpg'
|
|
if 'gif' in ct:
|
|
return '.gif'
|
|
if 'svg' in ct:
|
|
return '.svg'
|
|
if 'icon' in ct:
|
|
return '.ico'
|
|
if data[:4] == b'\x00\x00\x01\x00':
|
|
return '.ico'
|
|
if data[:8] == b'\x89PNG\r\n\x1a\n':
|
|
return '.png'
|
|
if data[:3] == b'GIF':
|
|
return '.gif'
|
|
return '.ico'
|
|
|
|
|
|
def _resolve_icon(domain: str, letter: str):
|
|
"""返回 (file_path, mimetype)。优先读本地缓存，否则拉取/生成并落盘。"""
|
|
safe = _safe_name(domain)
|
|
d = _icon_dir()
|
|
# 1) 命中本地缓存（任意已知扩展名）
|
|
for ext in ('.ico', '.png', '.jpg', '.jpeg', '.gif', '.svg'):
|
|
p = os.path.join(d, safe + ext)
|
|
if os.path.isfile(p):
|
|
return p, _ICON_EXT_MIME.get(ext, 'application/octet-stream')
|
|
# 2) 加锁后再次检查（避免并发重复拉取）
|
|
with _lock_for(safe):
|
|
for ext in ('.ico', '.png', '.jpg', '.jpeg', '.gif', '.svg'):
|
|
p = os.path.join(d, safe + ext)
|
|
if os.path.isfile(p):
|
|
return p, _ICON_EXT_MIME.get(ext, 'application/octet-stream')
|
|
# 3) 循序尝试：站点自身 https → http（含内网）→ DuckDuckGo
|
|
candidates = [
|
|
'https://%s/favicon.ico' % domain,
|
|
'http://%s/favicon.ico' % domain,
|
|
'https://icons.duckduckgo.com/ip3/%s.ico' % domain,
|
|
]
|
|
for url in candidates:
|
|
res = _fetch_bytes(url)
|
|
if not res:
|
|
continue
|
|
data, ctype = res
|
|
if not data or len(data) < 32: # 过小多半是错误页/空响应
|
|
continue
|
|
# 必须是真正的图片（按 Content-Type 或二进制魔数判断），
|
|
# 否则跳过——避免把 200 的 HTML 错误页当图标缓存下来
|
|
ct = (ctype or '').lower()
|
|
is_image = (
|
|
ct.startswith('image/')
|
|
or data[:4] == b'\x00\x00\x01\x00'
|
|
or data[:8] == b'\x89PNG\r\n\x1a\n'
|
|
or data[:3] == b'GIF'
|
|
)
|
|
if not is_image:
|
|
continue
|
|
ext = _ext_from(ctype, data)
|
|
p = os.path.join(d, safe + ext)
|
|
try:
|
|
with open(p, 'wb') as f:
|
|
f.write(data)
|
|
return p, _ICON_EXT_MIME.get(ext, 'application/octet-stream')
|
|
except OSError:
|
|
continue
|
|
# 4) 全部失败：生成字母头像 SVG 兜底并缓存
|
|
p = os.path.join(d, safe + '.svg')
|
|
with open(p, 'wb') as f:
|
|
f.write(_letter_svg(letter, domain))
|
|
return p, 'image/svg+xml'
|
|
|
|
|
|
@bp.get('/icon/<string:domain>')
|
|
def icon(domain):
|
|
"""卡片图标：本地缓存优先，拉取失败时返回字母头像 SVG。"""
|
|
letter = request.args.get('letter', '') or ''
|
|
path, mime = _resolve_icon(domain, letter)
|
|
return send_file(path, mimetype=mime, max_age=86400) |