app.py

import os
import sqlite3
import json
import datetime
import random
import requests
from flask import Flask, render_template, request, jsonify, g

app = Flask(__name__, instance_relative_config=True)
app.config["SECRET_KEY"] = "factory-flow-secret-key"
app.config["JSON_AS_ASCII"] = False

try:
    os.makedirs(app.instance_path, exist_ok=True)
except OSError:
    pass

DB_PATH = os.path.join(app.instance_path, "factory_flow.db")

SILICON_FLOW_API_URL = "https://api.siliconflow.cn/v1/chat/completions"
DEFAULT_MODEL = "Qwen/Qwen2.5-7B-Instruct"
DEFAULT_TEMPLATES = [
    {
        "title": "电池产线良率提升闭环",
        "scenario": "动力电池产线希望在 2 个月内提升良品率 1.5%，同时降低涂布段停机率。请给出闭环方案与验收指标。",
        "hint": "优先解决涂布段设备老化与质量波动问题。",
    },
    {
        "title": "仓配协同排产闭环",
        "scenario": "电子制造工厂需要将排产与仓配补料联动，降低缺料率并提高交付准时率。",
        "hint": "关注安全库存阈值与跨班次协同。",
    },
    {
        "title": "冷链质量追溯闭环",
        "scenario": "冷链食品工厂的温控波动导致返工率上升，需要建立温控预警与批次追溯体系。",
        "hint": "明确温控指标和追溯链路。",
    },
    {
        "title": "设备预防性维护闭环",
        "scenario": "设备故障频发导致产能波动，要求建立预防性维护节奏与停机风险预测。",
        "hint": "补齐设备状态监测与维护日志闭环。",
    },
]


def get_db():
    db = getattr(g, "_database", None)
    if db is None:
        db = g._database = sqlite3.connect(DB_PATH)
        db.row_factory = sqlite3.Row
    return db


@app.teardown_appcontext
def close_connection(exception):
    db = getattr(g, "_database", None)
    if db is not None:
        db.close()


def init_db():
    db = get_db()
    db.execute(
        """
        CREATE TABLE IF NOT EXISTS sessions (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT,
            scenario TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    db.execute(
        """
        CREATE TABLE IF NOT EXISTS steps (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            session_id INTEGER NOT NULL,
            step_order INTEGER NOT NULL,
            role TEXT NOT NULL,
            step_type TEXT NOT NULL,
            content TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY(session_id) REFERENCES sessions(id) ON DELETE CASCADE
        )
        """
    )
    db.execute(
        """
        CREATE TABLE IF NOT EXISTS assets (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            category TEXT NOT NULL,
            content TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    db.execute(
        """
        CREATE TABLE IF NOT EXISTS chats (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            role TEXT NOT NULL,
            content TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    db.commit()
    seed_default_data(db)


def seed_default_data(db):
    row = db.execute("SELECT COUNT(*) AS c FROM sessions").fetchone()
    if row and row["c"]:
        return

    demos = [
        {
            "name": "示例会话 · 新能源电池产线降本增效",
            "scenario": (
                "一家动力电池工厂计划在 3 个月内将良品率提升 2%，并降低单位能耗 8%。"
                "当前产线瓶颈集中在涂布与化成段，设备老化导致停机频繁。"
                "要求输出：可执行的改造路径、数据采集方案、质量验证与验收指标。"
            ),
            "hint": "优先解决涂布段产能与能耗问题，同时建立快速验收闭环。",
        },
        {
            "name": "示例会话 · 智能仓配协同工厂排产",
            "scenario": (
                "一家电子制造企业需要将产线排产与仓配补料联动。"
                "目标是让关键物料缺料率下降 50%，并保证交付准时率 > 95%。"
                "请给出排产策略、补料触发规则、异常处理与复盘机制。"
            ),
            "hint": "关注关键物料的安全库存阈值与跨班次协同。",
        },
        {
            "name": "示例会话 · 食品工厂冷链质量闭环",
            "scenario": (
                "一家冷链食品工厂需要降低返工率并提升出货稳定性。"
                "当前问题集中在冷链温控波动、包装漏检与批次追溯不完整。"
                "请给出数据采集与批次追溯闭环方案，并输出验收指标。"
            ),
            "hint": "优先建立温控波动预警与批次追溯链路。",
        },
    ]

    for item in demos:
        cursor = db.execute(
            "INSERT INTO sessions (name, scenario) VALUES (?, ?)",
            (item["name"], item["scenario"]),
        )
        session_id = cursor.lastrowid
        result = run_factory_workflow(item["scenario"], extra_hint=item["hint"])
        for s in result["steps"]:
            db.execute(
                """
                INSERT INTO steps (session_id, step_order, role, step_type, content)
                VALUES (?, ?, ?, ?, ?)
                """,
                (session_id, s["step_order"], s["role"], s["step_type"], s["content"]),
            )

    db.execute(
        """
        INSERT INTO assets (title, category, content)
        VALUES (?, ?, ?)
        """,
        (
            "基准产线画像模板",
            "流程资产",
            "用于沉淀关键工序、设备台账与质量指标的基础模板，可直接复用为后续方案的输入。",
        ),
    )
    db.execute(
        """
        INSERT INTO assets (title, category, content)
        VALUES (?, ?, ?)
        """,
        (
            "关键工序异常响应卡",
            "运营资产",
            "当检测到异常停机、良率波动或温控偏移时，按照分级响应流程快速锁定问题并记录复盘。",
        ),
    )
    db.execute(
        "INSERT INTO chats (role, content) VALUES (?, ?)",
        ("assistant", "你好，我可以帮你评估产线瓶颈、成本与交付风险。"),
    )
    db.commit()


def call_llm(messages, model=DEFAULT_MODEL, max_tokens=1200):
    api_key = os.getenv("SILICONFLOW_API_KEY", "").strip()

    def mock_completion():
        now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
        last_user = ""
        for m in reversed(messages):
            if m.get("role") == "user":
                last_user = m.get("content", "")[:300]
                break
        return (
            f"## 本地模拟智能体回复\n\n"
            f"- 时间：{now}\n"
            f"- 说明：当前未连接硅基流 API，系统使用本地规则进行推演。\n\n"
            f"### 输入摘要\n\n"
            f"{last_user}\n\n"
            f"### 建议\n\n"
            f"- 先建立数据采集与异常预警，再扩展到全链路优化。"
        )

    if not api_key or not api_key.startswith("sk-"):
        return mock_completion()

    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": model,
        "messages": messages,
        "stream": False,
        "max_tokens": max_tokens,
    }

    try:
        resp = requests.post(SILICON_FLOW_API_URL, headers=headers, json=payload, timeout=20)
        if resp.status_code == 200:
            data = resp.json()
            return data["choices"][0]["message"]["content"]
        return mock_completion()
    except Exception:
        return mock_completion()


def simulate_factory_tools(scenario_text):
    length = len(scenario_text)
    demand_index = min(96, max(55, 55 + length % 42))
    energy_cost_index = round(0.62 + (length % 36) * 0.01, 2)
    utilization = min(93, 66 + (length % 24))
    defect_risk = min(88, 35 + (length % 52))

    bottlenecks = ["涂布段", "化成段", "贴片段", "包装段", "检测段", "仓配补料"]
    weighted = []
    for item in bottlenecks:
        weight = 1
        if item in scenario_text:
            weight += 2
        weighted.extend([item] * weight)
    bottleneck = random.choice(weighted)

    keywords = ["冷链", "良品率", "缺料", "停机", "能耗", "交付", "追溯", "排产"]
    risk_score = 0
    for k in keywords:
        if k in scenario_text:
            risk_score += 7
    risk_score = min(92, risk_score + length % 12)

    sla_score = min(98, 72 + (100 - risk_score) // 4)
    oee = min(90, max(58, 60 + (length % 28)))
    throughput_gain = min(22, 6 + (length % 16))
    cost_reduction = min(20, 5 + (length % 14))
    energy_saving = min(18, 4 + (length % 12))
    inventory_turnover = min(14, 6 + (length % 9))

    return {
        "demand_index": demand_index,
        "energy_cost_index": energy_cost_index,
        "capacity_utilization": utilization,
        "defect_risk": defect_risk,
        "bottleneck": bottleneck,
        "risk_score": risk_score,
        "sla_score": sla_score,
        "oee": oee,
        "throughput_gain": throughput_gain,
        "cost_reduction": cost_reduction,
        "energy_saving": energy_saving,
        "inventory_turnover": inventory_turnover,
    }


def build_state_memory(scenario_text, tool_data):
    return (
        "### 产线状态记忆快照\n\n"
        f"- 当前瓶颈：{tool_data['bottleneck']}\n"
        f"- 产能利用率：{tool_data['capacity_utilization']}%\n"
        f"- 缺陷风险指数：{tool_data['defect_risk']}\n"
        f"- OEE 综合效率：{tool_data['oee']}%\n"
        f"- 能耗成本指数：{tool_data['energy_cost_index']}\n\n"
        "### 可复用资产\n\n"
        "- 产线瓶颈诊断模板\n"
        "- 工序 KPI 看板结构\n"
        "- 质量验收与复盘清单"
    )


def build_acceptance(tool_data):
    score = round(
        (tool_data["sla_score"] * 0.5)
        + (100 - tool_data["defect_risk"]) * 0.3
        + (100 - tool_data["risk_score"]) * 0.2,
        1,
    )
    verdict = "通过" if score >= 75 else "需整改"
    return {
        "acceptance_score": score,
        "verdict": verdict,
        "key_checks": [
            "瓶颈段产能提升是否达到目标",
            "能耗与成本下降是否有数据闭环",
            "质量缺陷率是否稳定下降",
            "异常处理与复盘是否形成闭环机制",
        ],
    }

def derive_risk_level(score):
    if score <= 35:
        return "低"
    if score <= 70:
        return "中"
    return "高"


def build_insight(tool_data):
    risk_level = derive_risk_level(tool_data["risk_score"])
    gaps = []
    if tool_data["defect_risk"] >= 60:
        gaps.append("质量波动偏高")
    if tool_data["capacity_utilization"] < 75:
        gaps.append("产能利用率偏低")
    if tool_data["energy_cost_index"] > 0.75:
        gaps.append("能耗成本偏高")
    if not gaps:
        gaps.append("暂无明显短板")
    return {
        "risk_level": risk_level,
        "gaps": gaps,
        "roi_hint": f"预计成本下降 {tool_data['cost_reduction']}%，产能提升 {tool_data['throughput_gain']}%。",
    }


def build_action_plan(tool_data):
    return (
        "### 执行路线图\n\n"
        "#### 第 1 周：数据基线建立\n"
        "- 梳理瓶颈工序与关键设备状态\n"
        "- 建立良率、能耗、停机与交付基线\n\n"
        "#### 第 2-4 周：瓶颈突破与协同排产\n"
        f"- 重点优化工序：{tool_data['bottleneck']}\n"
        "- 建立补料触发规则与异常响应流程\n"
        "- 引入批次追溯与质量预警\n\n"
        "#### 第 5-8 周：质量闭环与节能优化\n"
        "- 固化标准作业与质量验收清单\n"
        "- 建立节能计划与设备维护节奏\n\n"
        "#### 第 9-12 周：规模化复制与复盘\n"
        "- 复制闭环方案到相似产线\n"
        "- 形成复盘报告与资产沉淀"
    )


def run_factory_workflow(scenario_text, extra_hint=None):
    system_prompt = (
        "你是产线闭环运营智能体，负责从推理决策到工具行动、验收与复盘。"
        "输出内容必须是可执行的中文 Markdown。"
    )
    hint = extra_hint or "请给出可落地的改造步骤与量化指标。"

    reasoning = call_llm(
        [
            {"role": "system", "content": system_prompt},
            {
                "role": "user",
                "content": f"场景：{scenario_text}\n要求：{hint}\n请输出决策推理与行动路径。",
            },
        ]
    )

    tool_data = simulate_factory_tools(scenario_text)
    tool_payload = json.dumps(tool_data, ensure_ascii=False, indent=2)

    state_memory = build_state_memory(scenario_text, tool_data)
    acceptance = build_acceptance(tool_data)
    action_plan = build_action_plan(tool_data)
    acceptance_md = (
        "### 验收与校验\n\n"
        f"- 综合得分：{acceptance['acceptance_score']}\n"
        f"- 验收结论：{acceptance['verdict']}\n\n"
        "#### 关键校验点\n\n"
        + "\n".join([f"- {c}" for c in acceptance["key_checks"]])
    )

    iteration = call_llm(
        [
            {"role": "system", "content": system_prompt},
            {
                "role": "user",
                "content": (
                    f"基于以下产线工具结果与验收结论，输出迭代与复盘计划：\n\n"
                    f"工具结果：{tool_payload}\n\n"
                    f"验收结论：{acceptance['verdict']}\n"
                ),
            },
        ]
    )

    insight = build_insight(tool_data)
    return {
        "steps": [
            {
                "step_order": 1,
                "role": "决策官",
                "step_type": "reasoning",
                "content": reasoning,
            },
            {
                "step_order": 2,
                "role": "计划官",
                "step_type": "plan",
                "content": action_plan,
            },
            {
                "step_order": 3,
                "role": "工具引擎",
                "step_type": "tool_action",
                "content": tool_payload,
            },
            {
                "step_order": 4,
                "role": "状态记忆",
                "step_type": "state_memory",
                "content": state_memory,
            },
            {
                "step_order": 5,
                "role": "验收官",
                "step_type": "verification",
                "content": acceptance_md,
            },
            {
                "step_order": 6,
                "role": "复盘官",
                "step_type": "iteration",
                "content": iteration,
            },
        ],
        "tool_data": tool_data,
        "acceptance": acceptance,
        "insight": insight,
        "state_memory": state_memory,
    }


@app.route("/")
def index():
    return render_template("index.html")


@app.route("/api/sessions", methods=["GET"])
def list_sessions():
    db = get_db()
    rows = db.execute(
        "SELECT id, name, scenario, created_at FROM sessions ORDER BY id DESC LIMIT 50"
    ).fetchall()
    sessions = [
        {
            "id": r["id"],
            "name": r["name"],
            "scenario": r["scenario"],
            "created_at": r["created_at"],
        }
        for r in rows
    ]
    return jsonify({"sessions": sessions})


@app.route("/api/templates", methods=["GET"])
def list_templates():
    return jsonify({"templates": DEFAULT_TEMPLATES})


@app.route("/api/overview", methods=["GET"])
def overview():
    db = get_db()
    session_count = db.execute("SELECT COUNT(*) AS c FROM sessions").fetchone()["c"]
    asset_count = db.execute("SELECT COUNT(*) AS c FROM assets").fetchone()["c"]
    chat_count = db.execute("SELECT COUNT(*) AS c FROM chats").fetchone()["c"]
    latest_tool = db.execute(
        "SELECT content FROM steps WHERE step_type = 'tool_action' ORDER BY id DESC LIMIT 1"
    ).fetchone()
    tool_data = {}
    if latest_tool:
        try:
            tool_data = json.loads(latest_tool["content"])
        except Exception:
            tool_data = {}
    return jsonify(
        {
            "session_count": session_count,
            "asset_count": asset_count,
            "chat_count": chat_count,
            "latest_tool": tool_data,
        }
    )


@app.route("/api/sessions/<int:session_id>", methods=["GET"])
def get_session_detail(session_id):
    db = get_db()
    steps = db.execute(
        """
        SELECT step_order, role, step_type, content, created_at
        FROM steps WHERE session_id = ?
        ORDER BY step_order ASC
        """,
        (session_id,),
    ).fetchall()
    step_list = [
        {
            "step_order": s["step_order"],
            "role": s["role"],
            "step_type": s["step_type"],
            "content": s["content"],
            "created_at": s["created_at"],
        }
        for s in steps
    ]
    return jsonify({"steps": step_list})


@app.route("/api/run", methods=["POST"])
def run_workflow():
    try:
        data = request.get_json(force=True)
    except Exception:
        return jsonify({"error": "请求体必须为 JSON"}), 400

    scenario = (data or {}).get("scenario", "").strip()
    name = (data or {}).get("name", "").strip() or "产线闭环会话"
    hint = (data or {}).get("hint", "").strip()

    if not scenario:
        return jsonify({"error": "请提供场景描述"}), 400

    db = get_db()
    cursor = db.execute(
        "INSERT INTO sessions (name, scenario) VALUES (?, ?)", (name, scenario)
    )
    session_id = cursor.lastrowid

    result = run_factory_workflow(scenario, extra_hint=hint)
    for s in result["steps"]:
        db.execute(
            """
            INSERT INTO steps (session_id, step_order, role, step_type, content)
            VALUES (?, ?, ?, ?, ?)
            """,
            (session_id, s["step_order"], s["role"], s["step_type"], s["content"]),
        )
    if result["acceptance"]["verdict"] == "通过":
        db.execute(
            "INSERT INTO assets (title, category, content) VALUES (?, ?, ?)",
            (
                f"{name} · 闭环方案摘要",
                "闭环资产",
                result["state_memory"],
            ),
        )
    db.commit()
    return jsonify(
        {
            "session_id": session_id,
            "tool_data": result["tool_data"],
            "acceptance": result["acceptance"],
            "insight": result["insight"],
        }
    ), 201


@app.route("/api/assets", methods=["GET"])
def list_assets():
    db = get_db()
    rows = db.execute(
        "SELECT id, title, category, content, created_at FROM assets ORDER BY id DESC"
    ).fetchall()
    assets = [
        {
            "id": r["id"],
            "title": r["title"],
            "category": r["category"],
            "content": r["content"],
            "created_at": r["created_at"],
        }
        for r in rows
    ]
    return jsonify({"assets": assets})


@app.route("/api/assets", methods=["POST"])
def create_asset():
    data = request.get_json(force=True)
    title = (data.get("title") or "").strip()
    category = (data.get("category") or "").strip()
    content = (data.get("content") or "").strip()

    if not title or not category or not content:
        return jsonify({"error": "标题、分类与内容不能为空"}), 400

    db = get_db()
    db.execute(
        "INSERT INTO assets (title, category, content) VALUES (?, ?, ?)",
        (title, category, content),
    )
    db.commit()
    return jsonify({"status": "ok"})


@app.route("/api/chats", methods=["GET"])
def list_chats():
    db = get_db()
    rows = db.execute(
        "SELECT role, content, created_at FROM chats ORDER BY id DESC LIMIT 50"
    ).fetchall()
    chats = [
        {"role": r["role"], "content": r["content"], "created_at": r["created_at"]}
        for r in rows[::-1]
    ]
    return jsonify({"messages": chats})


@app.route("/api/chat", methods=["POST"])
def chat():
    data = request.get_json(force=True)
    message = (data.get("message") or "").strip()
    if not message:
        return jsonify({"error": "消息不能为空"}), 400

    system_prompt = (
        "你是制造业产线运营顾问，擅长在成本、质量、交付与风险之间做平衡决策。"
        "你的回复必须是中文 Markdown。"
    )

    reply = call_llm(
        [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": message},
        ]
    )

    db = get_db()
    db.execute(
        "INSERT INTO chats (role, content) VALUES (?, ?)", ("user", message)
    )
    db.execute(
        "INSERT INTO chats (role, content) VALUES (?, ?)", ("assistant", reply)
    )
    db.commit()
    return jsonify({"reply": reply})


with app.app_context():
    init_db()


if __name__ == "__main__":
    port = int(os.getenv("PORT", "7868"))
    app.run(host="0.0.0.0", port=port, debug=False)