scripts/seed_demo.py

#!/usr/bin/env python3
"""示範資料 seed（僅供公開示範用，請勿放真實資料）。

建立：
  1. 可直接登入的示範帳號（已知密碼、免 2FA、免強制改密碼）—— 管理者 / 主管 / 一般使用者
  2. 幾篇公開範例文件，讓知識庫不是空的
  3. 啟用「機密」偵測規則，讓多層 PII 防護在示範時看得到效果

特性：冪等（重複執行不會重複建立）。於容器啟動、uvicorn 之前執行。
"""
from __future__ import annotations

import json
import os
import sys

# 讓 `app.*` 可被 import（本檔在 scripts/，後端在 ../backend/）
_HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(_HERE, "..", "backend"))

from app.auth.passwords import hash_password  # noqa: E402
from app.db import _utcnow, get_conn  # noqa: E402

DEMO_PASSWORD = os.environ.get("DEMO_PASSWORD", "Demo1234!")

# (username, role, department)
DEMO_USERS = [
    ("demo_admin", "admin", "管理部"),
    ("demo_manager", "manager", "業務部"),
    ("demo_user", "user", "研發部"),
]

# 純通用辦公範例（不含任何真實產品/公司內容；公開示範用）
# (title, category, tags, summary, body)
SAMPLE_DOCS = [
    ("新進人員報到指南", "流程/SOP", ["新人", "報到", "onboarding"],
     "新進同仁第一週的報到與必讀事項。",
     "## 第一天\n* 領取系統帳號與識別證\n* 閱讀資訊安全與保密規範\n* 認識部門同事\n\n"
     "## 第一週\n* 熟悉知識庫『提交 → AI 整理 → 審核 → 入庫』流程\n* 完成必讀文件與線上課程\n"
     "* 與主管確認近期工作目標"),
    ("員工差勤管理辦法", "規範/制度", ["差勤", "請假", "出勤"],
     "公司請假、加班與出勤規定摘要。",
     "## 請假類別\n* 特休假\n* 病假\n* 事假\n* 公假\n\n"
     "## 申請流程\n1. 於系統提出申請\n2. 主管線上核准\n3. 系統自動更新出勤紀錄\n\n"
     "## 注意事項\n* 請假需事前申請；急病得事後補件"),
    ("會議室預約與使用規則", "流程/SOP", ["會議室", "預約", "行政"],
     "公司會議室的預約方式與使用須知。",
     "## 預約方式\n1. 於行政系統選擇會議室與時段\n2. 填寫會議主題與參與人數\n\n"
     "## 使用規則\n* 會議結束請整理桌面、關閉設備\n* 逾時未到視同放棄，釋出給他人\n* 茶水請自行清理"),
]


def seed_demo_users(conn) -> int:
    created = 0
    for uname, role, dept in DEMO_USERS:
        if conn.execute("SELECT id FROM users WHERE username=?", (uname,)).fetchone():
            continue
        conn.execute(
            "INSERT INTO users (username, email, password_hash, role, department, created_at, "
            "is_active, force_password_change, totp_enabled, jwt_version) "
            "VALUES (?,?,?,?,?,?,1,0,0,1)",
            (uname, f"{uname}@demo.local", hash_password(DEMO_PASSWORD), role, dept, _utcnow()),
        )
        created += 1
    return created


def seed_sample_docs(conn) -> int:
    author = conn.execute("SELECT id FROM users WHERE username='demo_admin'").fetchone()
    if not author:
        return 0
    aid = author["id"]
    from app.services import vault
    created = 0
    for title, cat, tags, summary, body in SAMPLE_DOCS:
        if conn.execute("SELECT id FROM documents WHERE title=? AND is_deleted=0",
                        (title,)).fetchone():
            continue
        now = _utcnow()
        cur = conn.execute(
            "INSERT INTO documents (title, category, tags, summary, body, author_id, "
            "owner_department, visibility, status, created_at, updated_at) "
            "VALUES (?,?,?,?,?,?,?, 'public', 'active', ?, ?)",
            (title, cat, json.dumps(tags, ensure_ascii=False), summary, body, aid, "管理部", now, now),
        )
        try:
            vault.write_document(conn, cur.lastrowid)  # 物化 .md（事實來源）
        except Exception as e:  # noqa: BLE001
            print(f"[seed_demo] vault 寫入略過：{e}")
        created += 1
    return created


def enable_demo_pii_rule(conn) -> None:
    """啟用「機密」keyword 偵測規則，讓多層 PII 防護示範看得到效果。"""
    if conn.execute("SELECT id FROM detection_rules WHERE pattern='機密' "
                    "AND rule_type='keyword' AND is_deleted=0").fetchone():
        return
    from app.services import variant_gen
    try:
        variants = json.dumps(variant_gen.generate_variants("機密"), ensure_ascii=False)
    except Exception:  # noqa: BLE001
        variants = "[]"
    conn.execute(
        "INSERT INTO detection_rules (rule_name, pattern, rule_type, layer, action, is_active, "
        "is_deleted, is_builtin, variants, created_at) "
        "VALUES ('機密','機密','keyword',1,'block',1,0,0,?,?)",
        (variants, _utcnow()),
    )


def backfill_embeddings(conn) -> int:
    """幫沒有 embedding 的現有文件補算（embeddings 可用時）。
    重疊偵測／AI 合併靠文件 embedding 相似度——沒有就比不到、不會建議合併。"""
    from app.services import embeddings
    if not embeddings.is_available():
        return 0
    from app.services.ai_pipeline import compute_document_embedding
    rows = conn.execute(
        "SELECT id FROM documents WHERE is_deleted=0 AND status='active' AND embedding IS NULL"
    ).fetchall()
    n = 0
    for r in rows:
        try:
            if compute_document_embedding(conn, r["id"]):
                n += 1
        except Exception:  # noqa: BLE001
            pass
    return n


def configure_demo_ai(conn) -> None:
    """示範用 AI 設定：Gemini 用 flash（免費方案有額度）；合併門檻調低讓功能容易展示。
    生產環境會用較高門檻（0.85）避免過度合併；示範刻意調低（0.6）方便看到合併效果。"""
    try:
        from app.services import ai_provider as ap
        ap.set_config(conn, models={"gemini": "gemini-flash-latest"},
                      merge_threshold=0.6, link_threshold=0.45, by=None)
    except Exception:  # noqa: BLE001
        pass


def main() -> None:
    conn = get_conn()  # 觸發 schema / migration / 標準 seed（KBV5_SEED=1）
    u = seed_demo_users(conn)
    d = seed_sample_docs(conn)
    enable_demo_pii_rule(conn)
    configure_demo_ai(conn)
    e = backfill_embeddings(conn)
    try:
        conn.commit()
    except Exception:  # noqa: BLE001
        pass
    print(f"[seed_demo] 完成：示範帳號 +{u}、範例文件 +{d}、補算 embedding {e} 篇、密碼={DEMO_PASSWORD}")


if __name__ == "__main__":
    main()