server-core/scripts/db/seed_hall_community.py

# -*- coding: utf-8 -*-
"""大厅与社区演示种子（非运行态，幂等）—— 0035 迁移后执行。

用法：uv run python scripts/db/seed_hall_community.py
灌入：人才档案 / 岗位 / OPC 服务 / 社区帖子演示数据（挂到演示账号 u_demo_01 / u_opc_01）。
"""
from __future__ import annotations

import asyncio
import json
import logging
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))

from app.infrastructure.hall_repositories import (
    CommunityPostRepository,
    JobRepository,
    OpcServiceRepository,
    TalentProfileRepository,
)
from app.infrastructure.repositories import Database, utcnow_iso

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("db.seed_hall")


TALENTS = [
    {"user_id": "u_demo_01", "talent_type": "opc", "display_name": "阿岩",
     "headline": "AI 应用开发 · 全栈 · 昆明", "region": "昆明",
     "bio": "专注 AI 应用落地：智能客服、内容生成、小程序全栈。带过 3 个创业团队技术线。",
     "fields_json": json.dumps(["AI应用", "技术开发"], ensure_ascii=False),
     "skills_json": json.dumps(["Python", "FastAPI", "小程序", "Prompt工程"], ensure_ascii=False),
     "work_years": 6, "cases_count": 12},
    {"user_id": "u_opc_01", "talent_type": "opc", "display_name": "小玉",
     "headline": "品牌设计 · 短视频剪辑 · 大理", "region": "大理",
     "bio": "服务过 30+ 农特产品牌的视觉与内容，擅长用 AI 把设计产能放大。",
     "fields_json": json.dumps(["设计创意", "音视频"], ensure_ascii=False),
     "skills_json": json.dumps(["PS", "AI绘画", "剪辑"], ensure_ascii=False),
     "work_years": 4, "cases_count": 30},
]

JOBS = [
    {"company_name": "云南某某甲方企业", "title": "小程序开发工程师", "category": "技术",
     "description": "负责公司电商小程序的功能迭代与性能优化。", "requirement": "1 年以上小程序开发经验，熟悉 Taro 或原生。",
     "tags": "小程序,前端", "salary_min": 6000, "salary_max": 10000, "location": "昆明·五华",
     "work_type": "fulltime", "headcount": 1, "contact": "hr@example.cn"},
    {"company_name": "大理文旅传媒", "title": "短视频运营", "category": "运营",
     "description": "负责文旅账号的内容策划与短视频运营。", "requirement": "有短视频账号运营案例者优先。",
     "tags": "短视频,文旅", "salary_min": 4500, "salary_max": 8000, "location": "大理",
     "work_type": "fulltime", "headcount": 2, "contact": "hr@wl.cn"},
]

SERVICES = [
    {"opc_id": "u_demo_01", "opc_name": "阿岩", "title": "企业官网 + AI 客服一站式搭建",
     "category": "软件开发", "description": "官网设计开发 + 接入 AI 智能客服，两周交付，含一年维护。",
     "price": 8000, "delivery_days": 14, "tags": "官网,AI客服,全栈", "status": "published"},
    {"opc_id": "u_opc_01", "opc_name": "小玉", "title": "农特产品牌全套视觉（Logo+包装+详情页）",
     "category": "设计创意", "description": "面向云南农特产品牌：Logo、包装、电商详情页一套齐。",
     "price": 5000, "delivery_days": 10, "tags": "Logo,包装,详情页", "status": "published"},
]

POSTS = [
    {"author_name": "阿岩", "topic": "经验", "title": "一个人接单两年，我踩过的 5 个坑",
     "content": "从报价到合同到尾款……每一条都是真金白银换来的经验，欢迎补充。", "pinned": True},
    {"author_name": "小玉", "topic": "资源", "title": "整理了一波云南本地可用的 AI 工具折扣",
     "content": "算力、设计、剪辑三类，评论区自取，需要邀请码的私信我。"},
    {"author_name": "李同学", "topic": "求助", "title": "应届生想入行超级个体，先从什么方向开始？",
     "content": "市场营销专业，会一点剪辑和文案，求过来人指条路。"},
]


async def run() -> None:
    db = Database()
    try:
        users = {u["username"]: u for u in []}   # 占位：直接按 id 取
        demo = await db.users.get_by_id("u_demo_01") or await db.users.get_by_username("u_demo_01") or {}
        opc = await db.users.get_by_id("u_opc_01") or await db.users.get_by_username("opc01") or {}
        talent_repo = TalentProfileRepository(db.session)
        job_repo = JobRepository(db.session)
        svc_repo = OpcServiceRepository(db.session)
        post_repo = CommunityPostRepository(db.session)
        now = utcnow_iso()

        for t in TALENTS:
            uid = t["user_id"]
            user = demo if uid == "u_demo_01" else (opc or None)
            if user is None:
                continue  # 演示账号不存在则跳过该档案
            patch = dict(t)
            patch.pop("user_id")
            patch["talentType"] = patch.pop("talent_type")
            patch["displayName"] = patch.get("displayName") or user.get("nickname")
            patch["avatar"] = user.get("avatar", "")
            patch["display_name"] = patch["displayName"]
            patch["fields"] = json.loads(patch.pop("fields_json"))
            patch["skills"] = json.loads(patch.pop("skills_json"))
            await talent_repo.upsert(uid, patch)
        log.info("人才档案种子完成")

        if not await job_repo.list(status="", q="小程序开发工程师"):
            for j in JOBS:
                await job_repo.create(status="published", published_at=now, **j)
            log.info("岗位种子完成")

        if not await svc_repo.list(q="企业官网"):
            for s in SERVICES:
                owner = demo if s["opc_id"] == "u_demo_01" else opc
                fields = {k: v for k, v in s.items() if k not in ("opc_id", "opc_name", "status")}
                await svc_repo.create(status="published",
                                      opc_id=owner.get("id") or s["opc_id"],
                                      opc_name=owner.get("nickname") or s["opc_name"], **fields)
            log.info("OPC 服务种子完成")

        if not await post_repo.list(status=""):
            author = demo or opc
            for p in POSTS:
                await post_repo.create(author_id=author.get("id"), avatar=author.get("avatar", ""),
                                       status="published", pinned=p.get("pinned", False),
                                       **{k: v for k, v in p.items() if k != "pinned"})
            log.info("社区帖子种子完成")
    finally:
        await db.close()


if __name__ == "__main__":
    asyncio.run(run())