455 lines
15 KiB
Python
455 lines
15 KiB
Python
import re
|
|
import secrets
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import DateTime, Integer, String, Text, func, select
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from app.db import Base
|
|
|
|
def _mysql_args() -> dict:
|
|
return {"mysql_charset": "utf8mb4", "mysql_collate": "utf8mb4_unicode_ci"}
|
|
|
|
|
|
COMPLAINT_STATUSES = ("pending", "processed")
|
|
DEFAULT_COMPLAINT_WORKDAYS = 30
|
|
SCOPE_EMPTY = "受理范围说明整理中"
|
|
PROCESS_EMPTY = "处理流程整理中"
|
|
MAX_EVIDENCE = 8
|
|
|
|
_MOBILE = re.compile(r"^1[3-9]\d{9}$")
|
|
_EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
|
_CODE = re.compile(r"^[0-9A-F]{12}$")
|
|
_NOTICE_PHONE = "discipline_phone"
|
|
_NOTICE_EMAIL = "discipline_email"
|
|
_NOTICE_SCOPE = "discipline_scope_html"
|
|
_NOTICE_PROCESS = "discipline_process_html"
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
class DisciplineComplaint(Base):
|
|
__tablename__ = "discipline_complaint"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
code: Mapped[str] = mapped_column(String(16), unique=True, nullable=False)
|
|
title: Mapped[str] = mapped_column(String(200), nullable=False, default="")
|
|
subject: Mapped[str] = mapped_column(String(200), nullable=False, default="")
|
|
body: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
contact_name: Mapped[str] = mapped_column(String(40), nullable=False, default="")
|
|
mobile_cipher: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
mobile_mask: Mapped[str] = mapped_column(String(32), nullable=False, default="")
|
|
email_cipher: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
email_mask: Mapped[str] = mapped_column(String(160), nullable=False, default="")
|
|
status: Mapped[str] = mapped_column(String(16), nullable=False, default="pending", index=True)
|
|
created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=_now, index=True)
|
|
processed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
|
|
processed_by: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
|
|
|
|
class DisciplineComplaintFile(Base):
|
|
__tablename__ = "discipline_complaint_file"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
complaint_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True)
|
|
file_id: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
def complaint_workdays(db) -> int:
|
|
from app.models.content import SiteSetting
|
|
|
|
row = db.scalar(select(SiteSetting).where(SiteSetting.key == "complaint_workdays"))
|
|
raw = (row.value if row else "").strip()
|
|
if not raw:
|
|
return DEFAULT_COMPLAINT_WORKDAYS
|
|
try:
|
|
value = int(raw)
|
|
except ValueError:
|
|
return DEFAULT_COMPLAINT_WORKDAYS
|
|
if value < 0:
|
|
return DEFAULT_COMPLAINT_WORKDAYS
|
|
return value
|
|
|
|
|
|
def _setting(db, key: str) -> str:
|
|
from app.models.content import SiteSetting
|
|
|
|
row = db.scalar(select(SiteSetting).where(SiteSetting.key == key))
|
|
return "" if row is None or row.value is None else str(row.value)
|
|
|
|
|
|
def _put_setting(db, key: str, value: str) -> None:
|
|
from app.models.content import SiteSetting
|
|
|
|
row = db.scalar(select(SiteSetting).where(SiteSetting.key == key))
|
|
if row is None:
|
|
db.add(SiteSetting(key=key, value=value))
|
|
else:
|
|
row.value = value
|
|
|
|
|
|
def _visible_text(html: str) -> str:
|
|
text = re.sub(r"<[^>]+>", "", html or "")
|
|
return text.replace(" ", " ").replace("\xa0", " ").strip()
|
|
|
|
|
|
def _filter_notice_html(db, html: str) -> str:
|
|
from app.core.config import get_settings
|
|
from app.models.content import AdminFile
|
|
from app.services.html_filter import filter_html
|
|
|
|
if len(html or "") > 20000:
|
|
raise ValueError("说明内容过长")
|
|
keys = list(db.scalars(select(AdminFile.object_key).where(AdminFile.bucket == "public")))
|
|
cleaned = filter_html(html or "", list(keys), get_settings().minio_base_url)
|
|
if not _visible_text(cleaned):
|
|
return ""
|
|
return cleaned
|
|
|
|
|
|
def _notice_phone(value: str) -> str:
|
|
from app.services.html_filter import is_javascript_url
|
|
|
|
text = " ".join((value or "").split())
|
|
if len(text) > 40 or is_javascript_url(text):
|
|
raise ValueError("举报电话不正确")
|
|
return text
|
|
|
|
|
|
def _notice_email(value: str) -> str:
|
|
text = (value or "").strip()
|
|
if not text:
|
|
return ""
|
|
if len(text) > 120 or not _EMAIL.fullmatch(text):
|
|
raise ValueError("专用邮箱格式不正确")
|
|
return text
|
|
|
|
|
|
def public_notice(db) -> dict:
|
|
scope = _setting(db, _NOTICE_SCOPE)
|
|
process = _setting(db, _NOTICE_PROCESS)
|
|
return {
|
|
"phone": _setting(db, _NOTICE_PHONE),
|
|
"email": _setting(db, _NOTICE_EMAIL),
|
|
"scopeHtml": scope,
|
|
"processHtml": process,
|
|
"scopePlaceholder": "" if scope else SCOPE_EMPTY,
|
|
"processPlaceholder": "" if process else PROCESS_EMPTY,
|
|
"workdays": complaint_workdays(db),
|
|
}
|
|
|
|
|
|
def admin_notice(db) -> dict:
|
|
return {
|
|
"phone": _setting(db, _NOTICE_PHONE),
|
|
"email": _setting(db, _NOTICE_EMAIL),
|
|
"scopeHtml": _setting(db, _NOTICE_SCOPE),
|
|
"processHtml": _setting(db, _NOTICE_PROCESS),
|
|
"workdays": complaint_workdays(db),
|
|
}
|
|
|
|
|
|
def save_notice(db, payload: dict) -> None:
|
|
phone = _notice_phone("" if payload.get("phone") is None else str(payload.get("phone")))
|
|
email = _notice_email("" if payload.get("email") is None else str(payload.get("email")))
|
|
scope = _filter_notice_html(db, "" if payload.get("scopeHtml") is None else str(payload.get("scopeHtml")))
|
|
process = _filter_notice_html(
|
|
db, "" if payload.get("processHtml") is None else str(payload.get("processHtml"))
|
|
)
|
|
_put_setting(db, _NOTICE_PHONE, phone)
|
|
_put_setting(db, _NOTICE_EMAIL, email)
|
|
_put_setting(db, _NOTICE_SCOPE, scope)
|
|
_put_setting(db, _NOTICE_PROCESS, process)
|
|
db.commit()
|
|
|
|
|
|
def _plain(value, *, label: str, limit: int) -> str:
|
|
text = "" if value is None else str(value)
|
|
text = text.replace("\r\n", "\n").replace("\r", "\n").strip()
|
|
if label != "违规事实":
|
|
text = " ".join(text.split())
|
|
if not text:
|
|
raise ValueError(f"请填写{label}")
|
|
if len(text) > limit:
|
|
raise ValueError(f"{label}过长")
|
|
return text
|
|
|
|
|
|
def _mobile(value: str) -> str:
|
|
text = re.sub(r"[\s-]", "", value or "")
|
|
if not _MOBILE.fullmatch(text):
|
|
raise ValueError("请填写有效的联系电话")
|
|
return text
|
|
|
|
|
|
def _reporter_email(value: str) -> str:
|
|
text = (value or "").strip()
|
|
if not text:
|
|
return ""
|
|
if len(text) > 120 or not _EMAIL.fullmatch(text):
|
|
raise ValueError("联系邮箱格式不正确")
|
|
return text
|
|
|
|
|
|
def _reject_sensitive(db, text: str) -> None:
|
|
from app.models.content import SensitiveWord
|
|
|
|
words = db.scalars(select(SensitiveWord.word)).all()
|
|
if any(word and word in text for word in words):
|
|
raise ValueError("内容包含敏感词，请修改后再提交")
|
|
|
|
|
|
def _status_label(status: str) -> str:
|
|
return "已处理" if status == "processed" else "待处理"
|
|
|
|
|
|
def _new_code(db) -> str:
|
|
for _ in range(5):
|
|
code = secrets.token_hex(6).upper()
|
|
exists = db.scalar(select(DisciplineComplaint.id).where(DisciplineComplaint.code == code))
|
|
if exists is None:
|
|
return code
|
|
raise ValueError("查询码生成失败，请稍后再试")
|
|
|
|
|
|
def normalize_code(value: str) -> str:
|
|
return re.sub(r"\s+", "", value or "").upper()
|
|
|
|
|
|
def _cleanup_objects(keys: list[str]) -> None:
|
|
from app.services.storage import remove_stored_object
|
|
|
|
for key in keys:
|
|
try:
|
|
remove_stored_object("private", key)
|
|
except Exception:
|
|
continue
|
|
|
|
|
|
def create_complaint(
|
|
db,
|
|
*,
|
|
title: str,
|
|
subject: str,
|
|
body: str,
|
|
contact_name: str,
|
|
mobile: str,
|
|
email: str,
|
|
files: list[tuple[str, str, bytes]],
|
|
) -> dict:
|
|
from app.core.field_crypto import encrypt_text
|
|
from app.core.masking import mask_email, mask_mobile
|
|
from app.models.content import AdminFile
|
|
from app.services.storage import make_object_key, safe_filename, upload_object
|
|
|
|
cleaned_title = _plain(title, label="举报标题", limit=200)
|
|
cleaned_subject = _plain(subject, label="被举报对象", limit=200)
|
|
cleaned_body = _plain(body, label="违规事实", limit=4000)
|
|
cleaned_name = _plain(contact_name, label="联系人", limit=40)
|
|
cleaned_mobile = _mobile(mobile)
|
|
cleaned_email = _reporter_email(email)
|
|
_reject_sensitive(db, "\n".join([cleaned_title, cleaned_subject, cleaned_body, cleaned_name]))
|
|
if not files:
|
|
raise ValueError("请上传证据材料")
|
|
if len(files) > MAX_EVIDENCE:
|
|
raise ValueError("证据材料不能超过 8 个")
|
|
for filename, _content_type, data in files:
|
|
if not data:
|
|
raise ValueError("证据材料不能为空")
|
|
if not safe_filename(filename):
|
|
raise ValueError("证据材料文件名不正确")
|
|
|
|
mobile_cipher = encrypt_text(cleaned_mobile)
|
|
email_cipher = encrypt_text(cleaned_email) if cleaned_email else ""
|
|
stored: list[str] = []
|
|
try:
|
|
file_rows = []
|
|
for filename, content_type, data in files:
|
|
object_key = make_object_key(filename)
|
|
upload_object("private", object_key, data, content_type or "application/octet-stream")
|
|
stored.append(object_key)
|
|
row = AdminFile(
|
|
bucket="private",
|
|
object_key=object_key,
|
|
filename=safe_filename(filename),
|
|
content_type=(content_type or "")[:128],
|
|
size=len(data),
|
|
created_by=None,
|
|
)
|
|
db.add(row)
|
|
db.flush()
|
|
file_rows.append(row)
|
|
complaint = DisciplineComplaint(
|
|
code=_new_code(db),
|
|
title=cleaned_title,
|
|
subject=cleaned_subject,
|
|
body=cleaned_body,
|
|
contact_name=cleaned_name,
|
|
mobile_cipher=mobile_cipher,
|
|
mobile_mask=mask_mobile(cleaned_mobile),
|
|
email_cipher=email_cipher,
|
|
email_mask=mask_email(cleaned_email),
|
|
status="pending",
|
|
created_at=_now(),
|
|
)
|
|
db.add(complaint)
|
|
db.flush()
|
|
for index, row in enumerate(file_rows):
|
|
db.add(
|
|
DisciplineComplaintFile(
|
|
complaint_id=complaint.id,
|
|
file_id=row.id,
|
|
position=index,
|
|
)
|
|
)
|
|
db.commit()
|
|
db.refresh(complaint)
|
|
except Exception:
|
|
db.rollback()
|
|
_cleanup_objects(stored)
|
|
raise
|
|
return {
|
|
"code": complaint.code,
|
|
"status": complaint.status,
|
|
"statusLabel": _status_label(complaint.status),
|
|
"workdays": complaint_workdays(db),
|
|
}
|
|
|
|
|
|
def _overdue(row: DisciplineComplaint, days: int) -> bool:
|
|
from app.models.expert import is_overdue
|
|
|
|
if row.status != "pending":
|
|
return False
|
|
return bool(is_overdue(row.created_at, days))
|
|
|
|
|
|
def _list_item(row: DisciplineComplaint, *, days: int, attachment_count: int) -> dict:
|
|
return {
|
|
"id": row.id,
|
|
"code": row.code,
|
|
"title": row.title,
|
|
"subject": row.subject,
|
|
"contactName": row.contact_name,
|
|
"mobileMask": row.mobile_mask,
|
|
"emailMask": row.email_mask,
|
|
"status": row.status,
|
|
"statusLabel": _status_label(row.status),
|
|
"createdAt": row.created_at.isoformat(timespec="seconds") if row.created_at else "",
|
|
"processedAt": row.processed_at.isoformat(timespec="seconds") if row.processed_at else "",
|
|
"overdue": _overdue(row, days),
|
|
"attachmentCount": attachment_count,
|
|
}
|
|
|
|
|
|
def admin_complaints(db, *, status: str | None = None) -> dict:
|
|
chosen = (status or "").strip()
|
|
if chosen and chosen not in COMPLAINT_STATUSES:
|
|
raise ValueError("状态不正确")
|
|
days = complaint_workdays(db)
|
|
filters = []
|
|
if chosen:
|
|
filters.append(DisciplineComplaint.status == chosen)
|
|
rows = db.scalars(
|
|
select(DisciplineComplaint)
|
|
.where(*filters)
|
|
.order_by(DisciplineComplaint.created_at.desc(), DisciplineComplaint.id.desc())
|
|
).all()
|
|
counts = dict(
|
|
db.execute(
|
|
select(DisciplineComplaintFile.complaint_id, func.count())
|
|
.group_by(DisciplineComplaintFile.complaint_id)
|
|
).all()
|
|
)
|
|
pending = db.scalar(
|
|
select(func.count())
|
|
.select_from(DisciplineComplaint)
|
|
.where(DisciplineComplaint.status == "pending")
|
|
) or 0
|
|
return {
|
|
"items": [
|
|
_list_item(row, days=days, attachment_count=int(counts.get(row.id, 0)))
|
|
for row in rows
|
|
],
|
|
"pendingCount": int(pending),
|
|
"workdays": days,
|
|
}
|
|
|
|
|
|
def _attachments(db, complaint_id: int) -> list[dict]:
|
|
from app.models.content import AdminFile
|
|
from app.services.storage import private_object_url
|
|
|
|
links = db.scalars(
|
|
select(DisciplineComplaintFile)
|
|
.where(DisciplineComplaintFile.complaint_id == complaint_id)
|
|
.order_by(DisciplineComplaintFile.position.asc(), DisciplineComplaintFile.id.asc())
|
|
).all()
|
|
items = []
|
|
for link in links:
|
|
row = db.get(AdminFile, link.file_id)
|
|
if row is None or row.bucket != "private":
|
|
continue
|
|
url, expires_at = private_object_url(row.object_key)
|
|
items.append(
|
|
{
|
|
"id": row.id,
|
|
"filename": row.filename,
|
|
"contentType": row.content_type or "",
|
|
"url": url,
|
|
"expiresAt": expires_at.isoformat(),
|
|
}
|
|
)
|
|
return items
|
|
|
|
|
|
def admin_complaint(db, complaint_id: int) -> dict | None:
|
|
from app.core.field_crypto import decrypt_text
|
|
|
|
row = db.get(DisciplineComplaint, complaint_id)
|
|
if row is None:
|
|
return None
|
|
days = complaint_workdays(db)
|
|
item = _list_item(row, days=days, attachment_count=0)
|
|
item["body"] = row.body
|
|
item["mobile"] = decrypt_text(row.mobile_cipher)
|
|
item["email"] = decrypt_text(row.email_cipher) if row.email_cipher else ""
|
|
item["attachments"] = _attachments(db, row.id)
|
|
item["attachmentCount"] = len(item["attachments"])
|
|
return item
|
|
|
|
|
|
def mark_processed(db, complaint_id: int, *, staff_id: int) -> None:
|
|
row = db.get(DisciplineComplaint, complaint_id)
|
|
if row is None:
|
|
raise ValueError("投诉不存在")
|
|
if row.status != "processed":
|
|
row.status = "processed"
|
|
row.processed_at = _now()
|
|
row.processed_by = staff_id
|
|
db.commit()
|
|
|
|
|
|
def public_complaint_status(db, code: str) -> dict | None:
|
|
normalized = normalize_code(code)
|
|
if not _CODE.fullmatch(normalized):
|
|
return None
|
|
row = db.scalar(select(DisciplineComplaint).where(DisciplineComplaint.code == normalized))
|
|
if row is None:
|
|
return None
|
|
days = complaint_workdays(db)
|
|
created_on = row.created_at.date().isoformat() if row.created_at else ""
|
|
return {
|
|
"code": row.code,
|
|
"status": row.status,
|
|
"statusLabel": _status_label(row.status),
|
|
"createdOn": created_on,
|
|
"workdays": days,
|
|
"overdue": _overdue(row, days),
|
|
}
|