"""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/', methods=['GET']) def get_item(item_id): """获取条目""" item = db.get_item(item_id) if not item: return jsonify({'success': False, 'error': '条目不存在'}), 404 # 获取邮件发送历史 email_logs = db.get_email_logs(item_id) item['email_logs'] = email_logs return jsonify({'success': True, 'data': item}) @app.route('/api/items/', 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/', 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//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/items//reopen', methods=['POST']) def reopen_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 # 获取请求中的目标状态,默认为 pending data = request.get_json() or {} new_status = data.get('status', 'pending') if new_status not in ['pending', 'in_progress']: return jsonify({'success': False, 'error': '无效状态'}), 400 db.update_item(item_id, status=new_status) item = db.get_item(item_id) return jsonify({'success': True, 'data': item}) @app.route('/api/items//convert', methods=['POST']) def convert_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 data = request.get_json() or {} mode = data.get('mode', 'convert') # convert 或 copy if mode not in ['convert', 'copy']: return jsonify({'success': False, 'error': '无效转换模式'}), 400 # 构建待办数据 todo_data = { 'type': 'todo', 'title': data.get('title') or item['title'] or f'{item["type"]}收藏', 'status': data.get('status', 'pending'), 'priority': data.get('priority', 'medium'), 'due_date': data.get('due_date'), 'tags': item['tags'], # 继承原标签 } # 根据原类型处理内容 if item['type'] == 'text': # 文本:content 放到 note todo_data['note'] = item['content'] or '' if item['note']: todo_data['note'] += '\n\n' + item['note'] elif item['type'] == 'link': # 链接:url 放到 note,保留链接可点击 todo_data['note'] = f'链接: {item["url"]}\n\n' if item['content']: todo_data['note'] += item['content'] + '\n\n' if item['note']: todo_data['note'] += item['note'] # url 字段也保留(方便后续操作) todo_data['content'] = item['url'] elif item['type'] == 'column': # 专栏:url + source 放到 note todo_data['note'] = f'专栏: {item["url"]}\n' if item['source']: todo_data['note'] += f'来源: {item["source"]}\n\n' if item['content']: todo_data['note'] += item['content'] + '\n\n' if item['note']: todo_data['note'] += item['note'] todo_data['content'] = item['url'] if mode == 'convert': # 直接转换:更新原条目 db.update_item(item_id, **todo_data) result = db.get_item(item_id) return jsonify({'success': True, 'data': result, 'mode': 'convert'}) else: # 复制创建:新建待办,原条目保留 new_id = db.create_item(**todo_data) result = db.get_item(new_id) return jsonify({'success': True, 'data': result, 'mode': 'copy', 'original_id': item_id}) @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/', methods=['PUT']) def update_tag(tag_id): """更新标签""" data = request.get_json() name = data.get('name', '').strip() if not name: return jsonify({'success': False, 'error': '标签名不能为空'}), 400 try: if db.update_tag(tag_id, name): return jsonify({'success': True, 'data': {'id': tag_id, 'name': name}}) return jsonify({'success': False, 'error': '标签不存在或名称已存在'}), 404 except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/tags/', 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/reminders', methods=['GET']) def get_reminders(): """获取提醒信息""" reminders = db.get_reminders() return jsonify({'success': True, 'data': reminders}) @app.route('/api/ai-process', methods=['POST']) def ai_process(): """AI处理文本""" import requests data = request.get_json() text = data.get('text', '').strip() if not text: return jsonify({'success': False, 'error': '请输入文本内容'}), 400 # 大模型配置 llm_url = "http://192.168.2.17:19007/v1/chat/completions" llm_key = "xxxx" prompt = f"""请分析以下文本内容,识别其类型并提取关键信息。 文本内容: {text} 请按以下JSON格式返回结果(只返回JSON,不要其他内容): {{ "type": "text/link/column/todo", "title": "提取的标题(简短概括)", "content": "主要内容(如果是文本类型)", "url": "如果是链接或专栏,提取URL", "source": "如果是专栏,提取来源", "tags": ["相关标签1", "标签2"], "note": "补充说明或备注", "status": "如果是待办,默认pending", "priority": "如果是待办,默认medium" }} 类型判断规则: - link: 包含http/https链接,且不是专栏订阅地址 - column: 专栏订阅地址或RSS链接 - todo: 包含任务、待办、提醒等关键词 - text: 其他文本内容""" try: response = requests.post( llm_url, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {llm_key}" }, json={ "model": "auto", "messages": [{"role": "user", "content": prompt}], "temperature": 0.3 }, timeout=30 ) if response.status_code != 200: return jsonify({'success': False, 'error': f'模型调用失败: {response.status_code}'}), 500 result = response.json() content = result['choices'][0]['message']['content'] # 解析JSON import json import re # 提取JSON部分 json_match = re.search(r'\{.*\}', content, re.DOTALL) if json_match: parsed = json.loads(json_match.group()) return jsonify({'success': True, 'data': parsed}) else: return jsonify({'success': False, 'error': '无法解析模型返回'}), 500 except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @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}) # ============ 邮箱管理 API ============ @app.route('/api/emails', methods=['GET']) def list_emails(): """列出所有邮箱""" emails = db.list_emails() return jsonify({'success': True, 'data': emails}) @app.route('/api/emails', methods=['POST']) def create_email(): """创建邮箱""" data = request.get_json() email_addr = data.get('email', '').strip() if not email_addr: return jsonify({'success': False, 'error': '邮箱地址不能为空'}), 400 # 验证邮箱格式 import re if not re.match(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$', email_addr): return jsonify({'success': False, 'error': '邮箱格式不正确'}), 400 try: email_id = db.create_email(email_addr, data.get('name')) return jsonify({'success': True, 'data': {'id': email_id, 'email': email_addr}}), 201 except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/emails/', methods=['PUT']) def update_email(email_id): """更新邮箱""" data = request.get_json() try: if db.update_email(email_id, email=data.get('email'), name=data.get('name')): email = db.get_email(email_id) return jsonify({'success': True, 'data': email}) return jsonify({'success': False, 'error': '邮箱不存在或地址已存在'}), 404 except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/emails/', methods=['DELETE']) def delete_email(email_id): """删除邮箱""" if db.delete_email(email_id): return jsonify({'success': True}) return jsonify({'success': False, 'error': '邮箱不存在'}), 404 @app.route('/api/send-email', methods=['POST']) def send_email(): """发送收藏内容到邮箱""" data = request.get_json() item_id = data.get('item_id') email_addr = data.get('email', '').strip() if not item_id or not email_addr: return jsonify({'success': False, 'error': '缺少参数'}), 400 # 获取收藏内容 item = db.get_item(item_id) if not item: return jsonify({'success': False, 'error': '收藏不存在'}), 404 # 如果是新邮箱,自动保存 import re if re.match(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$', email_addr): db.create_email(email_addr) # 构建邮件内容 type_labels = {'text': '文本笔记', 'link': '链接收藏', 'column': '专栏订阅', 'todo': '待办事项'} subject = f"【Xian Favor】{item['title'] or type_labels.get(item['type'], '收藏')}" body_lines = [ f"类型: {type_labels.get(item['type'], item['type'])}", f"标题: {item['title'] or '(无标题)'}", "" ] if item['url']: body_lines.append(f"链接: {item['url']}") body_lines.append("") if item['content']: body_lines.append("内容:") body_lines.append(item['content']) body_lines.append("") if item['source']: body_lines.append(f"来源: {item['source']}") body_lines.append("") if item['type'] == 'todo': status_labels = {'pending': '待处理', 'in_progress': '进行中', 'completed': '已完成'} priority_labels = {'low': '低', 'medium': '中', 'high': '高', 'urgent': '紧急'} body_lines.append(f"状态: {status_labels.get(item['status'], item['status'])}") body_lines.append(f"优先级: {priority_labels.get(item['priority'], item['priority'])}") if item['due_date']: body_lines.append(f"截止日期: {item['due_date']}") body_lines.append("") if item['tags']: body_lines.append(f"标签: {', '.join(item['tags'])}") body_lines.append("") if item['note']: body_lines.append("详情/备注:") body_lines.append(item['note']) body_lines.append("") body_lines.append(f"创建时间: {item['created_at']}") body_lines.append("---") body_lines.append("来自 Xian Favor 收藏系统") body = "\n".join(body_lines) # 调用邮件发送 try: import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from datetime import datetime # SMTP配置:端口587无SSL smtp_host = 'mail.tphai.com' smtp_port = 587 smtp_user = 'favor@tphai.com' smtp_pass = 'favor@!' msg = MIMEMultipart() msg['From'] = smtp_user msg['To'] = email_addr msg['Subject'] = subject msg['Date'] = datetime.now().strftime('%a, %d %b %Y %H:%M:%S +0800') msg['Reply-To'] = email_addr msg.attach(MIMEText(body, 'plain', 'utf-8')) # 直接连接,无SSL server = smtplib.SMTP(smtp_host, smtp_port) server.ehlo() server.login(smtp_user, smtp_pass) server.sendmail(smtp_user, email_addr, msg.as_string()) server.quit() # 记录发送日志 db.log_email_send(item_id, email_addr, success=True) return jsonify({'success': True, 'message': f'已发送到 {email_addr}'}) except Exception as e: # 记录失败日志 db.log_email_send(item_id, email_addr, success=False) return jsonify({'success': False, 'error': f'发送失败: {str(e)}'}), 500 # ============ Web 页面 ============ @app.route('/') def index(): """主页""" return render_template_string(INDEX_TEMPLATE) # ============ Web 模板 ============ INDEX_TEMPLATE = ''' Xian Favor - 收藏系统
总条目

0

待处理

0

进行中

0

已完成

0

''' def start_server(host: str = API_HOST, port: int = API_PORT): """启动服务""" app.run(host=host, port=port, debug=False)