snippet-notes/app.py

"""
碎片信息记录网站
- 实时保存到本地
- 大模型生成标题
- 搜索功能
- 编辑锁机制（防止并发编辑）
"""

from flask import Flask, render_template, jsonify, request
from flask_cors import CORS
import json
import time
import uuid
from datetime import datetime, timedelta
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'

# 编辑锁存储（内存，不持久化）
# 格式: {note_id: {"session_id": xxx, "locked_at": timestamp, "last_heartbeat": timestamp}}
EDIT_LOCKS = {}
LOCK_TIMEOUT = 30  # 锁超时时间（秒），无心跳自动释放

# 大模型配置 (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])

# ============ 编辑锁 API ============

def check_lock_valid(note_id):
    """检查锁是否有效（未超时）"""
    if note_id not in EDIT_LOCKS:
        return False
    
    lock = EDIT_LOCKS[note_id]
    now = time.time()
    
    # 超过LOCK_TIMEOUT秒无心跳，锁失效
    if now - lock['last_heartbeat'] > LOCK_TIMEOUT:
        del EDIT_LOCKS[note_id]
        return False
    
    return True

def get_lock_info(note_id):
    """获取锁信息"""
    if note_id in EDIT_LOCKS:
        return EDIT_LOCKS[note_id]
    return None

@app.route('/api/notes/<note_id>/lock', methods=['POST'])
def api_acquire_lock(note_id):
    """获取编辑锁"""
    data = request.get_json() or {}
    session_id = data.get('session_id', '')
    
    if not session_id:
        # 生成临时session_id
        session_id = uuid.uuid4().hex[:16]
    
    now = time.time()
    
    # 检查是否已被锁定
    if check_lock_valid(note_id):
        lock = EDIT_LOCKS[note_id]
        
        # 同一个session可以重复获取（刷新锁）
        if lock['session_id'] == session_id:
            lock['last_heartbeat'] = now
            return jsonify({
                'locked': True,
                'session_id': session_id,
                'is_owner': True
            })
        
        # 被其他人锁定
        return jsonify({
            'locked': True,
            'session_id': session_id,
            'is_owner': False,
            'lock_owner': lock['session_id'],
            'locked_at': lock['locked_at'],
            'message': '该记录正在被其他人编辑，暂时无法编辑'
        }), 423  # HTTP 423 Locked
    
    # 获取新锁
    EDIT_LOCKS[note_id] = {
        'session_id': session_id,
        'locked_at': now,
        'last_heartbeat': now
    }
    
    return jsonify({
        'locked': True,
        'session_id': session_id,
        'is_owner': True
    })

@app.route('/api/notes/<note_id>/unlock', methods=['POST'])
def api_release_lock(note_id):
    """释放编辑锁"""
    data = request.get_json() or {}
    session_id = data.get('session_id', '')
    
    if note_id in EDIT_LOCKS:
        lock = EDIT_LOCKS[note_id]
        
        # 只有锁持有者才能释放
        if lock['session_id'] == session_id:
            del EDIT_LOCKS[note_id]
            return jsonify({'success': True, 'message': '锁已释放'})
        
        return jsonify({'error': '非锁持有者，无法释放'}), 403
    
    return jsonify({'success': True, 'message': '锁已不存在'})

@app.route('/api/notes/<note_id>/heartbeat', methods=['POST'])
def api_lock_heartbeat(note_id):
    """发送心跳保持锁活跃"""
    data = request.get_json() or {}
    session_id = data.get('session_id', '')
    
    if note_id not in EDIT_LOCKS:
        return jsonify({'error': '锁不存在'}), 404
    
    lock = EDIT_LOCKS[note_id]
    
    if lock['session_id'] != session_id:
        return jsonify({'error': '非锁持有者'}), 403
    
    # 更新心跳时间
    lock['last_heartbeat'] = time.time()
    
    return jsonify({
        'success': True,
        'remaining_time': LOCK_TIMEOUT
    })

@app.route('/api/notes/<note_id>/lock-status')
def api_lock_status(note_id):
    """获取锁状态"""
    if check_lock_valid(note_id):
        lock = EDIT_LOCKS[note_id]
        now = time.time()
        remaining = LOCK_TIMEOUT - (now - lock['last_heartbeat'])
        
        return jsonify({
            'locked': True,
            'lock_owner': lock['session_id'],
            'locked_at': lock['locked_at'],
            'remaining_time': max(0, remaining)
        })
    
    return jsonify({'locked': False})

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)