cma-management/backend/app/database.py

"""数据库配置"""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import inspect
import os
import logging

logger = logging.getLogger("cma")

# 2026-09-12 库访问治理（wecom-ops）：统一从 backend/.env 读取凭据，
# 定时任务/脚本与 8010 API 同源（此前脚本吃硬编码 fallback，与 .env 不一致）。
try:
    from dotenv import load_dotenv as _load_dotenv
    _load_dotenv(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), ".env"))
except Exception:  # dotenv 缺失/文件不可读时不影响启动，由下方环境变量兜底
    pass

DB_USER = os.getenv("CMA_DB_USER", "cma_reader")   # 兜底=只读账号（最小权限）
DB_PASS = os.getenv("CMA_DB_PASS", "")             # 无明文兜底：缺失即连接失败并报错
DB_HOST = os.getenv("CMA_DB_HOST", "127.0.0.1")
DB_PORT = os.getenv("CMA_DB_PORT", "3306")
DB_NAME = os.getenv("CMA_DB_NAME", "cma")

DATABASE_URL = "mysql+pymysql://%(user)s:%(password)s@%(host)s:%(port)s/%(name)s?charset=utf8mb4" % {
    "user": DB_USER,
    "password": DB_PASS,
    "host": DB_HOST,
    "port": DB_PORT,
    "name": DB_NAME,
}

_engine = None
_SessionLocal = None
Base = declarative_base()


def get_engine():
    global _engine
    if _engine is None:
        _engine = create_engine(DATABASE_URL, echo=False, pool_size=5, max_overflow=10, pool_pre_ping=True)
    return _engine


def get_session_local():
    global _SessionLocal
    if _SessionLocal is None:
        _SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=get_engine())
    return _SessionLocal


def get_db():
    db = get_session_local()()
    try:
        yield db
    finally:
        db.close()


def auto_create_enabled() -> bool:
    """启动时建表开关（2026-09-16「启动 create_all × 运行库最小权限」根治）。

    **默认关闭**：生产 DDL 由部署链的特权迁移步
    `backend/scripts/db_migrate.py`（deploy.sh [4b]，应用重启**前**执行）承担，
    因此运行库账号（cma_api）无需 CREATE 权限。
    仅开发/测试可显式 `CMA_AUTO_CREATE=1` 打开。

    事故背景：2026-09-16 01:08 发布新增模型（tax_filing_log/treasury_*）→
    启动 create_all 被 `cma_api` 无 CREATE 拒绝 → `CREATE command denied` → 崩溃循环。
    """
    return os.getenv("CMA_AUTO_CREATE", "0").strip().lower() in ("1", "true", "yes", "on")


def init_db():
    import app.models  # 仅注册模型（无 DDL）；下方 seed 也复用该引用

    if auto_create_enabled():
        logger.warning(
            "CMA_AUTO_CREATE 已开启 → 启动时执行 create_all（仅限开发/测试；生产请走部署链迁移步）"
        )
        Base.metadata.create_all(bind=get_engine())
        logger.info("CMA数据库已初始化（启动建表）")
    else:
        logger.info(
            "启动建表已关闭（CMA_AUTO_CREATE 未开启）：DDL 由部署链 scripts/db_migrate.py 在应用重启前执行"
        )

    # ── 初始化组织层级示例数据 ──
    try:
        inspector = inspect(get_engine())
        if "org_nodes" in inspector.get_table_names():
            Session = get_session_local()
            session = Session()
            try:
                cnt = session.query(app.models.OrgNode).count()
                if cnt == 0:
                    _seed_org_data(session)
            finally:
                session.close()
    except Exception as e:
        logger.warning(f"组织数据初始化跳过: {e}")

    # ── user_entities 授权表初始化（账套模式平滑迁移）──
    # 首次建表（表空）时，给存量用户默认授权所有 active 企业，保证现有登录不丢权限
    try:
        inspector = inspect(get_engine())
        if "user_entities" in inspector.get_table_names():
            Session = get_session_local()
            session = Session()
            try:
                from app.models import UserEntity, User, Entity
                cnt = session.query(UserEntity).count()
                if cnt == 0:
                    users = session.query(User).all()
                    entities = session.query(Entity).filter(Entity.status == "active").all()
                    if users and entities:
                        for u in users:
                            for e in entities:
                                exists = session.query(UserEntity).filter(
                                    UserEntity.user_id == u.id,
                                    UserEntity.entity_id == e.id,
                                ).first()
                                if not exists:
                                    session.add(UserEntity(user_id=u.id, entity_id=e.id, granted_by=None))
                        session.commit()
                        logger.info(f"user_entities初始化: {len(users)}用户 × {len(entities)}企业")
            finally:
                session.close()
    except Exception as e:
        logger.warning(f"user_entities初始化跳过: {e}")

    # ── scenario_suggestions 告警场景建议 seed（2026-08-28 告警归因 P1-③）──
    # 幂等：仅补缺失的 alert_type，不覆盖已有模板
    try:
        inspector = inspect(get_engine())
        if "scenario_suggestions" in inspector.get_table_names():
            Session = get_session_local()
            session = Session()
            try:
                from app.models import ScenarioSuggestion
                existing_types = {s.alert_type for s in session.query(ScenarioSuggestion).all()}
                seeds = [
                    dict(alert_type="cash_low", title="现金流紧张 — 加强回款催收",
                         description="现金余额接近警戒线，建议优先处理应收款项，压缩非紧急支出。",
                         action_template="1. 列出未来30天应收清单，逐笔催收\n2. 暂停非紧急采购/费用支出\n3. 与银行沟通短期授信额度",
                         priority="high", sort_order=1),
                    dict(alert_type="cash_critical", title="现金流危急 — 立即止血",
                         description="现金余额已低于安全阈值，存在断流风险，需要立即采取止血措施。",
                         action_template="1. 冻结一切非必要支出\n2. 高管紧急复盘资金计划\n3. 启动应收账款特别催收\n4. 评估短期融资",
                         priority="high", sort_order=2),
                    dict(alert_type="cost_high", title="成本超支 — 核查费用构成",
                         description="实际成本超出预算，建议拆解到科目明细定位超支源头。",
                         action_template="1. 查看科目明细拆解，定位超支前3科目\n2. 分析价差/量差成因（单价上涨/用量增加）\n3. 对可控费用制定压降方案",
                         priority="medium", sort_order=3),
                    dict(alert_type="revenue_drop", title="收入下滑 — 追量提效",
                         description="实际收入低于预算，建议从子KPI量级分解查找差距来源。",
                         action_template="1. 查看子KPI拆解，定位量差最大维度\n2. 分析客户/渠道/产品线缺口\n3. 制定增量获客或转化提升方案",
                         priority="medium", sort_order=4),
                ]
                added = 0
                for s in seeds:
                    if s["alert_type"] not in existing_types:
                        session.add(ScenarioSuggestion(**s))
                        added += 1
                if added:
                    session.commit()
                    logger.info(f"scenario_suggestions seed: 新增{added}条场景建议模板")
            finally:
                session.close()
    except Exception as e:
        logger.warning(f"scenario_suggestions seed跳过: {e}")


def _seed_org_data(db_session):
    """插入5层级组织示例数据"""
    from app.models import OrgNode

    # 1. 集团
    g = OrgNode(id=1, parent_id=None, name="博海网络科技", code="BH", level=1, sort_order=1, enabled=1)
    db_session.add(g)
    db_session.flush()

    # 2. 事业部
    depts = [
        OrgNode(parent_id=1, name="技术事业部", code="TECH", level=2, sort_order=1, enabled=1),
        OrgNode(parent_id=1, name="销售事业部", code="SALES", level=2, sort_order=2, enabled=1),
        OrgNode(parent_id=1, name="财务事业部", code="FIN", level=2, sort_order=3, enabled=1),
    ]
    db_session.add_all(depts)
    db_session.flush()

    # 3. 区域/部门级
    regions = [
        OrgNode(parent_id=2, name="华南区域", code="SC", level=3, sort_order=1, enabled=1),
        OrgNode(parent_id=2, name="华东区域", code="EC", level=3, sort_order=2, enabled=1),
        OrgNode(parent_id=3, name="销售一部", code="S1", level=3, sort_order=1, enabled=1),
        OrgNode(parent_id=3, name="销售二部", code="S2", level=3, sort_order=2, enabled=1),
    ]
    db_session.add_all(regions)
    db_session.flush()

    # 4. 部门
    departs = [
        OrgNode(parent_id=5, name="研发部", code="RD", level=4, sort_order=1, enabled=1),
        OrgNode(parent_id=5, name="实施部", code="IMP", level=4, sort_order=2, enabled=1),
        OrgNode(parent_id=5, name="运维部", code="OPS", level=4, sort_order=3, enabled=1),
        OrgNode(parent_id=6, name="前端研发", code="FE", level=4, sort_order=1, enabled=1),
        OrgNode(parent_id=6, name="后端研发", code="BE", level=4, sort_order=2, enabled=1),
        OrgNode(parent_id=8, name="KA客户部", code="KA", level=4, sort_order=1, enabled=1),
    ]
    db_session.add_all(departs)
    db_session.flush()

    # 5. 班组
    teams = [
        OrgNode(parent_id=12, name="前端组", code="FE-TEAM", level=5, sort_order=1, enabled=1),
        OrgNode(parent_id=12, name="后端组", code="BE-TEAM", level=5, sort_order=2, enabled=1),
        OrgNode(parent_id=12, name="测试组", code="QA-TEAM", level=5, sort_order=3, enabled=1),
        OrgNode(parent_id=13, name="实施一组", code="IMP1", level=5, sort_order=1, enabled=1),
        OrgNode(parent_id=13, name="实施二组", code="IMP2", level=5, sort_order=2, enabled=1),
    ]
    db_session.add_all(teams)
    db_session.commit()
    logger.info("组织层级示例数据已初始化")