619 lines
21 KiB
Python
619 lines
21 KiB
Python
import os
|
||
import sqlite3
|
||
import unicodedata
|
||
from contextlib import contextmanager
|
||
|
||
DATA_DIR = os.environ.get('DATA_DIR', os.path.join(os.path.dirname(__file__), 'data'))
|
||
DATABASE_PATH = os.environ.get('DATABASE_PATH', os.path.join(DATA_DIR, 'site.db'))
|
||
_SCHEMA_PATH = os.path.join(os.path.dirname(__file__), 'schema.sql')
|
||
|
||
|
||
def _new_connection():
|
||
os.makedirs(os.path.dirname(DATABASE_PATH) or '.', exist_ok=True)
|
||
conn = sqlite3.connect(DATABASE_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute('PRAGMA foreign_keys = ON')
|
||
conn.execute('PRAGMA journal_mode = WAL')
|
||
return conn
|
||
|
||
|
||
@contextmanager
|
||
def get_connection():
|
||
conn = _new_connection()
|
||
try:
|
||
yield conn
|
||
conn.commit()
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _apply_schema(conn):
|
||
with open(_SCHEMA_PATH, 'r', encoding='utf-8') as f:
|
||
conn.executescript(f.read())
|
||
|
||
|
||
def _profile_is_empty(conn):
|
||
row = conn.execute('SELECT COUNT(*) AS c FROM profile').fetchone()
|
||
return row['c'] == 0
|
||
|
||
|
||
def _clean_tag_list(tags):
|
||
out = []
|
||
for t in tags or []:
|
||
s = str(t).strip()
|
||
if s:
|
||
out.append(s)
|
||
return out
|
||
|
||
|
||
def _tech_name_set(conn):
|
||
return {
|
||
str(r['name']).strip().lower()
|
||
for r in conn.execute('SELECT name FROM techstack_item')
|
||
if r['name'] and str(r['name']).strip()
|
||
}
|
||
|
||
|
||
def _split_merged_tags(conn, merged):
|
||
tech_set = _tech_name_set(conn)
|
||
cat, tech = [], []
|
||
for t in merged or []:
|
||
s = str(t).strip()
|
||
if not s:
|
||
continue
|
||
(tech if s.lower() in tech_set else cat).append(s)
|
||
return cat, tech
|
||
|
||
|
||
def _project_tag_payload_from_body(conn, data):
|
||
if 'techTags' in data:
|
||
return _clean_tag_list(data.get('tags')), _clean_tag_list(data.get('techTags'))
|
||
return _split_merged_tags(conn, data.get('tags') or [])
|
||
|
||
|
||
def _replace_project_tags(conn, project_id, tags):
|
||
conn.execute('DELETE FROM project_tag WHERE project_id = ?', (project_id,))
|
||
for j, tag in enumerate(_clean_tag_list(tags)):
|
||
conn.execute(
|
||
'INSERT INTO project_tag (project_id, sort_order, tag) VALUES (?, ?, ?)',
|
||
(project_id, j, tag),
|
||
)
|
||
|
||
|
||
def _replace_project_tech_tags(conn, project_id, tags):
|
||
conn.execute('DELETE FROM project_tech_tag WHERE project_id = ?', (project_id,))
|
||
for j, tag in enumerate(_clean_tag_list(tags)):
|
||
conn.execute(
|
||
'INSERT INTO project_tech_tag (project_id, sort_order, tag) VALUES (?, ?, ?)',
|
||
(project_id, j, tag),
|
||
)
|
||
|
||
|
||
def _migrate_legacy_project_tags_if_needed(conn):
|
||
tech_set = _tech_name_set(conn)
|
||
for row in conn.execute('SELECT id FROM project'):
|
||
pid = row['id']
|
||
if conn.execute(
|
||
'SELECT 1 FROM project_tech_tag WHERE project_id = ? LIMIT 1',
|
||
(pid,),
|
||
).fetchone():
|
||
continue
|
||
tag_rows = conn.execute(
|
||
'SELECT tag FROM project_tag WHERE project_id = ? ORDER BY sort_order, id',
|
||
(pid,),
|
||
).fetchall()
|
||
if not tag_rows:
|
||
continue
|
||
cat, tech = [], []
|
||
for r in tag_rows:
|
||
s = str(r['tag']).strip()
|
||
if not s:
|
||
continue
|
||
if s.lower() in tech_set:
|
||
tech.append(s)
|
||
else:
|
||
cat.append(s)
|
||
if not tech:
|
||
continue
|
||
conn.execute('DELETE FROM project_tag WHERE project_id = ?', (pid,))
|
||
for j, t in enumerate(cat):
|
||
conn.execute(
|
||
'INSERT INTO project_tag (project_id, sort_order, tag) VALUES (?,?,?)',
|
||
(pid, j, t),
|
||
)
|
||
for j, t in enumerate(tech):
|
||
conn.execute(
|
||
'INSERT INTO project_tech_tag (project_id, sort_order, tag) VALUES (?,?,?)',
|
||
(pid, j, t),
|
||
)
|
||
|
||
|
||
def _seed_minimal_defaults(conn):
|
||
"""空库时写入 profile 与技术栈区块占位行，内容通过管理后台或 SQL 维护。"""
|
||
conn.execute('INSERT INTO profile (id, nickname) VALUES (1, ?)', ('',))
|
||
conn.execute('INSERT INTO techstack_section (id, title) VALUES (1, ?)', ('技术栈',))
|
||
|
||
|
||
def _ensure_builtin_tech_items(conn):
|
||
"""内置技术栈项：按名称幂等追加（无则插入）。"""
|
||
builtin = (
|
||
(
|
||
'Vite',
|
||
'https://vitejs.dev/',
|
||
'https://pan.shumengya.top/d/smy/image/devkit/vite.svg',
|
||
'#646CFF',
|
||
),
|
||
(
|
||
'Tailwind CSS',
|
||
'https://tailwindcss.com/',
|
||
'https://pan.shumengya.top/d/smy/image/devkit/tailwind.svg',
|
||
'#06B6D4',
|
||
),
|
||
(
|
||
'RabbitMQ',
|
||
'https://www.rabbitmq.com/',
|
||
'https://pan.shumengya.top/d/smy/image/devkit/rabbitmq.svg',
|
||
'#FF6600',
|
||
),
|
||
)
|
||
for name, link, svg, color in builtin:
|
||
if conn.execute(
|
||
'SELECT 1 FROM techstack_item WHERE name = ? LIMIT 1',
|
||
(name,),
|
||
).fetchone():
|
||
continue
|
||
m = conn.execute('SELECT COALESCE(MAX(sort_order), 0) AS m FROM techstack_item').fetchone()['m']
|
||
sort_order = int(m) + 10
|
||
conn.execute(
|
||
'''INSERT INTO techstack_item (sort_order, name, link, svg, color, is_show)
|
||
VALUES (?, ?, ?, ?, ?, ?)''',
|
||
(sort_order, name, link, svg, color, 1),
|
||
)
|
||
|
||
|
||
def _ensure_contact_is_show_column(conn):
|
||
"""旧库 contact 表无 is_show 时追加列（默认展示）。"""
|
||
cols = {str(r['name']) for r in conn.execute('PRAGMA table_info(contact)').fetchall()}
|
||
if cols and 'is_show' not in cols:
|
||
conn.execute(
|
||
'ALTER TABLE contact ADD COLUMN is_show INTEGER NOT NULL DEFAULT 1'
|
||
)
|
||
|
||
|
||
def init_storage():
|
||
with get_connection() as conn:
|
||
_apply_schema(conn)
|
||
_ensure_contact_is_show_column(conn)
|
||
if _profile_is_empty(conn):
|
||
_seed_minimal_defaults(conn)
|
||
else:
|
||
_migrate_legacy_project_tags_if_needed(conn)
|
||
_ensure_builtin_tech_items(conn)
|
||
_apply_tech_item_display_patches(conn)
|
||
_purge_retired_tech_labels(conn)
|
||
|
||
|
||
def _normalize_tech_label_key(s):
|
||
"""仅保留字母数字并小写；NFKC 合并全角/兼容字形，匹配 d1SQL 各类写法。"""
|
||
if s is None:
|
||
return ''
|
||
t = unicodedata.normalize('NFKC', str(s)).strip()
|
||
return ''.join(c.lower() for c in t if c.isalnum())
|
||
|
||
|
||
def _purge_retired_tech_labels(conn):
|
||
"""移除已废弃的技术名：项目标签 + 技术栈表（幂等）。"""
|
||
retired_keys = {'d1sql'}
|
||
for table in ('project_tech_tag', 'project_tag'):
|
||
rows = conn.execute(f'SELECT id, tag FROM {table}').fetchall()
|
||
for r in rows:
|
||
if _normalize_tech_label_key(r['tag']) in retired_keys:
|
||
conn.execute(f'DELETE FROM {table} WHERE id = ?', (r['id'],))
|
||
rows = conn.execute('SELECT id, name FROM techstack_item').fetchall()
|
||
for r in rows:
|
||
if _normalize_tech_label_key(r['name']) in retired_keys:
|
||
conn.execute('DELETE FROM techstack_item WHERE id = ?', (r['id'],))
|
||
|
||
|
||
def _apply_tech_item_display_patches(conn):
|
||
"""微调技术栈徽章底色（图标为自带颜色的 SVG 时避免与底色撞色）。"""
|
||
patches = (
|
||
# Docker.svg 为蓝色填充：深蓝→亮蓝渐变，与白字、鲸鱼标层次更清晰
|
||
(
|
||
'Docker',
|
||
'linear-gradient(135deg, #052e4a 0%, #0d47a1 42%, #1e88e5 100%)',
|
||
),
|
||
# Flask.svg 为 #3babc3 青色，避免纯黑底；深青绿过渡到品牌青
|
||
(
|
||
'Flask',
|
||
'linear-gradient(135deg, #0f3a40 0%, #1e6570 45%, #2d8fa0 100%)',
|
||
),
|
||
# JSON / Markdown：Simple Icons 等品牌资料中的标准色（实心）
|
||
# JSON: https://simpleicons.org — #000000（与 Crockford JSON 标识一致）
|
||
('JSON', '#000000'),
|
||
# Markdown: Markdown 标识常用主色 — #083FA5
|
||
('Markdown', '#083FA5'),
|
||
)
|
||
for name, color in patches:
|
||
conn.execute(
|
||
'UPDATE techstack_item SET color = ? WHERE name = ?',
|
||
(color, name),
|
||
)
|
||
|
||
|
||
def _row_to_profile(row):
|
||
if row is None:
|
||
return None
|
||
keys = ('favicon', 'nickname', 'avatar', 'introduction', 'footer', 'site', 'homepage')
|
||
return {k: row[k] for k in keys if row[k] is not None}
|
||
|
||
|
||
def get_profile():
|
||
with get_connection() as conn:
|
||
row = conn.execute(
|
||
'''SELECT favicon, nickname, avatar, introduction, footer, site, homepage
|
||
FROM profile WHERE id = 1'''
|
||
).fetchone()
|
||
return _row_to_profile(row)
|
||
|
||
|
||
def get_projects():
|
||
with get_connection() as conn:
|
||
rows = conn.execute(
|
||
'SELECT id, title, description, link, icon, is_admin, is_show, is_develop FROM project ORDER BY sort_order, id'
|
||
).fetchall()
|
||
out = []
|
||
for row in rows:
|
||
tags = [
|
||
r['tag']
|
||
for r in conn.execute(
|
||
'SELECT tag FROM project_tag WHERE project_id = ? ORDER BY sort_order, id',
|
||
(row['id'],),
|
||
)
|
||
]
|
||
tech_tags = [
|
||
r['tag']
|
||
for r in conn.execute(
|
||
'SELECT tag FROM project_tech_tag WHERE project_id = ? ORDER BY sort_order, id',
|
||
(row['id'],),
|
||
)
|
||
]
|
||
out.append({
|
||
'id': row['id'],
|
||
'title': row['title'],
|
||
'description': row['description'] or '',
|
||
'link': row['link'] or '',
|
||
'icon': row['icon'] or '',
|
||
'tags': tags,
|
||
'techTags': tech_tags,
|
||
'admin': bool(row['is_admin']),
|
||
'show': bool(row['is_show']),
|
||
'develop': bool(row['is_develop']),
|
||
})
|
||
return {'projects': out}
|
||
|
||
|
||
def _contact_row_to_dict(row):
|
||
return {
|
||
'id': row['id'],
|
||
'type': row['type'],
|
||
'label': row['label'] or '',
|
||
'value': row['value'] or '',
|
||
'link': row['link'] or '',
|
||
'icon': row['icon'] or '',
|
||
'show': bool(row['is_show']),
|
||
}
|
||
|
||
|
||
def get_contacts():
|
||
"""前台公开接口：仅返回首页展示的联系方式。"""
|
||
with get_connection() as conn:
|
||
rows = conn.execute(
|
||
'''SELECT id, type, label, value, link, icon, is_show FROM contact
|
||
WHERE COALESCE(is_show, 1) = 1
|
||
ORDER BY sort_order, id'''
|
||
).fetchall()
|
||
return {
|
||
'contacts': [_contact_row_to_dict(row) for row in rows],
|
||
}
|
||
|
||
|
||
def get_techstack():
|
||
"""技术栈 svg 字段：可为 data/logo 下文件名，或 http(s) 外链 SVG 地址。"""
|
||
with get_connection() as conn:
|
||
sec = conn.execute('SELECT title FROM techstack_section WHERE id = 1').fetchone()
|
||
title = sec['title'] if sec else '技术栈'
|
||
rows = conn.execute(
|
||
'SELECT id, name, link, svg, color, is_show FROM techstack_item ORDER BY sort_order, id'
|
||
).fetchall()
|
||
return {
|
||
'title': title,
|
||
'items': [
|
||
{
|
||
'id': row['id'],
|
||
'name': row['name'],
|
||
'link': row['link'] or '',
|
||
'svg': row['svg'] or '',
|
||
'color': row['color'] or '',
|
||
'show': bool(row['is_show']),
|
||
}
|
||
for row in rows
|
||
],
|
||
}
|
||
|
||
|
||
def get_all(include_hidden_contacts=False):
|
||
with get_connection() as conn:
|
||
profile_row = conn.execute(
|
||
'''SELECT favicon, nickname, avatar, introduction, footer, site, homepage
|
||
FROM profile WHERE id = 1'''
|
||
).fetchone()
|
||
profile = _row_to_profile(profile_row)
|
||
|
||
tech_sec = conn.execute('SELECT title FROM techstack_section WHERE id = 1').fetchone()
|
||
tech_title = tech_sec['title'] if tech_sec else '技术栈'
|
||
tech_rows = conn.execute(
|
||
'SELECT id, name, link, svg, color, is_show FROM techstack_item ORDER BY sort_order, id'
|
||
).fetchall()
|
||
techstack = {
|
||
'title': tech_title,
|
||
'items': [
|
||
{
|
||
'id': r['id'],
|
||
'name': r['name'],
|
||
'link': r['link'] or '',
|
||
'svg': r['svg'] or '',
|
||
'color': r['color'] or '',
|
||
'show': bool(r['is_show']),
|
||
}
|
||
for r in tech_rows
|
||
],
|
||
}
|
||
|
||
proj_rows = conn.execute(
|
||
'SELECT id, title, description, link, icon, is_admin, is_show, is_develop FROM project ORDER BY sort_order, id'
|
||
).fetchall()
|
||
projects = []
|
||
for row in proj_rows:
|
||
tags = [
|
||
r['tag']
|
||
for r in conn.execute(
|
||
'SELECT tag FROM project_tag WHERE project_id = ? ORDER BY sort_order, id',
|
||
(row['id'],),
|
||
)
|
||
]
|
||
tech_tags = [
|
||
r['tag']
|
||
for r in conn.execute(
|
||
'SELECT tag FROM project_tech_tag WHERE project_id = ? ORDER BY sort_order, id',
|
||
(row['id'],),
|
||
)
|
||
]
|
||
projects.append({
|
||
'id': row['id'],
|
||
'title': row['title'],
|
||
'description': row['description'] or '',
|
||
'link': row['link'] or '',
|
||
'icon': row['icon'] or '',
|
||
'tags': tags,
|
||
'techTags': tech_tags,
|
||
'admin': bool(row['is_admin']),
|
||
'show': bool(row['is_show']),
|
||
'develop': bool(row['is_develop']),
|
||
})
|
||
|
||
contact_rows = conn.execute(
|
||
'SELECT id, type, label, value, link, icon, is_show FROM contact ORDER BY sort_order, id'
|
||
).fetchall()
|
||
contact_list = []
|
||
for cr in contact_rows:
|
||
if not include_hidden_contacts and not bool(cr['is_show']):
|
||
continue
|
||
contact_list.append(_contact_row_to_dict(cr))
|
||
contacts = {'contacts': contact_list}
|
||
|
||
return {
|
||
'profile': profile,
|
||
'techstack': techstack,
|
||
'projects': {'projects': projects},
|
||
'contacts': contacts,
|
||
}
|
||
|
||
|
||
def update_profile(payload):
|
||
allowed = {'favicon', 'nickname', 'avatar', 'introduction', 'footer', 'site', 'homepage'}
|
||
updates = {k: payload[k] for k in allowed if k in payload}
|
||
if not updates:
|
||
return
|
||
sets = ', '.join(f'{k} = ?' for k in updates)
|
||
vals = list(updates.values())
|
||
with get_connection() as conn:
|
||
conn.execute(f'UPDATE profile SET {sets} WHERE id = 1', vals)
|
||
|
||
|
||
def create_project(data):
|
||
with get_connection() as conn:
|
||
cat_tags, tech_tags = _project_tag_payload_from_body(conn, data)
|
||
m = conn.execute('SELECT COALESCE(MAX(sort_order), 0) AS m FROM project').fetchone()['m']
|
||
sort_order = int(m) + 10
|
||
cur = conn.execute(
|
||
'''INSERT INTO project (sort_order, title, description, link, icon, is_admin, is_show, is_develop)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)''',
|
||
(
|
||
sort_order,
|
||
data['title'],
|
||
data.get('description') or '',
|
||
data.get('link') or '',
|
||
data.get('icon') or '',
|
||
1 if data.get('admin') else 0,
|
||
1 if data.get('show', True) else 0,
|
||
1 if data.get('develop') else 0,
|
||
),
|
||
)
|
||
pid = cur.lastrowid
|
||
_replace_project_tags(conn, pid, cat_tags)
|
||
_replace_project_tech_tags(conn, pid, tech_tags)
|
||
return pid
|
||
|
||
|
||
def update_project(project_id, data):
|
||
with get_connection() as conn:
|
||
fields = []
|
||
vals = []
|
||
if 'title' in data:
|
||
fields.append('title = ?')
|
||
vals.append(data['title'])
|
||
if 'description' in data:
|
||
fields.append('description = ?')
|
||
vals.append(data['description'] or '')
|
||
if 'link' in data:
|
||
fields.append('link = ?')
|
||
vals.append(data['link'] or '')
|
||
if 'icon' in data:
|
||
fields.append('icon = ?')
|
||
vals.append(data['icon'] or '')
|
||
if 'admin' in data:
|
||
fields.append('is_admin = ?')
|
||
vals.append(1 if data['admin'] else 0)
|
||
if 'show' in data:
|
||
fields.append('is_show = ?')
|
||
vals.append(1 if data['show'] else 0)
|
||
if 'develop' in data:
|
||
fields.append('is_develop = ?')
|
||
vals.append(1 if data['develop'] else 0)
|
||
if fields:
|
||
vals.append(project_id)
|
||
conn.execute(
|
||
f"UPDATE project SET {', '.join(fields)} WHERE id = ?",
|
||
vals,
|
||
)
|
||
if 'tags' in data or 'techTags' in data:
|
||
if 'tags' in data and 'techTags' in data:
|
||
_replace_project_tags(conn, project_id, data['tags'])
|
||
_replace_project_tech_tags(conn, project_id, data['techTags'])
|
||
elif 'tags' in data:
|
||
cat, tech = _split_merged_tags(conn, data['tags'])
|
||
_replace_project_tags(conn, project_id, cat)
|
||
_replace_project_tech_tags(conn, project_id, tech)
|
||
else:
|
||
_replace_project_tech_tags(conn, project_id, data['techTags'])
|
||
|
||
|
||
def delete_project(project_id):
|
||
with get_connection() as conn:
|
||
conn.execute('DELETE FROM project WHERE id = ?', (project_id,))
|
||
|
||
|
||
def create_contact(data):
|
||
with get_connection() as conn:
|
||
m = conn.execute('SELECT COALESCE(MAX(sort_order), 0) AS m FROM contact').fetchone()['m']
|
||
sort_order = int(m) + 10
|
||
cur = conn.execute(
|
||
'''INSERT INTO contact (sort_order, type, label, value, link, icon, is_show)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)''',
|
||
(
|
||
sort_order,
|
||
data['type'],
|
||
data.get('label') or '',
|
||
data.get('value') or '',
|
||
data.get('link') or '',
|
||
data.get('icon') or '',
|
||
1 if data.get('show', True) else 0,
|
||
),
|
||
)
|
||
return cur.lastrowid
|
||
|
||
|
||
def update_contact(contact_id, data):
|
||
with get_connection() as conn:
|
||
fields = []
|
||
vals = []
|
||
for col, key in (
|
||
('type', 'type'),
|
||
('label', 'label'),
|
||
('value', 'value'),
|
||
('link', 'link'),
|
||
('icon', 'icon'),
|
||
):
|
||
if key in data:
|
||
fields.append(f'{col} = ?')
|
||
vals.append(data[key] or '')
|
||
if 'show' in data:
|
||
fields.append('is_show = ?')
|
||
vals.append(1 if data['show'] else 0)
|
||
if fields:
|
||
vals.append(contact_id)
|
||
conn.execute(
|
||
f"UPDATE contact SET {', '.join(fields)} WHERE id = ?",
|
||
vals,
|
||
)
|
||
|
||
|
||
def delete_contact(contact_id):
|
||
with get_connection() as conn:
|
||
conn.execute('DELETE FROM contact WHERE id = ?', (contact_id,))
|
||
|
||
|
||
def set_techstack_title(title):
|
||
t = title or '技术栈'
|
||
with get_connection() as conn:
|
||
row = conn.execute('SELECT id FROM techstack_section WHERE id = 1').fetchone()
|
||
if row:
|
||
conn.execute('UPDATE techstack_section SET title = ? WHERE id = 1', (t,))
|
||
else:
|
||
conn.execute('INSERT INTO techstack_section (id, title) VALUES (1, ?)', (t,))
|
||
|
||
|
||
def create_techstack_item(data):
|
||
with get_connection() as conn:
|
||
m = conn.execute('SELECT COALESCE(MAX(sort_order), 0) AS m FROM techstack_item').fetchone()['m']
|
||
sort_order = int(m) + 10
|
||
cur = conn.execute(
|
||
'''INSERT INTO techstack_item (sort_order, name, link, svg, color, is_show)
|
||
VALUES (?, ?, ?, ?, ?, ?)''',
|
||
(
|
||
sort_order,
|
||
data['name'],
|
||
data.get('link') or '',
|
||
data.get('svg') or '',
|
||
data.get('color') or '',
|
||
1 if data.get('show', True) else 0,
|
||
),
|
||
)
|
||
return cur.lastrowid
|
||
|
||
|
||
def update_techstack_item(item_id, data):
|
||
with get_connection() as conn:
|
||
fields = []
|
||
vals = []
|
||
if 'name' in data:
|
||
fields.append('name = ?')
|
||
vals.append(data['name'])
|
||
if 'link' in data:
|
||
fields.append('link = ?')
|
||
vals.append(data['link'] or '')
|
||
if 'svg' in data:
|
||
fields.append('svg = ?')
|
||
vals.append(data['svg'] or '')
|
||
if 'color' in data:
|
||
fields.append('color = ?')
|
||
vals.append(data['color'] or '')
|
||
if 'show' in data:
|
||
fields.append('is_show = ?')
|
||
vals.append(1 if data['show'] else 0)
|
||
if fields:
|
||
vals.append(item_id)
|
||
conn.execute(
|
||
f"UPDATE techstack_item SET {', '.join(fields)} WHERE id = ?",
|
||
vals,
|
||
)
|
||
|
||
|
||
def delete_techstack_item(item_id):
|
||
with get_connection() as conn:
|
||
conn.execute('DELETE FROM techstack_item WHERE id = ?', (item_id,))
|