hubian
e8ebd83d3c
- 每条记录增加version字段，每次保存版本号递增 - 客户端保存时携带版本号，服务端检查是否一致 - 版本冲突时返回HTTP 409，包含服务端最新内容 - 前端显示冲突对话框，用户可选择放弃或强制覆盖 - 旧数据自动添加version=1 测试场景： - A和B同时加载版本1 - A保存成功 → 版本2 - B保存失败 → 409冲突 - B选择覆盖 → 版本3
311 lines
9.7 KiB
Python
311 lines
9.7 KiB
Python
"""
|
|
碎片信息记录网站
|
|
- 实时保存到本地
|
|
- 大模型生成标题
|
|
- 搜索功能
|
|
"""
|
|
|
|
from flask import Flask, render_template, jsonify, request
|
|
from flask_cors import CORS
|
|
import json
|
|
import time
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
import requests
|
|
import threading
|
|
|
|
app = Flask(__name__, static_folder='static', static_url_path='/static')
|
|
CORS(app)
|
|
|
|
# 数据目录
|
|
DATA_DIR = Path(__file__).parent / 'data'
|
|
DATA_DIR.mkdir(exist_ok=True)
|
|
NOTES_FILE = DATA_DIR / 'notes.json'
|
|
|
|
# 大模型配置 (LLM Proxy)
|
|
LLM_CONFIG = {
|
|
'base_url': 'http://192.168.2.17:19007/v1',
|
|
'api_key': 'xxxx',
|
|
'model': 'auto',
|
|
}
|
|
|
|
def load_notes():
|
|
"""加载所有笔记"""
|
|
if NOTES_FILE.exists():
|
|
notes = json.loads(NOTES_FILE.read_text(encoding='utf-8'))
|
|
|
|
# 自动为旧数据添加版本号
|
|
for note in notes:
|
|
if 'version' not in note:
|
|
note['version'] = 1
|
|
|
|
# 按置顶和更新时间排序（置顶在前，更新时间降序）
|
|
return sorted(notes, key=lambda x: (-int(x.get('pinned', False)), x.get('updated_at', '')), reverse=True)
|
|
return []
|
|
|
|
def save_notes(notes):
|
|
"""保存笔记"""
|
|
NOTES_FILE.write_text(json.dumps(notes, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
|
|
def generate_title(content):
|
|
"""用大模型生成标题"""
|
|
if not content or len(content.strip()) < 10:
|
|
return content[:20] if content else "新记录"
|
|
|
|
try:
|
|
response = requests.post(
|
|
f"{LLM_CONFIG['base_url']}/chat/completions",
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {LLM_CONFIG['api_key']}"
|
|
},
|
|
json={
|
|
"model": LLM_CONFIG['model'],
|
|
"messages": [
|
|
{"role": "system", "content": "你是一个标题生成助手。请根据用户的内容，生成一个简洁的标题（不超过15个字）。只返回标题，不要其他内容。"},
|
|
{"role": "user", "content": content[:500]}
|
|
],
|
|
"max_tokens": 50,
|
|
"temperature": 0.3
|
|
},
|
|
timeout=10
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
title = data['choices'][0]['message']['content'].strip()
|
|
# 清理可能的引号或多余字符
|
|
title = title.replace('"', '').replace("'", '').strip()
|
|
if len(title) > 20:
|
|
title = title[:20]
|
|
return title
|
|
except Exception as e:
|
|
print(f"生成标题失败: {e}")
|
|
|
|
# 降级：取前20个字
|
|
return content[:20].strip()
|
|
|
|
def generate_title_async(note_id, content):
|
|
"""异步生成标题"""
|
|
title = generate_title(content)
|
|
|
|
# 更新笔记标题
|
|
notes = load_notes()
|
|
for note in notes:
|
|
if note['id'] == note_id:
|
|
note['title'] = title
|
|
break
|
|
save_notes(notes)
|
|
|
|
# ============ 页面路由 ============
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
# ============ API路由 ============
|
|
|
|
@app.route('/api/notes')
|
|
def api_notes():
|
|
"""获取笔记列表"""
|
|
keyword = request.args.get('q', '').strip().lower()
|
|
notes = load_notes()
|
|
|
|
# 搜索过滤
|
|
if keyword:
|
|
notes = [n for n in notes if keyword in n.get('title', '').lower() or keyword in n.get('content', '').lower()]
|
|
|
|
# 按置顶和更新时间排序（置顶在前，然后按更新时间降序）
|
|
notes = sorted(notes, key=lambda x: (-int(x.get('pinned', False)), x.get('updated_at', '')), reverse=True)
|
|
|
|
# 返回列表信息
|
|
return jsonify([{
|
|
'id': n['id'],
|
|
'title': n['title'],
|
|
'updated_at': n['updated_at'],
|
|
'created_at': n['created_at'],
|
|
'preview': n['content'][:50] if n['content'] else '',
|
|
'pinned': n.get('pinned', False),
|
|
} for n in notes])
|
|
|
|
@app.route('/api/notes/<note_id>')
|
|
def api_note_detail(note_id):
|
|
"""获取单个笔记详情"""
|
|
notes = load_notes()
|
|
note = next((n for n in notes if n['id'] == note_id), None)
|
|
|
|
if not note:
|
|
return jsonify({'error': 'Note not found'}), 404
|
|
|
|
# 确保有版本号
|
|
if 'version' not in note:
|
|
note['version'] = 1
|
|
|
|
return jsonify(note)
|
|
|
|
@app.route('/api/notes', methods=['POST'])
|
|
def api_create_note():
|
|
"""创建新笔记"""
|
|
notes = load_notes()
|
|
|
|
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
new_note = {
|
|
'id': uuid.uuid4().hex[:12],
|
|
'title': '新记录',
|
|
'content': '',
|
|
'created_at': now,
|
|
'updated_at': now,
|
|
}
|
|
|
|
notes.append(new_note)
|
|
save_notes(notes)
|
|
|
|
return jsonify(new_note)
|
|
|
|
@app.route('/api/notes/<note_id>', methods=['PUT'])
|
|
def api_update_note(note_id):
|
|
"""更新笔记内容（带版本检查）"""
|
|
data = request.get_json()
|
|
content = data.get('content', '')
|
|
client_version = data.get('version', 0) # 客户端提交的版本号
|
|
|
|
notes = load_notes()
|
|
note = next((n for n in notes if n['id'] == note_id), None)
|
|
|
|
if not note:
|
|
return jsonify({'error': 'Note not found'}), 404
|
|
|
|
# 确保服务端版本号存在
|
|
server_version = note.get('version', 1)
|
|
|
|
# 版本冲突检查：如果客户端版本 != 服务端版本，说明有并发编辑
|
|
if client_version > 0 and client_version != server_version:
|
|
# 返回冲突响应，包含服务端最新内容
|
|
return jsonify({
|
|
'error': 'version_conflict',
|
|
'message': '该记录已被其他设备修改，请先查看最新内容',
|
|
'server_version': server_version,
|
|
'server_content': note.get('content', ''),
|
|
'server_title': note.get('title', ''),
|
|
'server_updated_at': note.get('updated_at', '')
|
|
}), 409 # HTTP 409 Conflict
|
|
|
|
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
# 新记录第一次输入内容时，自动生成标题
|
|
old_content = note.get('content', '')
|
|
need_generate_title = (note['title'] == '新记录' and len(content) >= 20)
|
|
|
|
note['content'] = content
|
|
note['updated_at'] = now
|
|
note['version'] = server_version + 1 # 版本号递增
|
|
|
|
# 如果是新记录第一次输入内容，异步生成标题
|
|
if need_generate_title:
|
|
threading.Thread(target=generate_title_async, args=(note_id, content)).start()
|
|
|
|
save_notes(notes)
|
|
|
|
return jsonify(note)
|
|
|
|
|
|
@app.route('/api/notes/<note_id>/rename', methods=['POST'])
|
|
def api_rename_note(note_id):
|
|
"""重命名笔记标题"""
|
|
data = request.get_json()
|
|
new_title = data.get('title', '').strip()
|
|
|
|
if not new_title:
|
|
return jsonify({'error': '标题不能为空'}), 400
|
|
|
|
notes = load_notes()
|
|
note = next((n for n in notes if n['id'] == note_id), None)
|
|
|
|
if not note:
|
|
return jsonify({'error': 'Note not found'}), 404
|
|
|
|
note['title'] = new_title
|
|
note['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
# 重新排序
|
|
notes = sorted(notes, key=lambda x: (-int(x.get('pinned', False)), x.get('updated_at', '')), reverse=True)
|
|
save_notes(notes)
|
|
|
|
return jsonify({'success': True, 'title': new_title, 'note': note})
|
|
|
|
@app.route('/api/notes/<note_id>/title', methods=['POST'])
|
|
def api_regenerate_title(note_id):
|
|
"""手动重新生成标题"""
|
|
notes = load_notes()
|
|
note = next((n for n in notes if n['id'] == note_id), None)
|
|
|
|
if not note:
|
|
return jsonify({'error': 'Note not found'}), 404
|
|
|
|
title = generate_title(note['content'])
|
|
note['title'] = title
|
|
note['updated_at'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
# 重新排序
|
|
notes = sorted(notes, key=lambda x: (-int(x.get('pinned', False)), x.get('updated_at', '')), reverse=True)
|
|
save_notes(notes)
|
|
|
|
return jsonify({'success': True, 'title': title, 'note': note})
|
|
|
|
@app.route('/api/notes/<note_id>', methods=['DELETE'])
|
|
def api_delete_note(note_id):
|
|
"""删除笔记"""
|
|
notes = load_notes()
|
|
notes = [n for n in notes if n['id'] != note_id]
|
|
save_notes(notes)
|
|
|
|
return jsonify({'success': True})
|
|
|
|
@app.route('/api/notes/<note_id>/pin', methods=['POST'])
|
|
def api_pin_note(note_id):
|
|
"""置顶/取消置顶笔记"""
|
|
notes = load_notes()
|
|
note = next((n for n in notes if n['id'] == note_id), None)
|
|
|
|
if not note:
|
|
return jsonify({'error': 'Note not found'}), 404
|
|
|
|
note['pinned'] = not note.get('pinned', False)
|
|
|
|
# 重新排序
|
|
notes = sorted(notes, key=lambda x: (-int(x.get('pinned', False)), x.get('updated_at', '')), reverse=True)
|
|
save_notes(notes)
|
|
|
|
return jsonify({'success': True, 'pinned': note['pinned'], 'note': note})
|
|
|
|
@app.route('/api/search')
|
|
def api_search():
|
|
"""搜索笔记"""
|
|
keyword = request.args.get('q', '').strip().lower()
|
|
|
|
if not keyword:
|
|
return jsonify([])
|
|
|
|
notes = load_notes()
|
|
results = [n for n in notes if keyword in n.get('title', '').lower() or keyword in n.get('content', '').lower()]
|
|
|
|
# 按置顶和更新时间排序（置顶在前）
|
|
results = sorted(results, key=lambda x: (-int(x.get('pinned', False)), x.get('updated_at', '')), reverse=True)
|
|
|
|
return jsonify([{
|
|
'id': n['id'],
|
|
'title': n['title'],
|
|
'updated_at': n['updated_at'],
|
|
'preview': n['content'][:100] if n['content'] else '',
|
|
'pinned': n.get('pinned', False),
|
|
} for n in results])
|
|
|
|
if __name__ == '__main__':
|
|
print("=" * 50)
|
|
print("碎片信息记录网站")
|
|
print("=" * 50)
|
|
print(f"访问地址: http://localhost:19009")
|
|
print("=" * 50)
|
|
|
|
app.run(host='0.0.0.0', port=19009, debug=True) |