825 lines
27 KiB
Python
825 lines
27 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
ap_ds 代码仓库系统
|
||
完整的代码托管与文档展示平台
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import shutil
|
||
import zipfile
|
||
import tarfile
|
||
import hashlib
|
||
import socket
|
||
import mimetypes
|
||
import threading
|
||
import subprocess
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from functools import wraps
|
||
from flask import (
|
||
Flask, render_template, send_file, request,
|
||
redirect, url_for, session, jsonify, abort
|
||
)
|
||
from markupsafe import Markup
|
||
from werkzeug.utils import secure_filename
|
||
import markdown
|
||
from pygments import highlight
|
||
from pygments.lexers import get_lexer_by_name, guess_lexer
|
||
from pygments.formatters import HtmlFormatter
|
||
|
||
# ============================================
|
||
# 应用初始化
|
||
# ============================================
|
||
app = Flask(__name__)
|
||
app.secret_key = "ap_ds_code_repo_secret_key_2026"
|
||
app.config['MAX_CONTENT_LENGTH'] = 500 * 1024 * 1024 # 500MB
|
||
|
||
# 目录配置
|
||
BASE_DIR = Path(__file__).parent
|
||
CODE_DIR = BASE_DIR / "code"
|
||
PYPI_DIR = BASE_DIR / "pypi"
|
||
STATIC_DIR = BASE_DIR / "static"
|
||
TEMPLATES_DIR = BASE_DIR / "templates"
|
||
DATA_DIR = BASE_DIR / "data"
|
||
UPLOAD_DIR = BASE_DIR / "uploads"
|
||
|
||
# 创建必要目录
|
||
for d in [CODE_DIR, PYPI_DIR, DATA_DIR, UPLOAD_DIR]:
|
||
d.mkdir(exist_ok=True)
|
||
|
||
# 管理员密码
|
||
ADMIN_PASSWORD_HASH = ""
|
||
|
||
def get_version_announcement(branch, version):
|
||
"""获取版本的公告内容"""
|
||
ann_path = CODE_DIR / branch / version / "ANNCMNT.md"
|
||
if ann_path.exists():
|
||
with open(ann_path, 'r', encoding='utf-8', errors='ignore') as f:
|
||
content = f.read()
|
||
return render_markdown(content)
|
||
return None
|
||
|
||
def get_version_readme(branch, version):
|
||
"""获取版本的 README 内容"""
|
||
readme_path = CODE_DIR / branch / version / "README.md"
|
||
if readme_path.exists():
|
||
with open(readme_path, 'r', encoding='utf-8', errors='ignore') as f:
|
||
content = f.read()
|
||
return render_markdown(content)
|
||
return None
|
||
|
||
# ============================================
|
||
# 数据存储辅助函数
|
||
# ============================================
|
||
def load_json(filepath, default=None):
|
||
"""加载JSON文件"""
|
||
if default is None:
|
||
default = []
|
||
try:
|
||
with open(filepath, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except (FileNotFoundError, json.JSONDecodeError):
|
||
return default
|
||
|
||
def save_json(filepath, data):
|
||
"""保存JSON文件"""
|
||
with open(filepath, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
def load_feedbacks():
|
||
"""加载反馈数据"""
|
||
return load_json(DATA_DIR / "feedbacks.json", [])
|
||
|
||
def save_feedbacks(feedbacks):
|
||
"""保存反馈数据"""
|
||
save_json(DATA_DIR / "feedbacks.json", feedbacks)
|
||
|
||
def load_branches():
|
||
"""加载分支列表"""
|
||
branches = load_json(DATA_DIR / "branches.json", {})
|
||
if not branches:
|
||
# 扫描现有目录初始化
|
||
for branch_dir in CODE_DIR.iterdir():
|
||
if branch_dir.is_dir():
|
||
versions = []
|
||
for version_dir in branch_dir.iterdir():
|
||
if version_dir.is_dir():
|
||
versions.append(version_dir.name)
|
||
branches[branch_dir.name] = sorted(versions, reverse=True)
|
||
save_json(DATA_DIR / "branches.json", branches)
|
||
return branches
|
||
|
||
def save_branches(branches):
|
||
"""保存分支列表"""
|
||
save_json(DATA_DIR / "branches.json", branches)
|
||
|
||
# ============================================
|
||
# 邮件发送函数
|
||
# ============================================
|
||
def send_feedback_email(user_email, feedback_content):
|
||
print(f"The original email sending function contains sensitive content. If you need it, please figure it out yourself.")
|
||
|
||
|
||
# ============================================
|
||
# 文件操作辅助函数
|
||
# ============================================
|
||
def get_branches_and_versions():
|
||
"""获取所有分支和版本"""
|
||
branches_data = {}
|
||
for branch_dir in CODE_DIR.iterdir():
|
||
if branch_dir.is_dir():
|
||
versions = []
|
||
for version_dir in sorted(branch_dir.iterdir(), reverse=True):
|
||
if version_dir.is_dir():
|
||
versions.append({
|
||
"name": version_dir.name,
|
||
"path": str(version_dir.relative_to(CODE_DIR)).replace('\\', '/'),
|
||
"created": datetime.fromtimestamp(version_dir.stat().st_ctime).strftime("%Y-%m-%d")
|
||
})
|
||
if versions:
|
||
branches_data[branch_dir.name] = versions
|
||
return branches_data
|
||
|
||
def get_file_tree(branch, version):
|
||
"""获取文件树结构"""
|
||
version_path = CODE_DIR / branch / version
|
||
if not version_path.exists():
|
||
return []
|
||
|
||
tree = []
|
||
for item in sorted(version_path.iterdir()):
|
||
# 使用正斜杠统一路径分隔符
|
||
path_str = str(item.relative_to(CODE_DIR)).replace('\\', '/')
|
||
tree.append({
|
||
"name": item.name,
|
||
"type": "dir" if item.is_dir() else "file",
|
||
"path": path_str,
|
||
"size": item.stat().st_size if item.is_file() else 0,
|
||
"modified": datetime.fromtimestamp(item.stat().st_mtime).strftime("%Y-%m-%d %H:%M")
|
||
})
|
||
return tree
|
||
|
||
def get_file_content(file_path):
|
||
"""获取文件内容"""
|
||
full_path = CODE_DIR / file_path
|
||
if not full_path.exists() or full_path.is_dir():
|
||
return None
|
||
|
||
with open(full_path, 'r', encoding='utf-8', errors='ignore') as f:
|
||
return f.read()
|
||
|
||
def get_file_metadata(file_path):
|
||
"""获取文件元数据"""
|
||
full_path = CODE_DIR / file_path
|
||
if not full_path.exists():
|
||
return None
|
||
|
||
# 使用正斜杠统一路径分隔符
|
||
path_str = str(full_path.relative_to(CODE_DIR)).replace('\\', '/')
|
||
|
||
return {
|
||
"name": full_path.name,
|
||
"path": path_str,
|
||
"size": full_path.stat().st_size,
|
||
"modified": datetime.fromtimestamp(full_path.stat().st_mtime).strftime("%Y-%m-%d %H:%M:%S"),
|
||
"extension": full_path.suffix.lower()
|
||
}
|
||
|
||
def render_markdown(content):
|
||
"""渲染Markdown为HTML"""
|
||
extensions = ['extra', 'codehilite', 'toc', 'nl2br']
|
||
return markdown.markdown(content, extensions=extensions)
|
||
|
||
def syntax_highlight(code, language=None):
|
||
"""语法高亮"""
|
||
if not language:
|
||
try:
|
||
lexer = guess_lexer(code)
|
||
except:
|
||
lexer = get_lexer_by_name("text")
|
||
else:
|
||
try:
|
||
lexer = get_lexer_by_name(language)
|
||
except:
|
||
lexer = get_lexer_by_name("text")
|
||
|
||
formatter = HtmlFormatter(style="monokai", linenos=True)
|
||
return highlight(code, lexer, formatter)
|
||
|
||
def create_zip_archive(branch, version):
|
||
"""创建ZIP压缩包"""
|
||
source_dir = CODE_DIR / branch / version
|
||
if not source_dir.exists():
|
||
return None
|
||
|
||
zip_name = f"{branch}_{version}.zip"
|
||
zip_path = UPLOAD_DIR / zip_name
|
||
|
||
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||
for file_path in source_dir.rglob("*"):
|
||
if file_path.is_file():
|
||
arcname = file_path.relative_to(source_dir)
|
||
zf.write(file_path, arcname)
|
||
|
||
return zip_path
|
||
|
||
def create_tar_archive(branch, version):
|
||
"""创建TAR.GZ压缩包"""
|
||
source_dir = CODE_DIR / branch / version
|
||
if not source_dir.exists():
|
||
return None
|
||
|
||
tar_name = f"{branch}_{version}.tar.gz"
|
||
tar_path = UPLOAD_DIR / tar_name
|
||
|
||
with tarfile.open(tar_path, 'w:gz') as tf:
|
||
tf.add(source_dir, arcname=version)
|
||
|
||
return tar_path
|
||
|
||
def find_whl_file(version):
|
||
"""查找对应版本的whl文件"""
|
||
for whl_file in PYPI_DIR.glob(f"*{version}*.whl"):
|
||
return whl_file
|
||
return None
|
||
|
||
# ============================================
|
||
# 管理员认证装饰器
|
||
# ============================================
|
||
def admin_required(f):
|
||
@wraps(f)
|
||
def decorated_function(*args, **kwargs):
|
||
if not session.get('admin_logged_in'):
|
||
return redirect(url_for('admin_login'))
|
||
return f(*args, **kwargs)
|
||
return decorated_function
|
||
|
||
# ============================================
|
||
# 路由 - 首页
|
||
# ============================================
|
||
@app.route('/')
|
||
def index():
|
||
"""首页"""
|
||
branches_data = get_branches_and_versions()
|
||
|
||
# 获取最新发行版
|
||
releases = []
|
||
for branch, versions in branches_data.items():
|
||
for version in versions[:1]: # 每个分支最新版本
|
||
whl_file = find_whl_file(version['name'])
|
||
releases.append({
|
||
"branch": branch,
|
||
"version": version['name'],
|
||
"date": version['created'],
|
||
"has_whl": whl_file is not None
|
||
})
|
||
|
||
# 获取公告内容
|
||
announcement = None
|
||
for branch, versions in branches_data.items():
|
||
for version in versions:
|
||
ann_path = CODE_DIR / branch / version['name'] / "ANNCMNT.md"
|
||
if ann_path.exists():
|
||
with open(ann_path, 'r', encoding='utf-8', errors='ignore') as f:
|
||
announcement = render_markdown(f.read())
|
||
break
|
||
if announcement:
|
||
break
|
||
|
||
return render_template('index.html',
|
||
branches=branches_data,
|
||
releases=releases[:5],
|
||
announcement=announcement)
|
||
|
||
# ============================================
|
||
# 路由 - 代码浏览器
|
||
# ============================================
|
||
@app.route('/code')
|
||
def code_browser():
|
||
"""代码浏览器主页"""
|
||
branch = request.args.get('branch', '')
|
||
version = request.args.get('version', '')
|
||
|
||
branches_data = get_branches_and_versions()
|
||
|
||
# 如果没有指定分支，使用第一个
|
||
if not branch and branches_data:
|
||
branch = list(branches_data.keys())[0]
|
||
if branches_data[branch]:
|
||
version = branches_data[branch][0]['name']
|
||
|
||
file_tree = get_file_tree(branch, version) if branch and version else []
|
||
|
||
return render_template('code_browser.html',
|
||
branches=branches_data,
|
||
current_branch=branch,
|
||
current_version=version,
|
||
file_tree=file_tree)
|
||
|
||
# ============================================
|
||
# 路由 - 目录浏览
|
||
# ============================================
|
||
@app.route('/browse/<path:dir_path>')
|
||
def browse_directory(dir_path):
|
||
"""浏览子目录"""
|
||
branch = request.args.get('branch', '')
|
||
version = request.args.get('version', '')
|
||
|
||
full_dir = CODE_DIR / dir_path
|
||
if not full_dir.exists() or not full_dir.is_dir():
|
||
abort(404)
|
||
|
||
files = []
|
||
for item in sorted(full_dir.iterdir()):
|
||
files.append({
|
||
"name": item.name,
|
||
"type": "dir" if item.is_dir() else "file",
|
||
"path": str(item.relative_to(CODE_DIR)).replace('\\', '/'),
|
||
"size": item.stat().st_size if item.is_file() else 0,
|
||
"modified": datetime.fromtimestamp(item.stat().st_mtime).strftime("%Y-%m-%d %H:%M")
|
||
})
|
||
|
||
# 面包屑导航
|
||
breadcrumbs = []
|
||
parts = Path(dir_path).parts
|
||
current = ""
|
||
for part in parts:
|
||
current = str(Path(current) / part) if current else part
|
||
breadcrumbs.append({"name": part, "path": current.replace('\\', '/')})
|
||
|
||
return render_template('directory.html',
|
||
files=files,
|
||
current_path=dir_path,
|
||
breadcrumbs=breadcrumbs,
|
||
branch=branch,
|
||
version=version)
|
||
|
||
@app.route('/view/<path:file_path>')
|
||
def view_file(file_path):
|
||
"""查看文件内容"""
|
||
branch = request.args.get('branch', '')
|
||
version = request.args.get('version', '')
|
||
|
||
# 将反斜杠替换为正斜杠
|
||
file_path = file_path.replace('\\', '/')
|
||
|
||
# 构建完整路径
|
||
if branch and version:
|
||
# 如果 file_path 已经包含分支，直接使用
|
||
if file_path.startswith(f"{branch}/"):
|
||
full_path = CODE_DIR / file_path
|
||
else:
|
||
# 否则拼接分支和版本
|
||
full_path = CODE_DIR / branch / version / file_path
|
||
else:
|
||
full_path = CODE_DIR / file_path
|
||
|
||
if not full_path.exists() or full_path.is_dir():
|
||
abort(404)
|
||
|
||
metadata = get_file_metadata(str(full_path.relative_to(CODE_DIR)))
|
||
extension = full_path.suffix.lower()
|
||
|
||
# Python文件 - 语法高亮
|
||
if extension == '.py':
|
||
content = get_file_content(str(full_path.relative_to(CODE_DIR)))
|
||
highlighted = syntax_highlight(content, 'python')
|
||
return render_template('view_code.html',
|
||
metadata=metadata,
|
||
content=content,
|
||
highlighted_code=highlighted,
|
||
language='python',
|
||
branch=branch,
|
||
version=version)
|
||
|
||
# Markdown文件 - 渲染
|
||
elif extension in ['.md', '.markdown']:
|
||
content = get_file_content(str(full_path.relative_to(CODE_DIR)))
|
||
html_content = render_markdown(content)
|
||
return render_template('view_markdown.html',
|
||
metadata=metadata,
|
||
html_content=Markup(html_content),
|
||
branch=branch,
|
||
version=version)
|
||
|
||
# 文本文件
|
||
elif extension in ['.txt', '.json', '.xml', '.yaml', '.yml', '.conf', '.cfg', '.ini']:
|
||
content = get_file_content(str(full_path.relative_to(CODE_DIR)))
|
||
return render_template('view_file.html',
|
||
metadata=metadata,
|
||
content=content,
|
||
is_text=True,
|
||
branch=branch,
|
||
version=version)
|
||
|
||
# 其他文件 - 显示下载选项
|
||
else:
|
||
return render_template('view_file.html',
|
||
metadata=metadata,
|
||
is_text=False,
|
||
branch=branch,
|
||
version=version)
|
||
|
||
# ============================================
|
||
# 路由 - 下载文件
|
||
# ============================================
|
||
@app.route('/download/<path:file_path>')
|
||
def download_file(file_path):
|
||
"""下载单个文件"""
|
||
file_path = file_path.replace('\\', '/')
|
||
full_path = CODE_DIR / file_path
|
||
if not full_path.exists() or full_path.is_dir():
|
||
abort(404)
|
||
|
||
return send_file(full_path, as_attachment=True)
|
||
|
||
# ============================================
|
||
# 路由 - 下载压缩包
|
||
# ============================================
|
||
@app.route('/download/zip/<branch>/<version>')
|
||
def download_zip(branch, version):
|
||
"""下载ZIP压缩包"""
|
||
zip_path = create_zip_archive(branch, version)
|
||
if not zip_path:
|
||
abort(404)
|
||
|
||
return send_file(zip_path, as_attachment=True, download_name=f"{branch}_{version}.zip")
|
||
|
||
@app.route('/download/tar/<branch>/<version>')
|
||
def download_tar(branch, version):
|
||
"""下载TAR.GZ压缩包"""
|
||
tar_path = create_tar_archive(branch, version)
|
||
if not tar_path:
|
||
abort(404)
|
||
|
||
return send_file(tar_path, as_attachment=True, download_name=f"{branch}_{version}.tar.gz")
|
||
|
||
@app.route('/download/whl/<version>')
|
||
def download_whl(version):
|
||
"""下载whl文件 - 支持带v和不带v的版本号"""
|
||
# 去除版本号开头的 'v' 如果存在
|
||
clean_version = version.lstrip('v')
|
||
|
||
# 先尝试用原始版本号查找
|
||
whl_file = find_whl_file(version)
|
||
|
||
# 如果没找到，尝试用去掉v的版本号查找
|
||
if not whl_file and clean_version != version:
|
||
whl_file = find_whl_file(clean_version)
|
||
|
||
# 如果还没找到，尝试模糊匹配（直接搜索包含版本号的whl文件）
|
||
if not whl_file:
|
||
for whl in PYPI_DIR.glob(f"*{clean_version}*.whl"):
|
||
whl_file = whl
|
||
break
|
||
|
||
if not whl_file or not whl_file.exists():
|
||
abort(404)
|
||
|
||
return send_file(whl_file, as_attachment=True, download_name=whl_file.name)
|
||
|
||
@app.route('/api/version-info')
|
||
def api_version_info():
|
||
"""获取版本信息"""
|
||
return jsonify({
|
||
"version": "v3.0.0 LTS",
|
||
"title": "轻量级 Python 音频革命",
|
||
"subtitle": "2.5MB 终结 160MB 臃肿时代",
|
||
"size": "2.5MB",
|
||
"description": "极轻量级 Python 音频库，完整元数据解析，非阻塞播放。首个长期支持版本，永久免费技术支持。"
|
||
})
|
||
|
||
@app.route('/api/latest-announcement')
|
||
def api_latest_announcement():
|
||
"""获取最新公告（ANNCMNT.md）"""
|
||
# 查找最新的 ANNCMNT.md
|
||
branches_data = get_branches_and_versions()
|
||
announcement_html = None
|
||
|
||
for branch, versions in branches_data.items():
|
||
for version in versions[:1]:
|
||
ann_path = CODE_DIR / branch / version['name'] / "ANNCMNT.md"
|
||
if ann_path.exists():
|
||
with open(ann_path, 'r', encoding='utf-8', errors='ignore') as f:
|
||
content = f.read()
|
||
announcement_html = render_markdown(content)
|
||
break
|
||
if announcement_html:
|
||
break
|
||
|
||
return jsonify({"html": announcement_html})
|
||
|
||
# ============================================
|
||
# 路由 - 发行版页面
|
||
# ============================================
|
||
@app.route('/releases')
|
||
def releases():
|
||
"""发行版列表页"""
|
||
branches_data = get_branches_and_versions()
|
||
|
||
all_releases = []
|
||
for branch, versions in branches_data.items():
|
||
for version in versions:
|
||
whl_file = find_whl_file(version['name'])
|
||
# 获取该版本的公告内容
|
||
announcement = get_version_announcement(branch, version['name'])
|
||
all_releases.append({
|
||
"branch": branch,
|
||
"version": version['name'],
|
||
"date": version['created'],
|
||
"has_whl": whl_file is not None,
|
||
"announcement": announcement
|
||
})
|
||
|
||
# 按版本号排序（降序）
|
||
all_releases.sort(key=lambda x: x['version'], reverse=True)
|
||
|
||
return render_template('releases.html', releases=all_releases)
|
||
|
||
# ============================================
|
||
# 路由 - 搜索
|
||
# ============================================
|
||
@app.route('/search')
|
||
def search():
|
||
"""全目录搜索"""
|
||
query = request.args.get('q', '').strip()
|
||
branch = request.args.get('branch', '')
|
||
version = request.args.get('version', '')
|
||
|
||
results = []
|
||
if query and branch and version:
|
||
search_dir = CODE_DIR / branch / version
|
||
if search_dir.exists():
|
||
for file_path in search_dir.rglob("*"):
|
||
if file_path.is_file() and query.lower() in file_path.name.lower():
|
||
results.append({
|
||
"name": file_path.name,
|
||
"path": str(file_path.relative_to(CODE_DIR)).replace('\\', '/'),
|
||
"size": file_path.stat().st_size,
|
||
"modified": datetime.fromtimestamp(file_path.stat().st_mtime).strftime("%Y-%m-%d %H:%M")
|
||
})
|
||
|
||
return render_template('code_browser.html',
|
||
branches=get_branches_and_versions(),
|
||
current_branch=branch,
|
||
current_version=version,
|
||
search_query=query,
|
||
search_results=results)
|
||
|
||
# ============================================
|
||
# 路由 - 反馈
|
||
# ============================================
|
||
@app.route('/feedback', methods=['GET', 'POST'])
|
||
def feedback():
|
||
"""反馈页面"""
|
||
if request.method == 'POST':
|
||
email = request.form.get('email', '').strip()
|
||
content = request.form.get('content', '').strip()
|
||
|
||
if not email or not content:
|
||
return render_template('feedback.html', error="请填写邮箱和反馈内容")
|
||
|
||
# 保存反馈
|
||
feedbacks = load_feedbacks()
|
||
feedbacks.append({
|
||
"id": len(feedbacks) + 1,
|
||
"email": email,
|
||
"content": content,
|
||
"status": "unread",
|
||
"created": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
})
|
||
save_feedbacks(feedbacks)
|
||
|
||
# 发送邮件
|
||
send_feedback_email(email, content)
|
||
|
||
return render_template('feedback.html', success="反馈已提交，感谢您的建议！")
|
||
|
||
return render_template('feedback.html')
|
||
|
||
# ============================================
|
||
# 路由 - API
|
||
# ============================================
|
||
@app.route('/api/download')
|
||
def api_download():
|
||
"""API获取下载地址"""
|
||
branch = request.args.get('branch', '')
|
||
version = request.args.get('version', '')
|
||
|
||
if not branch or not version:
|
||
return jsonify({"error": "缺少 branch 或 version 参数"}), 400
|
||
|
||
# 检查版本是否存在
|
||
version_path = CODE_DIR / branch / version
|
||
if not version_path.exists():
|
||
return jsonify({"error": "版本不存在"}), 404
|
||
|
||
base_url = request.host_url.rstrip('/')
|
||
|
||
response = {
|
||
"branch": branch,
|
||
"version": version,
|
||
"downloads": {
|
||
"zip": f"{base_url}/download/zip/{branch}/{version}",
|
||
"tar": f"{base_url}/download/tar/{branch}/{version}"
|
||
}
|
||
}
|
||
|
||
whl_file = find_whl_file(version)
|
||
if whl_file:
|
||
response["downloads"]["whl"] = f"{base_url}/download/whl/{version}"
|
||
|
||
return jsonify(response)
|
||
|
||
# ============================================
|
||
# 管理员路由
|
||
# ============================================
|
||
@app.route('/admin/login', methods=['GET', 'POST'])
|
||
def admin_login():
|
||
"""管理员登录"""
|
||
if request.method == 'POST':
|
||
password = request.form.get('password', '')
|
||
password_hash = hashlib.sha256(password.encode()).hexdigest()
|
||
|
||
if password_hash == ADMIN_PASSWORD_HASH:
|
||
session['admin_logged_in'] = True
|
||
return redirect(url_for('admin_dashboard'))
|
||
else:
|
||
return render_template('admin_login.html', error="密码错误")
|
||
|
||
return render_template('admin_login.html')
|
||
|
||
@app.route('/admin/logout')
|
||
def admin_logout():
|
||
"""管理员登出"""
|
||
session.pop('admin_logged_in', None)
|
||
return redirect(url_for('admin_login'))
|
||
|
||
@app.route('/admin')
|
||
@admin_required
|
||
def admin_dashboard():
|
||
"""管理仪表板"""
|
||
branches_data = get_branches_and_versions()
|
||
feedbacks = load_feedbacks()
|
||
|
||
stats = {
|
||
"branches": len(branches_data),
|
||
"versions": sum(len(v) for v in branches_data.values()),
|
||
"feedbacks": len(feedbacks),
|
||
"unread_feedbacks": len([f for f in feedbacks if f.get('status') == 'unread'])
|
||
}
|
||
|
||
return render_template('admin_dashboard.html', stats=stats, recent_feedbacks=feedbacks[-10:])
|
||
|
||
@app.route('/admin/feedbacks')
|
||
@admin_required
|
||
def admin_feedbacks():
|
||
"""反馈管理"""
|
||
feedbacks = load_feedbacks()
|
||
return render_template('admin_feedbacks.html', feedbacks=feedbacks)
|
||
|
||
@app.route('/admin/feedback/mark_read/<int:feedback_id>')
|
||
@admin_required
|
||
def admin_feedback_mark_read(feedback_id):
|
||
"""标记反馈为已读"""
|
||
feedbacks = load_feedbacks()
|
||
for f in feedbacks:
|
||
if f.get('id') == feedback_id:
|
||
f['status'] = 'read'
|
||
break
|
||
save_feedbacks(feedbacks)
|
||
return redirect(url_for('admin_feedbacks'))
|
||
|
||
@app.route('/admin/feedback/delete/<int:feedback_id>')
|
||
@admin_required
|
||
def admin_feedback_delete(feedback_id):
|
||
"""删除反馈"""
|
||
feedbacks = load_feedbacks()
|
||
feedbacks = [f for f in feedbacks if f.get('id') != feedback_id]
|
||
save_feedbacks(feedbacks)
|
||
return redirect(url_for('admin_feedbacks'))
|
||
|
||
@app.route('/admin/branches')
|
||
@admin_required
|
||
def admin_branches():
|
||
"""分支管理"""
|
||
branches_data = get_branches_and_versions()
|
||
return render_template('admin_branches.html', branches=branches_data)
|
||
|
||
@app.route('/admin/branch/create', methods=['POST'])
|
||
@admin_required
|
||
def admin_branch_create():
|
||
"""创建分支"""
|
||
branch_name = request.form.get('branch_name', '').strip()
|
||
if not branch_name:
|
||
return jsonify({"error": "分支名不能为空"}), 400
|
||
|
||
# 安全处理分支名
|
||
branch_name = secure_filename(branch_name)
|
||
branch_path = CODE_DIR / branch_name
|
||
|
||
if branch_path.exists():
|
||
return jsonify({"error": "分支已存在"}), 400
|
||
|
||
branch_path.mkdir(parents=True)
|
||
|
||
# 更新分支数据
|
||
branches = load_branches()
|
||
branches[branch_name] = []
|
||
save_branches(branches)
|
||
|
||
return redirect(url_for('admin_branches'))
|
||
|
||
@app.route('/admin/version/upload', methods=['POST'])
|
||
@admin_required
|
||
def admin_version_upload():
|
||
"""上传新版本"""
|
||
branch = request.form.get('branch', '').strip()
|
||
version = request.form.get('version', '').strip()
|
||
|
||
if not branch or not version:
|
||
return jsonify({"error": "分支和版本不能为空"}), 400
|
||
|
||
version = secure_filename(version)
|
||
version_path = CODE_DIR / branch / version
|
||
|
||
if version_path.exists():
|
||
return jsonify({"error": "版本已存在"}), 400
|
||
|
||
version_path.mkdir(parents=True)
|
||
|
||
# 处理文件上传
|
||
if 'files' in request.files:
|
||
files = request.files.getlist('files')
|
||
for file in files:
|
||
if file and file.filename:
|
||
filename = secure_filename(file.filename)
|
||
file.save(version_path / filename)
|
||
|
||
# 更新分支数据
|
||
branches = load_branches()
|
||
if branch not in branches:
|
||
branches[branch] = []
|
||
if version not in branches[branch]:
|
||
branches[branch].append(version)
|
||
save_branches(branches)
|
||
|
||
return redirect(url_for('admin_branches'))
|
||
|
||
@app.route('/admin/version/delete/<branch>/<version>')
|
||
@admin_required
|
||
def admin_version_delete(branch, version):
|
||
"""删除版本"""
|
||
version_path = CODE_DIR / branch / version
|
||
if version_path.exists():
|
||
shutil.rmtree(version_path)
|
||
|
||
# 更新分支数据
|
||
branches = load_branches()
|
||
if branch in branches and version in branches[branch]:
|
||
branches[branch].remove(version)
|
||
save_branches(branches)
|
||
|
||
return redirect(url_for('admin_branches'))
|
||
|
||
@app.route('/admin/whl/upload', methods=['POST'])
|
||
@admin_required
|
||
def admin_whl_upload():
|
||
"""上传whl文件"""
|
||
if 'whl_file' not in request.files:
|
||
return jsonify({"error": "请选择文件"}), 400
|
||
|
||
file = request.files['whl_file']
|
||
if file.filename == '':
|
||
return jsonify({"error": "请选择文件"}), 400
|
||
|
||
if file.filename.endswith('.whl'):
|
||
filename = secure_filename(file.filename)
|
||
file.save(PYPI_DIR / filename)
|
||
|
||
return redirect(url_for('admin_dashboard'))
|
||
|
||
# ============================================
|
||
# 错误页面
|
||
# ============================================
|
||
@app.errorhandler(404)
|
||
def not_found(error):
|
||
return render_template('404.html'), 404
|
||
|
||
@app.errorhandler(500)
|
||
def internal_error(error):
|
||
return render_template('500.html'), 500
|
||
|
||
# ============================================
|
||
# 主程序入口
|
||
# ============================================
|
||
if __name__ == '__main__':
|
||
print("=" * 50)
|
||
print("ap_ds 代码仓库系统启动")
|
||
print("访问地址: http://127.0.0.1:8888")
|
||
print("管理员后台: http://127.0.0.1:8888/admin/login")
|
||
print("=" * 50)
|
||
app.run(host='0.0.0.0', port=8888, debug=True)
|