hubian
df23e4bd90
功能特性： - 用户认证：login/logout/register/whoami - 帖子操作：posts/post/create/reply/like - 主题操作：topics/topic/topic-create/follow - 其他功能：search/tags/status - Token持久化：登录状态保存在 ~/.tech-forum/token.json - 多种输入：参数输入或交互式输入 安装方式： cp cli/tech-forum.py ~/.local/bin/tech-forum && chmod +x ~/.local/bin/tech-forum
587 lines
18 KiB
Python
Executable File
587 lines
18 KiB
Python
Executable File
#!/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() |