1616 lines
58 KiB
Python
1616 lines
58 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import Boolean, DateTime, Float, Integer, String, Text, delete, func, or_, select, text
|
|
from sqlalchemy.dialects.mysql import MEDIUMTEXT
|
|
from sqlalchemy.exc import IntegrityError
|
|
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"}
|
|
|
|
TECH = "科技蓝"
|
|
BRAND = "LOGO蓝"
|
|
TEMPLATE_NAMES = (TECH, BRAND)
|
|
|
|
SLOGAN_EMPTY = "协会定位整理中"
|
|
VALUE_EMPTY = "价值主张整理中"
|
|
|
|
EMPTY_REGIONS = {
|
|
"carousel": "轮播内容准备中",
|
|
"news": "新闻动态即将发布",
|
|
"notices": "暂无新的通知公告",
|
|
"policies": "政策法规整理中",
|
|
"dashboard": "数据看板整理中",
|
|
"audiences": "估价师、机构与公众入口即将开放",
|
|
"hotLinks": "报告查询、报告审核与投诉举报入口即将开放",
|
|
"honors": "协会荣誉与资质整理中",
|
|
"members": "会员单位名录整理中",
|
|
"compliance": "隐私政策、使用条款与网站声明整理中",
|
|
"annual_reports": "暂无年度报告或工作计划",
|
|
"work_briefs": "暂无工作简报",
|
|
"qualifications": "暂无资质与荣誉",
|
|
"tech_seminars": "暂无技术研讨",
|
|
"tech_exchange": "暂无技术交流",
|
|
"party_activities": "暂无党建活动",
|
|
"appraisal_guides": "暂无估价指南",
|
|
"institution_directories": "暂无机构名录",
|
|
"policy_analyses": "暂无政策分析",
|
|
"case_library": "暂无案例库",
|
|
"edu_notices": "暂无教育公告",
|
|
"warning_cases": "暂无警示案例",
|
|
"integrity_records": "暂无诚信档案",
|
|
"discipline_results": "暂无结果公示",
|
|
}
|
|
|
|
HOT_LINK_SLOTS = (
|
|
("reportQuery", "报告查询", "/reports/query"),
|
|
("reportReview", "报告审核", "/reports/review"),
|
|
("complaint", "投诉举报", "/discipline"),
|
|
)
|
|
FIXED_HOT_URLS = {
|
|
"reportQuery": "/reports/query",
|
|
"reportReview": "/reports/review",
|
|
}
|
|
HOT_LINK_KEYS = {key for key, _label, _url in HOT_LINK_SLOTS}
|
|
|
|
|
|
class HomeSlide(Base):
|
|
__tablename__ = "home_slide"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
title: Mapped[str] = mapped_column(String(200), nullable=False, default="")
|
|
image_file_id: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
|
|
interval_seconds: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
published: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
|
body_html: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
|
|
|
|
class HomeSlideAttachment(Base):
|
|
__tablename__ = "home_slide_attachment"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
slide_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)
|
|
|
|
|
|
HOME_ARTICLE_CATEGORIES = ("news", "notices", "policies", "dashboard")
|
|
COLUMN_ARTICLE_CATEGORIES = (
|
|
"annual_reports",
|
|
"work_briefs",
|
|
"qualifications",
|
|
"tech_seminars",
|
|
"tech_exchange",
|
|
"party_activities",
|
|
"appraisal_guides",
|
|
"institution_directories",
|
|
"policy_analyses",
|
|
"case_library",
|
|
"edu_notices",
|
|
"warning_cases",
|
|
"integrity_records",
|
|
"discipline_results",
|
|
)
|
|
ARTICLE_CATEGORIES = HOME_ARTICLE_CATEGORIES + COLUMN_ARTICLE_CATEGORIES
|
|
|
|
|
|
def ensure_article_body_capacity(db) -> None:
|
|
bind = db.get_bind()
|
|
if bind.dialect.name != "mysql":
|
|
return
|
|
db.execute(text("ALTER TABLE home_article MODIFY body_html MEDIUMTEXT NOT NULL"))
|
|
db.commit()
|
|
|
|
|
|
RELATED_CATEGORIES = ("主管部门", "相关部门", "合作协会")
|
|
AUDIENCE_NAMES = ("估价师", "估价机构", "公众")
|
|
MENU_KEYS = ("member",)
|
|
EDU_PLATFORM_EMPTY = "暂无在线教育"
|
|
DEFAULT_DISPLAY_COUNT = 8
|
|
COLUMN_LIST_PAGE_SIZES = (10, 20, 100)
|
|
DEFAULT_COLUMN_PAGE_SIZE = 10
|
|
APPRAISAL_GUIDE_KINDS = ("residential", "commercial", "industrial", "specialty")
|
|
APPRAISAL_GUIDE_KIND_LABELS = {
|
|
"residential": "住宅",
|
|
"commercial": "商业",
|
|
"industrial": "工业",
|
|
"specialty": "特色地产",
|
|
}
|
|
POLICY_ANALYSIS_KINDS = ("national", "local", "industry", "expert", "digest")
|
|
POLICY_ANALYSIS_KIND_LABELS = {
|
|
"national": "国家级",
|
|
"local": "地方级",
|
|
"industry": "行业级",
|
|
"expert": "专家解读",
|
|
"digest": "政策汇编",
|
|
}
|
|
POLICY_SEARCH_EMPTY = "未找到相关政策"
|
|
|
|
|
|
class HomeArticleList(Base):
|
|
__tablename__ = "home_article_list"
|
|
__table_args__ = _mysql_args()
|
|
|
|
category: Mapped[str] = mapped_column(String(32), primary_key=True)
|
|
display_count: Mapped[int] = mapped_column(Integer, nullable=False, default=DEFAULT_DISPLAY_COUNT)
|
|
|
|
|
|
class HomeArticle(Base):
|
|
__tablename__ = "home_article"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
category: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
|
|
title: Mapped[str] = mapped_column(String(200), nullable=False, default="")
|
|
body_html: Mapped[str] = mapped_column(
|
|
Text().with_variant(MEDIUMTEXT(), "mysql"), nullable=False, default=""
|
|
)
|
|
image_file_id: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
|
published: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
|
published_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
|
|
|
|
|
|
class HomeArticleAttachment(Base):
|
|
__tablename__ = "home_article_attachment"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
article_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)
|
|
|
|
|
|
class HomeArticleKind(Base):
|
|
"""栏目文章的附加分类。估价指南与政策分析使用。"""
|
|
|
|
__tablename__ = "home_article_kind"
|
|
__table_args__ = _mysql_args()
|
|
|
|
article_id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
|
kind: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
|
|
|
|
|
|
class HomeAudienceLink(Base):
|
|
__tablename__ = "home_audience_link"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
category: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
|
|
label: Mapped[str] = mapped_column(String(200), nullable=False)
|
|
url: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
visible: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
class HomeHotLink(Base):
|
|
__tablename__ = "home_hot_link"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
key: Mapped[str] = mapped_column(String(32), nullable=False, default="", unique=True)
|
|
label: Mapped[str] = mapped_column(String(200), nullable=False)
|
|
url: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
visible: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
class HomeMenuItem(Base):
|
|
__tablename__ = "home_menu_item"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
key: Mapped[str] = mapped_column(String(32), unique=True, nullable=False)
|
|
label: Mapped[str] = mapped_column(String(200), nullable=False, default="")
|
|
url: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
visible: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
class HomeMarquee(Base):
|
|
__tablename__ = "home_marquee"
|
|
__table_args__ = _mysql_args()
|
|
|
|
kind: Mapped[str] = mapped_column(String(16), primary_key=True)
|
|
interval_seconds: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
|
|
|
|
class HomeHonor(Base):
|
|
__tablename__ = "home_honor"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
image_file_id: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
class HomeMember(Base):
|
|
__tablename__ = "home_member"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
name: Mapped[str] = mapped_column(String(200), nullable=False)
|
|
url: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
image_file_id: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
class EduPlatform(Base):
|
|
__tablename__ = "edu_platform"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
name: Mapped[str] = mapped_column(String(200), nullable=False)
|
|
url: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
image_file_id: Mapped[int] = mapped_column(Integer, nullable=False)
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
class HomeContact(Base):
|
|
__tablename__ = "home_contact"
|
|
__table_args__ = _mysql_args()
|
|
|
|
template_name: Mapped[str] = mapped_column(String(32), primary_key=True)
|
|
address: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
phone: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
email: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
latitude: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
longitude: Mapped[float | None] = mapped_column(Float, nullable=True)
|
|
|
|
|
|
class HomeRelatedLink(Base):
|
|
__tablename__ = "home_related_link"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
template_name: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
|
|
category: Mapped[str] = mapped_column(String(32), nullable=False)
|
|
label: Mapped[str] = mapped_column(String(200), nullable=False)
|
|
url: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
visible: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
class HomeComplianceLink(Base):
|
|
__tablename__ = "home_compliance_link"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
template_name: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
|
|
label: Mapped[str] = mapped_column(String(200), nullable=False)
|
|
url: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
visible: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
|
|
sort_order: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
|
|
|
|
|
class HomeTemplate(Base):
|
|
__tablename__ = "home_template"
|
|
__table_args__ = _mysql_args()
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
|
|
name: Mapped[str] = mapped_column(String(32), unique=True, nullable=False)
|
|
slogan: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
value_proposition: Mapped[str] = mapped_column(Text, nullable=False, default="")
|
|
is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
|
|
|
|
|
def ensure_home_templates(db) -> None:
|
|
existing = set(db.scalars(select(HomeTemplate.name)))
|
|
missing = [
|
|
HomeTemplate(name=name, slogan="", value_proposition="", is_active=name == TECH)
|
|
for name in TEMPLATE_NAMES
|
|
if name not in existing
|
|
]
|
|
if missing:
|
|
db.add_all(missing)
|
|
try:
|
|
db.commit()
|
|
except IntegrityError:
|
|
db.rollback()
|
|
rows = db.scalars(select(HomeTemplate)).all()
|
|
if rows and not any(row.is_active for row in rows):
|
|
for row in rows:
|
|
row.is_active = row.name == TECH
|
|
db.commit()
|
|
|
|
|
|
def _text(value: str, empty: str) -> str:
|
|
text = "" if value is None else str(value).strip()
|
|
return text or empty
|
|
|
|
|
|
def public_home(db) -> dict:
|
|
from app.models.content import settings_view
|
|
|
|
ensure_home_templates(db)
|
|
rows = db.scalars(select(HomeTemplate)).all()
|
|
active = next((row for row in rows if row.is_active), None)
|
|
if active is None and rows:
|
|
active = next(row for row in rows if row.name == TECH)
|
|
settings = settings_view(db)
|
|
payload = {
|
|
"templateName": active.name if active else TECH,
|
|
"slogan": _text(active.slogan if active else "", SLOGAN_EMPTY),
|
|
"valueProposition": _text(active.value_proposition if active else "", VALUE_EMPTY),
|
|
"icpText": settings["icp_text"],
|
|
"policeText": settings["police_text"],
|
|
}
|
|
template_name = active.name if active else TECH
|
|
for key, sentence in EMPTY_REGIONS.items():
|
|
if key == "carousel" or key in HOME_ARTICLE_CATEGORIES or key in COLUMN_ARTICLE_CATEGORIES or key in {
|
|
"audiences",
|
|
"hotLinks",
|
|
"honors",
|
|
"members",
|
|
"compliance",
|
|
}:
|
|
continue
|
|
payload[key] = {"placeholder": sentence}
|
|
payload["carousel"] = _carousel(db)
|
|
for key in HOME_ARTICLE_CATEGORIES:
|
|
payload[key] = _articles(db, key, EMPTY_REGIONS[key])
|
|
payload["audiences"] = _public_audiences(db)
|
|
payload["hotLinks"] = _public_hot_links(db)
|
|
payload["shortcutMenu"] = _shortcut_menu(db)
|
|
payload["honors"] = _public_honors(db)
|
|
payload["members"] = _public_members(db)
|
|
payload["contact"] = _public_contact(db, template_name)
|
|
payload["relatedLinks"] = _public_related(db, template_name)
|
|
payload["compliance"] = _public_compliance(db, template_name)
|
|
return payload
|
|
|
|
|
|
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 _carousel(db) -> dict:
|
|
from app.services.storage import public_object_url
|
|
|
|
rows = db.scalars(
|
|
select(HomeSlide)
|
|
.where(HomeSlide.published.is_(True), HomeSlide.enabled.is_(True))
|
|
.order_by(HomeSlide.sort_order.asc(), HomeSlide.id.asc())
|
|
).all()
|
|
items = []
|
|
for row in rows:
|
|
image = _public_file(db, row.image_file_id)
|
|
if image is None:
|
|
continue
|
|
items.append(
|
|
{
|
|
"id": row.id,
|
|
"title": row.title,
|
|
"imageUrl": public_object_url(image.object_key),
|
|
"sortOrder": row.sort_order,
|
|
"intervalSeconds": row.interval_seconds,
|
|
}
|
|
)
|
|
if not items:
|
|
return {"placeholder": EMPTY_REGIONS["carousel"]}
|
|
return {"items": items}
|
|
|
|
|
|
def _prepare_slide(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
|
|
|
|
image = _public_file(db, payload["imageFileId"])
|
|
if image is None:
|
|
raise ValueError("轮播图必须是已上传的公开图片")
|
|
interval = payload["intervalSeconds"]
|
|
if isinstance(interval, bool) or not isinstance(interval, int) or interval <= 0:
|
|
raise ValueError("间隔秒数必须是正整数")
|
|
attachments = []
|
|
for file_id in payload["attachmentFileIds"]:
|
|
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 = {
|
|
"title": "" if payload.get("title") is None else str(payload["title"]),
|
|
"image_file_id": image.id,
|
|
"sort_order": int(payload.get("sortOrder") or 0),
|
|
"enabled": bool(payload.get("enabled", True)),
|
|
"interval_seconds": interval,
|
|
"published": bool(payload.get("published", False)),
|
|
"body_html": cleaned,
|
|
}
|
|
return fields, attachments
|
|
|
|
|
|
def _replace_slide_attachments(db, slide_id: int, attachments: list) -> None:
|
|
links = db.scalars(
|
|
select(HomeSlideAttachment).where(HomeSlideAttachment.slide_id == slide_id)
|
|
).all()
|
|
for link in links:
|
|
db.delete(link)
|
|
db.flush()
|
|
for index, row in enumerate(attachments):
|
|
db.add(HomeSlideAttachment(slide_id=slide_id, file_id=row.id, position=index))
|
|
|
|
|
|
def create_slide(db, payload: dict) -> int:
|
|
fields, attachments = _prepare_slide(db, payload)
|
|
slide = HomeSlide(**fields)
|
|
db.add(slide)
|
|
db.flush()
|
|
_replace_slide_attachments(db, slide.id, attachments)
|
|
db.commit()
|
|
db.refresh(slide)
|
|
return slide.id
|
|
|
|
|
|
def update_slide(db, slide_id: int, payload: dict) -> None:
|
|
slide = db.get(HomeSlide, slide_id)
|
|
if slide is None:
|
|
raise ValueError("轮播不存在")
|
|
fields, attachments = _prepare_slide(db, payload)
|
|
for key, value in fields.items():
|
|
setattr(slide, key, value)
|
|
_replace_slide_attachments(db, slide.id, attachments)
|
|
db.commit()
|
|
|
|
|
|
def public_slide(db, slide_id: int) -> dict | None:
|
|
from app.services.storage import public_object_url
|
|
|
|
row = db.get(HomeSlide, slide_id)
|
|
if row is None or not row.published or not row.enabled:
|
|
return None
|
|
image = _public_file(db, row.image_file_id)
|
|
links = db.scalars(
|
|
select(HomeSlideAttachment)
|
|
.where(HomeSlideAttachment.slide_id == row.id)
|
|
.order_by(HomeSlideAttachment.position.asc(), HomeSlideAttachment.id.asc())
|
|
).all()
|
|
attachments = []
|
|
for link in links:
|
|
file_row = _public_file(db, link.file_id)
|
|
if file_row is None:
|
|
continue
|
|
attachments.append(
|
|
{"filename": file_row.filename, "url": public_object_url(file_row.object_key)}
|
|
)
|
|
return {
|
|
"id": row.id,
|
|
"title": row.title,
|
|
"bodyHtml": row.body_html,
|
|
"imageUrl": public_object_url(image.object_key) if image else "",
|
|
"intervalSeconds": row.interval_seconds,
|
|
"attachments": attachments,
|
|
}
|
|
|
|
|
|
def delete_slide(db, slide_id: int) -> None:
|
|
row = db.get(HomeSlide, slide_id)
|
|
if row is None:
|
|
raise ValueError("轮播不存在")
|
|
links = db.scalars(
|
|
select(HomeSlideAttachment).where(HomeSlideAttachment.slide_id == row.id)
|
|
).all()
|
|
for link in links:
|
|
db.delete(link)
|
|
db.delete(row)
|
|
db.commit()
|
|
|
|
|
|
def _display_count(db, category: str) -> int:
|
|
row = db.get(HomeArticleList, category)
|
|
if row is None or row.display_count <= 0:
|
|
return DEFAULT_DISPLAY_COUNT
|
|
return row.display_count
|
|
|
|
|
|
def set_article_display(db, limits: dict) -> None:
|
|
unknown = [key for key in limits if key not in HOME_ARTICLE_CATEGORIES]
|
|
if unknown:
|
|
raise ValueError("文章类别不正确")
|
|
for category, count in limits.items():
|
|
if count is None:
|
|
continue
|
|
if isinstance(count, bool) or not isinstance(count, int) or count <= 0:
|
|
raise ValueError("展示条数必须是正整数")
|
|
row = db.get(HomeArticleList, category)
|
|
if row is None:
|
|
db.add(HomeArticleList(category=category, display_count=count))
|
|
else:
|
|
row.display_count = count
|
|
db.commit()
|
|
|
|
|
|
def _published_at(value) -> datetime | None:
|
|
if value is None or value == "":
|
|
return None
|
|
if not isinstance(value, str):
|
|
raise ValueError("发布时间格式不正确")
|
|
try:
|
|
parsed = datetime.fromisoformat(value)
|
|
except ValueError as exc:
|
|
raise ValueError("发布时间格式不正确") from exc
|
|
if parsed.tzinfo is not None:
|
|
parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None)
|
|
return parsed
|
|
|
|
|
|
def _articles(db, category: str, placeholder: str, *, limit: int | None = None) -> dict:
|
|
from app.services.storage import public_object_url
|
|
|
|
query = (
|
|
select(HomeArticle)
|
|
.where(HomeArticle.category == category, HomeArticle.published.is_(True))
|
|
.order_by(HomeArticle.published_at.desc(), HomeArticle.id.desc())
|
|
)
|
|
if limit is None:
|
|
query = query.limit(_display_count(db, category))
|
|
else:
|
|
query = query.limit(limit)
|
|
rows = db.scalars(query).all()
|
|
items = []
|
|
for row in rows:
|
|
item = {"id": row.id, "title": row.title}
|
|
image = _public_file(db, row.image_file_id) if row.image_file_id else None
|
|
if image is not None:
|
|
item["imageUrl"] = public_object_url(image.object_key)
|
|
items.append(item)
|
|
if not items:
|
|
return {"placeholder": placeholder}
|
|
return {"items": items}
|
|
|
|
|
|
def _like_contains(value: str) -> str:
|
|
escaped = value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
|
return f"%{escaped}%"
|
|
|
|
|
|
def public_column_articles(
|
|
db,
|
|
category: str,
|
|
*,
|
|
current: int = 1,
|
|
size: int = DEFAULT_COLUMN_PAGE_SIZE,
|
|
kind: str | None = None,
|
|
q: str | None = None,
|
|
) -> dict | None:
|
|
if category not in COLUMN_ARTICLE_CATEGORIES:
|
|
return None
|
|
from app.services.storage import public_object_url
|
|
|
|
page_size = size if size in COLUMN_LIST_PAGE_SIZES else DEFAULT_COLUMN_PAGE_SIZE
|
|
page = current if current >= 1 else 1
|
|
placeholder = EMPTY_REGIONS[category]
|
|
filters = [HomeArticle.category == category, HomeArticle.published.is_(True)]
|
|
kind_filter = (kind or "").strip()
|
|
kind_sets = {
|
|
"appraisal_guides": APPRAISAL_GUIDE_KINDS,
|
|
"policy_analyses": POLICY_ANALYSIS_KINDS,
|
|
}
|
|
allowed_kinds = kind_sets.get(category)
|
|
if allowed_kinds is not None and kind_filter and kind_filter != "all":
|
|
if kind_filter not in allowed_kinds:
|
|
return None
|
|
filters.append(
|
|
HomeArticle.id.in_(
|
|
select(HomeArticleKind.article_id).where(HomeArticleKind.kind == kind_filter)
|
|
)
|
|
)
|
|
keyword = (q or "").strip()[:100]
|
|
if keyword and category == "policy_analyses":
|
|
needle = _like_contains(keyword)
|
|
filters.append(
|
|
or_(
|
|
HomeArticle.title.like(needle, escape="\\"),
|
|
HomeArticle.body_html.like(needle, escape="\\"),
|
|
)
|
|
)
|
|
total = db.scalar(select(func.count()).select_from(HomeArticle).where(*filters)) or 0
|
|
if total == 0:
|
|
if keyword and category == "policy_analyses":
|
|
placeholder = POLICY_SEARCH_EMPTY
|
|
return {"placeholder": placeholder, "items": [], "total": 0, "current": page, "size": page_size}
|
|
pages = max(1, (total + page_size - 1) // page_size)
|
|
if page > pages:
|
|
page = pages
|
|
rows = db.scalars(
|
|
select(HomeArticle)
|
|
.where(*filters)
|
|
.order_by(HomeArticle.published_at.desc(), HomeArticle.id.desc())
|
|
.offset((page - 1) * page_size)
|
|
.limit(page_size)
|
|
).all()
|
|
kinds = _article_kinds(db, [row.id for row in rows])
|
|
items = []
|
|
for row in rows:
|
|
item = {"id": row.id, "title": row.title}
|
|
image = _public_file(db, row.image_file_id) if row.image_file_id else None
|
|
if image is not None:
|
|
item["imageUrl"] = public_object_url(image.object_key)
|
|
article_kind = kinds.get(row.id)
|
|
if article_kind:
|
|
item["kind"] = article_kind
|
|
item["kindLabel"] = _kind_label(category, article_kind)
|
|
items.append(item)
|
|
return {"items": items, "total": total, "current": page, "size": page_size}
|
|
|
|
|
|
def admin_column_articles(db, category: str) -> dict | None:
|
|
if category not in COLUMN_ARTICLE_CATEGORIES:
|
|
return None
|
|
from app.services.storage import public_object_url
|
|
|
|
rows = db.scalars(
|
|
select(HomeArticle)
|
|
.where(HomeArticle.category == category)
|
|
.order_by(HomeArticle.published_at.desc(), HomeArticle.id.desc())
|
|
).all()
|
|
kinds = _article_kinds(db, [row.id for row in rows])
|
|
items = []
|
|
for row in rows:
|
|
image_id, image_url = _editor_image(db, row.image_file_id)
|
|
attachments = db.scalars(
|
|
select(HomeArticleAttachment)
|
|
.where(HomeArticleAttachment.article_id == row.id)
|
|
.order_by(HomeArticleAttachment.position.asc(), HomeArticleAttachment.id.asc())
|
|
).all()
|
|
files = _editor_attachments(db, attachments)
|
|
item = {
|
|
"id": row.id,
|
|
"category": row.category,
|
|
"title": row.title,
|
|
"publishedAt": row.published_at.isoformat(timespec="seconds") if row.published_at else "",
|
|
"published": bool(row.published),
|
|
"bodyHtml": row.body_html,
|
|
"imageFileId": image_id,
|
|
"imageUrl": image_url,
|
|
"attachmentFileIds": [item["id"] for item in files],
|
|
"attachments": files,
|
|
}
|
|
article_kind = kinds.get(row.id)
|
|
if article_kind:
|
|
item["kind"] = article_kind
|
|
item["kindLabel"] = _kind_label(category, article_kind)
|
|
items.append(item)
|
|
return {"category": category, "items": items}
|
|
|
|
|
|
def _kind_label(category: str, kind: str) -> str:
|
|
labels = {
|
|
"appraisal_guides": APPRAISAL_GUIDE_KIND_LABELS,
|
|
"policy_analyses": POLICY_ANALYSIS_KIND_LABELS,
|
|
}
|
|
return labels.get(category, {}).get(kind, kind)
|
|
|
|
|
|
def _article_kinds(db, article_ids: list[int]) -> dict[int, str]:
|
|
if not article_ids:
|
|
return {}
|
|
rows = db.scalars(select(HomeArticleKind).where(HomeArticleKind.article_id.in_(article_ids))).all()
|
|
return {row.article_id: row.kind for row in rows}
|
|
|
|
|
|
def _normalize_article_kind(category: str, payload: dict) -> str | None:
|
|
raw = payload.get("kind")
|
|
kind = "" if raw is None else str(raw).strip()
|
|
if category == "appraisal_guides":
|
|
if kind not in APPRAISAL_GUIDE_KINDS:
|
|
raise ValueError("估价指南分类不正确")
|
|
return kind
|
|
if category == "policy_analyses":
|
|
if kind not in POLICY_ANALYSIS_KINDS:
|
|
raise ValueError("政策分类不正确")
|
|
return kind
|
|
if raw not in (None, ""):
|
|
raise ValueError("该类文章不支持分类")
|
|
return None
|
|
|
|
|
|
def _set_article_kind(db, article_id: int, kind: str | None) -> None:
|
|
existing = db.get(HomeArticleKind, article_id)
|
|
if kind is None:
|
|
if existing is not None:
|
|
db.delete(existing)
|
|
return
|
|
if existing is None:
|
|
db.add(HomeArticleKind(article_id=article_id, kind=kind))
|
|
return
|
|
existing.kind = kind
|
|
|
|
|
|
def _prepare_article(db, payload: dict) -> tuple[dict, list, str | None]:
|
|
from app.core.config import get_settings
|
|
from app.models.content import AdminFile
|
|
from app.services.html_filter import filter_html
|
|
|
|
category = payload.get("category")
|
|
if category not in ARTICLE_CATEGORIES:
|
|
raise ValueError("文章类别不正确")
|
|
kind = _normalize_article_kind(category, payload)
|
|
image_id = payload.get("imageFileId")
|
|
image = None
|
|
if image_id is not None:
|
|
image = _public_file(db, image_id)
|
|
if image is None:
|
|
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 = {
|
|
"category": category,
|
|
"title": "" if payload.get("title") is None else str(payload["title"]),
|
|
"body_html": cleaned,
|
|
"image_file_id": image.id if image else None,
|
|
"published": bool(payload.get("published", False)),
|
|
"published_at": _published_at(payload.get("publishedAt")),
|
|
}
|
|
return fields, attachments, kind
|
|
|
|
|
|
def _replace_article_attachments(db, article_id: int, attachments: list) -> None:
|
|
links = db.scalars(
|
|
select(HomeArticleAttachment).where(HomeArticleAttachment.article_id == article_id)
|
|
).all()
|
|
for link in links:
|
|
db.delete(link)
|
|
db.flush()
|
|
for index, row in enumerate(attachments):
|
|
db.add(HomeArticleAttachment(article_id=article_id, file_id=row.id, position=index))
|
|
|
|
|
|
def create_article(db, payload: dict) -> int:
|
|
fields, attachments, kind = _prepare_article(db, payload)
|
|
article = HomeArticle(**fields)
|
|
db.add(article)
|
|
db.flush()
|
|
_replace_article_attachments(db, article.id, attachments)
|
|
_set_article_kind(db, article.id, kind)
|
|
db.commit()
|
|
db.refresh(article)
|
|
return article.id
|
|
|
|
|
|
def update_article(db, article_id: int, payload: dict) -> None:
|
|
article = db.get(HomeArticle, article_id)
|
|
if article is None:
|
|
raise ValueError("文章不存在")
|
|
fields, attachments, kind = _prepare_article(db, payload)
|
|
for key, value in fields.items():
|
|
setattr(article, key, value)
|
|
_replace_article_attachments(db, article.id, attachments)
|
|
_set_article_kind(db, article.id, kind)
|
|
db.commit()
|
|
|
|
|
|
def public_article(db, article_id: int) -> dict | None:
|
|
from app.services.storage import public_object_url
|
|
|
|
row = db.get(HomeArticle, article_id)
|
|
if row is None or not row.published:
|
|
return None
|
|
image = _public_file(db, row.image_file_id) if row.image_file_id else None
|
|
links = db.scalars(
|
|
select(HomeArticleAttachment)
|
|
.where(HomeArticleAttachment.article_id == row.id)
|
|
.order_by(HomeArticleAttachment.position.asc(), HomeArticleAttachment.id.asc())
|
|
).all()
|
|
attachments = []
|
|
for link in links:
|
|
file_row = _public_file(db, link.file_id)
|
|
if file_row is None:
|
|
continue
|
|
attachments.append(
|
|
{"filename": file_row.filename, "url": public_object_url(file_row.object_key)}
|
|
)
|
|
body = {
|
|
"id": row.id,
|
|
"category": row.category,
|
|
"title": row.title,
|
|
"bodyHtml": row.body_html,
|
|
"attachments": attachments,
|
|
}
|
|
if image is not None:
|
|
body["imageUrl"] = public_object_url(image.object_key)
|
|
kinds = _article_kinds(db, [row.id])
|
|
article_kind = kinds.get(row.id)
|
|
if article_kind:
|
|
body["kind"] = article_kind
|
|
body["kindLabel"] = APPRAISAL_GUIDE_KIND_LABELS.get(article_kind, article_kind)
|
|
return body
|
|
|
|
|
|
def delete_article(db, article_id: int) -> None:
|
|
row = db.get(HomeArticle, article_id)
|
|
if row is None:
|
|
raise ValueError("文章不存在")
|
|
links = db.scalars(
|
|
select(HomeArticleAttachment).where(HomeArticleAttachment.article_id == row.id)
|
|
).all()
|
|
for link in links:
|
|
db.delete(link)
|
|
kind_row = db.get(HomeArticleKind, row.id)
|
|
if kind_row is not None:
|
|
db.delete(kind_row)
|
|
db.delete(row)
|
|
db.commit()
|
|
|
|
|
|
def _link_label(link: dict) -> str:
|
|
label = link.get("label")
|
|
if not isinstance(label, str) or not label.strip():
|
|
raise ValueError("链接文字不能为空")
|
|
return label.strip()
|
|
|
|
|
|
def _link_url(value) -> str:
|
|
from app.services.html_filter import is_javascript_url
|
|
|
|
if value is None:
|
|
return ""
|
|
if not isinstance(value, str):
|
|
raise ValueError("链接地址不正确")
|
|
text = value.strip()
|
|
if is_javascript_url(text):
|
|
raise ValueError("链接地址不正确")
|
|
return text
|
|
|
|
|
|
def _sort_order(link: dict) -> int:
|
|
value = link.get("sortOrder", 0)
|
|
if isinstance(value, bool) or not isinstance(value, int):
|
|
raise ValueError("顺序必须是整数")
|
|
return value
|
|
|
|
|
|
def _navigation(url: str) -> str:
|
|
if not url:
|
|
return "none"
|
|
if url.startswith("/") and not url.startswith("//"):
|
|
return "sameWindow"
|
|
return "newWindow"
|
|
|
|
|
|
def _public_link(row) -> dict:
|
|
return {"label": row.label, "url": row.url, "navigation": _navigation(row.url)}
|
|
|
|
|
|
def _public_audiences(db) -> dict:
|
|
rows = db.scalars(
|
|
select(HomeAudienceLink)
|
|
.where(HomeAudienceLink.visible.is_(True))
|
|
.order_by(HomeAudienceLink.sort_order.asc(), HomeAudienceLink.id.asc())
|
|
).all()
|
|
if not rows:
|
|
return {"placeholder": EMPTY_REGIONS["audiences"]}
|
|
grouped = {name: [] for name in AUDIENCE_NAMES}
|
|
for row in rows:
|
|
if row.category in grouped:
|
|
grouped[row.category].append(_public_link(row))
|
|
return {
|
|
"categories": [{"name": name, "links": grouped[name]} for name in AUDIENCE_NAMES]
|
|
}
|
|
|
|
|
|
def _public_hot_link(row: HomeHotLink) -> dict:
|
|
url = FIXED_HOT_URLS.get(row.key, row.url)
|
|
return {"label": row.label, "url": url, "navigation": _navigation(url)}
|
|
|
|
|
|
def _public_hot_links(db) -> dict:
|
|
rows = db.scalars(
|
|
select(HomeHotLink)
|
|
.where(HomeHotLink.visible.is_(True))
|
|
.order_by(HomeHotLink.sort_order.asc(), HomeHotLink.id.asc())
|
|
).all()
|
|
if not rows:
|
|
return {"placeholder": EMPTY_REGIONS["hotLinks"]}
|
|
return {"links": [_public_hot_link(row) for row in rows]}
|
|
|
|
|
|
def _shortcut_menu(db) -> dict:
|
|
rows = db.scalars(
|
|
select(HomeMenuItem)
|
|
.where(HomeMenuItem.visible.is_(True), HomeMenuItem.key.in_(MENU_KEYS))
|
|
.order_by(HomeMenuItem.sort_order.asc(), HomeMenuItem.id.asc())
|
|
).all()
|
|
items = [{"label": "首页", "url": "/", "navigation": "sameWindow"}]
|
|
items.extend(_public_link(row) for row in rows)
|
|
return {"items": items}
|
|
|
|
|
|
def save_audiences(db, categories: list[dict]) -> None:
|
|
names = [item.get("name") for item in categories]
|
|
if names != list(AUDIENCE_NAMES):
|
|
raise ValueError("服务对象类别必须是估价师、估价机构、公众")
|
|
prepared: list[HomeAudienceLink] = []
|
|
for category in categories:
|
|
for link in category.get("links") or []:
|
|
prepared.append(
|
|
HomeAudienceLink(
|
|
category=category["name"],
|
|
label=_link_label(link),
|
|
url=_link_url(link.get("url")),
|
|
visible=bool(link.get("visible", True)),
|
|
sort_order=_sort_order(link),
|
|
)
|
|
)
|
|
db.execute(delete(HomeAudienceLink))
|
|
db.add_all(prepared)
|
|
db.commit()
|
|
|
|
|
|
def save_hot_links(db, links: list[dict]) -> None:
|
|
if len(links) != len(HOT_LINK_SLOTS):
|
|
raise ValueError("高频功能必须是报告查询、报告审核和投诉举报")
|
|
seen: set[str] = set()
|
|
prepared: list[HomeHotLink] = []
|
|
for link in links:
|
|
key = link.get("key")
|
|
if key not in HOT_LINK_KEYS or key in seen:
|
|
raise ValueError("高频功能必须是报告查询、报告审核和投诉举报")
|
|
seen.add(key)
|
|
url = FIXED_HOT_URLS.get(key)
|
|
if url is None:
|
|
url = _link_url(link.get("url"))
|
|
prepared.append(
|
|
HomeHotLink(
|
|
key=key,
|
|
label=_link_label(link),
|
|
url=url,
|
|
visible=bool(link.get("visible", True)),
|
|
sort_order=_sort_order(link),
|
|
)
|
|
)
|
|
db.execute(delete(HomeHotLink))
|
|
db.add_all(prepared)
|
|
db.commit()
|
|
|
|
|
|
def ensure_hot_link_slots(db) -> None:
|
|
bind = db.get_bind()
|
|
if bind.dialect.name == "mysql":
|
|
from sqlalchemy import inspect
|
|
|
|
columns = {column["name"] for column in inspect(bind).get_columns("home_hot_link")}
|
|
if "key" not in columns:
|
|
db.execute(text("ALTER TABLE home_hot_link ADD COLUMN `key` VARCHAR(32) NOT NULL DEFAULT ''"))
|
|
db.commit()
|
|
label_to_key = {label: key for key, label, _url in HOT_LINK_SLOTS}
|
|
rows = db.scalars(select(HomeHotLink).order_by(HomeHotLink.sort_order.asc(), HomeHotLink.id.asc())).all()
|
|
used: set[str] = set()
|
|
for row in rows:
|
|
if row.key in HOT_LINK_KEYS:
|
|
if row.key in used:
|
|
db.delete(row)
|
|
continue
|
|
used.add(row.key)
|
|
if row.key in FIXED_HOT_URLS:
|
|
row.url = FIXED_HOT_URLS[row.key]
|
|
continue
|
|
mapped = label_to_key.get(row.label)
|
|
if mapped and mapped not in used:
|
|
row.key = mapped
|
|
if mapped in FIXED_HOT_URLS:
|
|
row.url = FIXED_HOT_URLS[mapped]
|
|
used.add(mapped)
|
|
continue
|
|
db.delete(row)
|
|
db.commit()
|
|
if bind.dialect.name != "mysql":
|
|
return
|
|
from sqlalchemy import inspect
|
|
|
|
indexes = inspect(bind).get_indexes("home_hot_link")
|
|
has_unique = any(item.get("unique") and "key" in (item.get("column_names") or []) for item in indexes)
|
|
if not has_unique:
|
|
db.execute(text("ALTER TABLE home_hot_link ADD UNIQUE INDEX uq_home_hot_link_key (`key`)"))
|
|
db.commit()
|
|
|
|
|
|
def save_menu(db, items: list[dict]) -> None:
|
|
seen: set[str] = set()
|
|
for item in items:
|
|
key = item.get("key")
|
|
if key not in MENU_KEYS or key in seen:
|
|
raise ValueError("快捷菜单只能配置会员服务")
|
|
seen.add(key)
|
|
row = db.scalar(select(HomeMenuItem).where(HomeMenuItem.key == key))
|
|
if row is None:
|
|
row = HomeMenuItem(key=key)
|
|
db.add(row)
|
|
row.label = _link_label(item)
|
|
row.url = _link_url(item.get("url"))
|
|
row.visible = bool(item.get("visible", False))
|
|
row.sort_order = _sort_order(item)
|
|
db.commit()
|
|
|
|
|
|
def retire_removed_shortcut_keys(db) -> None:
|
|
db.execute(delete(HomeMenuItem).where(HomeMenuItem.key.notin_(MENU_KEYS)))
|
|
db.commit()
|
|
|
|
|
|
def _interval(value) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
|
|
raise ValueError("间隔秒数必须是正整数")
|
|
return value
|
|
|
|
|
|
def _image_id(db, file_id, message: str) -> int:
|
|
image = _public_file(db, file_id)
|
|
if image is None:
|
|
raise ValueError(message)
|
|
return image.id
|
|
|
|
|
|
def _store_interval(db, kind: str, seconds: int) -> None:
|
|
row = db.get(HomeMarquee, kind)
|
|
if row is None:
|
|
db.add(HomeMarquee(kind=kind, interval_seconds=seconds))
|
|
else:
|
|
row.interval_seconds = seconds
|
|
|
|
|
|
def _public_honors(db) -> dict:
|
|
from app.services.storage import public_object_url
|
|
|
|
rows = db.scalars(select(HomeHonor).order_by(HomeHonor.sort_order.asc(), HomeHonor.id.asc())).all()
|
|
items = []
|
|
for row in rows:
|
|
image = _public_file(db, row.image_file_id)
|
|
if image is None:
|
|
continue
|
|
items.append({"imageUrl": public_object_url(image.object_key)})
|
|
if not items:
|
|
return {"placeholder": EMPTY_REGIONS["honors"]}
|
|
setting = db.get(HomeMarquee, "honors")
|
|
return {"intervalSeconds": setting.interval_seconds if setting else 0, "items": items}
|
|
|
|
|
|
def _public_members(db) -> dict:
|
|
from app.services.storage import public_object_url
|
|
|
|
rows = db.scalars(
|
|
select(HomeMember).order_by(HomeMember.sort_order.asc(), HomeMember.id.asc())
|
|
).all()
|
|
items = []
|
|
for row in rows:
|
|
image = _public_file(db, row.image_file_id)
|
|
if image is None:
|
|
continue
|
|
items.append(
|
|
{
|
|
"name": row.name,
|
|
"url": row.url,
|
|
"navigation": _navigation(row.url),
|
|
"imageUrl": public_object_url(image.object_key),
|
|
}
|
|
)
|
|
if not items:
|
|
return {"placeholder": EMPTY_REGIONS["members"]}
|
|
setting = db.get(HomeMarquee, "members")
|
|
return {"intervalSeconds": setting.interval_seconds if setting else 0, "items": items}
|
|
|
|
|
|
def save_honors(db, interval_seconds: int, items: list[dict]) -> None:
|
|
seconds = _interval(interval_seconds)
|
|
prepared = [
|
|
HomeHonor(
|
|
image_file_id=_image_id(db, item.get("imageFileId"), "荣誉图片必须是已上传的公开图片"),
|
|
sort_order=_sort_order(item),
|
|
)
|
|
for item in items
|
|
]
|
|
db.execute(delete(HomeHonor))
|
|
db.add_all(prepared)
|
|
_store_interval(db, "honors", seconds)
|
|
db.commit()
|
|
|
|
|
|
def save_members(db, interval_seconds: int, items: list[dict]) -> None:
|
|
seconds = _interval(interval_seconds)
|
|
prepared = []
|
|
for item in items:
|
|
name = item.get("name")
|
|
if not isinstance(name, str) or not name.strip():
|
|
raise ValueError("会员单位名称不能为空")
|
|
prepared.append(
|
|
HomeMember(
|
|
name=name.strip(),
|
|
url=_link_url(item.get("url")),
|
|
image_file_id=_image_id(db, item.get("imageFileId"), "会员标志必须是已上传的公开图片"),
|
|
sort_order=_sort_order(item),
|
|
)
|
|
)
|
|
db.execute(delete(HomeMember))
|
|
db.add_all(prepared)
|
|
_store_interval(db, "members", seconds)
|
|
db.commit()
|
|
|
|
|
|
def _edu_site_name(item: dict) -> str:
|
|
name = item.get("name")
|
|
if not isinstance(name, str) or not name.strip():
|
|
raise ValueError("教育网站名称不能为空")
|
|
return name.strip()
|
|
|
|
|
|
def _edu_site_url(value) -> str:
|
|
text = _link_url(value)
|
|
if not text:
|
|
raise ValueError("教育网站地址不能为空")
|
|
if not text.lower().startswith(("http://", "https://")):
|
|
raise ValueError("教育网站地址不正确")
|
|
return text
|
|
|
|
|
|
def public_edu_platforms(db) -> dict:
|
|
from app.services.storage import public_object_url
|
|
|
|
rows = db.scalars(
|
|
select(EduPlatform).order_by(EduPlatform.sort_order.asc(), EduPlatform.id.asc())
|
|
).all()
|
|
items = []
|
|
for row in rows:
|
|
image = _public_file(db, row.image_file_id)
|
|
if image is None:
|
|
continue
|
|
items.append(
|
|
{
|
|
"name": row.name,
|
|
"url": row.url,
|
|
"navigation": "newWindow",
|
|
"imageUrl": public_object_url(image.object_key),
|
|
}
|
|
)
|
|
if not items:
|
|
return {"placeholder": EDU_PLATFORM_EMPTY}
|
|
return {"items": items}
|
|
|
|
|
|
def admin_edu_platforms(db) -> dict:
|
|
rows = db.scalars(
|
|
select(EduPlatform).order_by(EduPlatform.sort_order.asc(), EduPlatform.id.asc())
|
|
).all()
|
|
items = []
|
|
for row in rows:
|
|
image_id, image_url = _editor_image(db, row.image_file_id)
|
|
items.append(
|
|
{
|
|
"name": row.name,
|
|
"url": row.url,
|
|
"imageFileId": image_id,
|
|
"imageUrl": image_url,
|
|
"sortOrder": row.sort_order,
|
|
}
|
|
)
|
|
return {"items": items}
|
|
|
|
|
|
def save_edu_platforms(db, items: list[dict]) -> None:
|
|
prepared = [
|
|
EduPlatform(
|
|
name=_edu_site_name(item),
|
|
url=_edu_site_url(item.get("url")),
|
|
image_file_id=_image_id(db, item.get("imageFileId"), "教育网站图片必须是已上传的公开图片"),
|
|
sort_order=_sort_order(item),
|
|
)
|
|
for item in items
|
|
]
|
|
db.execute(delete(EduPlatform))
|
|
db.add_all(prepared)
|
|
db.commit()
|
|
|
|
|
|
def _coordinate(value, low: float, high: float, message: str) -> float | None:
|
|
if value is None or value == "":
|
|
return None
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise ValueError(message)
|
|
number = float(value)
|
|
if number < low or number > high:
|
|
raise ValueError(message)
|
|
return number
|
|
|
|
|
|
def _coordinate_pair(latitude, longitude) -> tuple[float | None, float | None]:
|
|
lat = _coordinate(latitude, -90, 90, "纬度不正确")
|
|
lng = _coordinate(longitude, -180, 180, "经度不正确")
|
|
if (lat is None) != (lng is None):
|
|
raise ValueError("纬度和经度需要同时填写")
|
|
return lat, lng
|
|
|
|
|
|
def staff_map_picker() -> dict | None:
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
key = settings.gaode_key.strip()
|
|
secret = settings.gaode_security_js_code.strip()
|
|
if not key or not secret:
|
|
return None
|
|
return {"key": key, "securityJsCode": secret}
|
|
|
|
|
|
def _public_map(row) -> str | dict:
|
|
from app.core.config import get_settings
|
|
|
|
settings = get_settings()
|
|
key = settings.gaode_key.strip()
|
|
secret = settings.gaode_security_js_code.strip()
|
|
if row is None or row.latitude is None or row.longitude is None or not key or not secret:
|
|
return "地图位置整理中"
|
|
return {
|
|
"latitude": float(row.latitude),
|
|
"longitude": float(row.longitude),
|
|
"key": key,
|
|
"securityJsCode": secret,
|
|
}
|
|
|
|
|
|
def _public_contact(db, template_name: str) -> dict:
|
|
row = db.get(HomeContact, template_name)
|
|
return {
|
|
"address": _text(row.address if row else "", "办公地址整理中"),
|
|
"phone": _text(row.phone if row else "", "联系电话整理中"),
|
|
"email": _text(row.email if row else "", "联系邮箱整理中"),
|
|
"map": _public_map(row),
|
|
}
|
|
|
|
|
|
def _visible_links(db, model, template_name: str, category: str | None = None) -> list:
|
|
stmt = select(model).where(model.template_name == template_name, model.visible.is_(True))
|
|
if category is not None:
|
|
stmt = stmt.where(model.category == category)
|
|
stmt = stmt.order_by(model.sort_order.asc(), model.id.asc())
|
|
return list(db.scalars(stmt).all())
|
|
|
|
|
|
def _public_related(db, template_name: str) -> dict:
|
|
result = {}
|
|
for category in RELATED_CATEGORIES:
|
|
rows = _visible_links(db, HomeRelatedLink, template_name, category)
|
|
if not rows:
|
|
result[category] = {"placeholder": f"{category}链接整理中"}
|
|
else:
|
|
result[category] = {"links": [_public_link(row) for row in rows]}
|
|
return result
|
|
|
|
|
|
def _public_compliance(db, template_name: str) -> dict:
|
|
# 页脚固定三项，正文由 legal_page 管理；不再使用外链列表占位。
|
|
del db, template_name
|
|
from app.models.legal import public_compliance_links
|
|
|
|
return public_compliance_links()
|
|
|
|
|
|
def _template_names(items: list[dict]) -> None:
|
|
names = [item.get("name") for item in items]
|
|
if sorted(names) != sorted(TEMPLATE_NAMES):
|
|
raise ValueError("模板必须是科技蓝和LOGO蓝")
|
|
|
|
|
|
def save_contact(db, templates: list[dict]) -> None:
|
|
_template_names(templates)
|
|
for item in templates:
|
|
latitude, longitude = _coordinate_pair(item.get("latitude"), item.get("longitude"))
|
|
row = db.get(HomeContact, item["name"])
|
|
if row is None:
|
|
row = HomeContact(template_name=item["name"])
|
|
db.add(row)
|
|
row.address = "" if item.get("address") is None else str(item.get("address"))
|
|
row.phone = "" if item.get("phone") is None else str(item.get("phone"))
|
|
row.email = "" if item.get("email") is None else str(item.get("email"))
|
|
row.latitude = latitude
|
|
row.longitude = longitude
|
|
db.commit()
|
|
|
|
|
|
def _prepared_links(db, model, template_name: str, links: list[dict], category: str | None = None):
|
|
prepared = []
|
|
for link in links:
|
|
values = {
|
|
"template_name": template_name,
|
|
"label": _link_label(link),
|
|
"url": _link_url(link.get("url")),
|
|
"visible": bool(link.get("visible", True)),
|
|
"sort_order": _sort_order(link),
|
|
}
|
|
if category is not None:
|
|
values["category"] = category
|
|
prepared.append(model(**values))
|
|
return prepared
|
|
|
|
|
|
def save_related_links(db, template_name: str, categories: list[dict]) -> None:
|
|
if template_name not in TEMPLATE_NAMES:
|
|
raise ValueError("模板必须是科技蓝和LOGO蓝")
|
|
names = [item.get("name") for item in categories]
|
|
if names != list(RELATED_CATEGORIES):
|
|
raise ValueError("相关链接类别必须是主管部门、相关部门、合作协会")
|
|
prepared = []
|
|
for category in categories:
|
|
prepared.extend(
|
|
_prepared_links(db, HomeRelatedLink, template_name, category.get("links") or [], category["name"])
|
|
)
|
|
db.execute(delete(HomeRelatedLink).where(HomeRelatedLink.template_name == template_name))
|
|
db.add_all(prepared)
|
|
db.commit()
|
|
|
|
|
|
def save_compliance(db, template_name: str, links: list[dict]) -> None:
|
|
if template_name not in TEMPLATE_NAMES:
|
|
raise ValueError("模板必须是科技蓝和LOGO蓝")
|
|
prepared = _prepared_links(db, HomeComplianceLink, template_name, links)
|
|
db.execute(delete(HomeComplianceLink).where(HomeComplianceLink.template_name == template_name))
|
|
db.add_all(prepared)
|
|
db.commit()
|
|
|
|
|
|
def save_home(db, active_name: str, templates: list[dict]) -> None:
|
|
ensure_home_templates(db)
|
|
names = [item["name"] for item in templates]
|
|
if active_name not in TEMPLATE_NAMES or sorted(names) != sorted(TEMPLATE_NAMES):
|
|
raise ValueError("模板必须是科技蓝和LOGO蓝")
|
|
by_name = {item["name"]: item for item in templates}
|
|
rows = db.scalars(select(HomeTemplate)).all()
|
|
found = {row.name: row for row in rows}
|
|
for name in TEMPLATE_NAMES:
|
|
row = found[name]
|
|
incoming = by_name[name]
|
|
row.slogan = "" if incoming["slogan"] is None else str(incoming["slogan"])
|
|
row.value_proposition = (
|
|
"" if incoming["valueProposition"] is None else str(incoming["valueProposition"])
|
|
)
|
|
row.is_active = name == active_name
|
|
db.commit()
|
|
|
|
|
|
def _editor_attachments(db, links) -> list[dict]:
|
|
from app.services.storage import public_object_url
|
|
|
|
items = []
|
|
for link in links:
|
|
row = _public_file(db, link.file_id)
|
|
if row is None:
|
|
continue
|
|
items.append(
|
|
{
|
|
"id": row.id,
|
|
"filename": row.filename,
|
|
"url": public_object_url(row.object_key),
|
|
}
|
|
)
|
|
return items
|
|
|
|
|
|
def _editor_image(db, file_id: int | None) -> tuple[int | None, str]:
|
|
from app.services.storage import public_object_url
|
|
|
|
row = _public_file(db, file_id) if file_id else None
|
|
if row is None:
|
|
return None, ""
|
|
return row.id, public_object_url(row.object_key)
|
|
|
|
|
|
def _link_editor(row) -> dict:
|
|
return {
|
|
"label": row.label,
|
|
"url": row.url,
|
|
"visible": bool(row.visible),
|
|
"sortOrder": row.sort_order,
|
|
}
|
|
|
|
|
|
def _hot_link_editor(rows: list[HomeHotLink]) -> list[dict]:
|
|
present: list[dict] = []
|
|
seen: set[str] = set()
|
|
for row in rows:
|
|
if row.key not in HOT_LINK_KEYS or row.key in seen:
|
|
continue
|
|
seen.add(row.key)
|
|
present.append(
|
|
{
|
|
"key": row.key,
|
|
"label": row.label,
|
|
"url": FIXED_HOT_URLS.get(row.key, row.url),
|
|
"visible": bool(row.visible),
|
|
"sortOrder": row.sort_order,
|
|
"urlLocked": row.key in FIXED_HOT_URLS,
|
|
}
|
|
)
|
|
for index, (key, label, url) in enumerate(HOT_LINK_SLOTS):
|
|
if key in seen:
|
|
continue
|
|
present.append(
|
|
{
|
|
"key": key,
|
|
"label": label,
|
|
"url": url,
|
|
"visible": False,
|
|
"sortOrder": index,
|
|
"urlLocked": key in FIXED_HOT_URLS,
|
|
}
|
|
)
|
|
return present
|
|
|
|
|
|
def admin_home_view(db) -> dict:
|
|
ensure_home_templates(db)
|
|
templates = db.scalars(select(HomeTemplate).order_by(HomeTemplate.id.asc())).all()
|
|
active = next((row.name for row in templates if row.is_active), TECH)
|
|
slides = []
|
|
for row in db.scalars(select(HomeSlide).order_by(HomeSlide.sort_order.asc(), HomeSlide.id.asc())).all():
|
|
image_id, image_url = _editor_image(db, row.image_file_id)
|
|
attachments = db.scalars(
|
|
select(HomeSlideAttachment)
|
|
.where(HomeSlideAttachment.slide_id == row.id)
|
|
.order_by(HomeSlideAttachment.position.asc(), HomeSlideAttachment.id.asc())
|
|
).all()
|
|
files = _editor_attachments(db, attachments)
|
|
slides.append(
|
|
{
|
|
"id": row.id,
|
|
"title": row.title,
|
|
"imageFileId": image_id,
|
|
"imageUrl": image_url,
|
|
"sortOrder": row.sort_order,
|
|
"enabled": bool(row.enabled),
|
|
"intervalSeconds": row.interval_seconds,
|
|
"published": bool(row.published),
|
|
"bodyHtml": row.body_html,
|
|
"attachmentFileIds": [item["id"] for item in files],
|
|
"attachments": files,
|
|
}
|
|
)
|
|
articles = []
|
|
for row in db.scalars(
|
|
select(HomeArticle)
|
|
.where(HomeArticle.category.in_(HOME_ARTICLE_CATEGORIES))
|
|
.order_by(HomeArticle.published_at.desc(), HomeArticle.id.desc())
|
|
).all():
|
|
image_id, image_url = _editor_image(db, row.image_file_id)
|
|
attachments = db.scalars(
|
|
select(HomeArticleAttachment)
|
|
.where(HomeArticleAttachment.article_id == row.id)
|
|
.order_by(HomeArticleAttachment.position.asc(), HomeArticleAttachment.id.asc())
|
|
).all()
|
|
files = _editor_attachments(db, attachments)
|
|
articles.append(
|
|
{
|
|
"id": row.id,
|
|
"category": row.category,
|
|
"title": row.title,
|
|
"publishedAt": row.published_at.isoformat(timespec="seconds") if row.published_at else "",
|
|
"published": bool(row.published),
|
|
"bodyHtml": row.body_html,
|
|
"imageFileId": image_id,
|
|
"imageUrl": image_url,
|
|
"attachmentFileIds": [item["id"] for item in files],
|
|
"attachments": files,
|
|
}
|
|
)
|
|
audience_rows = db.scalars(
|
|
select(HomeAudienceLink).order_by(HomeAudienceLink.sort_order.asc(), HomeAudienceLink.id.asc())
|
|
).all()
|
|
grouped = {name: [] for name in AUDIENCE_NAMES}
|
|
for row in audience_rows:
|
|
if row.category in grouped:
|
|
grouped[row.category].append(_link_editor(row))
|
|
hot_rows = db.scalars(select(HomeHotLink).order_by(HomeHotLink.sort_order.asc(), HomeHotLink.id.asc())).all()
|
|
menu_rows = db.scalars(select(HomeMenuItem).order_by(HomeMenuItem.sort_order.asc(), HomeMenuItem.id.asc())).all()
|
|
honors = db.scalars(select(HomeHonor).order_by(HomeHonor.sort_order.asc(), HomeHonor.id.asc())).all()
|
|
members = db.scalars(select(HomeMember).order_by(HomeMember.sort_order.asc(), HomeMember.id.asc())).all()
|
|
honor_setting = db.get(HomeMarquee, "honors")
|
|
member_setting = db.get(HomeMarquee, "members")
|
|
contacts = []
|
|
for name in TEMPLATE_NAMES:
|
|
row = db.get(HomeContact, name)
|
|
contacts.append(
|
|
{
|
|
"name": name,
|
|
"address": row.address if row else "",
|
|
"phone": row.phone if row else "",
|
|
"email": row.email if row else "",
|
|
"latitude": float(row.latitude) if row and row.latitude is not None else None,
|
|
"longitude": float(row.longitude) if row and row.longitude is not None else None,
|
|
}
|
|
)
|
|
related = []
|
|
compliance = []
|
|
for name in TEMPLATE_NAMES:
|
|
categories = []
|
|
for category in RELATED_CATEGORIES:
|
|
rows = db.scalars(
|
|
select(HomeRelatedLink)
|
|
.where(HomeRelatedLink.template_name == name, HomeRelatedLink.category == category)
|
|
.order_by(HomeRelatedLink.sort_order.asc(), HomeRelatedLink.id.asc())
|
|
).all()
|
|
categories.append({"name": category, "links": [_link_editor(item) for item in rows]})
|
|
related.append({"templateName": name, "categories": categories})
|
|
rows = db.scalars(
|
|
select(HomeComplianceLink)
|
|
.where(HomeComplianceLink.template_name == name)
|
|
.order_by(HomeComplianceLink.sort_order.asc(), HomeComplianceLink.id.asc())
|
|
).all()
|
|
compliance.append({"templateName": name, "links": [_link_editor(item) for item in rows]})
|
|
return {
|
|
"activeTemplate": active,
|
|
"templates": [
|
|
{
|
|
"name": row.name,
|
|
"slogan": row.slogan,
|
|
"valueProposition": row.value_proposition,
|
|
}
|
|
for row in templates
|
|
if row.name in TEMPLATE_NAMES
|
|
],
|
|
"slides": slides,
|
|
"articleDisplay": {key: _display_count(db, key) for key in HOME_ARTICLE_CATEGORIES},
|
|
"articles": articles,
|
|
"audiences": {
|
|
"categories": [{"name": name, "links": grouped[name]} for name in AUDIENCE_NAMES]
|
|
},
|
|
"hotLinks": {"links": _hot_link_editor(hot_rows)},
|
|
"menu": {
|
|
"items": [
|
|
{
|
|
"key": row.key,
|
|
"label": row.label,
|
|
"url": row.url,
|
|
"visible": bool(row.visible),
|
|
"sortOrder": row.sort_order,
|
|
}
|
|
for row in menu_rows
|
|
if row.key in MENU_KEYS
|
|
]
|
|
},
|
|
"honors": {
|
|
"intervalSeconds": honor_setting.interval_seconds if honor_setting else 8,
|
|
"items": [
|
|
{
|
|
"imageFileId": _editor_image(db, row.image_file_id)[0],
|
|
"imageUrl": _editor_image(db, row.image_file_id)[1],
|
|
"sortOrder": row.sort_order,
|
|
}
|
|
for row in honors
|
|
],
|
|
},
|
|
"members": {
|
|
"intervalSeconds": member_setting.interval_seconds if member_setting else 8,
|
|
"items": [
|
|
{
|
|
"name": row.name,
|
|
"url": row.url,
|
|
"imageFileId": _editor_image(db, row.image_file_id)[0],
|
|
"imageUrl": _editor_image(db, row.image_file_id)[1],
|
|
"sortOrder": row.sort_order,
|
|
}
|
|
for row in members
|
|
],
|
|
},
|
|
"contacts": contacts,
|
|
"relatedLinks": related,
|
|
"compliance": compliance,
|
|
"mapPicker": staff_map_picker(),
|
|
}
|