183 lines
8.9 KiB
Python
183 lines
8.9 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""【临时演示数据 · 手动运行 · 用完执行 del 清除】uv run python scripts/db/demo_content_tmp.py add|del
|
||
|
|
|
||
|
|
add —— 灌入演示数据：资讯 4 类别 × 25 条（大小图/官方·园区·OPC 混排）+ 活动 40 场（类型×线上线下×状态）
|
||
|
|
del —— 仅删除本脚本灌入的那批演示数据（按记录在 manifest 里的 id 精确删除，不动其它数据）
|
||
|
|
说明：id 全部确定性生成（cont_demo_* / E-DEMO-*），并同步写入 demo_content_tmp.ids.json 备查。
|
||
|
|
不入种子、不进迁移；演示完执行 del 即可。
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import json
|
||
|
|
import sqlite3
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
|
||
|
|
|
||
|
|
from app.infrastructure.db import AsyncSessionLocal # noqa: E402
|
||
|
|
from app.training.db import DB_PATH as TRAINING_DB # noqa: E402
|
||
|
|
from app.infrastructure.models import ContentItem # noqa: E402
|
||
|
|
from app.infrastructure.seed import utcnow_iso # noqa: E402
|
||
|
|
|
||
|
|
MANIFEST = Path(__file__).resolve().parent / "demo_content_tmp.ids.json"
|
||
|
|
|
||
|
|
# ── 生成参数 ────────────────────────────────────────────────────────────────
|
||
|
|
CATS = ["policy", "news", "skill", "dynamic"]
|
||
|
|
PER_CAT = 25 # 每类别条数（合计 100 条资讯）
|
||
|
|
EVENTS_N = 40 # 活动场数
|
||
|
|
SOURCES = [("official", "官方"), ("carrier", "园区运营方"), ("opc", "云南某某工作室")]
|
||
|
|
TOPICS = {
|
||
|
|
"policy": ["创业补贴", "创业担保贷款", "税收减免", "社保补贴", "场地扶持", "职称评定", "技能培训补贴", "青年见习", "创业导师", "园区入驻", "人才引进", "数字转型"],
|
||
|
|
"news": ["平台上线", "公益课回顾", "园区动态", "OPC 会员风采", "合作签约", "赛事启动", "政策解读会", "创业者访谈", "月度数据", "生态共建", "算力升级", "社区活动"],
|
||
|
|
"skill": ["AI 内容生产", "定价四步法", "路演 PPT", "财税入门", "私域运营", "短视频起号", "跨境选品", "合同避坑", "个人品牌", "效率工具", "数据分析", "直播带货"],
|
||
|
|
"dynamic": ["新成员入驻", "企业动态", "平台周更", "榜单发布", "活动排期", "服务升级", "导师入库", "资源对接", "空间改造", "认证公示", "积分商城", "周年活动"],
|
||
|
|
}
|
||
|
|
HINT = "（演示数据，可在运营端删除）"
|
||
|
|
|
||
|
|
TITLE_POOL = ["实战指南", "入门手册", "案例拆解", "常见问题", "申报要点", "趋势观察", "工具盘点", "流程图解", "避坑清单", "一手复盘"]
|
||
|
|
|
||
|
|
_all_titles = [(cat, i) for cat in CATS for i in range(PER_CAT)]
|
||
|
|
|
||
|
|
|
||
|
|
def _content_id(cat: str, i: int) -> str:
|
||
|
|
return f"cont_demo_{cat}_{i:03d}"
|
||
|
|
|
||
|
|
|
||
|
|
def _event_id(i: int) -> str:
|
||
|
|
return f"E-DEMO-{i:03d}"
|
||
|
|
|
||
|
|
|
||
|
|
def _iso(days_ago: float, hour: int = 9) -> str:
|
||
|
|
from datetime import datetime, timedelta, timezone
|
||
|
|
d = datetime.now(timezone.utc) - timedelta(days=days_ago)
|
||
|
|
return d.replace(hour=hour, minute=(int(days_ago * 60) % 60), second=0, microsecond=0).isoformat()
|
||
|
|
|
||
|
|
|
||
|
|
def _build_content() -> list[ContentItem]:
|
||
|
|
items: list[ContentItem] = []
|
||
|
|
for cat, i in _all_titles:
|
||
|
|
topic = TOPICS[cat][i % len(TOPICS[cat])]
|
||
|
|
title = f"{topic}·{TITLE_POOL[i % len(TITLE_POOL)]}{HINT}"
|
||
|
|
source, pub = SOURCES[i % 3]
|
||
|
|
mode = "big" if i % 3 == 0 else "small" # 1/3 大图、2/3 小图
|
||
|
|
prio = 200 - i # 越新越靠前
|
||
|
|
items.append(ContentItem(
|
||
|
|
id=_content_id(cat, i), type=cat, title=title,
|
||
|
|
summary=f"{topic}方向的演示摘要：覆盖要点、流程与常见疑问，帮助创业者快速上手。",
|
||
|
|
body=("<p>本条为<b>演示数据</b>，用于验证资讯中心在各类别、各发布方与大小图模式下的展示效果。</p>"
|
||
|
|
f"<p>类别：{cat} · 发布方：{pub} · 模式：{mode}</p>"
|
||
|
|
"<p>删除方式：uv run python scripts/db/demo_content_tmp.py del</p>"),
|
||
|
|
publisher_id="", publisher_name=pub, status="published", is_public=True,
|
||
|
|
card_mode=mode, source=source, priority=prio,
|
||
|
|
read_count=100 + i * 37, like_count=(i * 7) % 90, share_count=(i * 3) % 40,
|
||
|
|
published_at=_iso(i * 0.4 + 1), created_at=_iso(i * 0.4 + 1), updated_at=utcnow_iso(),
|
||
|
|
))
|
||
|
|
return items
|
||
|
|
|
||
|
|
|
||
|
|
def _build_events() -> list[dict]:
|
||
|
|
evs = []
|
||
|
|
types = [("free", "公益课"), ("salon", "沙龙")]
|
||
|
|
modes = [("offline", "昆明市大学生创业园"), ("online", "视频号直播")]
|
||
|
|
statuses = ["open", "open", "full", "done"]
|
||
|
|
for i in range(EVENTS_N):
|
||
|
|
etype, tlabel = types[i % 2]
|
||
|
|
mode, loc = modes[(i // 2) % 2]
|
||
|
|
status = statuses[i % 4]
|
||
|
|
topic = TOPICS["skill"][(i * 5) % len(TOPICS["skill"])]
|
||
|
|
evs.append({
|
||
|
|
"id": _event_id(i), "type": etype, "mode": mode,
|
||
|
|
"title": f"{topic}{tlabel}第 {i + 1} 场（演示）", "subtitle": f"{tlabel} · {'线上' if mode == 'online' else '线下'}场",
|
||
|
|
"desc": "演示场次：覆盖类型×模式×状态的组合维度。删除方式：uv run python scripts/db/demo_content_tmp.py del",
|
||
|
|
"location": loc, "host": "云超服 · OPC 培训",
|
||
|
|
"image": f"https://picsum.photos/seed/opc-demo-{i}/600/400", "link": "",
|
||
|
|
"start_at": _iso(-((i % 15) + 1), 14 if etype == "salon" else 19)[:19] + "+08:00",
|
||
|
|
"duration_min": 120 if etype == "salon" else 90,
|
||
|
|
"capacity": 30 if etype == "salon" else 60, "status": status,
|
||
|
|
})
|
||
|
|
return evs
|
||
|
|
|
||
|
|
|
||
|
|
async def _add() -> None:
|
||
|
|
items = _build_content()
|
||
|
|
async with AsyncSessionLocal() as session:
|
||
|
|
added = 0
|
||
|
|
for it in items:
|
||
|
|
if await session.get(ContentItem, it.id) is None:
|
||
|
|
session.add(it)
|
||
|
|
added += 1
|
||
|
|
await session.commit()
|
||
|
|
|
||
|
|
conn = sqlite3.connect(TRAINING_DB)
|
||
|
|
conn.row_factory = sqlite3.Row
|
||
|
|
ev_added = 0
|
||
|
|
try:
|
||
|
|
for e in _build_events():
|
||
|
|
dup = conn.execute("SELECT 1 FROM events WHERE id=:id", {"id": e["id"]}).fetchone()
|
||
|
|
if dup:
|
||
|
|
continue
|
||
|
|
conn.execute("INSERT INTO events (id,type,mode,title,subtitle,desc,location,host,image,link,start_at,duration_min,capacity,status) "
|
||
|
|
"VALUES (:id,:type,:mode,:title,:subtitle,:desc,:location,:host,:image,:link,:start_at,:duration_min,:capacity,:status)", e)
|
||
|
|
ev_added += 1
|
||
|
|
conn.commit()
|
||
|
|
finally:
|
||
|
|
conn.close()
|
||
|
|
|
||
|
|
MANIFEST.write_text(json.dumps({
|
||
|
|
"content_ids": [it.id for it in items],
|
||
|
|
"event_ids": [e["id"] for e in _build_events()],
|
||
|
|
"note": "demo_content_tmp.py 灌入的演示数据清单（del 时按此精确删除）",
|
||
|
|
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
|
print(f"add 完成：资讯新增 {added}/{len(items)}，活动新增 {ev_added}/{EVENTS_N}；清单已写入 {MANIFEST.name}")
|
||
|
|
|
||
|
|
|
||
|
|
async def _del() -> None:
|
||
|
|
if not MANIFEST.exists():
|
||
|
|
# 兜底：按确定性 id 规则删除
|
||
|
|
content_ids = [_content_id(cat, i) for cat, i in _all_titles]
|
||
|
|
event_ids = [_event_id(i) for i in range(EVENTS_N)]
|
||
|
|
print("未找到 manifest，按确定性 id 规则删除")
|
||
|
|
else:
|
||
|
|
m = json.loads(MANIFEST.read_text(encoding="utf-8"))
|
||
|
|
content_ids, event_ids = m["content_ids"], m["event_ids"]
|
||
|
|
# 兼容清理：早期手工批次的演示 id（同样属于本脚本的演示数据）
|
||
|
|
content_ids += ["cont_demo_pol_big_01", "cont_demo_pol_small_01", "cont_demo_pol_small_02",
|
||
|
|
"cont_demo_news_big_01", "cont_demo_news_small_01", "cont_demo_news_big_02", "cont_demo_news_small_02",
|
||
|
|
"cont_demo_skill_big_01", "cont_demo_skill_small_01", "cont_demo_skill_big_02", "cont_demo_skill_small_02",
|
||
|
|
"cont_demo_dyn_big_01", "cont_demo_dyn_small_01", "cont_demo_dyn_small_02"]
|
||
|
|
event_ids += ["E-S005", "E-F005"]
|
||
|
|
|
||
|
|
async with AsyncSessionLocal() as session:
|
||
|
|
removed = 0
|
||
|
|
for cid in content_ids:
|
||
|
|
obj = await session.get(ContentItem, cid)
|
||
|
|
if obj is not None:
|
||
|
|
await session.delete(obj)
|
||
|
|
removed += 1
|
||
|
|
await session.commit()
|
||
|
|
|
||
|
|
conn = sqlite3.connect(TRAINING_DB)
|
||
|
|
ev_removed = 0
|
||
|
|
try:
|
||
|
|
for eid in event_ids:
|
||
|
|
cur = conn.execute("DELETE FROM events WHERE id=:id", {"id": eid})
|
||
|
|
ev_removed += cur.rowcount
|
||
|
|
conn.commit()
|
||
|
|
finally:
|
||
|
|
conn.close()
|
||
|
|
|
||
|
|
MANIFEST.unlink(missing_ok=True)
|
||
|
|
print(f"del 完成：资讯删除 {removed}/{len(content_ids)}，活动删除 {ev_removed}/{len(event_ids)}")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else ""
|
||
|
|
if cmd == "add":
|
||
|
|
asyncio.run(_add())
|
||
|
|
elif cmd == "del":
|
||
|
|
asyncio.run(_del())
|
||
|
|
else:
|
||
|
|
print(__doc__)
|