From cb4b7d53638ecb4945123bdc360f4f11220966ca Mon Sep 17 00:00:00 2001 From: hubian <908234780@qq.com> Date: Sun, 12 Apr 2026 16:56:35 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20v1.1.0=20=E5=AE=89=E5=85=A8=E9=87=8D?= =?UTF-8?q?=E6=9E=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 后台添加登录验证（Session + JWT双重验证） - JSON存储改为SQLite数据库，解决并发问题 - API密钥移至config.py，支持环境变量覆盖 - SECRET_KEY改为随机生成 - 新增管理员登录页面 - 修复README.md乱码 - 更新.gitignore忽略敏感配置 --- .gitignore | 6 +- README.md | 77 +++++- admin/app.py | 343 +++++++++++++------------- admin/templates/login.html | 116 +++++++++ backend/app.py | 481 +++++++++++-------------------------- config.py | 25 ++ models.py | 442 ++++++++++++++++++++++++++++++++++ requirements.txt | 3 +- 8 files changed, 972 insertions(+), 521 deletions(-) create mode 100644 admin/templates/login.html create mode 100644 config.py create mode 100644 models.py diff --git a/.gitignore b/.gitignore index ed178cc..4293ff5 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ __pycache__/ # 数据 data/*.json +data/*.db !data/.gitkeep # 上传文件 @@ -12,4 +13,7 @@ uploads/* # 环境 venv/ -.env \ No newline at end of file +.env + +# 本地配置（敏感信息） +config.local.py \ No newline at end of file diff --git a/README.md b/README.md index 41f3dd6..f8ade80 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,13 @@ - 手机号（可选） - 密码确认 +### 🔐 后台管理 +- 管理员登录验证 +- 用户管理 +- 帖子管理（置顶、删除） +- 主题管理 +- 数据统计 + ## 快速开始 ### 安装依赖 @@ -30,6 +37,20 @@ pip install -r requirements.txt ``` +### 配置（可选） + +可以创建 `config.local.py` 覆盖默认配置： + +```python +# 管理员账户 +ADMIN_USERNAME = 'your_admin' +ADMIN_PASSWORD = 'your_password' + +# 或使用环境变量 +export TECH_FORUM_ADMIN_USER='your_admin' +export TECH_FORUM_ADMIN_PASS='your_password' +``` + ### 启动主服务 ```bash @@ -46,12 +67,16 @@ python admin/app.py 后台地址: http://localhost:19005 +默认账号: admin / admin123 + ## 项目结构 ``` tech-forum/ +├── config.py # 配置文件 +├── models.py # 数据库模型（SQLite） ├── backend/ -│ └── app.py # Flask后端 +│ └── app.py # Flask后端API ├── frontend/ │ ├── index.html # 首页 │ ├── login.html # 登录 @@ -60,12 +85,17 @@ tech-forum/ │ ├── post.html # 帖子详情 │ ├── topic.html # 主题详情 │ └── user.html # 用户主页 +├── admin/ +│ ├── app.py # 后台管理（带登录验证） +│ └── templates/ +│ ├── index.html # 仪表盘 +│ ├── login.html # 管理员登录 +│ ├── users.html # 用户管理 +│ ├── posts.html # 帖子管理 +│ └── topics.html # 主题管理 ├── data/ -│ ├── users.json # 用户数据 -│ ├── posts.json # 帖子数据 -│ └── topics.json # 主题数据 -├── uploads/ # 上传文件 -└── README.md +│ └ tech_forum.db # SQLite数据库（自动创建） +└── uploads/ # 上传文件 ``` ## API接口 @@ -95,8 +125,38 @@ tech-forum/ - GET /api/tags - 获取热门标签 - GET /api/search - 搜索 +## 安全改进 + +### v1.1.0 重构内容 + +1. **后台登录验证** + - 所有后台API需要管理员登录 + - Session + JWT双重验证 + - 未登录自动跳转登录页 + +2. **配置文件分离** + - 敏感信息移至 `config.py` + - 支持环境变量覆盖 + - SECRET_KEY 自动随机生成 + +3. **SQLite数据库** + - 替换JSON文件存储 + - 解决并发写入问题 + - 数据关系完整性 + +4. **API密钥保护** + - LLM密钥从代码移至配置 + - 支持环境变量设置 + ## 版本历史 +### v1.1.0 (2026-04-12) +- 重构：后台添加登录验证 +- 重构：JSON存储改为SQLite数据库 +- 重构：API密钥移至配置文件 +- 修复：SECRET_KEY改为随机生成 +- 新增：管理员登录页面 + ### v0.1.0 (2026-04-08) - 初始版本 - 技术交流帖子功能 @@ -106,9 +166,4 @@ tech-forum/ ## License -MIT录 -- 评论回复点赞 - -## License - MIT \ No newline at end of file diff --git a/admin/app.py b/admin/app.py index 969cd6c..351932c 100644 --- a/admin/app.py +++ b/admin/app.py @@ -1,90 +1,156 @@ """ -技术论坛 - 后台管理系统 +技术论坛 - 后台管理系统 (重构版 + 登录验证) """ -from flask import Flask, render_template, jsonify, request +from flask import Flask, render_template, jsonify, request, redirect, url_for, session, make_response from flask_cors import CORS -import json +import jwt +import datetime +import os +from functools import wraps from pathlib import Path -from datetime import datetime + +# 导入配置和模型 +import sys +sys.path.insert(0, str(Path(__file__).parent.parent)) +from config import SECRET_KEY, ADMIN_USERNAME, ADMIN_PASSWORD, DATABASE_PATH, ADMIN_PORT +from models import Database, UserModel, PostModel, ReplyModel, TopicModel app = Flask(__name__) CORS(app) +app.secret_key = SECRET_KEY -# 数据目录 -DATA_DIR = Path(__file__).parent.parent / 'data' -USERS_FILE = DATA_DIR / 'users.json' -POSTS_FILE = DATA_DIR / 'posts.json' -TOPICS_FILE = DATA_DIR / 'topics.json' +# 初始化数据库 +db = Database(DATABASE_PATH) +user_model = UserModel(db) +post_model = PostModel(db) +reply_model = ReplyModel(db) +topic_model = TopicModel(db) -def load_users(): - if USERS_FILE.exists(): - return json.loads(USERS_FILE.read_text(encoding='utf-8')) - return {} +# ============ 登录验证装饰器 ============ -def load_posts(): - if POSTS_FILE.exists(): - return json.loads(POSTS_FILE.read_text(encoding='utf-8')) - return {} +def admin_required(f): + @wraps(f) + def decorated_function(*args, **kwargs): + # 检查 session + if not session.get('admin_logged_in'): + # 检查 Authorization header + token = request.headers.get('Authorization', '').replace('Bearer ', '') + if token: + try: + data = jwt.decode(token, SECRET_KEY, algorithms=['HS256']) + if data.get('admin'): + return f(*args, **kwargs) + except: + pass + + # API请求返回401，页面请求跳转登录 + if request.path.startswith('/api/'): + return jsonify({'error': '请先登录', 'code': 401}), 401 + return redirect('/login') + return f(*args, **kwargs) + return decorated_function -def load_topics(): - if TOPICS_FILE.exists(): - return json.loads(TOPICS_FILE.read_text(encoding='utf-8')) - return {} +# ============ 登录相关 ============ -def save_users(users): - USERS_FILE.write_text(json.dumps(users, ensure_ascii=False, indent=2), encoding='utf-8') +@app.route('/login') +def login_page(): + return render_template('login.html') -def save_posts(posts): - POSTS_FILE.write_text(json.dumps(posts, ensure_ascii=False, indent=2), encoding='utf-8') +@app.route('/api/login', methods=['POST']) +def api_login(): + data = request.json + + username = data.get('username', '').strip() + password = data.get('password', '') + + if not username or not password: + return jsonify({'error': '请输入用户名和密码'}), 400 + + if username != ADMIN_USERNAME or password != ADMIN_PASSWORD: + return jsonify({'error': '用户名或密码错误'}), 400 + + # 设置session + session['admin_logged_in'] = True + session['admin_username'] = username + + # 生成token（可选，用于API调用） + token = jwt.encode({ + 'admin': True, + 'username': username, + 'exp': datetime.datetime.utcnow() + datetime.timedelta(hours=24) + }, SECRET_KEY, algorithm='HS256') + + return jsonify({ + 'success': True, + 'token': token, + 'message': '登录成功' + }) -def save_topics(topics): - TOPICS_FILE.write_text(json.dumps(topics, ensure_ascii=False, indent=2), encoding='utf-8') +@app.route('/api/logout', methods=['POST']) +def api_logout(): + session.pop('admin_logged_in', None) + session.pop('admin_username', None) + return jsonify({'success': True, 'message': '已退出登录'}) + +@app.route('/api/check-auth') +def api_check_auth(): + if session.get('admin_logged_in'): + return jsonify({ + 'logged_in': True, + 'username': session.get('admin_username') + }) + return jsonify({'logged_in': False}) # ============ 页面路由 ============ @app.route('/') +@admin_required def index(): return render_template('index.html') @app.route('/users') +@admin_required def users_page(): return render_template('users.html') @app.route('/posts') +@admin_required def posts_page(): return render_template('posts.html') @app.route('/topics') +@admin_required def topics_page(): return render_template('topics.html') # ============ API路由 ============ @app.route('/api/stats') +@admin_required def api_stats(): - users = load_users() - posts = load_posts() - topics = load_topics() + users = user_model.get_all() + posts, posts_total = post_model.get_all() + topics = topic_model.get_all() - # 统计 - total_messages = 0 - for post in posts.values(): - total_messages += len(post.get('replies', [])) + # 统计回复数 + total_replies = 0 + for post in posts: + total_replies += len(reply_model.get_by_post(post['id'])) - today = datetime.now().strftime('%Y-%m-%d') - today_posts = sum(1 for p in posts.values() if p.get('created_at', '').startswith(today)) - today_users = sum(1 for u in users.values() if u.get('created_at', '').startswith(today)) + today = datetime.datetime.now().strftime('%Y-%m-%d') + today_posts = sum(1 for p in posts if p.get('created_at', '').startswith(today)) + today_users = sum(1 for u in users if u.get('created_at', '').startswith(today)) # 帖子类型统计 - discussion_count = sum(1 for p in posts.values() if p.get('type') == 'discussion') - share_count = sum(1 for p in posts.values() if p.get('type') == 'share') + discussion_count = sum(1 for p in posts if p.get('type') == 'discussion') + share_count = sum(1 for p in posts if p.get('type') == 'share') return jsonify({ 'users_count': len(users), - 'posts_count': len(posts), + 'posts_count': posts_total, 'topics_count': len(topics), - 'messages_count': total_messages, + 'messages_count': total_replies, 'today_posts': today_posts, 'today_users': today_users, 'discussion_count': discussion_count, @@ -92,110 +158,74 @@ def api_stats(): }) @app.route('/api/users') +@admin_required def api_users(): - users = load_users() - posts = load_posts() + users = user_model.get_all() user_list = [] - for uid, user in users.items(): - # 统计用户帖子和回复数 - posts_count = len(user.get('posts', [])) - replies_count = 0 - for post in posts.values(): - for reply in post.get('replies', []): - if reply.get('author_id') == uid: - replies_count += 1 - + for user in users: user_list.append({ - 'id': uid, + 'id': user['id'], 'username': user.get('username', ''), 'email': user.get('email', ''), 'phone': user.get('phone', ''), - 'posts_count': posts_count, - 'replies_count': replies_count, + 'posts_count': user_model.get_posts_count(user['id']), + 'replies_count': user_model.get_replies_count(user['id']), 'created_at': user.get('created_at', ''), }) - user_list.sort(key=lambda x: x['created_at'], reverse=True) return jsonify(user_list) @app.route('/api/users/', methods=['DELETE']) +@admin_required def api_delete_user(user_id): - users = load_users() - posts = load_posts() - topics = load_topics() - - if user_id not in users: - return jsonify({'error': '用户不存在'}), 404 - - # 删除用户的帖子 - for post_id in users[user_id].get('posts', []): - if post_id in posts: - del posts[post_id] - - # 从主题关注中移除 - for topic in topics.values(): - if user_id in topic.get('followers', []): - topic['followers'].remove(user_id) - - # 删除用户 - del users[user_id] - - save_users(users) - save_posts(posts) - save_topics(topics) - + user_model.delete(user_id) return jsonify({'success': True}) @app.route('/api/posts') +@admin_required def api_posts(): - posts = load_posts() - users = load_users() - post_type = request.args.get('type') + posts, total = post_model.get_all(post_type) + post_list = [] - for pid, post in posts.items(): - if post_type and post['type'] != post_type: - continue - - author = users.get(post['author_id'], {}) + for post in posts: + author = user_model.get_by_id(post['author_id']) or {} post_list.append({ - 'id': pid, + 'id': post['id'], 'title': post['title'], 'type': post['type'], 'author': author.get('username', '未知'), 'author_id': post['author_id'], - 'likes': len(post.get('likes', [])), - 'replies': len(post.get('replies', [])), - 'views': post.get('views', 0), - 'is_pinned': post.get('is_pinned', False), + 'likes': len(post['likes']), + 'replies': len(reply_model.get_by_post(post['id'])), + 'views': post['views'], + 'is_pinned': post['is_pinned'], 'created_at': post['created_at'], }) - post_list.sort(key=lambda x: x['created_at'], reverse=True) return jsonify(post_list) @app.route('/api/posts/') +@admin_required def api_post_detail(post_id): - posts = load_posts() - users = load_users() - - post = posts.get(post_id) + post = post_model.get_by_id(post_id) if not post: return jsonify({'error': '帖子不存在'}), 404 - author = users.get(post['author_id'], {}) + author = user_model.get_by_id(post['author_id']) or {} # 获取回复 - replies = [] - for reply in post.get('replies', []): - reply_author = users.get(reply['author_id'], {}) - replies.append({ + replies = reply_model.get_by_post(post_id) + reply_list = [] + for reply in replies: + reply_author = user_model.get_by_id(reply['author_id']) or {} + reply_list.append({ 'id': reply['id'], 'content': reply['content'][:100] + '...' if len(reply['content']) > 100 else reply['content'], 'author': reply_author.get('username', '未知'), - 'likes': len(reply.get('likes', [])), + 'likes': len(reply['likes']), 'created_at': reply['created_at'], }) @@ -205,83 +235,58 @@ def api_post_detail(post_id): 'content': post['content'], 'type': post['type'], 'author': author.get('username', '未知'), - 'tags': post.get('tags', []), - 'likes': len(post.get('likes', [])), - 'replies': replies, - 'views': post.get('views', 0), - 'is_pinned': post.get('is_pinned', False), + 'tags': post['tags'], + 'likes': len(post['likes']), + 'replies': reply_list, + 'views': post['views'], + 'is_pinned': post['is_pinned'], 'created_at': post['created_at'], }) @app.route('/api/posts/', methods=['DELETE']) +@admin_required def api_delete_post(post_id): - posts = load_posts() - users = load_users() - - if post_id not in posts: - return jsonify({'error': '帖子不存在'}), 404 - - author_id = posts[post_id].get('author_id') - - del posts[post_id] - - # 从用户列表中移除 - if author_id and author_id in users: - if post_id in users[author_id].get('posts', []): - users[author_id]['posts'].remove(post_id) - - save_posts(posts) - save_users(users) - + post_model.delete(post_id) return jsonify({'success': True}) @app.route('/api/posts//pin', methods=['POST']) +@admin_required def api_pin_post(post_id): - posts = load_posts() - - if post_id not in posts: - return jsonify({'error': '帖子不存在'}), 404 - - posts[post_id]['is_pinned'] = not posts[post_id].get('is_pinned', False) - save_posts(posts) - + new_pin = post_model.toggle_pin(post_id) return jsonify({ 'success': True, - 'is_pinned': posts[post_id]['is_pinned'] + 'is_pinned': new_pin }) @app.route('/api/topics') +@admin_required def api_topics(): - topics = load_topics() - users = load_users() + topics = topic_model.get_all() topic_list = [] - for tid, topic in topics.items(): - author = users.get(topic['author_id'], {}) + for topic in topics: + author = user_model.get_by_id(topic['author_id']) or {} topic_list.append({ - 'id': tid, + 'id': topic['id'], 'name': topic['name'], 'icon': topic.get('icon', '🔧'), 'author': author.get('username', '未知'), - 'sub_topics_count': len(topic.get('sub_topics', [])), - 'questions_count': len(topic.get('questions', [])), - 'followers_count': len(topic.get('followers', [])), + 'sub_topics_count': len(topic_model.get_sub_topics(topic['id'])), + 'questions_count': len(topic_model.get_questions(topic['id'])), + 'followers_count': len(topic['followers']), 'created_at': topic['created_at'], }) - topic_list.sort(key=lambda x: x['followers_count'], reverse=True) return jsonify(topic_list) @app.route('/api/topics/') +@admin_required def api_topic_detail(topic_id): - topics = load_topics() - users = load_users() - - topic = topics.get(topic_id) + topic = topic_model.get_by_id(topic_id) if not topic: return jsonify({'error': '主题不存在'}), 404 - author = users.get(topic['author_id'], {}) + author = user_model.get_by_id(topic['author_id']) or {} return jsonify({ 'id': topic_id, @@ -289,41 +294,37 @@ def api_topic_detail(topic_id): 'description': topic.get('description', ''), 'icon': topic.get('icon', '🔧'), 'author': author.get('username', '未知'), - 'sub_topics': topic.get('sub_topics', []), - 'questions': topic.get('questions', []), - 'followers_count': len(topic.get('followers', [])), + 'sub_topics': topic_model.get_sub_topics(topic_id), + 'questions': topic_model.get_questions(topic_id), + 'followers_count': len(topic['followers']), 'created_at': topic['created_at'], }) @app.route('/api/topics/', methods=['DELETE']) +@admin_required def api_delete_topic(topic_id): - topics = load_topics() - - if topic_id not in topics: - return jsonify({'error': '主题不存在'}), 404 - - del topics[topic_id] - save_topics(topics) - + topic_model.delete(topic_id) return jsonify({'success': True}) @app.route('/api/tags') +@admin_required def api_tags(): - posts = load_posts() - - tag_counts = {} - for post in posts.values(): - for tag in post.get('tags', []): - tag_counts[tag] = tag_counts.get(tag, 0) + 1 - - tags = sorted(tag_counts.items(), key=lambda x: x[1], reverse=True) + tags = post_model.get_tags_stats() return jsonify([{'name': t[0], 'count': t[1]} for t in tags[:20]]) +# ============ 健康检查（无需登录） ============ + +@app.route('/api/health') +def api_health(): + return jsonify({'status': 'ok', 'service': 'tech-forum-admin', 'port': ADMIN_PORT}) + if __name__ == '__main__': print("=" * 50) print("技术论坛 - 后台管理系统") print("=" * 50) - print(f"访问地址: http://localhost:19005") + print(f"访问地址: http://localhost:{ADMIN_PORT}") + print(f"默认账号: {ADMIN_USERNAME}") + print(f"默认密码: {ADMIN_PASSWORD}") print("=" * 50) - app.run(host='0.0.0.0', port=19005, debug=True) \ No newline at end of file + app.run(host='0.0.0.0', port=ADMIN_PORT, debug=True) \ No newline at end of file diff --git a/admin/templates/login.html b/admin/templates/login.html new file mode 100644 index 0000000..ab93187 --- /dev/null +++ b/admin/templates/login.html @@ -0,0 +1,116 @@ + + + + + + 技术论坛 - 后台登录 + + + + + +
+
+
+
+ +
+

后台管理系统

+

请输入管理员账号登录

+
+ +
+
+ +
+ + +
+
+ +
+ +
+ + +
+
+ + + + +
+ +
+ + 访问前台网站 + +
+
+
+ + + + \ No newline at end of file diff --git a/backend/app.py b/backend/app.py index e81087d..7bca900 100644 --- a/backend/app.py +++ b/backend/app.py @@ -1,63 +1,31 @@ """ -技术论坛与技术分享网站 - 后端API +技术论坛与技术分享网站 - 后端API (重构版) """ from flask import Flask, request, jsonify, send_file from flask_cors import CORS -from werkzeug.security import generate_password_hash, check_password_hash import jwt import datetime -import json -import uuid import os from pathlib import Path +# 导入配置和模型 +import sys +sys.path.insert(0, str(Path(__file__).parent.parent)) +from config import SECRET_KEY, LLM_BASE_URL, LLM_API_KEY, LLM_MODEL, DATABASE_PATH, BACKEND_PORT +from models import Database, UserModel, PostModel, ReplyModel, TopicModel + app = Flask(__name__, static_folder='../frontend', static_url_path='') CORS(app) -# 配置 -SECRET_KEY = 'tech-forum-secret-2026' -LLM_BASE_URL = 'http://192.168.2.5:1234/v1' -LLM_API_KEY = 'sk-lm-fuP5tGU8:Hi7YU87jHyDP6Ay8Tl2j' -LLM_MODEL = 'qwen3.5-4b' +# 初始化数据库 +db = Database(DATABASE_PATH) +user_model = UserModel(db) +post_model = PostModel(db) +reply_model = ReplyModel(db) +topic_model = TopicModel(db) -# 数据目录 -DATA_DIR = Path(__file__).parent.parent / 'data' -USERS_FILE = DATA_DIR / 'users.json' -POSTS_FILE = DATA_DIR / 'posts.json' -TOPICS_FILE = DATA_DIR / 'topics.json' -UPLOAD_DIR = Path(__file__).parent.parent / 'uploads' -UPLOAD_DIR.mkdir(parents=True, exist_ok=True) - -# 初始化数据文件 -def init_data(): - if not USERS_FILE.exists(): - USERS_FILE.write_text(json.dumps({}, ensure_ascii=False)) - if not POSTS_FILE.exists(): - POSTS_FILE.write_text(json.dumps({}, ensure_ascii=False)) - if not TOPICS_FILE.exists(): - TOPICS_FILE.write_text(json.dumps({}, ensure_ascii=False)) - -init_data() - -# 辅助函数 -def load_users(): - return json.loads(USERS_FILE.read_text(encoding='utf-8')) - -def save_users(users): - USERS_FILE.write_text(json.dumps(users, ensure_ascii=False, indent=2), encoding='utf-8') - -def load_posts(): - return json.loads(POSTS_FILE.read_text(encoding='utf-8')) - -def save_posts(posts): - POSTS_FILE.write_text(json.dumps(posts, ensure_ascii=False, indent=2), encoding='utf-8') - -def load_topics(): - return json.loads(TOPICS_FILE.read_text(encoding='utf-8')) - -def save_topics(topics): - TOPICS_FILE.write_text(json.dumps(topics, ensure_ascii=False, indent=2), encoding='utf-8') +# ============ 辅助函数 ============ def generate_token(user_id): return jwt.encode({ @@ -78,8 +46,7 @@ def get_current_user(): data = verify_token(token) if not data: return None - users = load_users() - return users.get(data['user_id']) + return user_model.get_by_id(data['user_id']) # ============ 页面路由 ============ @@ -133,32 +100,15 @@ def api_register(): if password != confirm_password: return jsonify({'error': '两次密码不一致'}), 400 - users = load_users() - # 检查是否已存在 - for uid, user in users.items(): - if user['username'] == username: - return jsonify({'error': '用户名已存在'}), 400 - if user['email'] == email: - return jsonify({'error': '邮箱已注册'}), 400 + if user_model.get_by_username(username): + return jsonify({'error': '用户名已存在'}), 400 + if user_model.get_by_email(email): + return jsonify({'error': '邮箱已注册'}), 400 # 创建用户 - user_id = str(uuid.uuid4()) - users[user_id] = { - 'id': user_id, - 'username': username, - 'email': email, - 'phone': phone, - 'password': generate_password_hash(password), - 'avatar': f'https://api.dicebear.com/7.x/avataaars/svg?seed={username}', - 'bio': '', - 'created_at': datetime.datetime.now().isoformat(), - 'posts': [], - 'replies': [], - 'likes': [] - } - - save_users(users) + user_id = user_model.create(username, email, phone, password) + user = user_model.get_by_id(user_id) token = generate_token(user_id) @@ -169,7 +119,7 @@ def api_register(): 'id': user_id, 'username': username, 'email': email, - 'avatar': users[user_id]['avatar'] + 'avatar': user['avatar'] } }) @@ -183,27 +133,30 @@ def api_login(): if not login_name or not password: return jsonify({'error': '请输入用户名和密码'}), 400 - users = load_users() + # 查找用户 + user = user_model.get_by_username(login_name) + if not user: + user = user_model.get_by_email(login_name) - for user_id, user in users.items(): - if user['username'] == login_name or user['email'] == login_name: - if check_password_hash(user['password'], password): - token = generate_token(user_id) - return jsonify({ - 'success': True, - 'token': token, - 'user': { - 'id': user_id, - 'username': user['username'], - 'email': user['email'], - 'avatar': user['avatar'], - 'bio': user.get('bio', '') - } - }) - else: - return jsonify({'error': '密码错误'}), 400 + if not user: + return jsonify({'error': '用户不存在'}), 400 - return jsonify({'error': '用户不存在'}), 400 + if not user_model.verify_password(user, password): + return jsonify({'error': '密码错误'}), 400 + + token = generate_token(user['id']) + + return jsonify({ + 'success': True, + 'token': token, + 'user': { + 'id': user['id'], + 'username': user['username'], + 'email': user['email'], + 'avatar': user['avatar'], + 'bio': user.get('bio', '') + } + }) @app.route('/api/user') def api_current_user(): @@ -218,32 +171,28 @@ def api_current_user(): 'phone': user.get('phone', ''), 'avatar': user['avatar'], 'bio': user.get('bio', ''), - 'posts_count': len(user.get('posts', [])), - 'replies_count': len(user.get('replies', [])), + 'posts_count': user_model.get_posts_count(user['id']), + 'replies_count': user_model.get_replies_count(user['id']), 'created_at': user['created_at'] }) @app.route('/api/user/') def api_user_profile(user_id): - users = load_users() - posts = load_posts() - topics = load_topics() - - user = users.get(user_id) + user = user_model.get_by_id(user_id) if not user: return jsonify({'error': '用户不存在'}), 404 # 获取用户的帖子 + posts, total = post_model.get_all() user_posts = [] - for post_id in user.get('posts', []): - if post_id in posts: - post = posts[post_id] + for post in posts: + if post['author_id'] == user_id: user_posts.append({ - 'id': post_id, + 'id': post['id'], 'title': post['title'], 'type': post['type'], - 'likes': len(post.get('likes', [])), - 'replies': len(post.get('replies', [])), + 'likes': len(post['likes']), + 'replies': len(reply_model.get_by_post(post['id'])), 'created_at': post['created_at'] }) @@ -253,8 +202,8 @@ def api_user_profile(user_id): 'username': user['username'], 'avatar': user['avatar'], 'bio': user.get('bio', ''), - 'posts_count': len(user.get('posts', [])), - 'replies_count': len(user.get('replies', [])), + 'posts_count': user_model.get_posts_count(user['id']), + 'replies_count': user_model.get_replies_count(user['id']), 'created_at': user['created_at'] }, 'posts': user_posts @@ -264,24 +213,18 @@ def api_user_profile(user_id): @app.route('/api/posts') def api_posts(): - posts = load_posts() - users = load_users() - - post_type = request.args.get('type') # discussion, share + post_type = request.args.get('type') tag = request.args.get('tag') page = int(request.args.get('page', 1)) per_page = int(request.args.get('per_page', 20)) + posts, total = post_model.get_all(post_type, tag, page, per_page) + post_list = [] - for pid, post in posts.items(): - if post_type and post['type'] != post_type: - continue - if tag and tag not in post.get('tags', []): - continue - - author = users.get(post['author_id'], {}) + for post in posts: + author = user_model.get_by_id(post['author_id']) or {} post_list.append({ - 'id': pid, + 'id': post['id'], 'title': post['title'], 'type': post['type'], 'content_preview': post['content'][:150] + '...' if len(post['content']) > 150 else post['content'], @@ -290,24 +233,17 @@ def api_posts(): 'username': author.get('username', '未知'), 'avatar': author.get('avatar', '') }, - 'tags': post.get('tags', []), - 'likes': len(post.get('likes', [])), - 'replies': len(post.get('replies', [])), - 'views': post.get('views', 0), + 'tags': post['tags'], + 'likes': len(post['likes']), + 'replies': len(reply_model.get_by_post(post['id'])), + 'views': post['views'], 'created_at': post['created_at'], - 'is_pinned': post.get('is_pinned', False) + 'is_pinned': post['is_pinned'] }) - # 排序：置顶在前，然后按时间 - post_list.sort(key=lambda x: (not x['is_pinned'], x['created_at']), reverse=True) - - # 分页 - start = (page - 1) * per_page - end = start + per_page - return jsonify({ - 'posts': post_list[start:end], - 'total': len(post_list), + 'posts': post_list, + 'total': total, 'page': page, 'per_page': per_page }) @@ -322,7 +258,7 @@ def api_create_post(): title = data.get('title', '').strip() content = data.get('content', '').strip() - post_type = data.get('type', 'discussion') # discussion, share + post_type = data.get('type', 'discussion') tags = data.get('tags', []) if not title or len(title) < 5: @@ -330,31 +266,7 @@ def api_create_post(): if not content or len(content) < 10: return jsonify({'error': '内容至少10个字符'}), 400 - posts = load_posts() - users = load_users() - - post_id = str(uuid.uuid4()) - posts[post_id] = { - 'id': post_id, - 'title': title, - 'content': content, - 'type': post_type, - 'author_id': user['id'], - 'tags': tags, - 'likes': [], - 'replies': [], - 'views': 0, - 'is_pinned': False, - 'created_at': datetime.datetime.now().isoformat(), - 'updated_at': datetime.datetime.now().isoformat() - } - - save_posts(posts) - - # 更新用户的帖子列表 - if post_id not in users[user['id']]['posts']: - users[user['id']]['posts'].append(post_id) - save_users(users) + post_id = post_model.create(title, content, post_type, user['id'], tags) return jsonify({ 'success': True, @@ -363,24 +275,22 @@ def api_create_post(): @app.route('/api/posts/') def api_post_detail(post_id): - posts = load_posts() - users = load_users() - - post = posts.get(post_id) + post = post_model.get_by_id(post_id) if not post: return jsonify({'error': '帖子不存在'}), 404 # 增加浏览量 - post['views'] = post.get('views', 0) + 1 - save_posts(posts) + post_model.increment_views(post_id) + post['views'] += 1 - author = users.get(post['author_id'], {}) + author = user_model.get_by_id(post['author_id']) or {} # 获取回复 - replies = [] - for reply in post.get('replies', []): - reply_author = users.get(reply['author_id'], {}) - replies.append({ + replies = reply_model.get_by_post(post_id) + reply_list = [] + for reply in replies: + reply_author = user_model.get_by_id(reply['author_id']) or {} + reply_list.append({ 'id': reply['id'], 'content': reply['content'], 'author': { @@ -388,7 +298,7 @@ def api_post_detail(post_id): 'username': reply_author.get('username', '未知'), 'avatar': reply_author.get('avatar', '') }, - 'likes': len(reply.get('likes', [])), + 'likes': len(reply['likes']), 'created_at': reply['created_at'], 'reply_to': reply.get('reply_to') }) @@ -404,10 +314,10 @@ def api_post_detail(post_id): 'avatar': author.get('avatar', ''), 'bio': author.get('bio', '') }, - 'tags': post.get('tags', []), - 'likes': len(post.get('likes', [])), + 'tags': post['tags'], + 'likes': len(post['likes']), 'views': post['views'], - 'replies': replies, + 'replies': reply_list, 'created_at': post['created_at'], 'updated_at': post.get('updated_at', post['created_at']) }) @@ -420,36 +330,16 @@ def api_reply_post(post_id): data = request.json content = data.get('content', '').strip() - reply_to = data.get('reply_to') # 回复的评论ID + reply_to = data.get('reply_to') if not content: return jsonify({'error': '回复内容不能为空'}), 400 - posts = load_posts() - users = load_users() - - post = posts.get(post_id) + post = post_model.get_by_id(post_id) if not post: return jsonify({'error': '帖子不存在'}), 404 - reply_id = str(uuid.uuid4()) - reply = { - 'id': reply_id, - 'content': content, - 'author_id': user['id'], - 'likes': [], - 'reply_to': reply_to, - 'created_at': datetime.datetime.now().isoformat() - } - - post['replies'].append(reply) - post['updated_at'] = datetime.datetime.now().isoformat() - save_posts(posts) - - # 更新用户的回复列表 - if post_id not in users[user['id']]['replies']: - users[user['id']]['replies'].append(post_id) - save_users(users) + reply_id = reply_model.create(post_id, content, user['id'], reply_to) return jsonify({ 'success': True, @@ -462,39 +352,29 @@ def api_like_post(post_id): if not user: return jsonify({'error': '请先登录'}), 401 - posts = load_posts() - - post = posts.get(post_id) + post = post_model.get_by_id(post_id) if not post: return jsonify({'error': '帖子不存在'}), 404 - if user['id'] in post['likes']: - post['likes'].remove(user['id']) - liked = False - else: - post['likes'].append(user['id']) - liked = True - - save_posts(posts) + liked, likes_count = post_model.add_like(post_id, user['id']) return jsonify({ 'success': True, 'liked': liked, - 'likes_count': len(post['likes']) + 'likes_count': likes_count }) # ============ API: 工具分享主题 ============ @app.route('/api/topics') def api_topics(): - topics = load_topics() - users = load_users() + topics = topic_model.get_all() topic_list = [] - for tid, topic in topics.items(): - author = users.get(topic['author_id'], {}) + for topic in topics: + author = user_model.get_by_id(topic['author_id']) or {} topic_list.append({ - 'id': tid, + 'id': topic['id'], 'name': topic['name'], 'description': topic['description'][:100] + '...' if len(topic.get('description', '')) > 100 else topic.get('description', ''), 'icon': topic.get('icon', '🔧'), @@ -502,9 +382,9 @@ def api_topics(): 'id': topic['author_id'], 'username': author.get('username', '未知') }, - 'sub_topics_count': len(topic.get('sub_topics', [])), - 'questions_count': len(topic.get('questions', [])), - 'followers': len(topic.get('followers', [])), + 'sub_topics_count': len(topic_model.get_sub_topics(topic['id'])), + 'questions_count': len(topic_model.get_questions(topic['id'])), + 'followers': len(topic['followers']), 'created_at': topic['created_at'] }) @@ -527,22 +407,7 @@ def api_create_topic(): if not name: return jsonify({'error': '主题名称不能为空'}), 400 - topics = load_topics() - - topic_id = str(uuid.uuid4()) - topics[topic_id] = { - 'id': topic_id, - 'name': name, - 'description': description, - 'icon': icon, - 'author_id': user['id'], - 'sub_topics': [], - 'questions': [], - 'followers': [], - 'created_at': datetime.datetime.now().isoformat() - } - - save_topics(topics) + topic_id = topic_model.create(name, description, icon, user['id']) return jsonify({ 'success': True, @@ -551,33 +416,33 @@ def api_create_topic(): @app.route('/api/topics/') def api_topic_detail(topic_id): - topics = load_topics() - users = load_users() - - topic = topics.get(topic_id) + topic = topic_model.get_by_id(topic_id) if not topic: return jsonify({'error': '主题不存在'}), 404 - author = users.get(topic['author_id'], {}) + author = user_model.get_by_id(topic['author_id']) or {} # 获取子主题 - sub_topics = [] - for st in topic.get('sub_topics', []): - sub_topics.append({ + sub_topics = topic_model.get_sub_topics(topic_id) + sub_list = [] + for st in sub_topics: + st_author = user_model.get_by_id(st['author_id']) or {} + sub_list.append({ 'id': st['id'], 'title': st['title'], 'content': st['content'], - 'author': users.get(st['author_id'], {}), + 'author': st_author, 'created_at': st['created_at'] }) # 获取问题 - questions = [] - for q in topic.get('questions', []): - q_author = users.get(q['author_id'], {}) + questions = topic_model.get_questions(topic_id) + q_list = [] + for q in questions: + q_author = user_model.get_by_id(q['author_id']) or {} answers = [] - for a in q.get('answers', []): - a_author = users.get(a['author_id'], {}) + for a in q['answers']: + a_author = user_model.get_by_id(a['author_id']) or {} answers.append({ 'id': a['id'], 'content': a['content'], @@ -586,10 +451,10 @@ def api_topic_detail(topic_id): 'username': a_author.get('username', '未知'), 'avatar': a_author.get('avatar', '') }, - 'likes': len(a.get('likes', [])), + 'likes': len(a['likes']), 'created_at': a['created_at'] }) - questions.append({ + q_list.append({ 'id': q['id'], 'title': q['title'], 'content': q.get('content', ''), @@ -613,9 +478,9 @@ def api_topic_detail(topic_id): 'username': author.get('username', '未知'), 'avatar': author.get('avatar', '') }, - 'sub_topics': sub_topics, - 'questions': questions, - 'followers': len(topic.get('followers', [])), + 'sub_topics': sub_list, + 'questions': q_list, + 'followers': len(topic['followers']), 'created_at': topic['created_at'] }) @@ -632,24 +497,13 @@ def api_add_subtopic(topic_id): if not title: return jsonify({'error': '标题不能为空'}), 400 - topics = load_topics() - - topic = topics.get(topic_id) + topic = topic_model.get_by_id(topic_id) if not topic: return jsonify({'error': '主题不存在'}), 404 - subtopic = { - 'id': str(uuid.uuid4()), - 'title': title, - 'content': content, - 'author_id': user['id'], - 'created_at': datetime.datetime.now().isoformat() - } + subtopic_id = topic_model.add_sub_topic(topic_id, title, content, user['id']) - topic['sub_topics'].append(subtopic) - save_topics(topics) - - return jsonify({'success': True, 'subtopic_id': subtopic['id']}) + return jsonify({'success': True, 'subtopic_id': subtopic_id}) @app.route('/api/topics//question', methods=['POST']) def api_add_question(topic_id): @@ -664,26 +518,13 @@ def api_add_question(topic_id): if not title: return jsonify({'error': '问题标题不能为空'}), 400 - topics = load_topics() - - topic = topics.get(topic_id) + topic = topic_model.get_by_id(topic_id) if not topic: return jsonify({'error': '主题不存在'}), 404 - question = { - 'id': str(uuid.uuid4()), - 'title': title, - 'content': content, - 'author_id': user['id'], - 'answers': [], - 'views': 0, - 'created_at': datetime.datetime.now().isoformat() - } + question_id = topic_model.add_question(topic_id, title, content, user['id']) - topic['questions'].append(question) - save_topics(topics) - - return jsonify({'success': True, 'question_id': question['id']}) + return jsonify({'success': True, 'question_id': question_id}) @app.route('/api/topics//question//answer', methods=['POST']) def api_answer_question(topic_id, question_id): @@ -697,34 +538,13 @@ def api_answer_question(topic_id, question_id): if not content: return jsonify({'error': '回答内容不能为空'}), 400 - topics = load_topics() - - topic = topics.get(topic_id) + topic = topic_model.get_by_id(topic_id) if not topic: return jsonify({'error': '主题不存在'}), 404 - # 找到问题 - question = None - for q in topic['questions']: - if q['id'] == question_id: - question = q - break + answer_id = topic_model.add_answer(question_id, content, user['id']) - if not question: - return jsonify({'error': '问题不存在'}), 404 - - answer = { - 'id': str(uuid.uuid4()), - 'content': content, - 'author_id': user['id'], - 'likes': [], - 'created_at': datetime.datetime.now().isoformat() - } - - question['answers'].append(answer) - save_topics(topics) - - return jsonify({'success': True, 'answer_id': answer['id']}) + return jsonify({'success': True, 'answer_id': answer_id}) @app.route('/api/topics//follow', methods=['POST']) def api_follow_topic(topic_id): @@ -732,40 +552,23 @@ def api_follow_topic(topic_id): if not user: return jsonify({'error': '请先登录'}), 401 - topics = load_topics() - - topic = topics.get(topic_id) + topic = topic_model.get_by_id(topic_id) if not topic: return jsonify({'error': '主题不存在'}), 404 - if user['id'] in topic['followers']: - topic['followers'].remove(user['id']) - followed = False - else: - topic['followers'].append(user['id']) - followed = True - - save_topics(topics) + followed, followers_count = topic_model.add_follower(topic_id, user['id']) return jsonify({ 'success': True, 'followed': followed, - 'followers_count': len(topic['followers']) + 'followers_count': followers_count }) # ============ API: 标签 ============ @app.route('/api/tags') def api_tags(): - posts = load_posts() - - tag_counts = {} - for post in posts.values(): - for tag in post.get('tags', []): - tag_counts[tag] = tag_counts.get(tag, 0) + 1 - - tags = sorted(tag_counts.items(), key=lambda x: x[1], reverse=True) - + tags = post_model.get_tags_stats() return jsonify([{'name': t[0], 'count': t[1]} for t in tags]) # ============ API: 搜索 ============ @@ -777,17 +580,14 @@ def api_search(): if not query: return jsonify({'posts': [], 'topics': []}) - posts = load_posts() - topics = load_topics() - users = load_users() - # 搜索帖子 + posts, _ = post_model.get_all() matched_posts = [] - for pid, post in posts.items(): + for post in posts: if query in post['title'].lower() or query in post['content'].lower(): - author = users.get(post['author_id'], {}) + author = user_model.get_by_id(post['author_id']) or {} matched_posts.append({ - 'id': pid, + 'id': post['id'], 'title': post['title'], 'type': post['type'], 'author': author.get('username', '未知'), @@ -795,11 +595,12 @@ def api_search(): }) # 搜索主题 + topics = topic_model.get_all() matched_topics = [] - for tid, topic in topics.items(): + for topic in topics: if query in topic['name'].lower() or query in topic.get('description', '').lower(): matched_topics.append({ - 'id': tid, + 'id': topic['id'], 'name': topic['name'], 'icon': topic.get('icon', '🔧') }) @@ -809,11 +610,17 @@ def api_search(): 'topics': matched_topics[:20] }) +# ============ API: 健康检查 ============ + +@app.route('/api/health') +def api_health(): + return jsonify({'status': 'ok', 'service': 'tech-forum', 'port': BACKEND_PORT}) + if __name__ == '__main__': print("=" * 50) print("技术论坛与技术分享网站") print("=" * 50) - print(f"访问地址: http://localhost:19004") + print(f"访问地址: http://localhost:{BACKEND_PORT}") print("=" * 50) - app.run(host='0.0.0.0', port=19004, debug=True) \ No newline at end of file + app.run(host='0.0.0.0', port=BACKEND_PORT, debug=True) \ No newline at end of file diff --git a/config.py b/config.py new file mode 100644 index 0000000..6c4c284 --- /dev/null +++ b/config.py @@ -0,0 +1,25 @@ +""" +技术论坛配置文件 +""" + +import os +import secrets + +# 安全密钥（生产环境应使用环境变量） +SECRET_KEY = os.environ.get('TECH_FORUM_SECRET', secrets.token_hex(32)) + +# 管理员账户（后台登录） +ADMIN_USERNAME = os.environ.get('TECH_FORUM_ADMIN_USER', 'admin') +ADMIN_PASSWORD = os.environ.get('TECH_FORUM_ADMIN_PASS', 'admin123') + +# LLM 配置（可选，用于AI功能） +LLM_BASE_URL = os.environ.get('LLM_BASE_URL', 'http://192.168.2.5:1234/v1') +LLM_API_KEY = os.environ.get('LLM_API_KEY', '') +LLM_MODEL = os.environ.get('LLM_MODEL', 'qwen3.5-4b') + +# 数据库路径 +DATABASE_PATH = os.environ.get('TECH_FORUM_DB', 'data/tech_forum.db') + +# 服务端口 +BACKEND_PORT = 19004 +ADMIN_PORT = 19005 \ No newline at end of file diff --git a/models.py b/models.py new file mode 100644 index 0000000..e0155e8 --- /dev/null +++ b/models.py @@ -0,0 +1,442 @@ +""" +技术论坛数据库模型 - SQLite +""" + +import sqlite3 +import json +import uuid +import datetime +from pathlib import Path +from werkzeug.security import generate_password_hash, check_password_hash +from contextlib import contextmanager + +class Database: + def __init__(self, db_path='data/tech_forum.db'): + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self._init_tables() + + @contextmanager + def get_conn(self): + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + try: + yield conn + finally: + conn.close() + + def _init_tables(self): + with self.get_conn() as conn: + # 用户表 + conn.execute(''' + CREATE TABLE IF NOT EXISTS users ( + id TEXT PRIMARY KEY, + username TEXT UNIQUE NOT NULL, + email TEXT UNIQUE NOT NULL, + phone TEXT, + password TEXT NOT NULL, + avatar TEXT, + bio TEXT, + created_at TEXT, + updated_at TEXT + ) + ''') + + # 帖子表 + conn.execute(''' + CREATE TABLE IF NOT EXISTS posts ( + id TEXT PRIMARY KEY, + title TEXT NOT NULL, + content TEXT, + type TEXT DEFAULT 'discussion', + author_id TEXT, + tags TEXT, + likes TEXT DEFAULT '[]', + views INTEGER DEFAULT 0, + is_pinned INTEGER DEFAULT 0, + created_at TEXT, + updated_at TEXT, + FOREIGN KEY (author_id) REFERENCES users(id) + ) + ''') + + # 回复表 + conn.execute(''' + CREATE TABLE IF NOT EXISTS replies ( + id TEXT PRIMARY KEY, + post_id TEXT, + content TEXT, + author_id TEXT, + likes TEXT DEFAULT '[]', + reply_to TEXT, + created_at TEXT, + FOREIGN KEY (post_id) REFERENCES posts(id), + FOREIGN KEY (author_id) REFERENCES users(id) + ) + ''') + + # 主题表（工具分享） + conn.execute(''' + CREATE TABLE IF NOT EXISTS topics ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL, + description TEXT, + icon TEXT DEFAULT '🔧', + author_id TEXT, + followers TEXT DEFAULT '[]', + created_at TEXT, + FOREIGN KEY (author_id) REFERENCES users(id) + ) + ''') + + # 子主题表 + conn.execute(''' + CREATE TABLE IF NOT EXISTS sub_topics ( + id TEXT PRIMARY KEY, + topic_id TEXT, + title TEXT, + content TEXT, + author_id TEXT, + created_at TEXT, + FOREIGN KEY (topic_id) REFERENCES topics(id), + FOREIGN KEY (author_id) REFERENCES users(id) + ) + ''') + + # 问题表 + conn.execute(''' + CREATE TABLE IF NOT EXISTS questions ( + id TEXT PRIMARY KEY, + topic_id TEXT, + title TEXT, + content TEXT, + author_id TEXT, + views INTEGER DEFAULT 0, + created_at TEXT, + FOREIGN KEY (topic_id) REFERENCES topics(id), + FOREIGN KEY (author_id) REFERENCES users(id) + ) + ''') + + # 回答表 + conn.execute(''' + CREATE TABLE IF NOT EXISTS answers ( + id TEXT PRIMARY KEY, + question_id TEXT, + content TEXT, + author_id TEXT, + likes TEXT DEFAULT '[]', + created_at TEXT, + FOREIGN KEY (question_id) REFERENCES questions(id), + FOREIGN KEY (author_id) REFERENCES users(id) + ) + ''') + + conn.commit() + + +class UserModel: + def __init__(self, db): + self.db = db + + def create(self, username, email, phone, password): + user_id = str(uuid.uuid4()) + avatar = f'https://api.dicebear.com/7.x/avataaars/svg?seed={username}' + now = datetime.datetime.now().isoformat() + + with self.db.get_conn() as conn: + conn.execute(''' + INSERT INTO users (id, username, email, phone, password, avatar, bio, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, '', ?, ?) + ''', (user_id, username, email, phone, generate_password_hash(password), avatar, now, now)) + conn.commit() + + return user_id + + def get_by_id(self, user_id): + with self.db.get_conn() as conn: + row = conn.execute('SELECT * FROM users WHERE id = ?', (user_id,)).fetchone() + return dict(row) if row else None + + def get_by_username(self, username): + with self.db.get_conn() as conn: + row = conn.execute('SELECT * FROM users WHERE username = ?', (username,)).fetchone() + return dict(row) if row else None + + def get_by_email(self, email): + with self.db.get_conn() as conn: + row = conn.execute('SELECT * FROM users WHERE email = ?', (email,)).fetchone() + return dict(row) if row else None + + def verify_password(self, user, password): + return check_password_hash(user['password'], password) + + def get_all(self): + with self.db.get_conn() as conn: + rows = conn.execute('SELECT * FROM users ORDER BY created_at DESC').fetchall() + return [dict(row) for row in rows] + + def delete(self, user_id): + with self.db.get_conn() as conn: + # 删除用户的所有帖子 + conn.execute('DELETE FROM replies WHERE author_id = ?', (user_id,)) + conn.execute('DELETE FROM posts WHERE author_id = ?', (user_id,)) + conn.execute('DELETE FROM users WHERE id = ?', (user_id,)) + conn.commit() + + def get_posts_count(self, user_id): + with self.db.get_conn() as conn: + return conn.execute('SELECT COUNT(*) FROM posts WHERE author_id = ?', (user_id,)).fetchone()[0] + + def get_replies_count(self, user_id): + with self.db.get_conn() as conn: + return conn.execute('SELECT COUNT(*) FROM replies WHERE author_id = ?', (user_id,)).fetchone()[0] + + +class PostModel: + def __init__(self, db): + self.db = db + + def create(self, title, content, post_type, author_id, tags): + post_id = str(uuid.uuid4()) + now = datetime.datetime.now().isoformat() + tags_json = json.dumps(tags) + + with self.db.get_conn() as conn: + conn.execute(''' + INSERT INTO posts (id, title, content, type, author_id, tags, likes, views, is_pinned, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, '[]', 0, 0, ?, ?) + ''', (post_id, title, content, post_type, author_id, tags_json, now, now)) + conn.commit() + + return post_id + + def get_by_id(self, post_id): + with self.db.get_conn() as conn: + row = conn.execute('SELECT * FROM posts WHERE id = ?', (post_id,)).fetchone() + if row: + post = dict(row) + post['tags'] = json.loads(post['tags'] or '[]') + post['likes'] = json.loads(post['likes'] or '[]') + return post + return None + + def get_all(self, post_type=None, tag=None, page=1, per_page=20): + with self.db.get_conn() as conn: + query = 'SELECT * FROM posts WHERE 1=1' + params = [] + + if post_type: + query += ' AND type = ?' + params.append(post_type) + + query += ' ORDER BY is_pinned DESC, created_at DESC' + + rows = conn.execute(query, params).fetchall() + posts = [] + for row in rows: + post = dict(row) + post['tags'] = json.loads(post['tags'] or '[]') + post['likes'] = json.loads(post['likes'] or '[]') + posts.append(post) + + # 分页 + start = (page - 1) * per_page + return posts[start:start + per_page], len(posts) + + def increment_views(self, post_id): + with self.db.get_conn() as conn: + conn.execute('UPDATE posts SET views = views + 1 WHERE id = ?', (post_id,)) + conn.commit() + + def add_like(self, post_id, user_id): + with self.db.get_conn() as conn: + post = self.get_by_id(post_id) + likes = post['likes'] + + if user_id in likes: + likes.remove(user_id) + liked = False + else: + likes.append(user_id) + liked = True + + conn.execute('UPDATE posts SET likes = ? WHERE id = ?', (json.dumps(likes), post_id)) + conn.commit() + + return liked, len(likes) + + def delete(self, post_id): + with self.db.get_conn() as conn: + conn.execute('DELETE FROM replies WHERE post_id = ?', (post_id,)) + conn.execute('DELETE FROM posts WHERE id = ?', (post_id,)) + conn.commit() + + def toggle_pin(self, post_id): + with self.db.get_conn() as conn: + row = conn.execute('SELECT is_pinned FROM posts WHERE id = ?', (post_id,)).fetchone() + new_pin = 1 if row['is_pinned'] == 0 else 0 + conn.execute('UPDATE posts SET is_pinned = ? WHERE id = ?', (new_pin, post_id)) + conn.commit() + return new_pin + + def get_tags_stats(self): + with self.db.get_conn() as conn: + rows = conn.execute('SELECT tags FROM posts').fetchall() + tag_counts = {} + for row in rows: + tags = json.loads(row['tags'] or '[]') + for tag in tags: + tag_counts[tag] = tag_counts.get(tag, 0) + 1 + return sorted(tag_counts.items(), key=lambda x: x[1], reverse=True) + + +class ReplyModel: + def __init__(self, db): + self.db = db + + def create(self, post_id, content, author_id, reply_to=None): + reply_id = str(uuid.uuid4()) + now = datetime.datetime.now().isoformat() + + with self.db.get_conn() as conn: + conn.execute(''' + INSERT INTO replies (id, post_id, content, author_id, likes, reply_to, created_at) + VALUES (?, ?, ?, ?, '[]', ?, ?) + ''', (reply_id, post_id, content, author_id, reply_to, now)) + conn.commit() + + return reply_id + + def get_by_post(self, post_id): + with self.db.get_conn() as conn: + rows = conn.execute('SELECT * FROM replies WHERE post_id = ? ORDER BY created_at', (post_id,)).fetchall() + replies = [] + for row in rows: + reply = dict(row) + reply['likes'] = json.loads(reply['likes'] or '[]') + replies.append(reply) + return replies + + +class TopicModel: + def __init__(self, db): + self.db = db + + def create(self, name, description, icon, author_id): + topic_id = str(uuid.uuid4()) + now = datetime.datetime.now().isoformat() + + with self.db.get_conn() as conn: + conn.execute(''' + INSERT INTO topics (id, name, description, icon, author_id, followers, created_at) + VALUES (?, ?, ?, ?, ?, '[]', ?) + ''', (topic_id, name, description, icon, author_id, now)) + conn.commit() + + return topic_id + + def get_by_id(self, topic_id): + with self.db.get_conn() as conn: + row = conn.execute('SELECT * FROM topics WHERE id = ?', (topic_id,)).fetchone() + if row: + topic = dict(row) + topic['followers'] = json.loads(topic['followers'] or '[]') + return topic + return None + + def get_all(self): + with self.db.get_conn() as conn: + rows = conn.execute('SELECT * FROM topics ORDER BY created_at DESC').fetchall() + topics = [] + for row in rows: + topic = dict(row) + topic['followers'] = json.loads(topic['followers'] or '[]') + topics.append(topic) + return topics + + def delete(self, topic_id): + with self.db.get_conn() as conn: + conn.execute('DELETE FROM answers WHERE question_id IN (SELECT id FROM questions WHERE topic_id = ?)', (topic_id,)) + conn.execute('DELETE FROM questions WHERE topic_id = ?', (topic_id,)) + conn.execute('DELETE FROM sub_topics WHERE topic_id = ?', (topic_id,)) + conn.execute('DELETE FROM topics WHERE id = ?', (topic_id,)) + conn.commit() + + def add_follower(self, topic_id, user_id): + with self.db.get_conn() as conn: + topic = self.get_by_id(topic_id) + followers = topic['followers'] + + if user_id in followers: + followers.remove(user_id) + followed = False + else: + followers.append(user_id) + followed = True + + conn.execute('UPDATE topics SET followers = ? WHERE id = ?', (json.dumps(followers), topic_id)) + conn.commit() + + return followed, len(followers) + + def get_sub_topics(self, topic_id): + with self.db.get_conn() as conn: + rows = conn.execute('SELECT * FROM sub_topics WHERE topic_id = ? ORDER BY created_at', (topic_id,)).fetchall() + return [dict(row) for row in rows] + + def add_sub_topic(self, topic_id, title, content, author_id): + sub_id = str(uuid.uuid4()) + now = datetime.datetime.now().isoformat() + + with self.db.get_conn() as conn: + conn.execute(''' + INSERT INTO sub_topics (id, topic_id, title, content, author_id, created_at) + VALUES (?, ?, ?, ?, ?, ?) + ''', (sub_id, topic_id, title, content, author_id, now)) + conn.commit() + + return sub_id + + def get_questions(self, topic_id): + with self.db.get_conn() as conn: + rows = conn.execute('SELECT * FROM questions WHERE topic_id = ? ORDER BY created_at DESC', (topic_id,)).fetchall() + questions = [] + for row in rows: + q = dict(row) + # 获取回答 + ans_rows = conn.execute('SELECT * FROM answers WHERE question_id = ? ORDER BY created_at', (q['id'],)).fetchall() + answers = [] + for ans in ans_rows: + a = dict(ans) + a['likes'] = json.loads(a['likes'] or '[]') + answers.append(a) + q['answers'] = answers + questions.append(q) + return questions + + def add_question(self, topic_id, title, content, author_id): + q_id = str(uuid.uuid4()) + now = datetime.datetime.now().isoformat() + + with self.db.get_conn() as conn: + conn.execute(''' + INSERT INTO questions (id, topic_id, title, content, author_id, views, created_at) + VALUES (?, ?, ?, ?, ?, 0, ?) + ''', (q_id, topic_id, title, content, author_id, now)) + conn.commit() + + return q_id + + def add_answer(self, question_id, content, author_id): + ans_id = str(uuid.uuid4()) + now = datetime.datetime.now().isoformat() + + with self.db.get_conn() as conn: + conn.execute(''' + INSERT INTO answers (id, question_id, content, author_id, likes, created_at) + VALUES (?, ?, ?, ?, '[]', ?) + ''', (ans_id, question_id, content, author_id, now)) + conn.commit() + + return ans_id \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index d182a4d..1f1069b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,5 @@ flask>=2.3.0 flask-cors>=4.0.0 pyjwt>=2.8.0 werkzeug>=2.3.0 -requests>=2.28.0 \ No newline at end of file +requests>=2.28.0 +sqlite3 # Python内置，无需安装 \ No newline at end of file