Files
gsfdc/server/app/commands/seed_tech_exchange_sample.py

204 lines
7.1 KiB
Python

"""把现站相关公开摘录写入技术研讨 / 技术交流(模块),供演示库非空。
数据来自 docs/参考/现站。重复执行会清掉 tech_seminars / tech_exchange 后重写。
口令只从环境变量读取。
"""
from __future__ import annotations
from html import escape, unescape
from pathlib import Path
from sqlalchemy import delete, select
from app.db import Base, SessionLocal, get_engine
from app.models.content import AdminFile
from app.models.home import HomeArticle, HomeArticleAttachment, create_article
from app.services.storage import ensure_buckets, make_object_key, upload_object
ROOT = Path(__file__).resolve().parents[3]
SOURCE = ROOT / "docs" / "参考" / "现站" / "基础数据"
ABOUT = SOURCE / "协会介绍"
NEWS = SOURCE / "新闻动态"
CAROUSEL = SOURCE / "首页轮播"
TECH_CATEGORIES = ("tech_seminars", "tech_exchange")
def _paragraphs(path: Path) -> str:
text = unescape(path.read_text(encoding="utf-8"))
parts = []
for line in text.splitlines():
line = line.strip()
if not line:
continue
parts.append(f"<p>{escape(line)}</p>")
return "".join(parts)
def _content_type(path: Path) -> str:
suffix = path.suffix.lower()
if suffix == ".png":
return "image/png"
if suffix == ".webp":
return "image/webp"
if suffix == ".gif":
return "image/gif"
return "image/jpeg"
def _store_public(db, path: Path, filename: str | None = None) -> AdminFile | None:
if not path.is_file():
return None
name = filename or f"tech-sample-{path.name}"
row = db.scalar(
select(AdminFile).where(AdminFile.bucket == "public", AdminFile.filename == name)
)
if row is not None:
return row
data = path.read_bytes()
key = make_object_key(name)
upload_object("public", key, data, _content_type(path))
row = AdminFile(
bucket="public",
object_key=key,
filename=name,
content_type=_content_type(path),
size=len(data),
created_by=None,
)
db.add(row)
db.commit()
db.refresh(row)
return row
def _clear_tech(db) -> None:
article_ids = list(
db.scalars(select(HomeArticle.id).where(HomeArticle.category.in_(TECH_CATEGORIES))).all()
)
if article_ids:
db.execute(
delete(HomeArticleAttachment).where(HomeArticleAttachment.article_id.in_(article_ids))
)
db.execute(delete(HomeArticle).where(HomeArticle.id.in_(article_ids)))
db.commit()
def seed() -> None:
ensure_buckets()
engine = get_engine()
Base.metadata.create_all(bind=engine)
SessionLocal.configure(bind=engine)
db = SessionLocal()
try:
_clear_tech(db)
cover_a = _store_public(db, CAROUSEL / "58612080.jpg", "tech-cover-seminar.jpg")
cover_b = _store_public(db, CAROUSEL / "50933127.jpg", "tech-cover-report.jpg")
cover_c = _store_public(db, CAROUSEL / "49922210.jpg", "tech-cover-award.jpg")
urban = NEWS / "文章" / "11341437"
training = NEWS / "文章" / "10837751"
award = ABOUT / "文章" / "1209332"
create_article(
db,
{
"category": "tech_seminars",
"title": "城市更新中的房地产估价服务专题研讨",
"publishedAt": "2025-11-08T08:00:00",
"published": True,
"bodyHtml": (
_paragraphs(urban / "content.txt")
if (urban / "content.txt").is_file()
else "<p>城市更新估价服务专题研讨(现站摘录)。</p>"
)
+ "<p>整理自现站新闻动态公开报道,属技术研讨活动公示示意。</p>",
"imageFileId": cover_a.id if cover_a else None,
"attachmentFileIds": [],
},
)
create_article(
db,
{
"category": "tech_seminars",
"title": "涉房财报及企业价值评估专题研讨摘录",
"publishedAt": "2025-03-23T08:00:00",
"published": True,
"bodyHtml": (
_paragraphs(training / "content.txt")
if (training / "content.txt").is_file()
else "<p>继续教育与专题研讨公开摘录。</p>"
)
+ "<p>整理自现站培训报道,示意专项研讨发布形态。</p>",
"imageFileId": cover_b.id if cover_b else None,
"attachmentFileIds": [],
},
)
create_article(
db,
{
"category": "tech_seminars",
"title": "绿色建筑估价实践交流会报名通知(示意)",
"publishedAt": "2025-06-02T08:00:00",
"published": True,
"bodyHtml": "<p>本条无封面,示意无配图时列表仅显示标题。活动报名与附件由工作人员后续维护。</p>",
"imageFileId": None,
"attachmentFileIds": [],
},
)
create_article(
db,
{
"category": "tech_exchange",
"title": "2018年度优秀报告评选表彰与实践分享(公示)",
"publishedAt": "2019-04-26T08:00:00",
"published": True,
"bodyHtml": (
_paragraphs(award / "content.txt")
if (award / "content.txt").is_file()
else "<p>优秀报告评选表彰会议公开报道。</p>"
)
+ "<p>整理自现站协会介绍,属优秀报告评选公示示意;届次写在标题中。</p>",
"imageFileId": cover_c.id if cover_c else None,
"attachmentFileIds": [],
},
)
create_article(
db,
{
"category": "tech_exchange",
"title": "专家委员撰稿:涉房财报及企业价值评估专题摘录",
"publishedAt": "2025-03-24T08:00:00",
"published": True,
"bodyHtml": "<p>示意专家撰稿 / 期刊转载形态。正式正文由工作人员在管理端维护,可挂讲义附件。</p>",
"imageFileId": cover_b.id if cover_b else None,
"attachmentFileIds": [],
},
)
create_article(
db,
{
"category": "tech_exchange",
"title": "会员投稿:估价机构内部治理经验交流(代发)",
"publishedAt": "2025-01-10T08:00:00",
"published": True,
"bodyHtml": "<p>本期仅工作人员代发公示,无会员自助投稿入口。本条无封面。</p>",
"imageFileId": None,
"attachmentFileIds": [],
},
)
print(
"技术交流测试数据已写入:技术研讨 3 条,技术交流(模块)3 条。"
" 执行:python -m app.commands.seed_tech_exchange_sample"
)
finally:
db.close()
if __name__ == "__main__":
seed()