AI Agent
26940c4574
Features: - User authentication (login/register) - Ticket CRUD with status/priority/category - Rich text description with paste-to-image - File attachment upload (images/documents) - Multi-select CC recipients - Role-based permission control (admin/user) - In-app notifications - Statistics dashboard - Gunicorn production deployment
776 lines
25 KiB
Python
776 lines
25 KiB
Python
"""
|
||
轻量级工单系统 - Flask + SQLite
|
||
增强版：分类 / 优先级 / 标签 / 统计看板 / 登录 / 站内通知
|
||
"""
|
||
|
||
from datetime import datetime, timezone
|
||
import os
|
||
import uuid
|
||
|
||
from werkzeug.security import generate_password_hash, check_password_hash
|
||
from flask import (
|
||
Flask, request, redirect, render_template, session, abort, url_for,
|
||
jsonify, send_from_directory,
|
||
)
|
||
from flask_sqlalchemy import SQLAlchemy
|
||
|
||
app = Flask(__name__)
|
||
app.config["SECRET_KEY"] = "ticket-system-secret-key-change-in-production"
|
||
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///tickets.db"
|
||
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
|
||
app.config["UPLOAD_FOLDER"] = os.path.join(os.path.dirname(os.path.abspath(__file__)), "uploads")
|
||
app.config["MAX_CONTENT_LENGTH"] = 20 * 1024 * 1024 # 20 MB
|
||
|
||
db = SQLAlchemy(app)
|
||
|
||
# ── 枚举定义 ──────────────────────────────────────────
|
||
|
||
STATUS = {
|
||
"open": "新建",
|
||
"in_progress": "进行中",
|
||
"pending": "待处理",
|
||
"resolved": "已完成",
|
||
"closed": "已关闭",
|
||
}
|
||
|
||
PRIORITY = {
|
||
"low": "低",
|
||
"medium": "中",
|
||
"high": "高",
|
||
"urgent": "紧急",
|
||
}
|
||
|
||
CATEGORY = {
|
||
"bug": "技术故障",
|
||
"feature": "功能需求",
|
||
"consult": "咨询",
|
||
"incident": "事件处理",
|
||
"other": "其他",
|
||
}
|
||
|
||
|
||
# ── 数据模型 ──────────────────────────────────────────
|
||
|
||
class User(db.Model):
|
||
__tablename__ = "users"
|
||
|
||
id = db.Column(db.Integer, primary_key=True)
|
||
username = db.Column(db.String(80), unique=True, nullable=False)
|
||
password_hash = db.Column(db.String(200), nullable=False)
|
||
role = db.Column(db.String(20), default="user") # admin / user
|
||
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
||
|
||
def set_password(self, password):
|
||
self.password_hash = generate_password_hash(password)
|
||
|
||
def check_password(self, password):
|
||
return check_password_hash(self.password_hash, password)
|
||
|
||
|
||
class Notification(db.Model):
|
||
__tablename__ = "notifications"
|
||
|
||
id = db.Column(db.Integer, primary_key=True)
|
||
user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
|
||
title = db.Column(db.String(200), nullable=False)
|
||
content = db.Column(db.Text)
|
||
ticket_id = db.Column(db.Integer) # 关联工单（可选）
|
||
read = db.Column(db.Boolean, default=False)
|
||
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
||
|
||
user = db.relationship("User", backref=db.backref("notifications", lazy=True))
|
||
|
||
|
||
class Ticket(db.Model):
|
||
__tablename__ = "tickets"
|
||
|
||
id = db.Column(db.Integer, primary_key=True)
|
||
title = db.Column(db.String(200), nullable=False)
|
||
description = db.Column(db.Text) # 支持 HTML（内联图片）
|
||
status = db.Column(db.String(20), default="open", nullable=False)
|
||
priority = db.Column(db.String(20), default="medium", nullable=False)
|
||
category = db.Column(db.String(20), default="other", nullable=False)
|
||
assignee = db.Column(db.String(100))
|
||
requester = db.Column(db.String(100))
|
||
cc_users = db.Column(db.String(500), default="") # 抄送人（逗号分隔）
|
||
tags = db.Column(db.String(500), default="")
|
||
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
||
updated_at = db.Column(
|
||
db.DateTime,
|
||
default=lambda: datetime.now(timezone.utc),
|
||
onupdate=lambda: datetime.now(timezone.utc),
|
||
)
|
||
|
||
comments = db.relationship(
|
||
"Comment", backref="ticket", lazy=True, cascade="all, delete-orphan",
|
||
)
|
||
|
||
|
||
class Comment(db.Model):
|
||
__tablename__ = "comments"
|
||
|
||
id = db.Column(db.Integer, primary_key=True)
|
||
ticket_id = db.Column(db.Integer, db.ForeignKey("tickets.id"), nullable=False)
|
||
author = db.Column(db.String(100), nullable=False)
|
||
content = db.Column(db.Text, nullable=False)
|
||
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
||
|
||
|
||
class Attachment(db.Model):
|
||
"""工单附件（图片 / 文件）"""
|
||
__tablename__ = "attachments"
|
||
|
||
id = db.Column(db.Integer, primary_key=True)
|
||
ticket_id = db.Column(db.Integer, db.ForeignKey("tickets.id"), nullable=False)
|
||
filename = db.Column(db.String(200), nullable=False) # 存储文件名（uuid）
|
||
original_name = db.Column(db.String(300), nullable=False) # 原始文件名
|
||
content_type = db.Column(db.String(100)) # MIME type
|
||
size = db.Column(db.Integer, default=0) # 字节
|
||
is_image = db.Column(db.Boolean, default=False)
|
||
uploaded_by = db.Column(db.String(100))
|
||
uploaded_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
|
||
|
||
ticket = db.relationship("Ticket", backref=db.backref(
|
||
"attachments", lazy=True, cascade="all, delete-orphan",
|
||
))
|
||
|
||
|
||
# ── 工具函数 ──────────────────────────────────────────
|
||
|
||
def status_label(s):
|
||
return STATUS.get(s, s)
|
||
|
||
|
||
def priority_label(p):
|
||
return PRIORITY.get(p, p)
|
||
|
||
|
||
def category_label(c):
|
||
return CATEGORY.get(c, c)
|
||
|
||
|
||
def parse_tags(tags):
|
||
if not tags:
|
||
return []
|
||
return [t.strip() for t in tags.split(",") if t.strip()]
|
||
|
||
|
||
def current_user():
|
||
"""返回当前登录用户或 None"""
|
||
if "user_id" not in session:
|
||
return None
|
||
return db.session.get(User, session["user_id"])
|
||
|
||
|
||
def login_required(func):
|
||
"""登录守卫装饰器"""
|
||
def wrapper(*args, **kwargs):
|
||
if "user_id" not in session:
|
||
return redirect(url_for("login", next=request.path))
|
||
return func(*args, **kwargs)
|
||
wrapper.__name__ = func.__name__
|
||
return wrapper
|
||
|
||
|
||
def create_notification(user_id, title, content="", ticket_id=None):
|
||
"""为指定用户创建一条通知"""
|
||
n = Notification(
|
||
user_id=user_id,
|
||
title=title,
|
||
content=content,
|
||
ticket_id=ticket_id,
|
||
)
|
||
db.session.add(n)
|
||
|
||
|
||
def all_users():
|
||
"""返回所有用户列表，供下拉选择"""
|
||
return User.query.order_by(User.username).all()
|
||
|
||
|
||
def parse_cc(cc_str):
|
||
"""解析抄送人字段，返回列表"""
|
||
if not cc_str:
|
||
return []
|
||
return [u.strip() for u in cc_str.split(",") if u.strip()]
|
||
|
||
|
||
def can_access_ticket(user, ticket):
|
||
"""
|
||
权限控制：
|
||
- admin 可见所有工单
|
||
- 普通用户只能看到：自己是提交人 / 处理人 / 抄送人的工单
|
||
"""
|
||
if user.role == "admin":
|
||
return True
|
||
if ticket.requester == user.username:
|
||
return True
|
||
if ticket.assignee == user.username:
|
||
return True
|
||
if user.username in parse_cc(ticket.cc_users):
|
||
return True
|
||
return False
|
||
|
||
|
||
def accessible_tickets_query(me):
|
||
"""
|
||
返回当前用户可见的工单 Query。
|
||
- admin: 全部
|
||
- 普通用户: 提交人=me OR 处理人=me OR 在抄送人里
|
||
"""
|
||
if me.role == "admin":
|
||
return Ticket.query
|
||
return Ticket.query.filter(
|
||
db.or_(
|
||
Ticket.requester == me.username,
|
||
Ticket.assignee == me.username,
|
||
Ticket.cc_users.like(f"%{me.username}%"),
|
||
),
|
||
)
|
||
|
||
|
||
def _save_upload(file, ticket_id):
|
||
"""保存上传文件，返回 Attachment 实例或 None"""
|
||
me = current_user()
|
||
if not file or file.filename == "":
|
||
return None
|
||
|
||
ext = os.path.splitext(file.filename)[1].lower()
|
||
safe_name = f"{uuid.uuid4().hex}{ext}"
|
||
user_dir = os.path.join(app.config["UPLOAD_FOLDER"], str(ticket_id))
|
||
os.makedirs(user_dir, exist_ok=True)
|
||
path = os.path.join(user_dir, safe_name)
|
||
file.save(path)
|
||
|
||
content_type = file.content_type or "application/octet-stream"
|
||
is_image = content_type.startswith("image/")
|
||
|
||
att = Attachment(
|
||
ticket_id=ticket_id,
|
||
filename=safe_name,
|
||
original_name=file.filename,
|
||
content_type=content_type,
|
||
size=os.path.getsize(path),
|
||
is_image=is_image,
|
||
uploaded_by=me.username if me else "system",
|
||
)
|
||
db.session.add(att)
|
||
db.session.commit()
|
||
return att
|
||
|
||
|
||
@app.route("/api/upload", methods=["POST"])
|
||
@login_required
|
||
def api_upload():
|
||
"""JSON API：上传文件（含二进制流 / 粘贴图片）"""
|
||
ticket_id = request.args.get("t")
|
||
if not ticket_id:
|
||
return jsonify({"error": "缺少 ticket 参数"}), 400
|
||
ticket = db.get_or_404(Ticket, int(ticket_id))
|
||
user = current_user()
|
||
if not can_access_ticket(user, ticket):
|
||
return jsonify({"error": "无权限"}), 403
|
||
|
||
# 1) 普通文件上传 (multipart)
|
||
if "file" in request.files:
|
||
att = _save_upload(request.files["file"], ticket_id)
|
||
# 2) 二进制流（粘贴图片、拖拽）
|
||
elif request.data:
|
||
import base64
|
||
data = request.data
|
||
# 尝试 base64 解码（前端 paste 用 base64 传）
|
||
try:
|
||
if request.headers.get("Content-Type", "").startswith("image/"):
|
||
raw = data
|
||
elif request.headers.get("Content-Type") == "application/octet-stream":
|
||
raw = data
|
||
else:
|
||
raw = data
|
||
ext = ".png"
|
||
# 猜测扩展名
|
||
ct = request.headers.get("Content-Type", "")
|
||
ext_map = {
|
||
"image/jpeg": ".jpg", "image/png": ".png",
|
||
"image/gif": ".gif", "image/webp": ".webp",
|
||
}
|
||
ext = ext_map.get(ct, ".png")
|
||
safe_name = f"{uuid.uuid4().hex}{ext}"
|
||
user_dir = os.path.join(app.config["UPLOAD_FOLDER"], str(ticket_id))
|
||
os.makedirs(user_dir, exist_ok=True)
|
||
path = os.path.join(user_dir, safe_name)
|
||
with open(path, "wb") as f:
|
||
f.write(raw)
|
||
att = Attachment(
|
||
ticket_id=ticket_id, filename=safe_name,
|
||
original_name="pasted-image.png", content_type=ct or "image/png",
|
||
size=os.path.getsize(path), is_image=True,
|
||
uploaded_by=current_user().username if current_user() else "system",
|
||
)
|
||
db.session.add(att)
|
||
db.session.commit()
|
||
except Exception as e:
|
||
return jsonify({"error": str(e)}), 500
|
||
else:
|
||
return jsonify({"error": "没有上传文件"}), 400
|
||
|
||
return jsonify({
|
||
"ok": True,
|
||
"url": f"/uploads/{ticket_id}/{att.filename}",
|
||
"filename": att.original_name,
|
||
"is_image": att.is_image,
|
||
})
|
||
|
||
|
||
@app.route("/api/upload/delete/<int:aid>", methods=["POST"])
|
||
@login_required
|
||
def api_delete_attachment(aid):
|
||
"""删除附件"""
|
||
att = db.get_or_404(Attachment, aid)
|
||
os.remove(os.path.join(app.config["UPLOAD_FOLDER"], str(att.ticket_id), att.filename))
|
||
db.session.delete(att)
|
||
db.session.commit()
|
||
return jsonify({"ok": True})
|
||
|
||
|
||
@app.route("/uploads/<int:ticket_id>/<path:filename>")
|
||
@login_required
|
||
def serve_upload(ticket_id, filename):
|
||
"""提供上传文件下载"""
|
||
ticket = db.get_or_404(Ticket, ticket_id)
|
||
user = current_user()
|
||
if not can_access_ticket(user, ticket):
|
||
abort(403)
|
||
safe_dir = os.path.join(app.config["UPLOAD_FOLDER"], str(ticket_id))
|
||
return send_from_directory(safe_dir, filename)
|
||
|
||
@app.route("/login", methods=["GET", "POST"])
|
||
def login():
|
||
if "user_id" in session:
|
||
return redirect(url_for("dashboard"))
|
||
if request.method == "POST":
|
||
username = request.form.get("username", "").strip()
|
||
password = request.form.get("password", "")
|
||
user = User.query.filter_by(username=username).first()
|
||
if user and user.check_password(password):
|
||
session["user_id"] = user.id
|
||
session["username"] = user.username
|
||
next_page = request.args.get("next") or url_for("dashboard")
|
||
return redirect(next_page)
|
||
return render_template(
|
||
"login.html", error="用户名或密码错误", username=username,
|
||
)
|
||
return render_template("login.html")
|
||
|
||
|
||
@app.route("/register", methods=["GET", "POST"])
|
||
def register():
|
||
if "user_id" in session:
|
||
return redirect(url_for("dashboard"))
|
||
if request.method == "POST":
|
||
username = request.form.get("username", "").strip()
|
||
password = request.form.get("password", "")
|
||
password2 = request.form.get("password2", "")
|
||
if not username or not password:
|
||
return render_template(
|
||
"register.html", error="用户名和密码不能为空", username=username,
|
||
)
|
||
if password != password2:
|
||
return render_template(
|
||
"register.html", error="两次输入的密码不一致", username=username,
|
||
)
|
||
if User.query.filter_by(username=username).first():
|
||
return render_template(
|
||
"register.html", error="用户名已存在", username=username,
|
||
)
|
||
user = User(username=username, role="user")
|
||
user.set_password(password)
|
||
db.session.add(user)
|
||
db.session.commit()
|
||
session["user_id"] = user.id
|
||
session["username"] = user.username
|
||
return redirect(url_for("dashboard"))
|
||
return render_template("register.html")
|
||
|
||
|
||
@app.route("/logout")
|
||
def logout():
|
||
session.pop("user_id", None)
|
||
session.pop("username", None)
|
||
return redirect(url_for("login"))
|
||
|
||
|
||
# ── 通知路由 ──────────────────────────────────────────
|
||
|
||
@app.route("/notifications")
|
||
@login_required
|
||
def notifications():
|
||
user = current_user()
|
||
notifs = (
|
||
Notification.query.filter_by(user_id=user.id)
|
||
.order_by(Notification.created_at.desc())
|
||
.limit(50)
|
||
.all()
|
||
)
|
||
# 标记为已读
|
||
for n in notifs:
|
||
n.read = True
|
||
db.session.commit()
|
||
return render_template(
|
||
"notifications.html", notifications=notifs, datetime=datetime,
|
||
status_label=status_label,
|
||
)
|
||
|
||
|
||
@app.route("/notifications/count")
|
||
def notifications_count():
|
||
"""返回未读通知数（供前端角标使用）"""
|
||
count = 0
|
||
if "user_id" in session:
|
||
count = Notification.query.filter_by(
|
||
user_id=session["user_id"], read=False,
|
||
).count()
|
||
return str(count)
|
||
|
||
|
||
# ── 工单路由（全部需要登录） ──────────────────────────
|
||
|
||
@app.route("/")
|
||
@login_required
|
||
def dashboard():
|
||
"""统计看板"""
|
||
user = current_user()
|
||
me = user.username
|
||
base_query = accessible_tickets_query(user)
|
||
|
||
# 我的工单：提交人 + 处理人 + 抄送
|
||
my_tickets = base_query.count()
|
||
total = base_query.count()
|
||
by_status = {}
|
||
for key, label in STATUS.items():
|
||
by_status[key] = {"label": label, "count": base_query.filter_by(status=key).count()}
|
||
by_priority = {}
|
||
for key, label in PRIORITY.items():
|
||
by_priority[key] = {"label": label, "count": base_query.filter_by(priority=key).count()}
|
||
by_category = {}
|
||
for key, label in CATEGORY.items():
|
||
by_category[key] = {"label": label, "count": base_query.filter_by(category=key).count()}
|
||
recent = base_query.order_by(Ticket.updated_at.desc()).limit(10).all()
|
||
return render_template(
|
||
"dashboard.html",
|
||
total=total,
|
||
my_tickets=my_tickets,
|
||
by_status=by_status,
|
||
by_priority=by_priority,
|
||
by_category=by_category,
|
||
recent=recent,
|
||
STATUS=STATUS,
|
||
PRIORITY=PRIORITY,
|
||
CATEGORY=CATEGORY,
|
||
status_label=status_label,
|
||
priority_label=priority_label,
|
||
category_label=category_label,
|
||
)
|
||
|
||
|
||
@app.route("/tickets")
|
||
@login_required
|
||
def ticket_list():
|
||
"""工单列表（支持筛选）"""
|
||
user = current_user()
|
||
q_status = request.args.get("status", "")
|
||
q_priority = request.args.get("priority", "")
|
||
q_category = request.args.get("category", "")
|
||
q_tag = request.args.get("tag", "")
|
||
q_keyword = request.args.get("q", "")
|
||
|
||
query = accessible_tickets_query(user)
|
||
|
||
if q_status:
|
||
query = query.filter_by(status=q_status)
|
||
if q_priority:
|
||
query = query.filter_by(priority=q_priority)
|
||
if q_category:
|
||
query = query.filter_by(category=q_category)
|
||
if q_tag:
|
||
query = query.filter(Ticket.tags.like(f"%{q_tag}%"))
|
||
if q_keyword:
|
||
kw = f"%{q_keyword}%"
|
||
query = query.filter(
|
||
db.or_(Ticket.title.like(kw), Ticket.description.like(kw)),
|
||
)
|
||
|
||
tickets = query.order_by(Ticket.updated_at.desc()).all()
|
||
|
||
return render_template(
|
||
"ticket_list.html",
|
||
tickets=tickets,
|
||
STATUS=STATUS,
|
||
PRIORITY=PRIORITY,
|
||
CATEGORY=CATEGORY,
|
||
status_label=status_label,
|
||
priority_label=priority_label,
|
||
category_label=category_label,
|
||
parse_tags=parse_tags,
|
||
q_status=q_status,
|
||
q_priority=q_priority,
|
||
q_category=q_category,
|
||
q_tag=q_tag,
|
||
q_keyword=q_keyword,
|
||
)
|
||
|
||
|
||
@app.route("/ticket/<int:tid>")
|
||
@login_required
|
||
def ticket_detail(tid):
|
||
t = db.get_or_404(Ticket, tid)
|
||
user = current_user()
|
||
if not can_access_ticket(user, t):
|
||
abort(403)
|
||
atts = Attachment.query.filter_by(ticket_id=tid).order_by(
|
||
Attachment.uploaded_at.desc()
|
||
).all()
|
||
return render_template(
|
||
"ticket_detail.html",
|
||
ticket=t,
|
||
attachments=atts,
|
||
STATUS=STATUS,
|
||
PRIORITY=PRIORITY,
|
||
CATEGORY=CATEGORY,
|
||
status_label=status_label,
|
||
priority_label=priority_label,
|
||
category_label=category_label,
|
||
parse_tags=parse_tags,
|
||
parse_cc=parse_cc,
|
||
datetime=datetime,
|
||
all_users=all_users,
|
||
)
|
||
|
||
|
||
@app.route("/ticket/new", methods=["GET", "POST"])
|
||
@login_required
|
||
def ticket_new():
|
||
return _ticket_form(None)
|
||
|
||
|
||
@app.route("/ticket/<int:tid>/edit", methods=["GET", "POST"])
|
||
@login_required
|
||
def ticket_edit(tid):
|
||
t = db.get_or_404(Ticket, tid)
|
||
user = current_user()
|
||
if not can_access_ticket(user, t):
|
||
abort(403)
|
||
return _ticket_form(t)
|
||
|
||
|
||
def _ticket_form(ticket):
|
||
me = current_user()
|
||
if request.method == "POST":
|
||
title = request.form.get("title", "").strip()
|
||
desc = request.form.get("description", "").strip()
|
||
status = request.form.get("status", "open")
|
||
priority = request.form.get("priority", "medium")
|
||
category = request.form.get("category", "other")
|
||
assignee = request.form.get("assignee", "").strip()
|
||
requester = request.form.get("requester", "").strip()
|
||
cc_str = request.form.get("cc_users", "").strip()
|
||
tags = request.form.get("tags", "").strip()
|
||
|
||
if not title:
|
||
return "标题不能为空", 400
|
||
|
||
if ticket:
|
||
ticket.title = title
|
||
ticket.description = desc
|
||
ticket.status = status
|
||
ticket.priority = priority
|
||
ticket.category = category
|
||
ticket.assignee = assignee or None
|
||
ticket.requester = requester or None
|
||
ticket.cc_users = cc_str
|
||
ticket.tags = tags
|
||
# 分配人变更时通知新处理人
|
||
if assignee and assignee != me.username:
|
||
u = User.query.filter_by(username=assignee).first()
|
||
if u:
|
||
create_notification(
|
||
u.id,
|
||
f"工单 #{ticket.id} 已分配给你",
|
||
title,
|
||
ticket_id=ticket.id,
|
||
)
|
||
else:
|
||
ticket = Ticket(
|
||
title=title,
|
||
description=desc,
|
||
status=status,
|
||
priority=priority,
|
||
category=category,
|
||
assignee=assignee or None,
|
||
requester=me.username,
|
||
cc_users=cc_str,
|
||
tags=tags,
|
||
)
|
||
db.session.add(ticket)
|
||
db.session.commit()
|
||
|
||
# 通知新增的抄送人
|
||
cc_list = parse_cc(cc_str)
|
||
for uname in cc_list:
|
||
u = User.query.filter_by(username=uname).first()
|
||
if u and u.username != me.username:
|
||
create_notification(
|
||
u.id,
|
||
f"工单 #{ticket.id} 被抄送给你",
|
||
title,
|
||
ticket_id=ticket.id,
|
||
)
|
||
|
||
return redirect(f"/ticket/{ticket.id}")
|
||
|
||
return render_template(
|
||
"ticket_form.html",
|
||
ticket=ticket,
|
||
STATUS=STATUS,
|
||
PRIORITY=PRIORITY,
|
||
CATEGORY=CATEGORY,
|
||
is_edit=bool(ticket),
|
||
all_users=all_users,
|
||
current_username=me.username if me else "",
|
||
parse_cc=parse_cc,
|
||
)
|
||
|
||
|
||
@app.route("/ticket/<int:tid>/comment", methods=["POST"])
|
||
@login_required
|
||
def add_comment(tid):
|
||
t = db.get_or_404(Ticket, tid)
|
||
me = current_user()
|
||
content = request.form.get("content", "").strip()
|
||
if not content:
|
||
return "评论内容不能为空", 400
|
||
c = Comment(ticket_id=tid, author=me.username, content=content)
|
||
db.session.add(c)
|
||
|
||
# 通知：工单处理人 + 提交人 + 抄送人（避免通知自己）
|
||
notify_targets = set()
|
||
if t.assignee and t.assignee != me.username:
|
||
notify_targets.add(t.assignee)
|
||
if t.requester and t.requester != me.username:
|
||
notify_targets.add(t.requester)
|
||
for uname in parse_cc(t.cc_users):
|
||
if uname != me.username:
|
||
notify_targets.add(uname)
|
||
for uname in notify_targets:
|
||
u = User.query.filter_by(username=uname).first()
|
||
if u:
|
||
create_notification(
|
||
u.id,
|
||
f"工单 #{tid} 有新评论",
|
||
f"{me.username}: {content[:50]}{'...' if len(content) > 50 else ''}",
|
||
ticket_id=tid,
|
||
)
|
||
|
||
db.session.commit()
|
||
return redirect(f"/ticket/{tid}")
|
||
|
||
|
||
@app.route("/ticket/<int:tid>/close", methods=["POST"])
|
||
@login_required
|
||
def close_ticket(tid):
|
||
t = db.get_or_404(Ticket, tid)
|
||
t.status = "closed"
|
||
db.session.commit()
|
||
return redirect(f"/ticket/{tid}")
|
||
|
||
|
||
@app.route("/ticket/<int:tid>/delete", methods=["POST"])
|
||
@login_required
|
||
def delete_ticket(tid):
|
||
t = db.get_or_404(Ticket, tid)
|
||
db.session.delete(t)
|
||
db.session.commit()
|
||
return redirect("/tickets")
|
||
|
||
|
||
@app.template_filter("render_desc")
|
||
def render_desc_html(html):
|
||
"""
|
||
渲染描述中的内联图片，处理 contenteditable 的 div 包装。
|
||
只允许 <img> 和 <br> 标签，其他 HTML 转义。
|
||
contenteditable 默认会用 <div> 包裹段落，需要去掉。
|
||
"""
|
||
if not html:
|
||
return ""
|
||
from markupsafe import Markup, escape
|
||
import re
|
||
|
||
# Step 1: 将 <div>...</div> 替换为内容 + <br>，模拟内容editable的行为
|
||
def replace_div(m):
|
||
content = m.group(1)
|
||
return content + "\n"
|
||
html = re.sub(r'<div[^>]*>(.*?)</div>', replace_div, html, flags=re.DOTALL)
|
||
|
||
# Step 2: 将 <p>...</p> 替换为内容 + <br>
|
||
html = re.sub(r'<p[^>]*>(.*?)</p>', replace_div, html, flags=re.DOTALL)
|
||
|
||
# Step 3: 将 <br> 和 <br/> 替换为 \n
|
||
html = re.sub(r'<br\s*/?>', '\n', html)
|
||
|
||
# Step 4: 转义所有文本，但保留 <img> 标签
|
||
parts = []
|
||
last = 0
|
||
for m in re.finditer(r'<img[^>]*>', html):
|
||
if m.start() > last:
|
||
parts.append(escape(html[last:m.start()]))
|
||
parts.append(Markup(m.group(0)))
|
||
last = m.end()
|
||
if last < len(html):
|
||
parts.append(escape(html[last:]))
|
||
|
||
result = "".join(parts)
|
||
# 替换 \n 为 <br> 用于显示
|
||
return Markup(result).replace("\n", "<br>")
|
||
|
||
|
||
@app.errorhandler(403)
|
||
def forbidden(e):
|
||
return render_template("error.html", code=403, message="无权限访问此工单"), 403
|
||
|
||
|
||
@app.errorhandler(404)
|
||
def not_found(e):
|
||
return render_template("error.html", code=404, message="页面不存在"), 404
|
||
|
||
|
||
# ── 全局模板上下文 ────────────────────────────────────
|
||
|
||
@app.context_processor
|
||
def inject_globals():
|
||
"""每个模板都能访问的用户信息 + 未读通知数"""
|
||
unread = 0
|
||
if "user_id" in session:
|
||
unread = Notification.query.filter_by(
|
||
user_id=session["user_id"], read=False,
|
||
).count()
|
||
return dict(current_user=current_user, unread_count=unread)
|
||
|
||
|
||
# ── 启动 ──────────────────────────────────────────────
|
||
|
||
def seed_admin():
|
||
"""确保存在 admin 用户（密码 admin）"""
|
||
if not User.query.filter_by(username="admin").first():
|
||
admin = User(username="admin", role="admin")
|
||
admin.set_password("admin")
|
||
db.session.add(admin)
|
||
db.session.commit()
|
||
print("✅ 已创建默认管理员: admin / admin")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
os.makedirs(app.config["UPLOAD_FOLDER"], exist_ok=True)
|
||
with app.app_context():
|
||
db.create_all()
|
||
seed_admin()
|
||
app.run(host="0.0.0.0", port=5000, debug=True)
|