feat: 收藏关注系统 v1.0.0 - 支持CLI/API/Web三种操作模式

This commit is contained in:
2026-04-12 01:34:13 +08:00
commit 184cc5b56b
26 changed files with 2953 additions and 0 deletions

6
xian_favor/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
"""
Xian Favor - 收藏关注系统
支持命令行、API、Web多种操作模式
"""
__version__ = "1.0.0"

7
xian_favor/__main__.py Normal file
View File

@@ -0,0 +1,7 @@
#!/usr/bin/env python3
"""命令行入口"""
from xian_favor.cli import main
if __name__ == "__main__":
main()

567
xian_favor/api.py Normal file
View File

@@ -0,0 +1,567 @@
"""API服务"""
from flask import Flask, request, jsonify, render_template_string
from flask_cors import CORS
import os
from .db import db
from .config import API_HOST, API_PORT, ITEM_TYPES, TODO_STATUS, PRIORITY_LEVELS
app = Flask(__name__,
template_folder=os.path.join(os.path.dirname(__file__), '../web/templates'),
static_folder=os.path.join(os.path.dirname(__file__), '../web/static'))
CORS(app)
# ============ API 路由 ============
@app.route('/api/items', methods=['GET'])
def list_items():
"""列出条目"""
items = db.list_items(
type=request.args.get('type'),
status=request.args.get('status'),
tag=request.args.get('tag'),
keyword=request.args.get('keyword'),
limit=int(request.args.get('limit', 50)),
offset=int(request.args.get('offset', 0))
)
return jsonify({'success': True, 'data': items})
@app.route('/api/items', methods=['POST'])
def create_item():
"""创建条目"""
data = request.get_json()
if not data:
return jsonify({'success': False, 'error': '无数据'}), 400
item_type = data.get('type', 'text')
if item_type not in ITEM_TYPES:
return jsonify({'success': False, 'error': f'无效类型: {item_type}'}), 400
try:
item_id = db.create_item(
type=item_type,
title=data.get('title'),
content=data.get('content'),
url=data.get('url'),
source=data.get('source'),
status=data.get('status', 'pending'),
priority=data.get('priority', 'medium'),
due_date=data.get('due_date'),
note=data.get('note'),
tags=data.get('tags', [])
)
item = db.get_item(item_id)
return jsonify({'success': True, 'data': item}), 201
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/items/<int:item_id>', methods=['GET'])
def get_item(item_id):
"""获取条目"""
item = db.get_item(item_id)
if not item:
return jsonify({'success': False, 'error': '条目不存在'}), 404
return jsonify({'success': True, 'data': item})
@app.route('/api/items/<int:item_id>', methods=['PUT'])
def update_item(item_id):
"""更新条目"""
data = request.get_json()
if not data:
return jsonify({'success': False, 'error': '无数据'}), 400
try:
if db.update_item(item_id, **data):
item = db.get_item(item_id)
return jsonify({'success': True, 'data': item})
else:
return jsonify({'success': False, 'error': '条目不存在或无变化'}), 404
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/items/<int:item_id>', methods=['DELETE'])
def delete_item(item_id):
"""删除条目"""
if db.delete_item(item_id):
return jsonify({'success': True})
return jsonify({'success': False, 'error': '条目不存在'}), 404
@app.route('/api/items/<int:item_id>/done', methods=['POST'])
def complete_item(item_id):
"""完成待办"""
item = db.get_item(item_id)
if not item:
return jsonify({'success': False, 'error': '条目不存在'}), 404
if item['type'] != 'todo':
return jsonify({'success': False, 'error': '不是待办事项'}), 400
db.update_item(item_id, status='completed')
item = db.get_item(item_id)
return jsonify({'success': True, 'data': item})
@app.route('/api/tags', methods=['GET'])
def list_tags():
"""列出标签"""
tags = db.list_tags()
return jsonify({'success': True, 'data': tags})
@app.route('/api/tags', methods=['POST'])
def create_tag():
"""创建标签"""
data = request.get_json()
name = data.get('name', '').strip()
if not name:
return jsonify({'success': False, 'error': '标签名不能为空'}), 400
try:
tag_id = db.create_tag(name, data.get('color', '#3498db'))
return jsonify({'success': True, 'data': {'id': tag_id, 'name': name}}), 201
except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/tags/<int:tag_id>', methods=['DELETE'])
def delete_tag(tag_id):
"""删除标签"""
if db.delete_tag(tag_id=tag_id):
return jsonify({'success': True})
return jsonify({'success': False, 'error': '标签不存在'}), 404
@app.route('/api/stats', methods=['GET'])
def get_stats():
"""获取统计"""
stats = db.stats()
return jsonify({'success': True, 'data': stats})
@app.route('/api/search', methods=['GET'])
def search_items():
"""搜索条目"""
keyword = request.args.get('q', '')
if not keyword:
return jsonify({'success': False, 'error': '请提供搜索关键词'}), 400
items = db.list_items(
keyword=keyword,
type=request.args.get('type'),
limit=int(request.args.get('limit', 50))
)
return jsonify({'success': True, 'data': items})
# ============ Web 页面 ============
@app.route('/')
def index():
"""主页"""
return render_template_string(INDEX_TEMPLATE)
# ============ Web 模板 ============
INDEX_TEMPLATE = '''
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Xian Favor - 收藏系统</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
<link href="https://cdn.jsdelivr.net/npm/bootstrap-icons@1.11.0/font/bootstrap-icons.css" rel="stylesheet">
<style>
body { background: #f8f9fa; }
.sidebar { height: 100vh; background: #343a40; color: #fff; }
.sidebar a { color: #adb5bd; text-decoration: none; padding: 10px 20px; display: block; }
.sidebar a:hover, .sidebar a.active { background: #495057; color: #fff; }
.content { padding: 20px; }
.card { margin-bottom: 15px; transition: transform 0.2s; }
.card:hover { transform: translateY(-2px); box-shadow: 0 4px 12px rgba(0,0,0,0.15); }
.tag { margin-right: 5px; }
.type-text { border-left: 4px solid #17a2b8; }
.type-link { border-left: 4px solid #28a745; }
.type-column { border-left: 4px solid #6f42c1; }
.type-todo { border-left: 4px solid #ffc107; }
.status-pending { color: #ffc107; }
.status-in_progress { color: #17a2b8; }
.status-completed { color: #28a745; text-decoration: line-through; }
.priority-low { color: #6c757d; }
.priority-medium { color: #ffc107; }
.priority-high { color: #fd7e14; }
.priority-urgent { color: #dc3545; }
</style>
</head>
<body>
<div class="container-fluid">
<div class="row">
<!-- 侧边栏 -->
<div class="col-md-2 sidebar p-0">
<div class="p-3 border-bottom border-secondary">
<h5><i class="bi bi-bookmark-star"></i> Xian Favor</h5>
</div>
<nav>
<a href="#" class="active" data-filter="all"><i class="bi bi-inbox"></i> 全部</a>
<a href="#" data-filter="text"><i class="bi bi-file-text"></i> 文本</a>
<a href="#" data-filter="link"><i class="bi bi-link-45deg"></i> 链接</a>
<a href="#" data-filter="column"><i class="bi bi-newspaper"></i> 专栏</a>
<a href="#" data-filter="todo"><i class="bi bi-check2-square"></i> 待办</a>
<hr class="border-secondary">
<a href="#" data-filter="pending"><i class="bi bi-clock"></i> 待处理</a>
<a href="#" data-filter="in_progress"><i class="bi bi-arrow-repeat"></i> 进行中</a>
<a href="#" data-filter="completed"><i class="bi bi-check-circle"></i> 已完成</a>
</nav>
</div>
<!-- 主内容 -->
<div class="col-md-10 content">
<!-- 顶部操作栏 -->
<div class="d-flex justify-content-between align-items-center mb-4">
<div class="d-flex gap-2">
<input type="text" id="searchInput" class="form-control" placeholder="搜索..." style="width: 300px;">
<select id="typeFilter" class="form-select" style="width: 120px;">
<option value="">全部类型</option>
<option value="text">文本</option>
<option value="link">链接</option>
<option value="column">专栏</option>
<option value="todo">待办</option>
</select>
</div>
<button class="btn btn-primary" data-bs-toggle="modal" data-bs-target="#addModal">
<i class="bi bi-plus-lg"></i> 添加
</button>
</div>
<!-- 统计卡片 -->
<div class="row mb-4" id="statsCards">
<div class="col-md-3">
<div class="card bg-primary text-white">
<div class="card-body">
<h6>总条目</h6>
<h3 id="statTotal">0</h3>
</div>
</div>
</div>
<div class="col-md-3">
<div class="card bg-warning text-dark">
<div class="card-body">
<h6>待处理</h6>
<h3 id="statPending">0</h3>
</div>
</div>
</div>
<div class="col-md-3">
<div class="card bg-info text-white">
<div class="card-body">
<h6>进行中</h6>
<h3 id="statProgress">0</h3>
</div>
</div>
</div>
<div class="col-md-3">
<div class="card bg-success text-white">
<div class="card-body">
<h6>已完成</h6>
<h3 id="statCompleted">0</h3>
</div>
</div>
</div>
</div>
<!-- 列表 -->
<div id="itemList"></div>
</div>
</div>
</div>
<!-- 添加模态框 -->
<div class="modal fade" id="addModal" tabindex="-1">
<div class="modal-dialog">
<div class="modal-content">
<div class="modal-header">
<h5 class="modal-title">添加条目</h5>
<button type="button" class="btn-close" data-bs-dismiss="modal"></button>
</div>
<div class="modal-body">
<form id="addForm">
<div class="mb-3">
<label class="form-label">类型</label>
<select id="addType" class="form-select">
<option value="text">📝 文本</option>
<option value="link">🔗 链接</option>
<option value="column">📰 专栏</option>
<option value="todo">✅ 待办</option>
</select>
</div>
<div class="mb-3">
<label class="form-label">标题</label>
<input type="text" id="addTitle" class="form-control" placeholder="可选">
</div>
<div class="mb-3" id="contentGroup">
<label class="form-label">内容</label>
<textarea id="addContent" class="form-control" rows="3"></textarea>
</div>
<div class="mb-3" id="urlGroup" style="display:none;">
<label class="form-label">URL</label>
<input type="url" id="addUrl" class="form-control">
</div>
<div class="mb-3" id="sourceGroup" style="display:none;">
<label class="form-label">来源</label>
<input type="text" id="addSource" class="form-control">
</div>
<div class="mb-3" id="todoFields" style="display:none;">
<div class="row">
<div class="col">
<label class="form-label">状态</label>
<select id="addStatus" class="form-select">
<option value="pending">⏳ 待处理</option>
<option value="in_progress">🔄 进行中</option>
<option value="completed">✅ 已完成</option>
</select>
</div>
<div class="col">
<label class="form-label">优先级</label>
<select id="addPriority" class="form-select">
<option value="low">🟢 低</option>
<option value="medium" selected>🟡 中</option>
<option value="high">🟠 高</option>
<option value="urgent">🔴 紧急</option>
</select>
</div>
</div>
<div class="mt-3">
<label class="form-label">截止日期</label>
<input type="date" id="addDueDate" class="form-control">
</div>
</div>
<div class="mb-3">
<label class="form-label">标签 (逗号分隔)</label>
<input type="text" id="addTags" class="form-control" placeholder="标签1, 标签2">
</div>
<div class="mb-3">
<label class="form-label">备注</label>
<input type="text" id="addNote" class="form-control">
</div>
</form>
</div>
<div class="modal-footer">
<button type="button" class="btn btn-secondary" data-bs-dismiss="modal">取消</button>
<button type="button" class="btn btn-primary" onclick="addItem()">添加</button>
</div>
</div>
</div>
</div>
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/js/bootstrap.bundle.min.js"></script>
<script>
const API_BASE = '/api';
let currentFilter = { type: '', status: '' };
// 初始化
document.addEventListener('DOMContentLoaded', () => {
loadItems();
loadStats();
// 类型切换时显示/隐藏字段
document.getElementById('addType').addEventListener('change', (e) => {
const type = e.target.value;
document.getElementById('contentGroup').style.display = type === 'text' ? 'block' : 'none';
document.getElementById('urlGroup').style.display = ['link', 'column'].includes(type) ? 'block' : 'none';
document.getElementById('sourceGroup').style.display = type === 'column' ? 'block' : 'none';
document.getElementById('todoFields').style.display = type === 'todo' ? 'block' : 'none';
});
// 搜索
document.getElementById('searchInput').addEventListener('input', debounce(loadItems, 300));
// 类型过滤
document.getElementById('typeFilter').addEventListener('change', (e) => {
currentFilter.type = e.target.value;
loadItems();
});
// 侧边栏过滤
document.querySelectorAll('.sidebar a').forEach(a => {
a.addEventListener('click', (e) => {
e.preventDefault();
document.querySelectorAll('.sidebar a').forEach(x => x.classList.remove('active'));
a.classList.add('active');
const filter = a.dataset.filter;
if (['text', 'link', 'column', 'todo'].includes(filter)) {
currentFilter = { type: filter, status: '' };
} else if (['pending', 'in_progress', 'completed'].includes(filter)) {
currentFilter = { type: 'todo', status: filter };
} else {
currentFilter = { type: '', status: '' };
}
loadItems();
});
});
});
// 加载列表
async function loadItems() {
const keyword = document.getElementById('searchInput').value;
let url = `${API_BASE}/items?limit=100`;
if (currentFilter.type) url += `&type=${currentFilter.type}`;
if (currentFilter.status) url += `&status=${currentFilter.status}`;
if (keyword) url += `&keyword=${encodeURIComponent(keyword)}`;
const res = await fetch(url);
const data = await res.json();
if (data.success) {
renderItems(data.data);
}
}
// 渲染列表
function renderItems(items) {
const container = document.getElementById('itemList');
if (!items.length) {
container.innerHTML = '<div class="text-center text-muted py-5">暂无数据</div>';
return;
}
container.innerHTML = items.map(item => `
<div class="card type-${item.type}">
<div class="card-body">
<div class="d-flex justify-content-between align-items-start">
<div>
<h6 class="card-title">
${getTypeIcon(item.type)} ${item.title || truncate(item.content || item.url, 50)}
</h6>
<p class="card-text text-muted small mb-2">
${item.url ? `<a href="${item.url}" target="_blank">${item.url}</a><br>` : ''}
${item.content ? truncate(item.content, 100) : ''}
</p>
<div>
${item.tags.map(t => `<span class="badge bg-secondary tag">${t}</span>`).join('')}
${item.type === 'todo' ? `
<span class="badge status-${item.status}">${getStatusLabel(item.status)}</span>
<span class="badge priority-${item.priority}">${getPriorityLabel(item.priority)}</span>
${item.due_date ? `<span class="badge bg-light text-dark">📅 ${item.due_date}</span>` : ''}
` : ''}
</div>
</div>
<div class="btn-group btn-group-sm">
${item.type === 'todo' && item.status !== 'completed' ?
`<button class="btn btn-outline-success" onclick="completeItem(${item.id})" title="完成"><i class="bi bi-check-lg"></i></button>` : ''}
<button class="btn btn-outline-danger" onclick="deleteItem(${item.id})" title="删除"><i class="bi bi-trash"></i></button>
</div>
</div>
<div class="text-muted small mt-2">${formatDate(item.created_at)}</div>
</div>
</div>
`).join('');
}
// 加载统计
async function loadStats() {
const res = await fetch(`${API_BASE}/stats`);
const data = await res.json();
if (data.success) {
document.getElementById('statTotal').textContent = data.data.total;
document.getElementById('statPending').textContent = data.data.todo_status?.pending || 0;
document.getElementById('statProgress').textContent = data.data.todo_status?.in_progress || 0;
document.getElementById('statCompleted').textContent = data.data.todo_status?.completed || 0;
}
}
// 添加条目
async function addItem() {
const type = document.getElementById('addType').value;
const data = {
type,
title: document.getElementById('addTitle').value,
content: type === 'text' ? document.getElementById('addContent').value : null,
url: ['link', 'column'].includes(type) ? document.getElementById('addUrl').value : null,
source: type === 'column' ? document.getElementById('addSource').value : null,
status: type === 'todo' ? document.getElementById('addStatus').value : null,
priority: type === 'todo' ? document.getElementById('addPriority').value : null,
due_date: type === 'todo' ? document.getElementById('addDueDate').value : null,
note: document.getElementById('addNote').value,
tags: document.getElementById('addTags').value.split(',').map(t => t.trim()).filter(t => t)
};
const res = await fetch(`${API_BASE}/items`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(data)
});
if (res.ok) {
bootstrap.Modal.getInstance(document.getElementById('addModal')).hide();
document.getElementById('addForm').reset();
loadItems();
loadStats();
}
}
// 完成待办
async function completeItem(id) {
await fetch(`${API_BASE}/items/${id}/done`, { method: 'POST' });
loadItems();
loadStats();
}
// 删除条目
async function deleteItem(id) {
if (!confirm('确认删除?')) return;
await fetch(`${API_BASE}/items/${id}`, { method: 'DELETE' });
loadItems();
loadStats();
}
// 工具函数
function getTypeIcon(type) {
const icons = { text: '📝', link: '🔗', column: '📰', todo: '' };
return icons[type] || '📄';
}
function getStatusLabel(status) {
const labels = { pending: '待处理', in_progress: '进行中', completed: '已完成' };
return labels[status] || status;
}
function getPriorityLabel(priority) {
const labels = { low: '', medium: '', high: '', urgent: '紧急' };
return labels[priority] || priority;
}
function truncate(str, len) {
return str && str.length > len ? str.substring(0, len) + '...' : str || '';
}
function formatDate(dateStr) {
return new Date(dateStr).toLocaleString('zh-CN');
}
function debounce(fn, delay) {
let timer;
return function(...args) {
clearTimeout(timer);
timer = setTimeout(() => fn.apply(this, args), delay);
};
}
</script>
</body>
</html>
'''
def start_server(host: str = API_HOST, port: int = API_PORT):
"""启动服务"""
app.run(host=host, port=port, debug=False)

430
xian_favor/cli.py Normal file
View File

@@ -0,0 +1,430 @@
"""命令行工具"""
import argparse
import sys
import json
from datetime import datetime
from typing import List
from .db import db
from .config import ITEM_TYPES, TODO_STATUS, PRIORITY_LEVELS
def main():
parser = argparse.ArgumentParser(
description="Xian Favor - 收藏关注系统",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
示例:
xian_favor add text "这是我的笔记" -t "笔记,重要"
xian_favor add link "https://example.com" --title "示例网站" -t "技术"
xian_favor add todo "完成项目" -p high -d "2024-12-31" -t "工作"
xian_favor add column "https://column.example.com/feed" --title "技术专栏" -t "RSS"
xian_favor list
xian_favor list --type todo --status pending
xian_favor search "关键词"
xian_favor show 1
xian_favor edit 1 --status completed --note "已完成"
xian_favor done 1
xian_favor delete 1
xian_favor tags
xian_favor stats
"""
)
subparsers = parser.add_subparsers(dest="command", help="命令")
# ============ add 命令 ============
add_parser = subparsers.add_parser("add", help="添加新条目")
add_parser.add_argument("type", choices=ITEM_TYPES, help="类型: text/link/column/todo")
add_parser.add_argument("content", help="内容或URL")
add_parser.add_argument("--title", "-T", help="标题")
add_parser.add_argument("--url", "-u", help="URL (link/column类型)")
add_parser.add_argument("--source", "-s", help="来源")
add_parser.add_argument("--status", choices=TODO_STATUS, default="pending", help="状态 (todo)")
add_parser.add_argument("--priority", "-p", choices=PRIORITY_LEVELS, default="medium", help="优先级")
add_parser.add_argument("--due-date", "-d", help="截止日期 (YYYY-MM-DD)")
add_parser.add_argument("--note", "-n", help="备注")
add_parser.add_argument("--tags", "-t", help="标签 (逗号分隔)")
# ============ list 命令 ============
list_parser = subparsers.add_parser("list", help="列出条目")
list_parser.add_argument("--type", choices=ITEM_TYPES, help="类型过滤")
list_parser.add_argument("--status", choices=TODO_STATUS, help="状态过滤")
list_parser.add_argument("--tag", help="标签过滤")
list_parser.add_argument("--limit", "-l", type=int, default=20, help="数量限制")
list_parser.add_argument("--json", "-j", action="store_true", help="JSON输出")
# ============ show 命令 ============
show_parser = subparsers.add_parser("show", help="查看详情")
show_parser.add_argument("id", type=int, help="条目ID")
show_parser.add_argument("--json", "-j", action="store_true", help="JSON输出")
# ============ edit 命令 ============
edit_parser = subparsers.add_parser("edit", help="编辑条目")
edit_parser.add_argument("id", type=int, help="条目ID")
edit_parser.add_argument("--title", "-T", help="标题")
edit_parser.add_argument("--content", "-c", help="内容")
edit_parser.add_argument("--url", "-u", help="URL")
edit_parser.add_argument("--source", "-s", help="来源")
edit_parser.add_argument("--status", choices=TODO_STATUS, help="状态")
edit_parser.add_argument("--priority", "-p", choices=PRIORITY_LEVELS, help="优先级")
edit_parser.add_argument("--due-date", "-d", help="截止日期")
edit_parser.add_argument("--note", "-n", help="备注")
edit_parser.add_argument("--tags", "-t", help="标签 (逗号分隔, 覆盖)")
# ============ done 命令 ============
done_parser = subparsers.add_parser("done", help="完成待办")
done_parser.add_argument("id", type=int, help="条目ID")
# ============ delete 命令 ============
delete_parser = subparsers.add_parser("delete", help="删除条目")
delete_parser.add_argument("id", type=int, help="条目ID")
delete_parser.add_argument("-f", "--force", action="store_true", help="强制删除不确认")
# ============ search 命令 ============
search_parser = subparsers.add_parser("search", help="搜索条目")
search_parser.add_argument("keyword", help="关键词")
search_parser.add_argument("--type", choices=ITEM_TYPES, help="类型过滤")
search_parser.add_argument("--limit", "-l", type=int, default=20, help="数量限制")
search_parser.add_argument("--json", "-j", action="store_true", help="JSON输出")
# ============ tags 命令 ============
tags_parser = subparsers.add_parser("tags", help="标签管理")
tags_parser.add_argument("--delete", "-d", help="删除标签")
# ============ stats 命令 ============
subparsers.add_parser("stats", help="统计信息")
# ============ serve 命令 ============
serve_parser = subparsers.add_parser("serve", help="启动API服务")
serve_parser.add_argument("--host", default="0.0.0.0", help="主机")
serve_parser.add_argument("--port", type=int, default=19014, help="端口")
# 解析参数
args = parser.parse_args()
if not args.command:
parser.print_help()
return
# 执行命令
try:
if args.command == "add":
cmd_add(args)
elif args.command == "list":
cmd_list(args)
elif args.command == "show":
cmd_show(args)
elif args.command == "edit":
cmd_edit(args)
elif args.command == "done":
cmd_done(args)
elif args.command == "delete":
cmd_delete(args)
elif args.command == "search":
cmd_search(args)
elif args.command == "tags":
cmd_tags(args)
elif args.command == "stats":
cmd_stats(args)
elif args.command == "serve":
cmd_serve(args)
except Exception as e:
print(f"❌ 错误: {e}", file=sys.stderr)
sys.exit(1)
# ============ 命令实现 ============
def parse_tags(tags_str: str) -> List[str]:
"""解析标签字符串"""
if not tags_str:
return []
return [t.strip() for t in tags_str.split(",") if t.strip()]
def format_item(item: dict, brief: bool = True) -> str:
"""格式化条目显示"""
type_icons = {
"text": "📝",
"link": "🔗",
"column": "📰",
"todo": ""
}
status_icons = {
"pending": "",
"in_progress": "🔄",
"completed": ""
}
priority_icons = {
"low": "🟢",
"medium": "🟡",
"high": "🟠",
"urgent": "🔴"
}
icon = type_icons.get(item['type'], "📄")
status = status_icons.get(item.get('status', ''), '')
priority = priority_icons.get(item.get('priority', ''), '')
title = item.get('title') or item.get('content', '')[:50]
tags_str = f" [{', '.join(item['tags'])}]" if item.get('tags') else ""
if brief:
return f" {icon} [{item['id']}] {title}{tags_str}"
else:
lines = [
f"{icon} [{item['id']}] {title}",
f" 类型: {item['type']}",
]
if item.get('content'):
lines.append(f" 内容: {item['content'][:200]}")
if item.get('url'):
lines.append(f" URL: {item['url']}")
if item.get('source'):
lines.append(f" 来源: {item['source']}")
if item['type'] == 'todo':
lines.append(f" 状态: {status} {item.get('status', 'pending')}")
lines.append(f" 优先级: {priority} {item.get('priority', 'medium')}")
if item.get('due_date'):
lines.append(f" 截止: {item['due_date']}")
if item.get('note'):
lines.append(f" 备注: {item['note']}")
if item.get('tags'):
lines.append(f" 标签: {', '.join(item['tags'])}")
lines.append(f" 创建: {item['created_at'][:19]}")
return "\n".join(lines)
def cmd_add(args):
"""添加条目"""
tags = parse_tags(args.tags)
# 根据类型处理内容
content = args.content
url = args.url
if args.type == "link":
url = url or args.content
content = None
elif args.type == "column":
url = url or args.content
content = None
item_id = db.create_item(
type=args.type,
title=args.title,
content=content,
url=url,
source=args.source,
status=args.status,
priority=args.priority,
due_date=args.due_date,
note=args.note,
tags=tags
)
item = db.get_item(item_id)
print(f"✅ 创建成功 (ID: {item_id})")
print(format_item(item, brief=False))
def cmd_list(args):
"""列出条目"""
items = db.list_items(
type=args.type,
status=args.status,
tag=args.tag,
limit=args.limit
)
if args.json:
print(json.dumps(items, ensure_ascii=False, indent=2))
return
if not items:
print("📭 没有找到条目")
return
type_labels = {"text": "文本", "link": "链接", "column": "专栏", "todo": "待办"}
filter_desc = []
if args.type:
filter_desc.append(f"类型: {type_labels.get(args.type, args.type)}")
if args.status:
filter_desc.append(f"状态: {args.status}")
if args.tag:
filter_desc.append(f"标签: {args.tag}")
if filter_desc:
print(f"📋 筛选: {' | '.join(filter_desc)}")
else:
print(f"📋 全部条目 ({len(items)} 条)")
print("-" * 50)
for item in items:
print(format_item(item))
def cmd_show(args):
"""查看详情"""
item = db.get_item(args.id)
if not item:
print(f"❌ 条目不存在: {args.id}")
return
if args.json:
print(json.dumps(item, ensure_ascii=False, indent=2))
return
print(format_item(item, brief=False))
def cmd_edit(args):
"""编辑条目"""
update_data = {}
if args.title is not None:
update_data['title'] = args.title
if args.content is not None:
update_data['content'] = args.content
if args.url is not None:
update_data['url'] = args.url
if args.source is not None:
update_data['source'] = args.source
if args.status is not None:
update_data['status'] = args.status
if args.priority is not None:
update_data['priority'] = args.priority
if args.due_date is not None:
update_data['due_date'] = args.due_date
if args.note is not None:
update_data['note'] = args.note
if args.tags is not None:
update_data['tags'] = parse_tags(args.tags)
if not update_data:
print("❌ 没有指定要更新的字段")
return
if db.update_item(args.id, **update_data):
item = db.get_item(args.id)
print(f"✅ 更新成功 (ID: {args.id})")
print(format_item(item, brief=False))
else:
print(f"❌ 更新失败: 条目不存在或没有变化")
def cmd_done(args):
"""完成待办"""
item = db.get_item(args.id)
if not item:
print(f"❌ 条目不存在: {args.id}")
return
if item['type'] != 'todo':
print(f"❌ 不是待办事项: {args.id}")
return
if db.update_item(args.id, status='completed'):
print(f"✅ 已完成待办 (ID: {args.id})")
else:
print(f"❌ 操作失败")
def cmd_delete(args):
"""删除条目"""
if not args.force:
item = db.get_item(args.id)
if not item:
print(f"❌ 条目不存在: {args.id}")
return
print(format_item(item, brief=False))
confirm = input("确认删除? [y/N] ")
if confirm.lower() != 'y':
print("❌ 取消删除")
return
if db.delete_item(args.id):
print(f"✅ 已删除 (ID: {args.id})")
else:
print(f"❌ 删除失败")
def cmd_search(args):
"""搜索条目"""
items = db.list_items(
type=args.type,
keyword=args.keyword,
limit=args.limit
)
if args.json:
print(json.dumps(items, ensure_ascii=False, indent=2))
return
if not items:
print(f"🔍 没有找到匹配 '{args.keyword}' 的条目")
return
print(f"🔍 搜索 '{args.keyword}' ({len(items)} 条)")
print("-" * 50)
for item in items:
print(format_item(item))
def cmd_tags(args):
"""标签管理"""
if args.delete:
if db.delete_tag(name=args.delete):
print(f"✅ 已删除标签: {args.delete}")
else:
print(f"❌ 标签不存在: {args.delete}")
return
tags = db.list_tags()
if not tags:
print("🏷️ 没有标签")
return
print(f"🏷️ 标签列表 ({len(tags)} 个)")
print("-" * 50)
for tag in tags:
print(f"{tag['name']} ({tag['item_count']} 条)")
def cmd_stats(args):
"""统计信息"""
stats = db.stats()
print("📊 收藏统计")
print("-" * 30)
print(f"总条目: {stats['total']}")
if stats.get('by_type'):
print("\n按类型:")
type_labels = {"text": "文本", "link": "链接", "column": "专栏", "todo": "待办"}
for t, count in stats['by_type'].items():
print(f"{type_labels.get(t, t)}: {count}")
if stats.get('todo_status'):
print("\n待办状态:")
status_labels = {"pending": "待处理", "in_progress": "进行中", "completed": "已完成"}
for s, count in stats['todo_status'].items():
print(f"{status_labels.get(s, s)}: {count}")
print(f"\n标签数: {stats['tags']}")
def cmd_serve(args):
"""启动API服务"""
from .api import start_server
print(f"🚀 启动API服务: http://{args.host}:{args.port}")
start_server(host=args.host, port=args.port)
if __name__ == "__main__":
main()

26
xian_favor/config.py Normal file
View File

@@ -0,0 +1,26 @@
"""配置文件"""
import os
from pathlib import Path
# 数据目录 - 使用用户可访问的路径
# 默认在 ~/.xian_favor/ 目录下
DEFAULT_DATA_DIR = Path.home() / ".xian_favor"
DATA_DIR = Path(os.getenv("XIAN_FAVOR_DATA_DIR", str(DEFAULT_DATA_DIR)))
DATA_DIR.mkdir(parents=True, exist_ok=True)
# 数据库
DATABASE_URL = os.getenv("XIAN_FAVOR_DB", str(DATA_DIR / "xian_favor.db"))
# API服务
API_HOST = os.getenv("XIAN_FAVOR_HOST", "0.0.0.0")
API_PORT = int(os.getenv("XIAN_FAVOR_PORT", "19014"))
# 内容类型
ITEM_TYPES = ["text", "link", "column", "todo"]
# 待办状态
TODO_STATUS = ["pending", "in_progress", "completed"]
# 优先级
PRIORITY_LEVELS = ["low", "medium", "high", "urgent"]

321
xian_favor/db.py Normal file
View File

@@ -0,0 +1,321 @@
"""数据库操作"""
import sqlite3
import json
from datetime import datetime
from typing import Optional, List, Dict, Any
from contextlib import contextmanager
from .config import DATABASE_URL, TODO_STATUS, PRIORITY_LEVELS
class Database:
"""SQLite数据库管理"""
def __init__(self, db_path: str = DATABASE_URL):
self.db_path = db_path
self._initialized = False
def _ensure_init(self):
"""确保数据库已初始化"""
if self._initialized:
return
self._init_db()
self._initialized = True
@contextmanager
def get_conn(self):
"""获取数据库连接"""
conn = sqlite3.connect(self.db_path, timeout=30.0)
conn.row_factory = sqlite3.Row
# 启用WAL模式提高并发性能
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA busy_timeout=30000")
try:
yield conn
finally:
conn.close()
def _init_db(self):
"""初始化数据库表"""
with self.get_conn() as conn:
cursor = conn.cursor()
# 主内容表
cursor.execute("""
CREATE TABLE IF NOT EXISTS items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
type TEXT NOT NULL DEFAULT 'text',
title TEXT,
content TEXT,
url TEXT,
source TEXT,
status TEXT DEFAULT 'pending',
priority TEXT DEFAULT 'medium',
due_date TEXT,
note TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""")
# 标签表
cursor.execute("""
CREATE TABLE IF NOT EXISTS tags (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
color TEXT DEFAULT '#3498db',
created_at TEXT NOT NULL
)
""")
# 内容-标签关联表
cursor.execute("""
CREATE TABLE IF NOT EXISTS item_tags (
item_id INTEGER NOT NULL,
tag_id INTEGER NOT NULL,
PRIMARY KEY (item_id, tag_id),
FOREIGN KEY (item_id) REFERENCES items(id) ON DELETE CASCADE,
FOREIGN KEY (tag_id) REFERENCES tags(id) ON DELETE CASCADE
)
""")
# 创建索引
cursor.execute("CREATE INDEX IF NOT EXISTS idx_items_type ON items(type)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_items_status ON items(status)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_items_created ON items(created_at)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_item_tags_item ON item_tags(item_id)")
cursor.execute("CREATE INDEX IF NOT EXISTS idx_item_tags_tag ON item_tags(tag_id)")
conn.commit()
# ============ Item 操作 ============
def create_item(self, type: str = "text", title: str = None, content: str = None,
url: str = None, source: str = None, status: str = "pending",
priority: str = "medium", due_date: str = None, note: str = None,
tags: List[str] = None) -> int:
"""创建新条目"""
self._ensure_init()
now = datetime.now().isoformat()
# 验证状态
if type == "todo" and status not in TODO_STATUS:
status = "pending"
if priority not in PRIORITY_LEVELS:
priority = "medium"
with self.get_conn() as conn:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO items (type, title, content, url, source, status, priority, due_date, note, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (type, title, content, url, source, status, priority, due_date, note, now, now))
item_id = cursor.lastrowid
# 添加标签
if tags:
self._add_tags_to_item(conn, item_id, tags)
conn.commit()
return item_id
def get_item(self, item_id: int) -> Optional[Dict[str, Any]]:
"""获取单个条目"""
with self.get_conn() as conn:
cursor = conn.cursor()
cursor.execute("SELECT * FROM items WHERE id = ?", (item_id,))
row = cursor.fetchone()
if not row:
return None
item = dict(row)
item['tags'] = self._get_item_tags(conn, item_id)
return item
def list_items(self, type: str = None, status: str = None, tag: str = None,
keyword: str = None, limit: int = 50, offset: int = 0) -> List[Dict[str, Any]]:
"""列出条目"""
with self.get_conn() as conn:
cursor = conn.cursor()
query = "SELECT DISTINCT i.* FROM items i"
params = []
conditions = []
# 标签过滤需要JOIN
if tag:
query += " JOIN item_tags it ON i.id = it.item_id JOIN tags t ON it.tag_id = t.id"
conditions.append("t.name = ?")
params.append(tag)
if type:
conditions.append("i.type = ?")
params.append(type)
if status:
conditions.append("i.status = ?")
params.append(status)
if keyword:
conditions.append("(i.title LIKE ? OR i.content LIKE ? OR i.note LIKE ?)")
keyword_pattern = f"%{keyword}%"
params.extend([keyword_pattern, keyword_pattern, keyword_pattern])
if conditions:
query += " WHERE " + " AND ".join(conditions)
query += " ORDER BY i.created_at DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
cursor.execute(query, params)
items = []
for row in cursor.fetchall():
item = dict(row)
item['tags'] = self._get_item_tags(conn, item['id'])
items.append(item)
return items
def update_item(self, item_id: int, **kwargs) -> bool:
"""更新条目"""
allowed_fields = ['type', 'title', 'content', 'url', 'source', 'status', 'priority', 'due_date', 'note']
update_fields = {k: v for k, v in kwargs.items() if k in allowed_fields}
if not update_fields and 'tags' not in kwargs:
return False
now = datetime.now().isoformat()
with self.get_conn() as conn:
cursor = conn.cursor()
if update_fields:
set_clause = ", ".join(f"{k} = ?" for k in update_fields.keys())
set_clause += ", updated_at = ?"
values = list(update_fields.values()) + [now, item_id]
cursor.execute(f"UPDATE items SET {set_clause} WHERE id = ?", values)
if 'tags' in kwargs:
# 先删除旧标签关联
cursor.execute("DELETE FROM item_tags WHERE item_id = ?", (item_id,))
# 添加新标签
if kwargs['tags']:
self._add_tags_to_item(conn, item_id, kwargs['tags'])
conn.commit()
return cursor.rowcount > 0
def delete_item(self, item_id: int) -> bool:
"""删除条目"""
with self.get_conn() as conn:
cursor = conn.cursor()
cursor.execute("DELETE FROM items WHERE id = ?", (item_id,))
conn.commit()
return cursor.rowcount > 0
# ============ Tag 操作 ============
def create_tag(self, name: str, color: str = "#3498db") -> int:
"""创建标签"""
now = datetime.now().isoformat()
with self.get_conn() as conn:
cursor = conn.cursor()
try:
cursor.execute("INSERT INTO tags (name, color, created_at) VALUES (?, ?, ?)",
(name, color, now))
conn.commit()
return cursor.lastrowid
except sqlite3.IntegrityError:
# 标签已存在
cursor.execute("SELECT id FROM tags WHERE name = ?", (name,))
return cursor.fetchone()['id']
def list_tags(self) -> List[Dict[str, Any]]:
"""列出所有标签"""
with self.get_conn() as conn:
cursor = conn.cursor()
cursor.execute("""
SELECT t.*, COUNT(it.item_id) as item_count
FROM tags t
LEFT JOIN item_tags it ON t.id = it.tag_id
GROUP BY t.id
ORDER BY t.name
""")
return [dict(row) for row in cursor.fetchall()]
def delete_tag(self, tag_id: int = None, name: str = None) -> bool:
"""删除标签"""
with self.get_conn() as conn:
cursor = conn.cursor()
if name:
cursor.execute("DELETE FROM tags WHERE name = ?", (name,))
elif tag_id:
cursor.execute("DELETE FROM tags WHERE id = ?", (tag_id,))
conn.commit()
return cursor.rowcount > 0
# ============ 辅助方法 ============
def _add_tags_to_item(self, conn, item_id: int, tags: List[str]):
"""为条目添加标签"""
cursor = conn.cursor()
for tag_name in tags:
tag_name = tag_name.strip()
if not tag_name:
continue
# 确保标签存在 - 使用同一个连接
cursor.execute("SELECT id FROM tags WHERE name = ?", (tag_name,))
row = cursor.fetchone()
if row:
tag_id = row['id']
else:
# 创建新标签
now = datetime.now().isoformat()
cursor.execute("INSERT INTO tags (name, color, created_at) VALUES (?, '#3498db', ?)",
(tag_name, now))
tag_id = cursor.lastrowid
# 创建关联
cursor.execute("INSERT OR IGNORE INTO item_tags (item_id, tag_id) VALUES (?, ?)",
(item_id, tag_id))
def _get_item_tags(self, conn, item_id: int) -> List[str]:
"""获取条目的标签"""
cursor = conn.cursor()
cursor.execute("""
SELECT t.name FROM tags t
JOIN item_tags it ON t.id = it.tag_id
WHERE it.item_id = ?
ORDER BY t.name
""", (item_id,))
return [row['name'] for row in cursor.fetchall()]
def stats(self) -> Dict[str, Any]:
"""获取统计信息"""
self._ensure_init()
with self.get_conn() as conn:
cursor = conn.cursor()
stats = {}
# 总数
cursor.execute("SELECT COUNT(*) as count FROM items")
stats['total'] = cursor.fetchone()['count']
# 按类型统计
cursor.execute("SELECT type, COUNT(*) as count FROM items GROUP BY type")
stats['by_type'] = {row['type']: row['count'] for row in cursor.fetchall()}
# 待办状态统计
cursor.execute("SELECT status, COUNT(*) as count FROM items WHERE type = 'todo' GROUP BY status")
stats['todo_status'] = {row['status']: row['count'] for row in cursor.fetchall()}
# 标签数
cursor.execute("SELECT COUNT(*) as count FROM tags")
stats['tags'] = cursor.fetchone()['count']
return stats
# 全局数据库实例
db = Database()