diff --git a/cli/tech-forum.py b/cli/tech-forum.py new file mode 100755 index 0000000..b810aa9 --- /dev/null +++ b/cli/tech-forum.py @@ -0,0 +1,587 @@ +#!/usr/bin/env python3 +""" +技术论坛命令行工具 +支持用户认证和多种操作 +""" + +import argparse +import sys +import os +import json +import urllib.request +import urllib.error +from pathlib import Path +from datetime import datetime + +# 配置 +API_BASE = "http://localhost:19004/api" +CONFIG_DIR = Path.home() / ".tech-forum" +TOKEN_FILE = CONFIG_DIR / "token.json" + +# 确保配置目录存在 +CONFIG_DIR.mkdir(parents=True, exist_ok=True) + + +def api_request(method, path, data=None, token=None): + """发送API请求""" + url = f"{API_BASE}{path}" + headers = {"Content-Type": "application/json"} + + if token: + headers["Authorization"] = f"Bearer {token}" + + body = json.dumps(data).encode('utf-8') if data else None + + try: + req = urllib.request.Request(url, data=body, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=10) as response: + result = json.loads(response.read().decode('utf-8')) + return True, result + except urllib.error.HTTPError as e: + try: + error_body = json.loads(e.read().decode('utf-8')) + return False, error_body.get('error', f'HTTP错误: {e.code}') + except: + return False, f'HTTP错误: {e.code}' + except urllib.error.URLError as e: + return False, f'连接失败: {e.reason}' + except Exception as e: + return False, f'请求异常: {e}' + + +def save_token(username, token, user_id): + """保存登录token""" + data = { + "username": username, + "token": token, + "user_id": user_id, + "login_time": datetime.now().isoformat() + } + TOKEN_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2)) + print(f"✅ 登录成功!Token已保存到 {TOKEN_FILE}") + + +def load_token(): + """读取保存的token""" + if not TOKEN_FILE.exists(): + return None + + try: + data = json.loads(TOKEN_FILE.read_text()) + return data + except: + return None + + +def clear_token(): + """清除token""" + if TOKEN_FILE.exists(): + TOKEN_FILE.unlink() + print("✅ 已退出登录") + + +def check_login(): + """检查是否已登录""" + token_data = load_token() + if not token_data: + print("❌ 未登录,请先使用 tech-forum login 登录") + sys.exit(1) + + # 验证token是否有效 + ok, result = api_request("GET", "/user", token=token_data['token']) + if not ok: + print(f"❌ Token已过期: {result}") + clear_token() + sys.exit(1) + + return token_data + + +# ============ 命令实现 ============ + +def cmd_login(args): + """登录命令""" + username = args.username + password = args.password + + if not username: + username = input("用户名: ") + if not password: + password = input("密码: ") + + ok, result = api_request("POST", "/login", { + "username": username, + "password": password + }) + + if ok: + save_token(username, result['token'], result['user']['id']) + print(f"欢迎回来,{username}!") + else: + print(f"❌ 登录失败: {result}") + + +def cmd_logout(args): + """退出登录""" + clear_token() + + +def cmd_register(args): + """注册命令""" + username = args.username + email = args.email + password = args.password + + if not username: + username = input("用户名: ") + if not email: + email = input("邮箱: ") + if not password: + password = input("密码: ") + + ok, result = api_request("POST", "/register", { + "username": username, + "email": email, + "password": password + }) + + if ok: + print(f"✅ 注册成功!用户ID: {result['user']['id']}") + print(f"现在可以使用 tech-forum login 登录") + else: + print(f"❌ 注册失败: {result}") + + +def cmd_whoami(args): + """查看当前登录用户""" + token_data = check_login() + + ok, result = api_request("GET", "/user", token=token_data['token']) + if ok: + print(f"用户名: {result['username']}") + print(f"邮箱: {result['email']}") + print(f"用户ID: {result['id']}") + else: + print(f"❌ 获取用户信息失败: {result}") + + +def cmd_posts(args): + """查看帖子列表""" + page = args.page or 1 + post_type = args.type + + params = f"?page={page}" + if post_type: + params += f"&type={post_type}" + + ok, result = api_request("GET", f"/posts{params}") + + if ok: + posts = result['posts'] + total = result.get('total', len(posts)) + + print(f"\n📝 帖子列表 (第{page}页,共{total}篇)") + print("-" * 60) + + for post in posts: + pinned = "📌 " if post.get('is_pinned') else "" + # 适配API格式: author是嵌套对象, likes是整数 + author_info = post.get('author', {}) + author_name = author_info.get('username', '未知') if isinstance(author_info, dict) else '未知' + likes_count = post.get('likes', 0) if isinstance(post.get('likes'), int) else len(post.get('likes', [])) + replies_count = post.get('replies', 0) + print(f"{pinned}[{post['id'][:8]}] {post['title']}") + print(f" 类型: {post['type']} | 作者: {author_name} | 浏览: {post['views']} | 点赞: {likes_count} | 回复: {replies_count}") + tags = post.get('tags', []) + if tags: + print(f" 标签: {', '.join(tags)}") + print() + + if not posts: + print("暂无帖子") + else: + print(f"❌ 获取帖子失败: {result}") + + +def cmd_post(args): + """查看帖子详情""" + post_id = args.id + + ok, result = api_request("GET", f"/posts/{post_id}") + + if ok: + post = result + print(f"\n📝 {post['title']}") + print("-" * 60) + # 适配API格式: author是嵌套对象 + author_info = post.get('author', {}) + author_name = author_info.get('username', '未知') if isinstance(author_info, dict) else '未知' + print(f"作者: {author_name}") + print(f"类型: {post['type']}") + likes_count = post.get('likes', 0) if isinstance(post.get('likes'), int) else len(post.get('likes', [])) + print(f"浏览: {post['views']} | 点赞: {likes_count}") + tags = post.get('tags', []) + if tags: + print(f"标签: {', '.join(tags)}") + print("-" * 60) + print(post.get('content', '(无内容)')) + print() + + # 显示回复 + replies = post.get('replies', []) + if replies: + print(f"\n💬 回复 ({len(replies)}条)") + print("-" * 40) + for reply in replies: + reply_author = reply.get('author', {}) + reply_name = reply_author.get('username', '未知') if isinstance(reply_author, dict) else '未知' + print(f"[{reply['id'][:8]}] {reply_name}:") + print(f" {reply['content']}") + print() + else: + print(f"❌ 获取帖子失败: {result}") + + +def cmd_create(args): + """创建帖子""" + token_data = check_login() + + title = args.title + content = args.content + post_type = args.type or "discussion" + tags = args.tags or [] + + if not title: + title = input("标题: ") + if not content: + print("内容(输入空行结束):") + lines = [] + while True: + line = input() + if not line: + break + lines.append(line) + content = "\n".join(lines) + + ok, result = api_request("POST", "/posts", { + "title": title, + "content": content, + "type": post_type, + "tags": tags + }, token=token_data['token']) + + if ok: + print(f"✅ 帖子发布成功!ID: {result['post']['id']}") + print(f"查看: tech-forum post {result['post']['id']}") + else: + print(f"❌ 发布失败: {result}") + + +def cmd_reply(args): + """回复帖子""" + token_data = check_login() + + post_id = args.id + content = args.content + + if not content: + print("回复内容(输入空行结束):") + lines = [] + while True: + line = input() + if not line: + break + lines.append(line) + content = "\n".join(lines) + + ok, result = api_request("POST", f"/posts/{post_id}/reply", { + "content": content + }, token=token_data['token']) + + if ok: + print(f"✅ 回复成功!") + else: + print(f"❌ 回复失败: {result}") + + +def cmd_like(args): + """点赞帖子""" + token_data = check_login() + + post_id = args.id + + ok, result = api_request("POST", f"/posts/{post_id}/like", token=token_data['token']) + + if ok: + liked = result.get('liked', False) + count = result.get('count', 0) + action = "点赞" if liked else "取消点赞" + print(f"✅ 已{action}!当前点赞数: {count}") + else: + print(f"❌ 操作失败: {result}") + + +def cmd_topics(args): + """查看主题列表""" + ok, result = api_request("GET", "/topics") + + if ok: + # API直接返回数组 + topics = result if isinstance(result, list) else result.get('topics', []) + + print(f"\n🔧 主题列表 (共{len(topics)}个)") + print("-" * 60) + + for topic in topics: + followers = len(topic.get('followers', [])) if isinstance(topic.get('followers'), list) else topic.get('followers', 0) + print(f"[{topic['id'][:8]}] {topic.get('icon', '🔧')} {topic['name']}") + print(f" {topic.get('description', '无描述')}") + print(f" 关注者: {followers}") + print() + + if not topics: + print("暂无主题") + else: + print(f"❌ 获取主题失败: {result}") + + +def cmd_topic(args): + """查看主题详情""" + topic_id = args.id + + ok, result = api_request("GET", f"/topics/{topic_id}") + + if ok: + print(f"\n{result['icon']} {result['name']}") + print("-" * 60) + print(result.get('description', '无描述')) + print(f"关注者: {len(result.get('followers', []))}") + print() + + # 显示子主题 + sub_topics = result.get('sub_topics', []) + if sub_topics: + print(f"📂 子主题 ({len(sub_topics)}个)") + print("-" * 40) + for sub in sub_topics: + print(f"- {sub['title']}") + + # 显示问题 + questions = result.get('questions', []) + if questions: + print(f"\n❓ 问题 ({len(questions)}个)") + print("-" * 40) + for q in questions: + answers = len(q.get('answers', [])) + print(f"[{q['id'][:8]}] {q['title']} ({answers}个回答)") + else: + print(f"❌ 获取主题失败: {result}") + + +def cmd_topic_create(args): + """创建主题""" + token_data = check_login() + + name = args.name + description = args.description or "" + icon = args.icon or "🔧" + + if not name: + name = input("主题名称: ") + if not description: + description = input("主题描述: ") + + ok, result = api_request("POST", "/topics", { + "name": name, + "description": description, + "icon": icon + }, token=token_data['token']) + + if ok: + print(f"✅ 主题创建成功!ID: {result['topic']['id']}") + else: + print(f"❌ 创建失败: {result}") + + +def cmd_follow(args): + """关注主题""" + token_data = check_login() + + topic_id = args.id + + ok, result = api_request("POST", f"/topics/{topic_id}/follow", token=token_data['token']) + + if ok: + followed = result.get('followed', False) + count = result.get('count', 0) + action = "关注" if followed else "取消关注" + print(f"✅ 已{action}!当前关注者: {count}") + else: + print(f"❌ 操作失败: {result}") + + +def cmd_search(args): + """搜索""" + query = args.query + + if not query: + query = input("搜索关键词: ") + + ok, result = api_request("GET", f"/search?q={query}") + + if ok: + posts = result.get('posts', []) + topics = result.get('topics', []) + + print(f"\n🔍 搜索结果: {query}") + print("-" * 60) + + if posts: + print(f"\n📝 帖子 ({len(posts)}个)") + for post in posts: + print(f" [{post['id'][:8]}] {post['title']}") + + if topics: + print(f"\n🔧 主题 ({len(topics)}个)") + for topic in topics: + print(f" [{topic['id'][:8]}] {topic['icon']} {topic['name']}") + + if not posts and not topics: + print("未找到相关内容") + else: + print(f"❌ 搜索失败: {result}") + + +def cmd_tags(args): + """查看热门标签""" + ok, result = api_request("GET", "/tags") + + if ok: + # API直接返回数组 [[tag, count], ...] + tags = result if isinstance(result, list) else result.get('tags', []) + + print(f"\n🏷️ 热门标签") + print("-" * 40) + + for item in tags[:20]: + if isinstance(item, (list, tuple)) and len(item) >= 2: + tag, count = item[0], item[1] + elif isinstance(item, dict): + tag = item.get('name', '未知') + count = item.get('count', 1) + else: + tag = str(item) + count = 1 + print(f" {tag}: {count}篇") + + if not tags: + print("暂无标签") + else: + print(f"❌ 获取标签失败: {result}") + + +def cmd_status(args): + """查看服务状态""" + ok, result = api_request("GET", "/health") + + if ok: + print("✅ 服务运行正常") + print(f"API地址: {API_BASE}") + else: + print(f"❌ 服务异常: {result}") + + +def main(): + parser = argparse.ArgumentParser( + description="技术论坛命令行工具", + prog="tech-forum" + ) + + subparsers = parser.add_subparsers(dest="command", help="可用命令") + + # 认证相关 + login_parser = subparsers.add_parser("login", help="登录") + login_parser.add_argument("-u", "--username", help="用户名") + login_parser.add_argument("-p", "--password", help="密码") + login_parser.set_defaults(func=cmd_login) + + logout_parser = subparsers.add_parser("logout", help="退出登录") + logout_parser.set_defaults(func=cmd_logout) + + register_parser = subparsers.add_parser("register", help="注册") + register_parser.add_argument("-u", "--username", help="用户名") + register_parser.add_argument("-e", "--email", help="邮箱") + register_parser.add_argument("-p", "--password", help="密码") + register_parser.set_defaults(func=cmd_register) + + whoami_parser = subparsers.add_parser("whoami", help="查看当前用户") + whoami_parser.set_defaults(func=cmd_whoami) + + # 帖子相关 + posts_parser = subparsers.add_parser("posts", help="查看帖子列表") + posts_parser.add_argument("-p", "--page", type=int, help="页码") + posts_parser.add_argument("-t", "--type", help="帖子类型 (discussion/share/question)") + posts_parser.set_defaults(func=cmd_posts) + + post_parser = subparsers.add_parser("post", help="查看帖子详情") + post_parser.add_argument("id", help="帖子ID") + post_parser.set_defaults(func=cmd_post) + + create_parser = subparsers.add_parser("create", help="发布帖子") + create_parser.add_argument("-t", "--title", help="标题") + create_parser.add_argument("-c", "--content", help="内容") + create_parser.add_argument("--type", default="discussion", help="类型") + create_parser.add_argument("--tags", nargs="+", help="标签") + create_parser.set_defaults(func=cmd_create) + + reply_parser = subparsers.add_parser("reply", help="回复帖子") + reply_parser.add_argument("id", help="帖子ID") + reply_parser.add_argument("-c", "--content", help="回复内容") + reply_parser.set_defaults(func=cmd_reply) + + like_parser = subparsers.add_parser("like", help="点赞帖子") + like_parser.add_argument("id", help="帖子ID") + like_parser.set_defaults(func=cmd_like) + + # 主题相关 + topics_parser = subparsers.add_parser("topics", help="查看主题列表") + topics_parser.set_defaults(func=cmd_topics) + + topic_parser = subparsers.add_parser("topic", help="查看主题详情") + topic_parser.add_argument("id", help="主题ID") + topic_parser.set_defaults(func=cmd_topic) + + topic_create_parser = subparsers.add_parser("topic-create", help="创建主题") + topic_create_parser.add_argument("-n", "--name", help="主题名称") + topic_create_parser.add_argument("-d", "--description", help="描述") + topic_create_parser.add_argument("--icon", default="🔧", help="图标") + topic_create_parser.set_defaults(func=cmd_topic_create) + + follow_parser = subparsers.add_parser("follow", help="关注主题") + follow_parser.add_argument("id", help="主题ID") + follow_parser.set_defaults(func=cmd_follow) + + # 其他 + search_parser = subparsers.add_parser("search", help="搜索") + search_parser.add_argument("query", nargs="?", help="搜索关键词") + search_parser.set_defaults(func=cmd_search) + + tags_parser = subparsers.add_parser("tags", help="热门标签") + tags_parser.set_defaults(func=cmd_tags) + + status_parser = subparsers.add_parser("status", help="服务状态") + status_parser.set_defaults(func=cmd_status) + + # 解析参数 + args = parser.parse_args() + + if not args.command: + parser.print_help() + sys.exit(0) + + # 执行命令 + args.func(args) + + +if __name__ == "__main__": + main() \ No newline at end of file