Wecom Cs Automation

#!/bin/bash
# 企业微信客服自动化系统 - 一键安装脚本

set -e

echo "🚀 开始安装企业微信客服自动化系统..."
echo ""

# 颜色定义
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
RED='\033[0;31m'
NC='\033[0m' # No Color

# 检查依赖
check_dependencies() {
    echo -e "${YELLOW}检查系统依赖...${NC}"

    # Python 3
    if ! command -v python3 &> /dev/null; then
        echo -e "${RED}❌ 未安装 Python 3${NC}"
        exit 1
    fi

    # pip
    if ! command -v pip3 &> /dev/null; then
        echo -e "${RED}❌ 未安装 pip3${NC}"
        exit 1
    fi

    # PostgreSQL
    if ! command -v psql &> /dev/null; then
        echo -e "${YELLOW}⚠️  未安装 PostgreSQL，正在安装...${NC}"
        sudo apt update
        sudo apt install -y postgresql postgresql-contrib
    fi

    echo -e "${GREEN}✓ 依赖检查完成${NC}"
    echo ""
}

# 安装 Python 包
install_python_packages() {
    echo -e "${YELLOW}安装 Python 依赖...${NC}"

    pip3 install --user \
        openai \
        psycopg2-binary \
        requests \
        fastapi \
        uvicorn \
        python-dotenv \
        numpy || {
        echo -e "${RED}❌ Python 包安装失败${NC}"
        exit 1
    }

    echo -e "${GREEN}✓ Python 包安装完成${NC}"
    echo ""
}

# 配置数据库
setup_database() {
    echo -e "${YELLOW}配置数据库...${NC}"

    # 启动 PostgreSQL
    sudo service postgresql start

    # 创建数据库
    sudo -u postgres createdb wecom_kb 2>/dev/null || echo "数据库已存在"

    # 启用 pgvector 扩展
    echo -e "${YELLOW}检查 pgvector 扩展...${NC}"

    if ! sudo -u postgres psql -d wecom_kb -c "SELECT * FROM pg_extension WHERE extname = 'vector';" | grep -q vector; then
        echo -e "${YELLOW}安装 pgvector...${NC}"

        # 检测 PostgreSQL 版本
        PG_VERSION=$(sudo -u postgres psql -t -c "SELECT version()" | grep -oP 'PostgreSQL \K[0-9.]+' | head -1)
        PG_MAJOR=$(echo $PG_VERSION | cut -d. -f1)

        echo "检测到 PostgreSQL $PG_VERSION"

        # 安装 pgvector
        if [ ! -d "/tmp/pgvector" ]; then
            cd /tmp
            git clone --branch v0.5.1 https://github.com/pgvector/pgvector.git
            cd pgvector
            sudo apt install -y build-essential libpq-dev
            make
            sudo make install
        fi

        # 启用扩展
        sudo -u postgres psql -d wecom_kb -c "CREATE EXTENSION vector;"
        echo -e "${GREEN}✓ pgvector 扩展已启用${NC}"
    else
        echo -e "${GREEN}✓ pgvector 扩展已存在${NC}"
    fi

    # 初始化表结构
    echo -e "${YELLOW}初始化数据库表...${NC}"
    sudo -u postgres psql -d wecom_kb -f ~/clawd/skills/wecom-cs-automation/schema.sql

    echo -e "${GREEN}✓ 数据库配置完成${NC}"
    echo ""
}

# 配置环境变量
setup_env() {
    echo -e "${YELLOW}配置环境变量...${NC}"

    ENV_FILE="$HOME/clawd/skills/wecom-cs-automation/.env"

    if [ -f "$ENV_FILE" ]; then
        echo -e "${YELLOW}⚠️  .env 文件已存在，跳过${NC}"
    else
        cat > "$ENV_FILE" << EOF
# 企业微信配置
WECOM_CORP_ID=
WECOM_AGENT_ID=1000002
WECOM_AGENT_SECRET=
WECOM_TOKEN=
WECOM_ENCODING_AES_KEY=

# 数据库
KB_DB_URL=postgresql://postgres@localhost/wecom_kb

# LLM
LLM_PROVIDER=kimi
LLM_API_KEY=\$(pass show api/kimi)
LLM_API_BASE=https://api.moonshot.cn/v1
LLM_MODEL=moonshot-v1-8k

# 知识库搜索
KB_SIMILARITY_THRESHOLD=0.7
KB_TOP_K=3

# 人工介入
NOTIFICATION_ENABLED=true
NOTIFICATION_CHANNEL=telegram:REDACTED_TG_USER_ID
EOF

        echo -e "${GREEN}✓ 已创建 .env 模板${NC}"
        echo -e "${YELLOW}⚠️  请编辑 $ENV_FILE 填入企业微信配置${NC}"
    fi

    echo ""
}

# 导入示例知识库
import_sample_kb() {
    echo -e "${YELLOW}导入示例知识库...${NC}"

    KB_FILE="$HOME/clawd/skills/wecom-cs-automation/knowledge/sample.md"

    if [ -f "$KB_FILE" ]; then
        python3 ~/clawd/skills/wecom-cs-automation/scripts/import_kb.py \
            --input "$KB_FILE" \
            --category "示例知识" \
            --tags "示例,测试" \
            --key "$(pass show api/kimi)" || {
            echo -e "${RED}❌ 知识库导入失败（可能需要先配置 Kimi API Key）${NC}"
            echo "可以稍后手动导入："
            echo "python3 ~/clawd/skills/wecom-cs-automation/scripts/import_kb.py --input knowledge/sample.md --key YOUR_KIMI_KEY"
        }
        echo -e "${GREEN}✓ 知识库导入完成${NC}"
    else
        echo -e "${YELLOW}⚠️  示例知识库文件不存在${NC}"
    fi

    echo ""
}

# 打印后续步骤
print_next_steps() {
    echo -e "${GREEN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
    echo -e "${GREEN}✅ 安装完成！${NC}"
    echo -e "${GREEN}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
    echo ""
    echo "📋 后续步骤："
    echo ""
    echo "1️⃣  配置企业微信应用"
    echo "   - 登录企业微信管理后台"
    echo "   - 创建'微信客服'应用"
    echo "   - 配置回调地址：https://your-domain.com/wecom/callback"
    echo ""
    echo "2️⃣  填写环境变量"
    echo "   - 编辑 ~/clawd/skills/wecom-cs-automation/.env"
    echo "   - 填入 WECOM_CORP_ID、WECOM_AGENT_SECRET 等"
    echo ""
    echo "3️⃣  导入知识库"
    echo "   python3 ~/clawd/skills/wecom-cs-automation/scripts/import_kb.py \\"
    echo "     --input your_knowledge.md \\"
    echo "     --category \"常见问题\" \\"
    echo "     --tags \"售后,FAQ\""
    echo ""
    echo "4️⃣  测试功能"
    echo "   # 测试知识库搜索"
    echo "   python3 ~/clawd/skills/wecom-cs-automation/scripts/search_kb.py \"如何退款？\""
    echo ""
    echo "   # 测试新好友处理"
    echo "   python3 ~/clawd/skills/wecom-cs-automation/workflows/on_friend_add.py \\"
    echo "     --user-id test_user --name \"测试用户\""
    echo ""
    echo "   # 测试问答"
    echo "   python3 ~/clawd/skills/wecom-cs-automation/workflows/answer_question.py \\"
    echo "     --user-id test_user --question \"如何退货？\""
    echo ""
    echo "5️⃣  启动回调服务"
    echo "   uvicorn ~/clawd/skills/wecom-cs-automation/server/main.py:app \\"
    echo "     --host 0.0.0.0 --port 8000"
    echo ""
    echo "📚 更多信息："
    echo "   cat ~/clawd/skills/wecom-cs-automation/SKILL.md"
    echo ""
}

# 主流程
main() {
    check_dependencies
    install_python_packages
    setup_database
    setup_env
    import_sample_kb
    print_next_steps
}

# 运行安装
main

-- 企业微信客服自动化系统 - 数据库表结构

-- 知识库片段表
CREATE TABLE IF NOT EXISTS knowledge_chunks (
    id SERIAL PRIMARY KEY,
    content TEXT NOT NULL,
    embedding vector(1024),  -- Kimi 嵌入维度
    category VARCHAR(50),
    tags TEXT[],
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- 未解决问题记录
CREATE TABLE IF NOT EXISTS unknown_questions (
    id SERIAL PRIMARY KEY,
    user_id VARCHAR(100),
    user_name VARCHAR(100),
    question TEXT NOT NULL,
    conversation_context JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

-- 对话历史
CREATE TABLE IF NOT EXISTS conversation_history (
    id SERIAL PRIMARY KEY,
    user_id VARCHAR(100),
    role VARCHAR(20) NOT NULL,  -- user/assistant/system
    content TEXT NOT NULL,
    msg_id VARCHAR(100),
    created_at TIMESTAMP DEFAULT NOW()
);

-- 人工介入记录
CREATE TABLE IF NOT EXISTS escalation_log (
    id SERIAL PRIMARY KEY,
    user_id VARCHAR(100),
    user_name VARCHAR(100),
    reason TEXT,
    question TEXT,
    notified BOOLEAN DEFAULT FALSE,
    resolved BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT NOW(),
    resolved_at TIMESTAMP
);

-- 用户信息表
CREATE TABLE IF NOT EXISTS users (
    id SERIAL PRIMARY KEY,
    external_userid VARCHAR(100) UNIQUE,
    name VARCHAR(100),
    avatar TEXT,
    tags TEXT[],
    first_contact TIMESTAMP DEFAULT NOW(),
    last_contact TIMESTAMP DEFAULT NOW(),
    total_messages INTEGER DEFAULT 0
);

-- 索引优化
CREATE INDEX idx_chunks_category ON knowledge_chunks(category);
CREATE INDEX idx_chunks_tags ON knowledge_chunks USING GIN(tags);
CREATE INDEX idx_unknown_user ON unknown_questions(user_id);
CREATE INDEX idx_conv_user ON conversation_history(user_id);
CREATE INDEX idx_conv_created ON conversation_history(created_at DESC);
CREATE INDEX idx_users_external ON users(external_userid);
CREATE INDEX idx_users_tags ON users USING GIN(tags);

-- 向量相似度搜索索引（HNSW 算法）
CREATE INDEX idx_chunks_embedding ON knowledge_chunks
    USING hnsw (embedding vector_cosine_ops)
    WITH (m = 16, ef_construction = 64);

-- 更新时间触发器
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = NOW();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trigger_knowledge_chunks_updated
    BEFORE UPDATE ON knowledge_chunks
    FOR EACH ROW
    EXECUTE FUNCTION update_updated_at();

#!/usr/bin/env python3
"""
企业微信客服知识库导入工具
支持 Markdown 文件导入，自动切片并向量化
"""

import os
import sys
import json
import psycopg2
from psycopg2.extras import execute_values
from pathlib import Path
import re
from typing import List, Dict
import argparse

# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent.parent))

try:
    from openai import OpenAI
except ImportError:
    print("❌ 缺少依赖: pip install openai psycopg2-binary")
    sys.exit(1)


def read_markdown(file_path: str) -> str:
    """读取 Markdown 文件"""
    with open(file_path, 'r', encoding='utf-8') as f:
        return f.read()


def split_text(text: str, max_length: int = 500, overlap: int = 50) -> List[str]:
    """
    智能切分文本
    - 按段落切分
    - 保持语义完整
    - 添加重叠区域
    """
    # 按标题和段落分割
    sections = re.split(r'\n(#{1,3}\s.+)\n', text)

    chunks = []
    current_chunk = ""
    current_title = ""

    for section in sections:
        if section.startswith('#'):
            # 这是标题
            if current_chunk:
                chunks.append(current_title + "\n" + current_chunk.strip())
            current_title = section
            current_chunk = ""
        else:
            # 这是内容
            paragraphs = section.split('\n\n')
            for para in paragraphs:
                if len(current_chunk) + len(para) > max_length:
                    if current_chunk:
                        chunks.append((current_title + "\n" + current_chunk.strip()) if current_title else current_chunk.strip())
                    current_chunk = para + "\n\n"
                    current_title = ""
                else:
                    current_chunk += para + "\n\n"

    if current_chunk:
        chunks.append((current_title + "\n" + current_chunk.strip()) if current_title else current_chunk.strip())

    return chunks


def get_embedding(text: str, api_key: str, api_base: str = "https://api.moonshot.cn/v1") -> List[float]:
    """
    获取文本向量嵌入
    使用 Kimi API
    """
    client = OpenAI(
        api_key=api_key,
        base_url=api_base
    )

    try:
        response = client.embeddings.create(
            model="embedding-v1",  # Kimi 嵌入模型
            input=text
        )
        return response.data[0].embedding
    except Exception as e:
        print(f"⚠️  嵌入失败: {e}")
        return None


def import_to_db(
    chunks: List[str],
    category: str = "default",
    tags: List[str] = None,
    db_url: str = "postgresql://postgres@localhost/wecom_kb",
    api_key: str = None,
    refresh: bool = False
):
    """导入到数据库"""

    conn = psycopg2.connect(db_url)
    cur = conn.cursor()

    # 如果是刷新模式，先清空该分类
    if refresh:
        cur.execute("DELETE FROM knowledge_chunks WHERE category = %s", (category,))
        print(f"🗑️  已清空分类: {category}")

    # 批量插入
    inserted = 0
    skipped = 0

    for i, chunk in enumerate(chunks):
        print(f"📝 处理 {i+1}/{len(chunks)}... ", end='')

        # 检查是否已存在
        cur.execute("SELECT id FROM knowledge_chunks WHERE content = %s LIMIT 1", (chunk,))
        if cur.fetchone():
            print("✓ (已存在)")
            skipped += 1
            continue

        # 获取嵌入向量
        embedding = get_embedding(chunk, api_key)
        if not embedding:
            print("✗ (嵌入失败)")
            skipped += 1
            continue

        # 插入数据库
        cur.execute("""
            INSERT INTO knowledge_chunks (content, embedding, category, tags, metadata)
            VALUES (%s, %s, %s, %s, %s)
        """, (
            chunk,
            embedding,
            category,
            tags or [],
            json.dumps({"source": "import_script"})
        ))

        inserted += 1
        print("✓")

    conn.commit()
    cur.close()
    conn.close()

    print(f"\n✅ 导入完成！")
    print(f"   新增: {inserted}")
    print(f"   跳过: {skipped}")


def main():
    parser = argparse.ArgumentParser(description="导入知识库到 PostgreSQL")
    parser.add_argument("--input", "-i", required=True, help="Markdown 文件路径")
    parser.add_argument("--category", "-c", default="default", help="分类名称")
    parser.add_argument("--tags", "-t", nargs="*", default=[], help="标签列表")
    parser.add_argument("--db", default="postgresql://postgres@localhost/wecom_kb", help="数据库连接")
    parser.add_argument("--key", help="Kimi API Key (默认从环境变量读取)")
    parser.add_argument("--refresh", action="store_true", help="刷新模式（清空该分类后重新导入）")
    parser.add_argument("--rebuild", action="store_true", help="重建所有嵌入")

    args = parser.parse_args()

    # 获取 API Key
    api_key = args.key or os.environ.get("KIMI_API_KEY") or os.environ.get("LLM_API_KEY")
    if not api_key:
        print("❌ 请提供 API Key (--key 或环境变量)")
        sys.exit(1)

    # 读取文件
    if not os.path.exists(args.input):
        print(f"❌ 文件不存在: {args.input}")
        sys.exit(1)

    print(f"📖 读取文件: {args.input}")
    text = read_markdown(args.input)

    # 切分文本
    print(f"✂️  切分文本...")
    chunks = split_text(text)
    print(f"   共 {len(chunks)} 个片段")

    # 导入数据库
    print(f"💾 导入数据库...")
    import_to_db(
        chunks=chunks,
        category=args.category,
        tags=args.tags,
        db_url=args.db,
        api_key=api_key,
        refresh=args.refresh
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
企业微信 - 问答处理流程
基于知识库搜索 + LLM 生成答案
"""

import os
import sys
from pathlib import Path
from datetime import datetime
import logging
import json

# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent.parent))

try:
    import requests
    from openai import OpenAI
    from psycopg2 import extensions
    import numpy as np
except ImportError as e:
    print(f"❌ 缺少依赖: {e}")
    print("请安装: pip install openai psycopg2-binary numpy")
    sys.exit(1)


# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class KnowledgeBase:
    """知识库搜索"""

    def __init__(self):
        self.db_url = os.environ.get("KB_DB_URL", "postgresql://postgres@localhost/wecom_kb")
        self.top_k = int(os.environ.get("KB_TOP_K", "3"))
        self.similarity_threshold = float(os.environ.get("KB_SIMILARITY_THRESHOLD", "0.7"))

    def get_embedding(self, text: str, api_key: str, api_base: str = "https://api.moonshot.cn/v1"):
        """获取文本向量嵌入"""
        client = OpenAI(api_key=api_key, base_url=api_base)

        try:
            response = client.embeddings.create(
                model="embedding-v1",
                input=text
            )
            return response.data[0].embedding
        except Exception as e:
            logger.error(f"获取嵌入失败: {e}")
            return None

    def search(self, query: str, api_key: str = None, api_base: str = None) -> list:
        """
        搜索知识库

        Returns:
            List[dict]: [{
                "content": str,
                "similarity": float,
                "category": str,
                "metadata": dict
            }]
        """
        # 获取查询向量
        query_embedding = self.get_embedding(query, api_key, api_base)
        if not query_embedding:
            return []

        # 查询数据库
        conn = extensions.connect(self.db_url)
        cur = conn.cursor()

        try:
            # 向量相似度搜索
            cur.execute("""
                SELECT
                    content,
                    1 - (embedding <=> %s::vector) as similarity,
                    category,
                    metadata
                FROM knowledge_chunks
                WHERE embedding IS NOT NULL
                ORDER BY embedding <=> %s::vector
                LIMIT %s
            """, (query_embedding, query_embedding, self.top_k))

            results = []
            for content, similarity, category, metadata in cur.fetchall():
                results.append({
                    "content": content,
                    "similarity": float(similarity),
                    "category": category,
                    "metadata": json.loads(metadata) if metadata else {}
                })

            logger.info(f"搜索到 {len(results)} 个结果，最高相似度: {results[0]['similarity'] if results else 0:.2f}")
            return results

        except Exception as e:
            logger.error(f"搜索失败: {e}")
            return []
        finally:
            cur.close()
            conn.close()


class AnswerGenerator:
    """答案生成器"""

    def __init__(self):
        self.api_key = os.environ.get("LLM_API_KEY") or os.environ.get("KIMI_API_KEY")
        self.api_base = os.environ.get("LLM_API_BASE", "https://api.moonshot.cn/v1")
        self.model = os.environ.get("LLM_MODEL", "moonshot-v1-8k")
        self.client = OpenAI(api_key=self.api_key, base_url=self.api_base)

    def generate(self, query: str, kb_results: list) -> dict:
        """
        基于知识库生成答案

        Returns:
            dict: {
                "answer": str,
                "confidence": float,
                "sources": list,
                "need_escalation": bool
            }
        """
        if not kb_results:
            return {
                "answer": "抱歉，这个问题我暂时无法回答，已为您转接人工客服。",
                "confidence": 0.0,
                "sources": [],
                "need_escalation": True
            }

        # 构建上下文
        context_parts = []
        for i, result in enumerate(kb_results, 1):
            context_parts.append(f"[来源 {i}] {result['content'].strip()}")

        context = "\n\n".join(context_parts)

        # 最高相似度
        max_similarity = kb_results[0]["similarity"]

        # 构建提示词
        system_prompt = """你是一个专业的客服助手。请基于提供的知识库内容回答用户问题。

**重要规则**：
1. 优先使用知识库中的信息
2. 保持语言自然、专业、友好
3. 如果知识库中没有直接答案，诚实地说明无法回答
4. 可以适当引用多个来源
5. 保持回答简洁明了，通常不超过200字

**回答格式**：
- 直接给出答案
- 如有多个要点，用序号列出
- 必要时提供额外建议"""

        user_prompt = f"""知识库内容：
{context}

用户问题：{query}

请给出专业的回答。"""

        # 调用 LLM
        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt}
                ],
                temperature=0.7,
                max_tokens=500
            )

            answer = response.choices[0].message.content.strip()

            # 判断是否需要人工介入
            need_escalation = max_similarity < float(os.environ.get("KB_SIMILARITY_THRESHOLD", "0.7"))

            return {
                "answer": answer,
                "confidence": max_similarity,
                "sources": [r["category"] for r in kb_results],
                "need_escalation": need_escalation
            }

        except Exception as e:
            logger.error(f"生成答案失败: {e}")
            return {
                "answer": "抱歉，我遇到了一些技术问题，已为您转接人工客服。",
                "confidence": 0.0,
                "sources": [],
                "need_escalation": True
            }


class WecomClient:
    """企业微信客户端（简化版）"""

    def __init__(self):
        self.corp_id = os.environ.get("WECOM_CORP_ID")
        self.agent_id = os.environ.get("WECOM_AGENT_ID", "1000002")
        self.secret = os.environ.get("WECOM_AGENT_SECRET")
        self.access_token = None
        self.base_url = "https://qyapi.weixin.qq.com/cgi-bin"

    def send_text(self, external_userid: str, text: str) -> bool:
        """发送文本消息"""
        # 简化实现：只记录日志
        logger.info(f"📤 发送消息给 {external_userid}: {text[:50]}...")
        return True


def handle_question(
    external_userid: str,
    question: str,
    user_name: str = "客户"
) -> dict:
    """
    处理用户问题

    Returns:
        dict: {
            "success": bool,
            "answer": str,
            "escalated": bool,
            "confidence": float
        }
    """

    logger.info(f"❓ 收到问题: {question[:50]}... from {user_name} ({external_userid})")

    # 初始化组件
    kb = KnowledgeBase()
    generator = AnswerGenerator()
    wecom = WecomClient()

    # 1. 搜索知识库
    kb_results = kb.search(question, api_key=generator.api_key, api_base=generator.api_base)

    # 2. 生成答案
    result = generator.generate(question, kb_results)

    logger.info(f"✓ 生成答案 (置信度: {result['confidence']:.2f})")

    # 3. 发送回复
    if result["need_escalation"]:
        # 需要人工介入
        escalation_msg = f"""⏳ {result['answer']}

请您稍候，我立即为您联系人工客服。如有急事，可直接致电：400-XXX-XXXX"""

        wecom.send_text(external_userid, escalation_msg)

        logger.warning(f"⚠️  转人工: {user_name} - {question}")

        return {
            "success": True,
            "answer": escalation_msg,
            "escalated": True,
            "confidence": result["confidence"]
        }
    else:
        # 自动回复
        wecom.send_text(external_userid, result["answer"])

        logger.info(f"✅ 自动回复: {result['answer'][:50]}...")

        return {
            "success": True,
            "answer": result["answer"],
            "escalated": False,
            "confidence": result["confidence"]
        }


# 测试入口
if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="测试问答流程")
    parser.add_argument("--user-id", required=True, help="外部用户 ID")
    parser.add_argument("--question", required=True, help="用户问题")
    parser.add_argument("--name", default="测试用户", help="用户姓名")

    args = parser.parse_args()

    # 检查环境变量
    required_vars = ["KB_DB_URL", "LLM_API_KEY"]
    missing = [var for var in required_vars if not os.environ.get(var)]

    if missing:
        print(f"❌ 缺少环境变量: {', '.join(missing)}")
        sys.exit(1)

    # 执行
    result = handle_question(
        external_userid=args.user_id,
        question=args.question,
        user_name=args.name
    )

    print(f"\n📊 结果:")
    print(f"   成功: {result['success']}")
    print(f"   转人工: {result['escalated']}")
    print(f"   置信度: {result['confidence']:.2f}")
    print(f"   回复: {result['answer']}")

    sys.exit(0 if result["success"] else 1)

#!/usr/bin/env python3
"""
人工介入提醒模块
当 AI 无法回答时，通知人工客服
"""

import os
import sys
from pathlib import Path
from datetime import datetime
import logging
import json

# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent.parent))

try:
    import requests
    from psycopg2 import extensions
except ImportError as e:
    print(f"❌ 缺少依赖: {e}")
    sys.exit(1)


# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class NotificationChannel:
    """通知渠道"""

    def __init__(self):
        self.channel_config = os.environ.get("NOTIFICATION_CHANNEL", "telegram:REDACTED_TG_USER_ID")

    def parse_channel(self) -> tuple:
        """解析渠道配置: "telegram:REDACTED_TG_USER_ID" -> ("telegram", "REDACTED_TG_USER_ID")"""
        parts = self.channel_config.split(":")
        if len(parts) != 2:
            logger.error(f"无效的渠道配置: {self.channel_config}")
            return None, None
        return parts[0], parts[1]

    def send(self, message: str) -> bool:
        """发送通知"""
        channel, target = self.parse_channel()

        if not channel:
            return False

        if channel == "telegram":
            return self._send_telegram(target, message)
        elif channel == "feishu":
            return self._send_feishu(target, message)
        else:
            logger.error(f"不支持的渠道: {channel}")
            return False

    def _send_telegram(self, chat_id: str, message: str) -> bool:
        """发送 Telegram 通知"""
        token = os.environ.get("TELEGRAM_BOT_TOKEN")

        if not token:
            logger.error("缺少 TELEGRAM_BOT_TOKEN")
            return False

        url = f"https://api.telegram.org/bot{token}/sendMessage"
        payload = {
            "chat_id": chat_id,
            "text": message,
            "parse_mode": "Markdown"
        }

        try:
            resp = requests.post(url, json=payload, timeout=10)
            data = resp.json()

            if data.get("ok"):
                logger.info("✓ Telegram 通知发送成功")
                return True
            else:
                logger.error(f"✗ Telegram 通知失败: {data}")
                return False

        except Exception as e:
            logger.error(f"✗ Telegram 通知异常: {e}")
            return False

    def _send_feishu(self, webhook_url: str, message: str) -> bool:
        """发送飞书通知"""
        payload = {
            "msg_type": "text",
            "content": {"text": message}
        }

        try:
            resp = requests.post(webhook_url, json=payload, timeout=10)

            if resp.status_code == 200:
                logger.info("✓ 飞书通知发送成功")
                return True
            else:
                logger.error(f"✗ 飞书通知失败: {resp.text}")
                return False

        except Exception as e:
            logger.error(f"✗ 飞书通知异常: {e}")
            return False


class Database:
    """数据库操作"""

    def __init__(self):
        self.db_url = os.environ.get("KB_DB_URL", "postgresql://postgres@localhost/wecom_kb")

    def save_unknown_question(self, user_id: str, user_name: str, question: str, context: dict = None):
        """保存未解决的问题"""
        conn = extensions.connect(self.db_url)
        cur = conn.cursor()

        try:
            cur.execute("""
                INSERT INTO unknown_questions (user_id, user_name, question, conversation_context)
                VALUES (%s, %s, %s, %s)
            """, (user_id, user_name, question, json.dumps(context or {})))

            conn.commit()
            logger.info("✓ 已记录未解决问题")
            return True

        except Exception as e:
            logger.error(f"✗ 保存未解决问题失败: {e}")
            conn.rollback()
            return False
        finally:
            cur.close()
            conn.close()

    def log_escalation(self, user_id: str, user_name: str, question: str, reason: str = "低置信度"):
        """记录人工介入日志"""
        conn = extensions.connect(self.db_url)
        cur = conn.cursor()

        try:
            cur.execute("""
                INSERT INTO escalation_log (user_id, user_name, reason, question, notified)
                VALUES (%s, %s, %s, %s, %s)
            """, (user_id, user_name, reason, question, True))

            conn.commit()
            logger.info("✓ 已记录介入日志")
            return True

        except Exception as e:
            logger.error(f"✗ 记录介入日志失败: {e}")
            conn.rollback()
            return False
        finally:
            cur.close()
            conn.close()


def escalate_to_human(
    user_id: str,
    user_name: str,
    question: str,
    context: dict = None
) -> bool:
    """
    转接人工客服

    Returns:
        bool: 是否成功
    """

    logger.info(f"🚨 转人工: {user_name} ({user_id})")

    # 初始化组件
    notifier = NotificationChannel()
    db = Database()

    # 1. 构建通知消息
    notification = f"""🚨 *需要人工介入*

👤 用户：{user_name}
🆔 ID: {user_id}
❓ 问题：{question}
⏰ 时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}

请及时处理，避免用户等待过久。

---
由企业微信客服 AI 助手自动推送"""

    # 2. 发送通知
    notified = notifier.send(notification)

    # 3. 记录到数据库
    db.save_unknown_question(user_id, user_name, question, context)
    db.log_escalation(user_id, user_name, question, "低置信度")

    return notified


def get_escalation_stats() -> dict:
    """获取人工介入统计"""
    db_url = os.environ.get("KB_DB_URL", "postgresql://postgres@localhost/wecom_kb")
    conn = extensions.connect(db_url)
    cur = conn.cursor()

    try:
        # 今日介入次数
        cur.execute("""
            SELECT COUNT(*)
            FROM escalation_log
            WHERE DATE(created_at) = CURRENT_DATE
        """)
        today_count = cur.fetchone()[0]

        # 未解决问题 Top 5
        cur.execute("""
            SELECT question, COUNT(*) as count
            FROM unknown_questions
            WHERE created_at > NOW() - INTERVAL '7 days'
            GROUP BY question
            ORDER BY count DESC
            LIMIT 5
        """)
        top_questions = [(row[0][:30] + "...", row[1]) for row in cur.fetchall()]

        return {
            "today_escalations": today_count,
            "top_questions": top_questions
        }

    except Exception as e:
        logger.error(f"获取统计失败: {e}")
        return {}
    finally:
        cur.close()
        conn.close()


# 测试入口
if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="测试人工介入流程")
    parser.add_argument("--user-id", required=True, help="外部用户 ID")
    parser.add_argument("--name", default="测试用户", help="用户姓名")
    parser.add_argument("--question", required=True, help="用户问题")
    parser.add_argument("--stats", action="store_true", help="显示统计信息")

    args = parser.parse_args()

    if args.stats:
        stats = get_escalation_stats()
        print(f"\n📊 今日介入次数: {stats.get('today_escalations', 0)}")
        print("\n🔝 Top 未解决问题:")
        for q, c in stats.get('top_questions', []):
            print(f"   {c}x - {q}")
    else:
        # 执行转人工
        result = escalate_to_human(
            user_id=args.user_id,
            user_name=args.name,
            question=args.question
        )

        print(f"\n{'✅' if result else '❌'} 转人工{'成功' if result else '失败'}")

#!/usr/bin/env python3
"""
企业微信 - 新好友添加处理流程
自动通过、添加标签、发送欢迎消息
"""

import os
import sys
from pathlib import Path
from datetime import datetime
import logging

# 添加项目路径
sys.path.insert(0, str(Path(__file__).parent.parent))

try:
    import requests
    from psycopg2 import extensions
except ImportError as e:
    print(f"❌ 缺少依赖: {e}")
    sys.exit(1)


# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class WecomClient:
    """企业微信 API 客户端"""

    def __init__(self):
        self.corp_id = os.environ.get("WECOM_CORP_ID")
        self.agent_id = os.environ.get("WECOM_AGENT_ID", "1000002")
        self.secret = os.environ.get("WECOM_AGENT_SECRET")
        self.access_token = None
        self.base_url = "https://qyapi.weixin.qq.com/cgi-bin"

    def get_access_token(self) -> str:
        """获取 access_token"""
        if self.access_token:
            return self.access_token

        url = f"{self.base_url}/gettoken"
        params = {
            "corpid": self.corp_id,
            "corpsecret": self.secret
        }

        try:
            resp = requests.get(url, params=params, timeout=10)
            data = resp.json()

            if data.get("errcode") == 0:
                self.access_token = data["access_token"]
                logger.info("✓ 获取 access_token 成功")
                return self.access_token
            else:
                logger.error(f"✗ 获取 access_token 失败: {data}")
                return None

        except Exception as e:
            logger.error(f"✗ 获取 access_token 异常: {e}")
            return None

    def send_message(self, external_userid: str, text: str) -> bool:
        """发送文本消息"""
        token = self.get_access_token()
        if not token:
            return False

        url = f"{self.base_url}/kf/send_msg?access_token={token}"
        payload = {
            "external_userid": external_userid,
            "open_kfid": self.agent_id,
            "msgtype": "text",
            "text": {
                "content": text
            }
        }

        try:
            resp = requests.post(url, json=payload, timeout=10)
            data = resp.json()

            if data.get("errcode") == 0:
                logger.info(f"✓ 发送消息成功: {external_userid}")
                return True
            else:
                logger.error(f"✗ 发送消息失败: {data}")
                return False

        except Exception as e:
            logger.error(f"✗ 发送消息异常: {e}")
            return False

    def add_tag(self, external_userid: str, tag_name: str) -> bool:
        """添加用户标签"""
        # 这里需要调用企业微信标签管理 API
        # 简化实现，仅记录日志
        logger.info(f"🏷️  添加标签: {tag_name} -> {external_userid}")
        return True


class Database:
    """数据库操作"""

    def __init__(self):
        self.db_url = os.environ.get("KB_DB_URL", "postgresql://postgres@localhost/wecom_kb")
        self.conn = None

    def connect(self):
        """连接数据库"""
        try:
            self.conn = extensions.connect(self.db_url)
            logger.info("✓ 数据库连接成功")
            return True
        except Exception as e:
            logger.error(f"✗ 数据库连接失败: {e}")
            return False

    def save_user(self, external_userid: str, name: str, tags: list = None):
        """保存用户信息"""
        if not self.conn:
            if not self.connect():
                return False

        cur = self.conn.cursor()

        try:
            # 检查用户是否存在
            cur.execute("SELECT id FROM users WHERE external_userid = %s", (external_userid,))
            user_exists = cur.fetchone()

            if user_exists:
                # 更新
                cur.execute("""
                    UPDATE users
                    SET name = %s, tags = %s, last_contact = NOW(), total_messages = total_messages + 1
                    WHERE external_userid = %s
                """, (name, tags or [], external_userid))
                logger.info(f"✓ 更新用户: {name}")
            else:
                # 新增
                cur.execute("""
                    INSERT INTO users (external_userid, name, tags, first_contact, last_contact, total_messages)
                    VALUES (%s, %s, %s, NOW(), NOW(), 1)
                """, (external_userid, name, tags or ["新客户"]))
                logger.info(f"✓ 新增用户: {name}")

            self.conn.commit()
            return True

        except Exception as e:
            logger.error(f"✗ 保存用户失败: {e}")
            self.conn.rollback()
            return False
        finally:
            cur.close()


def send_welcome_message(wecom: WecomClient, external_userid: str, name: str) -> bool:
    """发送欢迎消息"""

    welcome_text = f"""👋 欢迎来到{name}！

我是智能客服小助手 🤖，可以帮您：

📋 查询订单状态
❓ 解答常见问题
🔄 处理售后问题
💰 查询物流信息

如需帮助，请直接回复问题。
如有复杂需求，我会转接人工客服为您服务。

如有急事，也可致电人工客服：400-XXX-XXXX"""

    return wecom.send_message(external_userid, welcome_text)


def handle_friend_add(
    external_userid: str,
    name: str = "客户",
    avatar: str = None,
    source: str = "未知"
) -> bool:
    """
    处理好友添加事件

    Args:
        external_userid: 外部联系人 UserID
        name: 用户姓名
        avatar: 头像 URL
        source: 来源渠道

    Returns:
        bool: 处理是否成功
    """

    logger.info(f"🆕 新好友添加: {name} ({external_userid}) from {source}")

    # 初始化客户端
    wecom = WecomClient()
    db = Database()

    # 1. 添加用户标签
    wecom.add_tag(external_userid, "新客户")

    # 2. 保存用户到数据库
    tags = ["新客户", f"来源:{source}"]
    db.save_user(external_userid, name, tags)

    # 3. 发送欢迎消息
    success = send_welcome_message(wecom, external_userid, name)

    if success:
        logger.info(f"✅ 新好友处理完成: {name}")
    else:
        logger.error(f"✗ 新好友处理失败: {name}")

    return success


# 测试入口
if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="测试新好友添加流程")
    parser.add_argument("--user-id", required=True, help="外部用户 ID")
    parser.add_argument("--name", default="测试用户", help="用户姓名")
    parser.add_argument("--source", default="测试", help="来源")

    args = parser.parse_args()

    # 检查环境变量
    required_vars = ["WECOM_CORP_ID", "WECOM_AGENT_SECRET"]
    missing = [var for var in required_vars if not os.environ.get(var)]

    if missing:
        print(f"❌ 缺少环境变量: {', '.join(missing)}")
        sys.exit(1)

    # 执行
    result = handle_friend_add(
        external_userid=args.user_id,
        name=args.name,
        source=args.source
    )

    sys.exit(0 if result else 1)