284 lines
8.5 KiB
Python
284 lines
8.5 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_CONFIG = {
|
||
'base_url': 'http://192.168.2.5:1234/v1',
|
||
'api_key': 'sk-lm-fuP5tGU8:Hi7YU87jHyDP6Ay8Tl2j',
|
||
'model': 'qwen3.5-4b',
|
||
}
|
||
|
||
def load_notes():
|
||
"""加载所有笔记"""
|
||
if NOTES_FILE.exists():
|
||
notes = json.loads(NOTES_FILE.read_text(encoding='utf-8'))
|
||
# 按置顶和更新时间排序
|
||
return sorted(notes, key=lambda x: (not 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: (not 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
|
||
|
||
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', '')
|
||
|
||
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
|
||
|
||
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
|
||
|
||
# 如果是新记录第一次输入内容，异步生成标题
|
||
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: (not 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: (not 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: (not 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: (not 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) |