e105b573da
- 新增 cli.py 命令行工具
- 支持所有核心功能操作
- 新增 CLI_USAGE.md 使用文档
- 所有命令测试通过 ✅
功能列表:
- login: 登录认证
- provinces: 获取省份
- article: 文章管理
- service: 服务管理
- audit: AI 审核
423 lines
12 KiB
Python
423 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
城市手册 - 命令行接口工具
|
||
|
||
用法:
|
||
python cli.py <命令> [参数]
|
||
|
||
示例:
|
||
python cli.py login admin Admin123!
|
||
python cli.py provinces
|
||
python cli.py article list
|
||
python cli.py article create "标题" "内容" 1
|
||
python cli.py audit article "标题" "内容"
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import urllib.request
|
||
import urllib.error
|
||
|
||
# Django 设置
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.prod')
|
||
|
||
import django
|
||
django.setup()
|
||
|
||
from django.contrib.auth import get_user_model
|
||
from rest_framework_simplejwt.tokens import RefreshToken
|
||
|
||
# 配置
|
||
BASE_URL = 'http://localhost:8000'
|
||
ACCESS_TOKEN = None
|
||
|
||
|
||
def get_token(username='admin'):
|
||
"""获取 JWT Token"""
|
||
User = get_user_model()
|
||
user = User.objects.filter(username=username).first()
|
||
if not user:
|
||
print(f'❌ 用户 {username} 不存在')
|
||
return None
|
||
|
||
refresh = RefreshToken.for_user(user)
|
||
return str(refresh.access_token)
|
||
|
||
|
||
def api_request(endpoint, method='GET', data=None, token=None):
|
||
"""发送 API 请求"""
|
||
url = f'{BASE_URL}{endpoint}'
|
||
headers = {'Content-Type': 'application/json'}
|
||
|
||
if token:
|
||
headers['Authorization'] = f'Bearer {token}'
|
||
|
||
if data:
|
||
data = json.dumps(data, ensure_ascii=False).encode('utf-8')
|
||
|
||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as response:
|
||
return json.loads(response.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
error_body = e.read().decode()
|
||
try:
|
||
error_data = json.loads(error_body)
|
||
return {'error': e.code, 'message': error_data}
|
||
except:
|
||
return {'error': e.code, 'message': error_body}
|
||
except Exception as e:
|
||
return {'error': 'network', 'message': str(e)}
|
||
|
||
|
||
def cmd_login(username, password):
|
||
"""登录获取 Token"""
|
||
result = api_request('/api/auth/login/', 'POST', {
|
||
'username': username,
|
||
'password': password
|
||
})
|
||
|
||
if 'access' in result:
|
||
print(f'✅ 登录成功')
|
||
print(f'Token: {result["access"][:50]}...')
|
||
return result['access']
|
||
else:
|
||
print(f'❌ 登录失败：{result}')
|
||
return None
|
||
|
||
|
||
def cmd_provinces():
|
||
"""获取所有省份"""
|
||
result = api_request('/api/regions/provinces/')
|
||
|
||
if isinstance(result, list):
|
||
print(f'✅ 共 {len(result)} 个省份:')
|
||
for i, p in enumerate(result, 1):
|
||
print(f' {i}. {p["name"]} (ID: {p["id"]})')
|
||
return result
|
||
else:
|
||
print(f'❌ 获取失败：{result}')
|
||
return None
|
||
|
||
|
||
def cmd_region_detail(region_id):
|
||
"""获取省份详情"""
|
||
result = api_request(f'/api/regions/{region_id}/')
|
||
|
||
if isinstance(result, dict) and 'id' in result:
|
||
print(f'✅ 省份信息:')
|
||
print(f' 名称：{result.get("name")}')
|
||
print(f' 级别：{result.get("level")}')
|
||
print(f' 状态：{result.get("status")}')
|
||
return result
|
||
else:
|
||
print(f'❌ 获取失败：{result}')
|
||
return None
|
||
|
||
|
||
def cmd_article_list(limit=10):
|
||
"""获取文章列表"""
|
||
token = get_token()
|
||
result = api_request(f'/api/articles/?limit={limit}', token=token)
|
||
|
||
if isinstance(result, dict) and 'results' in result:
|
||
articles = result['results']
|
||
print(f'✅ 共 {result.get("count", 0)} 篇文章:')
|
||
for i, a in enumerate(articles, 1):
|
||
print(f' {i}. [{a.get("id")}] {a.get("title")} - {a.get("publish_status")}')
|
||
return result
|
||
else:
|
||
print(f'❌ 获取失败：{result}')
|
||
return None
|
||
|
||
|
||
def cmd_article_create(title, content, region_id, article_type='basic'):
|
||
"""创建文章"""
|
||
token = get_token()
|
||
if not token:
|
||
return None
|
||
|
||
result = api_request('/api/articles/', 'POST', {
|
||
'title': title,
|
||
'content': content,
|
||
'region': region_id,
|
||
'article_type': article_type
|
||
}, token=token)
|
||
|
||
if 'id' in result:
|
||
print(f'✅ 文章创建成功 (ID: {result["id"]})')
|
||
print(f' 标题：{result.get("title")}')
|
||
print(f' 状态：{result.get("publish_status")}')
|
||
return result
|
||
else:
|
||
print(f'❌ 创建失败：{result}')
|
||
return None
|
||
|
||
|
||
def cmd_article_submit(article_id):
|
||
"""提交文章审核"""
|
||
token = get_token()
|
||
if not token:
|
||
return None
|
||
|
||
result = api_request(f'/api/articles/{article_id}/submit/', 'POST', {}, token=token)
|
||
|
||
if 'id' in result:
|
||
print(f'✅ 文章已提交审核 (ID: {article_id})')
|
||
print(f' 版主审核状态：{result.get("moderator_status")}')
|
||
print(f' AI 审核状态：{result.get("ai_status")}')
|
||
return result
|
||
else:
|
||
print(f'❌ 提交失败：{result}')
|
||
return None
|
||
|
||
|
||
def cmd_audit_article(title, content):
|
||
"""AI 审核文章"""
|
||
token = get_token()
|
||
if not token:
|
||
return None
|
||
|
||
result = api_request('/api/audit/article/', 'POST', {
|
||
'title': title,
|
||
'content': content
|
||
}, token=token)
|
||
|
||
status = '✅ 通过' if result.get('approved') else '❌ 拒绝'
|
||
print(f'AI 审核结果：{status}')
|
||
print(f' 原因：{result.get("reason")}')
|
||
if 'details' in result:
|
||
print(f' 详情：{json.dumps(result["details"], ensure_ascii=False, indent=2)}')
|
||
return result
|
||
|
||
|
||
def cmd_audit_comment(content):
|
||
"""AI 审核评论"""
|
||
token = get_token()
|
||
if not token:
|
||
return None
|
||
|
||
result = api_request('/api/audit/comment/', 'POST', {
|
||
'content': content
|
||
}, token=token)
|
||
|
||
status = '✅ 通过' if result.get('approved') else '❌ 拒绝'
|
||
print(f'AI 审核结果：{status}')
|
||
print(f' 原因：{result.get("reason")}')
|
||
return result
|
||
|
||
|
||
def cmd_audit_service(name, description):
|
||
"""AI 审核服务"""
|
||
token = get_token()
|
||
if not token:
|
||
return None
|
||
|
||
result = api_request('/api/audit/service/', 'POST', {
|
||
'name': name,
|
||
'description': description
|
||
}, token=token)
|
||
|
||
status = '✅ 通过' if result.get('approved') else '❌ 拒绝'
|
||
print(f'AI 审核结果：{status}')
|
||
print(f' 原因：{result.get("reason")}')
|
||
return result
|
||
|
||
|
||
def cmd_audit_status():
|
||
"""AI 审核服务状态"""
|
||
token = get_token()
|
||
result = api_request('/api/audit/status/', token=token)
|
||
|
||
if 'status' in result:
|
||
print(f'✅ AI 审核服务状态：{result["status"]}')
|
||
print(f' 版本：{result.get("version")}')
|
||
print(f' 功能：{", ".join(result.get("features", []))}')
|
||
return result
|
||
else:
|
||
print(f'❌ 获取失败：{result}')
|
||
return None
|
||
|
||
|
||
def cmd_service_list(limit=10):
|
||
"""获取服务列表"""
|
||
token = get_token()
|
||
result = api_request(f'/api/services/?limit={limit}', token=token)
|
||
|
||
if isinstance(result, dict) and 'results' in result:
|
||
services = result['results']
|
||
print(f'✅ 共 {result.get("count", 0)} 个服务:')
|
||
for i, s in enumerate(services, 1):
|
||
print(f' {i}. [{s.get("id")}] {s.get("name")} - {s.get("category")}')
|
||
return result
|
||
else:
|
||
print(f'❌ 获取失败：{result}')
|
||
return None
|
||
|
||
|
||
def cmd_service_create(name, description, region_id, category='food'):
|
||
"""创建特色服务"""
|
||
token = get_token()
|
||
if not token:
|
||
return None
|
||
|
||
result = api_request('/api/services/', 'POST', {
|
||
'name': name,
|
||
'description': description,
|
||
'region': region_id,
|
||
'category': category
|
||
}, token=token)
|
||
|
||
if 'id' in result:
|
||
print(f'✅ 服务创建成功 (ID: {result["id"]})')
|
||
print(f' 名称：{result.get("name")}')
|
||
print(f' 分类：{result.get("category")}')
|
||
return result
|
||
else:
|
||
print(f'❌ 创建失败：{result}')
|
||
return None
|
||
|
||
|
||
def cmd_help():
|
||
"""显示帮助信息"""
|
||
help_text = """
|
||
城市手册 - 命令行接口
|
||
|
||
用法：python cli.py <命令> [参数]
|
||
|
||
认证命令:
|
||
login <用户名> <密码> 登录获取 Token
|
||
|
||
省份命令:
|
||
provinces 获取所有省份
|
||
region <ID> 获取省份详情
|
||
|
||
文章命令:
|
||
article list [limit] 获取文章列表
|
||
article create <标题> <内容> <省份 ID> [类型]
|
||
创建文章
|
||
article submit <ID> 提交文章审核
|
||
|
||
服务命令:
|
||
service list [limit] 获取服务列表
|
||
service create <名称> <描述> <省份 ID> [分类]
|
||
创建服务
|
||
|
||
AI 审核命令:
|
||
audit article <标题> <内容> AI 审核文章
|
||
audit comment <内容> AI 审核评论
|
||
audit service <名称> <描述> AI 审核服务
|
||
audit status AI 审核服务状态
|
||
|
||
示例:
|
||
python cli.py login admin Admin123!
|
||
python cli.py provinces
|
||
python cli.py article create "北京攻略" "北京是首都..." 1
|
||
python cli.py audit article "测试" "包含暴力内容"
|
||
python cli.py audit status
|
||
"""
|
||
print(help_text)
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) < 2:
|
||
cmd_help()
|
||
return
|
||
|
||
command = sys.argv[1]
|
||
|
||
# 登录命令
|
||
if command == 'login':
|
||
if len(sys.argv) < 4:
|
||
print('用法：python cli.py login <用户名> <密码>')
|
||
return
|
||
cmd_login(sys.argv[2], sys.argv[3])
|
||
|
||
# 省份命令
|
||
elif command == 'provinces':
|
||
cmd_provinces()
|
||
|
||
elif command == 'region':
|
||
if len(sys.argv) < 3:
|
||
print('用法：python cli.py region <ID>')
|
||
return
|
||
cmd_region_detail(sys.argv[2])
|
||
|
||
# 文章命令
|
||
elif command == 'article':
|
||
if len(sys.argv) < 3:
|
||
print('用法：python cli.py article <list|create|submit> [参数]')
|
||
return
|
||
|
||
subcommand = sys.argv[2]
|
||
if subcommand == 'list':
|
||
limit = sys.argv[3] if len(sys.argv) > 3 else 10
|
||
cmd_article_list(limit)
|
||
elif subcommand == 'create':
|
||
if len(sys.argv) < 6:
|
||
print('用法：python cli.py article create <标题> <内容> <省份 ID> [类型]')
|
||
return
|
||
cmd_article_create(sys.argv[3], sys.argv[4], sys.argv[5],
|
||
sys.argv[6] if len(sys.argv) > 6 else 'basic')
|
||
elif subcommand == 'submit':
|
||
if len(sys.argv) < 4:
|
||
print('用法：python cli.py article submit <ID>')
|
||
return
|
||
cmd_article_submit(sys.argv[3])
|
||
|
||
# 服务命令
|
||
elif command == 'service':
|
||
if len(sys.argv) < 3:
|
||
print('用法：python cli.py service <list|create> [参数]')
|
||
return
|
||
|
||
subcommand = sys.argv[2]
|
||
if subcommand == 'list':
|
||
limit = sys.argv[3] if len(sys.argv) > 3 else 10
|
||
cmd_service_list(limit)
|
||
elif subcommand == 'create':
|
||
if len(sys.argv) < 6:
|
||
print('用法：python cli.py service create <名称> <描述> <省份 ID> [分类]')
|
||
return
|
||
cmd_service_create(sys.argv[3], sys.argv[4], sys.argv[5],
|
||
sys.argv[6] if len(sys.argv) > 6 else 'food')
|
||
|
||
# AI 审核命令
|
||
elif command == 'audit':
|
||
if len(sys.argv) < 3:
|
||
print('用法：python cli.py audit <article|comment|service|status> [参数]')
|
||
return
|
||
|
||
subcommand = sys.argv[2]
|
||
if subcommand == 'article':
|
||
if len(sys.argv) < 5:
|
||
print('用法：python cli.py audit article <标题> <内容>')
|
||
return
|
||
cmd_audit_article(sys.argv[3], sys.argv[4])
|
||
elif subcommand == 'comment':
|
||
if len(sys.argv) < 4:
|
||
print('用法：python cli.py audit comment <内容>')
|
||
return
|
||
cmd_audit_comment(sys.argv[3])
|
||
elif subcommand == 'service':
|
||
if len(sys.argv) < 5:
|
||
print('用法：python cli.py audit service <名称> <描述>')
|
||
return
|
||
cmd_audit_service(sys.argv[3], sys.argv[4])
|
||
elif subcommand == 'status':
|
||
cmd_audit_status()
|
||
|
||
# 帮助命令
|
||
elif command == 'help' or command == '--help' or command == '-h':
|
||
cmd_help()
|
||
|
||
else:
|
||
print(f'❌ 未知命令：{command}')
|
||
print('使用 python cli.py help 查看帮助')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|