server-core/app/park/park_config.py

# -*- coding: utf-8 -*-
"""园区主数据源（park_config）—— 入驻企业 / 园区智能体 / 知识库 / 大屏数据。

园区端（admin-portal carrier）与园区大屏共同的主数据源，落 serverdata/park/park_config.json。
企业默认从 sim_engine.COMPANIES（39 家）一次性导入，园区端可增删改 + 状态流转。
"""
from __future__ import annotations

import json
import threading
import time
from pathlib import Path

from .config import PARK_DIR, settings
from .sim_engine import COMPANIES

CONFIG_FILE = Path(PARK_DIR) / "park_config.json"
_lock = threading.RLock()


def _now() -> str:
    return time.strftime("%Y-%m-%d %H:%M:%S")


def _default() -> dict:
    # 首次：从园区名录一次性导入企业（name 为准，其余缺省）
    companies = [
        {
            "id": f"PC-{i:03d}",
            "name": name,
            "zone": "",
            "room": "",
            "industry": "",
            "bio": "",
            "founder": "",
            "status": "active",
            "employees": None,
            "created_at": _now(),
        }
        for i, name in enumerate(COMPANIES, start=1)
    ]
    return {
        "companies": companies,
        "agent": {
            "system_prompt": settings.S2S_INSTRUCTIONS,
            "model": settings.LLM_MODEL,
            "enabled_tools": ["get_park_overview", "query_companies", "control_display", "get_time"],
            "preset_questions": [],
            "opening": "",
        },
        "kb": {"docs": []},
        "screen": {
            "name": "昆明市大学生创业园",
            "region": "云南·昆明",
            "capacity": 49,
            "invested": 560,
            "jobs": 213,
            "area": 3000,
            "founded": 2009,
        },
    }


def _load() -> dict:
    if CONFIG_FILE.exists():
        try:
            data = json.loads(CONFIG_FILE.read_text("utf-8"))
            # 规范化：缺字段补默认
            base = _default()
            base.update(data)
            return base
        except Exception:  # noqa: BLE001
            pass
    return _default()


def _save(data: dict):
    CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
    CONFIG_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2), "utf-8")


# ---------------- 入驻企业 ----------------

def list_companies(status: str | None = None) -> list[dict]:
    with _lock:
        data = _load()
    comps = data.get("companies", [])
    if status:
        comps = [c for c in comps if c.get("status") == status]
    return comps


def create_company(payload: dict) -> dict:
    with _lock:
        data = _load()
        comps = data.setdefault("companies", [])
        cid = f"PC-{int(time.time())%100000}-{len(comps)+1}"
        row = {
            "id": cid,
            "name": payload.get("name", ""),
            "zone": payload.get("zone", ""),
            "room": payload.get("room", ""),
            "industry": payload.get("industry", ""),
            "bio": payload.get("bio", ""),
            "founder": payload.get("founder", ""),
            "status": payload.get("status", "applying"),
            "employees": payload.get("employees"),
            "created_at": _now(),
        }
        comps.append(row)
        _save(data)
        return row


def update_company(cid: str, payload: dict) -> dict | None:
    with _lock:
        data = _load()
        comps = data.setdefault("companies", [])
        for c in comps:
            if c.get("id") == cid:
                for k, v in payload.items():
                    if v is not None and k in c:
                        c[k] = v
                _save(data)
                return c
    return None


def delete_company(cid: str) -> bool:
    with _lock:
        data = _load()
        comps = data.get("companies", [])
        before = len(comps)
        data["companies"] = [c for c in comps if c.get("id") != cid]
        changed = len(data["companies"]) != before
        if changed:
            _save(data)
        return changed


# ---------------- 园区智能体 ----------------

def get_agent() -> dict:
    with _lock:
        return _load().get("agent", _default()["agent"])


def update_agent(payload: dict) -> dict:
    with _lock:
        data = _load()
        agent = data.setdefault("agent", _default()["agent"])
        for k, v in payload.items():
            if v is not None:
                agent[k] = v
        _save(data)
        return agent


# ---------------- 园区知识库 ----------------

def list_kb_docs() -> list[dict]:
    with _lock:
        return _load().get("kb", {}).get("docs", [])


def create_kb_doc(payload: dict) -> dict:
    with _lock:
        data = _load()
        docs = data.setdefault("kb", {}).setdefault("docs", [])
        row = {
            "id": f"KB-{int(time.time())%100000}-{len(docs)+1}",
            "group": payload.get("group", "general"),
            "title": payload.get("title", ""),
            "content_md": payload.get("content_md", ""),
            "created_at": _now(),
        }
        docs.append(row)
        _save(data)
        return row


def update_kb_doc(did: str, payload: dict) -> dict | None:
    with _lock:
        data = _load()
        docs = data.get("kb", {}).get("docs", [])
        for d in docs:
            if d.get("id") == did:
                for k, v in payload.items():
                    if v is not None and k in d:
                        d[k] = v
                _save(data)
                return d
    return None


def delete_kb_doc(did: str) -> bool:
    with _lock:
        data = _load()
        docs = data.get("kb", {}).get("docs", [])
        before = len(docs)
        data["kb"]["docs"] = [d for d in docs if d.get("id") != did]
        changed = len(data["kb"]["docs"]) != before
        if changed:
            _save(data)
        return changed


# ---------------- 大屏数据 ----------------

def get_screen() -> dict:
    with _lock:
        return _load().get("screen", _default()["screen"])


def update_screen(payload: dict) -> dict:
    with _lock:
        data = _load()
        screen = data.setdefault("screen", _default()["screen"])
        for k, v in payload.items():
            if v is not None:
                screen[k] = v
        _save(data)
        return screen