320 lines
11 KiB
Python
320 lines
11 KiB
Python
from sqlalchemy import Boolean, Integer, String, Text, 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"}
|
|
|
|
|
|
ABOUT_PAGE_KEYS = ("intro", "org", "contact", "party_branch")
|
|
ABOUT_PAGE_PLACEHOLDERS = {
|
|
"intro": "协会简介整理中",
|
|
"org": "组织机构整理中",
|
|
"contact": "联系与合作信息整理中",
|
|
"party_branch": "协会党支部介绍整理中",
|
|
}
|
|
EXPERT_LIST_PLACEHOLDER = "暂无专家委员会信息"
|
|
|
|
|
|
class AboutPage(Base):
|
|
__tablename__ = "about_page"
|
|
__table_args__ = _mysql_args()
|
|
|
|
page_key: Mapped[str] = mapped_column(String(32), primary_key=True)
|
|
body_html: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
published: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
|
|
|
|
|
class AboutPageAttachment(Base):
|
|
__tablename__ = "about_page_attachment"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
page_key: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
|
|
file_id: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
def _public_file(db, file_id: int):
|
|
from app.models.content import AdminFile
|
|
|
|
if isinstance(file_id, bool) or not isinstance(file_id, int) or file_id <= 0:
|
|
return None
|
|
row = db.get(AdminFile, file_id)
|
|
if row is None or row.bucket != "public" or not row.object_key:
|
|
return None
|
|
return row
|
|
|
|
|
|
def _page_attachments(db, page_key: str) -> list:
|
|
from app.services.storage import public_object_url
|
|
|
|
links = db.scalars(
|
|
select(AboutPageAttachment)
|
|
.where(AboutPageAttachment.page_key == page_key)
|
|
.order_by(AboutPageAttachment.position.asc(), AboutPageAttachment.id.asc())
|
|
).all()
|
|
files = []
|
|
for link in links:
|
|
row = _public_file(db, link.file_id)
|
|
if row is None:
|
|
continue
|
|
files.append({"filename": row.filename, "url": public_object_url(row.object_key)})
|
|
return files
|
|
|
|
|
|
def _editor_attachments(db, page_key: str) -> list:
|
|
from app.services.storage import public_object_url
|
|
|
|
links = db.scalars(
|
|
select(AboutPageAttachment)
|
|
.where(AboutPageAttachment.page_key == page_key)
|
|
.order_by(AboutPageAttachment.position.asc(), AboutPageAttachment.id.asc())
|
|
).all()
|
|
files = []
|
|
for link in links:
|
|
row = _public_file(db, link.file_id)
|
|
if row is None:
|
|
continue
|
|
files.append(
|
|
{
|
|
"id": row.id,
|
|
"filename": row.filename,
|
|
"url": public_object_url(row.object_key),
|
|
}
|
|
)
|
|
return files
|
|
|
|
|
|
def _replace_attachments(db, page_key: str, attachments: list) -> None:
|
|
links = db.scalars(
|
|
select(AboutPageAttachment).where(AboutPageAttachment.page_key == page_key)
|
|
).all()
|
|
for link in links:
|
|
db.delete(link)
|
|
db.flush()
|
|
for index, row in enumerate(attachments):
|
|
db.add(AboutPageAttachment(page_key=page_key, file_id=row.id, position=index))
|
|
|
|
|
|
def _prepare_page(db, payload: dict) -> tuple[str, bool, list]:
|
|
from app.core.config import get_settings
|
|
from app.models.content import AdminFile
|
|
from app.services.html_filter import filter_html
|
|
|
|
attachments = []
|
|
for file_id in payload.get("attachmentFileIds") or []:
|
|
row = _public_file(db, file_id)
|
|
if row is None:
|
|
raise ValueError("附件必须是已上传的公开文件")
|
|
attachments.append(row)
|
|
keys = list(db.scalars(select(AdminFile.object_key).where(AdminFile.bucket == "public")))
|
|
cleaned = filter_html(payload.get("bodyHtml") or "", list(keys), get_settings().minio_base_url)
|
|
return cleaned, bool(payload.get("published", False)), attachments
|
|
|
|
|
|
def save_about_page(db, page_key: str, payload: dict) -> None:
|
|
if page_key not in ABOUT_PAGE_KEYS:
|
|
raise ValueError("单篇页键不正确")
|
|
body_html, published, attachments = _prepare_page(db, payload)
|
|
row = db.get(AboutPage, page_key)
|
|
if row is None:
|
|
row = AboutPage(page_key=page_key)
|
|
db.add(row)
|
|
row.body_html = body_html
|
|
row.published = published
|
|
db.flush()
|
|
_replace_attachments(db, page_key, attachments)
|
|
db.commit()
|
|
|
|
|
|
def admin_about_page(db, page_key: str) -> dict | None:
|
|
if page_key not in ABOUT_PAGE_KEYS:
|
|
return None
|
|
row = db.get(AboutPage, page_key)
|
|
files = _editor_attachments(db, page_key)
|
|
return {
|
|
"key": page_key,
|
|
"published": bool(row.published) if row else False,
|
|
"bodyHtml": row.body_html if row else "",
|
|
"attachmentFileIds": [item["id"] for item in files],
|
|
"attachments": files,
|
|
}
|
|
|
|
|
|
def public_about_page(db, page_key: str) -> dict | None:
|
|
if page_key not in ABOUT_PAGE_KEYS:
|
|
return None
|
|
placeholder = ABOUT_PAGE_PLACEHOLDERS[page_key]
|
|
row = db.get(AboutPage, page_key)
|
|
files = _page_attachments(db, page_key) if row else []
|
|
body = (row.body_html if row else "").strip()
|
|
if row is None or not row.published or (not body and not files):
|
|
return {"key": page_key, "placeholder": placeholder}
|
|
return {
|
|
"key": page_key,
|
|
"bodyHtml": row.body_html,
|
|
"attachments": files,
|
|
}
|
|
|
|
|
|
class AboutExpert(Base):
|
|
__tablename__ = "about_expert"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
name: Mapped[str] = mapped_column(String(200), nullable=False, default="")
|
|
body_html: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
published: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
|
|
|
|
|
class AboutExpertAttachment(Base):
|
|
__tablename__ = "about_expert_attachment"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
expert_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 _expert_attachments(db, expert_id: int, *, editor: bool = False) -> list:
|
|
from app.services.storage import public_object_url
|
|
|
|
links = db.scalars(
|
|
select(AboutExpertAttachment)
|
|
.where(AboutExpertAttachment.expert_id == expert_id)
|
|
.order_by(AboutExpertAttachment.position.asc(), AboutExpertAttachment.id.asc())
|
|
).all()
|
|
files = []
|
|
for link in links:
|
|
row = _public_file(db, link.file_id)
|
|
if row is None:
|
|
continue
|
|
item = {"filename": row.filename, "url": public_object_url(row.object_key)}
|
|
if editor:
|
|
item = {"id": row.id, **item}
|
|
files.append(item)
|
|
return files
|
|
|
|
|
|
def _replace_expert_attachments(db, expert_id: int, attachments: list) -> None:
|
|
links = db.scalars(
|
|
select(AboutExpertAttachment).where(AboutExpertAttachment.expert_id == expert_id)
|
|
).all()
|
|
for link in links:
|
|
db.delete(link)
|
|
db.flush()
|
|
for index, row in enumerate(attachments):
|
|
db.add(AboutExpertAttachment(expert_id=expert_id, file_id=row.id, position=index))
|
|
|
|
|
|
def _prepare_expert(db, payload: dict) -> tuple[dict, list]:
|
|
from app.core.config import get_settings
|
|
from app.models.content import AdminFile
|
|
from app.services.html_filter import filter_html
|
|
|
|
name = payload.get("name")
|
|
if not isinstance(name, str) or not name.strip():
|
|
raise ValueError("委员会名称不能为空")
|
|
sort_order = payload.get("sortOrder", 0)
|
|
if isinstance(sort_order, bool) or not isinstance(sort_order, int):
|
|
raise ValueError("排序必须是整数")
|
|
attachments = []
|
|
for file_id in payload.get("attachmentFileIds") or []:
|
|
row = _public_file(db, file_id)
|
|
if row is None:
|
|
raise ValueError("附件必须是已上传的公开文件")
|
|
attachments.append(row)
|
|
keys = list(db.scalars(select(AdminFile.object_key).where(AdminFile.bucket == "public")))
|
|
cleaned = filter_html(payload.get("bodyHtml") or "", list(keys), get_settings().minio_base_url)
|
|
fields = {
|
|
"name": name.strip(),
|
|
"body_html": cleaned,
|
|
"sort_order": sort_order,
|
|
"published": bool(payload.get("published", False)),
|
|
}
|
|
return fields, attachments
|
|
|
|
|
|
def create_expert(db, payload: dict) -> int:
|
|
fields, attachments = _prepare_expert(db, payload)
|
|
row = AboutExpert(**fields)
|
|
db.add(row)
|
|
db.flush()
|
|
_replace_expert_attachments(db, row.id, attachments)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return row.id
|
|
|
|
|
|
def update_expert(db, expert_id: int, payload: dict) -> None:
|
|
row = db.get(AboutExpert, expert_id)
|
|
if row is None:
|
|
raise ValueError("专家委员会不存在")
|
|
fields, attachments = _prepare_expert(db, payload)
|
|
for key, value in fields.items():
|
|
setattr(row, key, value)
|
|
_replace_expert_attachments(db, row.id, attachments)
|
|
db.commit()
|
|
|
|
|
|
def delete_expert(db, expert_id: int) -> None:
|
|
row = db.get(AboutExpert, expert_id)
|
|
if row is None:
|
|
raise ValueError("专家委员会不存在")
|
|
links = db.scalars(
|
|
select(AboutExpertAttachment).where(AboutExpertAttachment.expert_id == expert_id)
|
|
).all()
|
|
for link in links:
|
|
db.delete(link)
|
|
db.delete(row)
|
|
db.commit()
|
|
|
|
|
|
def admin_experts(db) -> dict:
|
|
rows = db.scalars(
|
|
select(AboutExpert).order_by(AboutExpert.sort_order.asc(), AboutExpert.id.asc())
|
|
).all()
|
|
items = []
|
|
for row in rows:
|
|
files = _expert_attachments(db, row.id, editor=True)
|
|
items.append(
|
|
{
|
|
"id": row.id,
|
|
"name": row.name,
|
|
"bodyHtml": row.body_html,
|
|
"sortOrder": row.sort_order,
|
|
"published": bool(row.published),
|
|
"attachmentFileIds": [item["id"] for item in files],
|
|
"attachments": files,
|
|
}
|
|
)
|
|
return {"items": items}
|
|
|
|
|
|
def public_experts(db) -> dict:
|
|
rows = db.scalars(
|
|
select(AboutExpert)
|
|
.where(AboutExpert.published.is_(True))
|
|
.order_by(AboutExpert.sort_order.asc(), AboutExpert.id.asc())
|
|
).all()
|
|
if not rows:
|
|
return {"placeholder": EXPERT_LIST_PLACEHOLDER}
|
|
return {"items": [{"id": row.id, "name": row.name} for row in rows]}
|
|
|
|
|
|
def public_expert(db, expert_id: int) -> dict | None:
|
|
row = db.get(AboutExpert, expert_id)
|
|
if row is None or not row.published:
|
|
return None
|
|
return {
|
|
"id": row.id,
|
|
"name": row.name,
|
|
"bodyHtml": row.body_html,
|
|
"attachments": _expert_attachments(db, row.id),
|
|
}
|