HebeCrawl：构建AI驱动的智能工作流系统深度实践

传统工作流：规则驱动 → 固定路径 → 机械执行
智能工作流：意图理解 → 动态决策 → 自适应执行

# 第一层：角色定义和任务边界
"""你是一个内容类型分析和URL提取助手。请分析用户提供的内容，判断其主要类型并提取URL。"""

# 第二层：判断规则的结构化描述
"""
判断规则：
1. **URL类型**：如果内容主要是URL链接（即使包含少量其他文字），应归类为URL
   - 单个或多个URL链接
   - URL + 简短描述文字
   - 复制粘贴时带来的少量无关文字 + URL
   - Markdown格式的链接和图片

2. **笔记类型**：如果内容主要是文字信息、笔记、想法等，应归类为笔记
   - 大段文字内容
   - 学习笔记、工作记录
   - 即使包含URL，但文字内容占主导
"""

# 第三层：技术实现要求
"""
URL提取要求：
- 仔细分析内容，提取所有完整的、有效的URL
- 支持各种格式：直接URL、Markdown链接[text](url)、Markdown图片![alt](url)
- 确保提取的URL是完整的，以http://或https://开头
- 移除URL中的多余字符（如末尾的括号、引号等）
- 只返回可以直接访问的URL
"""

# 第四层：输出格式约束
"""
请返回JSON格式：
{
  "type": "url" 或 "note",
  "urls": ["提取的所有完整URL列表"],
  "reason": "判断理由"
}
"""

# 第五层：示例驱动学习
"""
示例：
输入：[![](https://example.com/image.png)](https://example.com)
输出：{"type": "url", "urls": ["https://example.com/image.png", "https://example.com"], "reason": "检测到Markdown格式的图片和链接"}
"""

# 策略判断系统
"""
**总结策略判断：**
1. **深度总结**（适用于以下情况）：
   - 技术文档、教程、学术文章
   - 长篇分析、研究报告
   - 复杂的业务流程说明
   - 包含多个要点的详细内容

2. **简单概括**（适用于以下情况）：
   - 新闻资讯、简短文章
   - 单一主题的简单说明
   - 个人笔记、随想
   - 内容本身就比较简洁
"""

# 量化指标
"""
**总结要求：**
- 深度总结：200-300字，包含主要观点、关键细节、结论
- 简单概括：50-100字，提炼核心要点
- 保持客观准确，不添加个人观点
- 使用清晰的中文表达
- 如果是技术内容，保留重要的技术术语
"""

# 核心原则的优先级排序
"""
**核心原则：**
1. **完整保留原文内容** - 绝对不能删除、修改或省略任何原始信息
2. **先原文后总结** - 这是笔记保存，原文是最重要的
3. **优化可读性** - 通过格式化提升阅读体验
"""

# 具体操作的负面约束
"""
**格式化规则：**
1. **保持原文完整性**：
   - 所有原始文字必须完整保留
   - 不要改写或简化任何内容
   - 保持原作者的表达方式和语气

2. **结构化组织**：
   - 识别并创建合适的标题层级（# ## ###）
   - 将相关内容组织成段落
   - 使用列表（- 或 1.）整理要点
   - 用代码块（```）包装代码片段
   - 用引用块（>）标记重要观点
"""

# URL类型的上下文增强
if content_type == "url":
    title_info = f"\n\n原网页标题：{original_title}" if original_title else ""
    prompt = f"""你是一个文件命名助手。请根据网页内容生成一个简洁的文件名。{title_info}

要求：
1. 长度控制在10个字符以内（中文或英文）
2. 能够概括主要内容
3. 只返回文件名，不要其他内容
4. 使用中文或英文，避免特殊字符
5. 如果是技术文档，可以包含技术关键词
6. 优先考虑原网页标题的关键信息

示例：
- 新闻类：财经新闻、科技资讯
- 技术类：Python教程、API文档
- 生活类：美食攻略、旅游指南"""

# 笔记类型的主题提取
else:  # note
    prompt = """你是一个文件命名助手。请根据笔记内容生成一个简洁的文件名。
要求：
1. 长度控制在10个字符以内（中文或英文）
2. 能够概括主要内容
3. 只返回文件名，不要其他内容
4. 使用中文或英文，避免特殊字符
5. 突出重点主题

示例：
- 学习笔记：FastAPI学习、数据库设计
- 工作记录：项目总结、会议纪要
- 个人想法：读书心得、生活感悟"""

用户输入
    ↓
输入验证（长度、格式、安全性）
    ↓
AI内容分析（意图理解）
    ↓
决策分支
    ├─ URL分支
    │   ├─ 单URL处理
    │   └─ 多URL并发处理
    └─ 笔记分支
        └─ 格式化处理
    ↓
结果整合与输出

# 第一层：输入验证和预处理
async def process_content(request: ContentRequest):
    content = request.content.strip()

    # 多维度输入验证
    if not content:
        raise HTTPException(status_code=400, detail="内容不能为空")
    if len(content) > 10000:
        raise HTTPException(status_code=400, detail="内容过长，请控制在10000字符以内")

    # 第二层：AI内容分析
    content_analysis = await ai_service.judge_content_type(content)
    content_type = content_analysis["type"]
    urls = content_analysis["urls"]

    # 调试信息：完整的决策追踪
    print(f"🔍 AI 内容分析结果:")
    print(f"   类型: {content_type}")
    print(f"   URL 数量: {len(urls)}")
    print(f"   URL 列表: {urls}")
    print(f"   判断理由: {content_analysis.get('reason', '无')}")

if content_type == "url":
    if len(urls) == 1:
        # 单URL处理流水线
        url = urls[0]
        print(f"\n🔄 单个 URL 处理:")
        print(f"   URL: '{url}'")
        print(f"   URL 类型: {type(url)}")
        print(f"   URL 长度: {len(url)}")
        print(f"   URL repr: {repr(url)}")

        # 处理链：抓取 → 总结 → 命名 → 保存
        scrape_result = await firecrawl_service.scrape_url(url)

        if scrape_result["success"]:
            summary = await ai_service.summarize_content(scrape_result["markdown"])
            original_title = scrape_result.get("metadata", {}).get("title", "")
            smart_name = await ai_service.generate_smart_filename(
                scrape_result["markdown"], "url", original_title
            )
            file_path = await FileService.save_url_content(
                scrape_result["markdown"], summary, url, None, smart_name
            )

    elif len(urls) > 1:
        # 多URL并发处理流水线
        print(f"\n🔄 开始处理 {len(urls)} 个 URL:")
        file_paths = []
        success_count = 0
        error_messages = []

        for i, url in enumerate(urls):
            print(f"\n📋 处理第 {i+1}/{len(urls)} 个 URL:")

            try:
                # 独立处理每个URL，错误隔离
                scrape_result = await firecrawl_service.scrape_url(url)

                if scrape_result["success"]:
                    # 完整的处理流水线
                    summary = await ai_service.summarize_content(scrape_result["markdown"])
                    original_title = scrape_result.get("metadata", {}).get("title", "")
                    smart_name = await ai_service.generate_smart_filename(
                        scrape_result["markdown"], "url", original_title
                    )
                    file_path = await FileService.save_url_content(
                        scrape_result["markdown"], summary, url, None, smart_name
                    )

                    file_paths.append(file_path)
                    success_count += 1
                else:
                    error_messages.append(f"URL {i+1} 抓取失败: {scrape_result.get('error')}")

            except Exception as e:
                error_messages.append(f"URL {i+1} 处理异常: {str(e)}")

        # 结果聚合和反馈
        if success_count > 0:
            message = f"成功处理 {success_count}/{len(urls)} 个URL"
            if error_messages:
                message += f"，失败信息: {'; '.join(error_messages)}"

else:  # note分支
    print(f"\n📝 笔记内容处理:")
    print(f"   原始内容长度: {len(content)} 字符")

    # 先格式化（保留原文）
    formatted_content = await ai_service.format_note_content(content)
    print(f"   格式化完成")

    # 再生成总结
    summary = await ai_service.summarize_content(content)
    print(f"   总结生成完成")

    # 最后智能命名
    smart_name = await ai_service.generate_smart_filename(content, "note")
    print(f"   智能文件名: {smart_name}")

    # 保存完整内容（原文 + 格式化 + 总结）
    file_path = await FileService.save_note_content(
        content,           # 原始内容
        formatted_content, # 格式化内容
        summary,          # 智能总结
        None,
        smart_name
    )

# 笔记文件的标准模板
full_content = f"""# 笔记

**创建时间**: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}

## 内容总结
{summary}

---

## 原始内容
{formatted_content}

---

## 原文备份

"""

# 1. 标准图片上传接口
@app.post("/upload-image")
async def upload_image(file: UploadFile = File(...)):
    """标准的图片上传，包含完整的验证和处理"""

# 2. iOS快捷指令优化接口
@app.post("/ios-upload")
async def ios_upload_image(file: UploadFile = File(...)):
    """专为iOS快捷指令优化，支持更宽松的文件类型检测"""

# 3. 简化上传接口
@app.post("/simple-upload")
async def simple_upload_image(file: UploadFile = File(...)):
    """最简化的接口，返回简单JSON格式"""

# 4. 批量上传接口
@app.post("/multi-upload")
async def multi_upload_images(files: List[UploadFile] = File(...)):
    """支持一次上传多张图片"""

def detect_image_type(data: bytes) -> str:
    """通过文件头检测图片类型（比文件扩展名更安全）"""
    if data.startswith(b'\xff\xd8\xff'):
        return 'jpg'
    elif data.startswith(b'\x89PNG\r\n\x1a\n'):
        return 'png'
    elif data.startswith(b'GIF87a') or data.startswith(b'GIF89a'):
        return 'gif'
    elif data.startswith(b'RIFF') and b'WEBP' in data[:12]:
        return 'webp'
    elif data.startswith(b'\x00\x00\x01\x00') or data.startswith(b'\x00\x00\x02\x00'):
        return 'ico'
    elif data.startswith(b'BM'):
        return 'bmp'
    else:
        return None

def generate_filename(content_type: str, smart_name: str = None) -> str:
    """生成语义化的文件名"""
    date_str = datetime.now().strftime("%Y%m%d")

    if smart_name:
        # AI生成的智能文件名
        return f"{date_str}_{smart_name}"
    else:
        # 默认文件名
        return f"{date_str}_{content_type}"

# 图片文件的特殊命名
filename = f"{date_str}_{time_str}_{safe_filename}"

hebecrawl/
├── mark/          # URL抓取内容（结构化网页信息）
├── note/          # 笔记内容（格式化的文本信息）
├── photo/         # 图片文件（多媒体资源）
└── static/        # 前端静态文件

# URL内容的标准模板
url_template = f"""# 网页内容摘要

**URL**: {url}
**抓取时间**: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}

## 内容总结
{summary}

---

## 原始内容
{markdown_content}
"""

# 笔记内容的标准模板
note_template = f"""# 笔记

**创建时间**: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")}

## 内容总结
{summary}

---

## 原始内容
{formatted_content}

---

## 原文备份

"""

class AIConfigManager:
    """AI配置管理器 - 支持主配置和备份配置自动切换"""

    def __init__(self):
        # 主配置（从 .env 读取）
        self.primary_config = {
            "api_key": os.getenv("AI_API_KEY"),
            "base_url": os.getenv("AI_BASE_URL", "https://api.siliconflow.cn/v1"),
            "model": os.getenv("AI_MODEL", "deepseek-ai/DeepSeek-V3")
        }

        # 备份配置（从 .env 读取备份变量）
        self.backup_config = {
            "api_key": os.getenv("AI_BACKUP_API_KEY"),
            "base_url": os.getenv("AI_BACKUP_BASE_URL", "https://api.siliconflow.cn/v1"),
            "model": os.getenv("AI_BACKUP_MODEL", "deepseek-ai/DeepSeek-V3")
        }

        # 当前使用的配置
        self.current_config = self.primary_config.copy()
        self.using_backup = False

async def test_config(self, config):
    """测试配置是否可用"""
    try:
        client = openai.OpenAI(
            api_key=config["api_key"],
            base_url=config["base_url"]
        )

        # 轻量级测试请求
        def sync_test():
            return client.chat.completions.create(
                model=config["model"],
                messages=[{"role": "user", "content": "test"}],
                max_tokens=5,
                timeout=10  # 快速超时检测
            )

        await asyncio.to_thread(sync_test)
        return True
    except Exception as e:
        print(f"❌ 配置测试失败: {str(e)}")
        return False

async def ensure_working_config(self):
    """确保使用可工作的配置"""
    # 如果当前使用主配置，先测试主配置
    if not self.using_backup:
        print("🔍 测试主配置...")
        if await self.test_config(self.primary_config):
            print("✅ 主配置正常")
            return self.current_config
        else:
            print("❌ 主配置失败，切换到备份配置")
            self.current_config = self.backup_config.copy()
            self.using_backup = True

    # 测试备份配置
    print("🔍 测试备份配置...")
    if await self.test_config(self.backup_config):
        print("✅ 备份配置正常")
        return self.current_config
    else:
        print("❌ 备份配置也失败")
        raise Exception("所有AI配置都无法使用")

async def _call_with_fallback(self, sync_call_func, operation_name="AI调用"):
    """带备份切换的AI调用"""
    max_retries = 2

    for attempt in range(max_retries):
        try:
            # 确保使用可工作的配置
            if attempt > 0:
                await self.config_manager.ensure_working_config()
                self._init_client()  # 重新初始化客户端

            # 执行调用
            response = await asyncio.to_thread(sync_call_func)
            return response

        except Exception as e:
            print(f"❌ {operation_name} 失败 (尝试 {attempt + 1}/{max_retries}): {str(e)}")

            if attempt < max_retries - 1:
                # 不是最后一次尝试，切换配置
                if not self.config_manager.using_backup:
                    print("🔄 切换到备份配置重试...")
                    self.config_manager.switch_to_backup()
                else:
                    print("🔄 切换到主配置重试...")
                    self.config_manager.switch_to_primary()
            else:
                # 最后一次尝试也失败了
                raise Exception(f"{operation_name} 所有配置都失败: {str(e)}")

# 在具体AI调用中的使用
async def judge_content_type(self, content: str) -> dict:
    def sync_call():
        return self.client.chat.completions.create(
            model=self.config_manager.get_current_config()["model"],
            messages=[...],  # 具体的prompt
            max_tokens=300,
            temperature=0.1
        )

    # 使用带容错的调用
    response = await self._call_with_fallback(sync_call, "内容类型判断")
    return self._parse_response(response)

async def judge_content_type(self, content: str) -> dict:
    try:
        # 第一层：LLM智能分析
        response = await self._call_with_fallback(sync_call, "内容类型判断")
        result_text = response.choices[0].message.content.strip()

        try:
            result = json.loads(result_text)
            # LLM分析成功，进行URL验证
            if result.get("type") == "url" and "urls" in result:
                original_urls = result["urls"]
                valid_urls = self.simple_validate_urls(original_urls)
                result["urls"] = valid_urls
            return result

        except json.JSONDecodeError:
            # 第二层：JSON解析失败，使用正则表达式备用
            print(f"❌ JSON 解析失败，使用备用逻辑...")
            url_pattern = r'https?://[^\s]+'
            urls = re.findall(url_pattern, content)
            valid_urls = self.simple_validate_urls(urls)

            if valid_urls:
                return {"type": "url", "urls": valid_urls, "reason": "JSON解析失败，使用备用逻辑检测到URL"}
            else:
                return {"type": "note", "urls": [], "reason": "JSON解析失败，备用逻辑未检测到有效URL，判断为笔记"}

    except Exception as e:
        # 第三层：AI调用完全失败，纯正则表达式处理
        print(f"❌ AI 判断异常: {e}，使用最终备用逻辑...")
        url_pattern = r'https?://[^\s]+'
        urls = re.findall(url_pattern, content)
        valid_urls = self.simple_validate_urls(urls)

        if valid_urls:
            return {"type": "url", "urls": valid_urls, "reason": f"AI判断失败，使用备用逻辑: {str(e)}"}
        else:
            return {"type": "note", "urls": [], "reason": f"AI判断失败，使用备用逻辑: {str(e)}"}

def simple_validate_urls(self, urls):
    """简单验证 LLM 提取的 URL 列表"""
    valid_urls = []
    for url in urls:
        if not url or not isinstance(url, str):
            continue

        # 基本清理
        clean_url = url.strip()
        print(f"🔧 验证 URL: '{clean_url}'")

        # 基本格式验证
        if not clean_url.startswith(('http://', 'https://')):
            print(f"❌ 无效协议: {clean_url}")
            continue

        # 长度验证
        if len(clean_url) < 10:
            print(f"❌ URL 太短: {clean_url}")
            continue

        # 使用 urlparse 验证
        try:
            parsed = urlparse(clean_url)
            if parsed.netloc and parsed.scheme in ['http', 'https'] and '.' in parsed.netloc:
                valid_urls.append(clean_url)
                print(f"✅ 有效 URL: {clean_url}")
            else:
                print(f"❌ 无效 URL: {clean_url}")
        except Exception as e:
            print(f"❌ URL 解析异常: {clean_url} - {e}")

    print(f"🎯 验证结果: {len(valid_urls)}/{len(urls)} 个有效 URL")
    return valid_urls

// 智能提交处理
async function submitContent() {
    const content = document.getElementById('content').value.trim();

    // 前端基础验证
    if (!content) {
        showStatus('请输入内容', 'error');
        return;
    }

    if (content.length > 10000) {
        showStatus('内容过长，请控制在10000字符以内', 'error');
        return;
    }

    // 显示处理状态
    showStatus('正在智能分析内容类型...', 'info');

    try {
        const response = await fetch('/process-content', {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
            },
            body: JSON.stringify({ content: content })
        });

        const result = await response.json();

        if (result.success) {
            // 成功处理
            showStatus(`处理成功！文件已保存到: ${result.file_path}`, 'success');

            // 如果是多文件，显示详细信息
            if (result.file_paths && result.file_paths.length > 1) {
                const fileList = result.file_paths.map(path => `• ${path}`).join('\n');
                showStatus(`${result.message}\n\n保存的文件：\n${fileList}`, 'success');
            }
        } else {
            showStatus(`处理失败: ${result.message}`, 'error');
        }

    } catch (error) {
        showStatus(`网络错误: ${error.message}`, 'error');
    }
}

// 状态显示的用户体验优化
function showStatus(message, type = 'info') {
    const statusDiv = document.getElementById('status');
    const timestamp = new Date().toLocaleTimeString();

    // 根据类型设置不同的图标和样式
    const icons = {
        'info': '🔄',
        'success': '✅',
        'error': '❌',
        'warning': '⚠️'
    };

    statusDiv.innerHTML = `
        <div class="status-${type}">
            ${icons[type]} ${message}
            <small class="timestamp">${timestamp}</small>
        </div>
    `;

    // 自动滚动到状态区域
    statusDiv.scrollIntoView({ behavior: 'smooth' });

    // 成功状态自动清除
    if (type === 'success') {
        setTimeout(() => {
            statusDiv.innerHTML = '';
        }, 5000);
    }
}

// 拖拽上传的完整实现
function setupDragAndDrop() {
    const dropZone = document.getElementById('imageUpload');

    // 拖拽进入
    dropZone.addEventListener('dragover', (e) => {
        e.preventDefault();
        dropZone.classList.add('drag-over');
        dropZone.style.borderColor = '#007bff';
        dropZone.style.backgroundColor = '#f8f9fa';
    });

    // 拖拽离开
    dropZone.addEventListener('dragleave', (e) => {
        e.preventDefault();
        dropZone.classList.remove('drag-over');
        dropZone.style.borderColor = '#ced4da';
        dropZone.style.backgroundColor = 'white';
    });

    // 文件放下
    dropZone.addEventListener('drop', (e) => {
        e.preventDefault();
        dropZone.classList.remove('drag-over');
        dropZone.style.borderColor = '#ced4da';
        dropZone.style.backgroundColor = 'white';

        const files = e.dataTransfer.files;
        if (files.length > 0) {
            handleImageUpload(files[0]);
        }
    });
}

// 统一的图片上传处理
async function handleImageUpload(file) {
    // 客户端验证
    if (!file.type.startsWith('image/')) {
        showStatus('请选择图片文件', 'error');
        return;
    }

    if (file.size > 10 * 1024 * 1024) {
        showStatus('图片大小不能超过10MB', 'error');
        return;
    }

    showStatus('正在上传图片...', 'info');

    const formData = new FormData();
    formData.append('file', file);

    try {
        const response = await fetch('/upload-image', {
            method: 'POST',
            body: formData
        });

        const result = await response.json();

        if (result.success) {
            showStatus(`图片上传成功！保存路径: ${result.file_path}`, 'success');
        } else {
            showStatus(`上传失败: ${result.message}`, 'error');
        }

    } catch (error) {
        showStatus(`上传错误: ${error.message}`, 'error');
    }
}

/* 基础移动端样式 */
.container {
    max-width: 800px;
    margin: 0 auto;
    padding: 20px;
    font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', system-ui, sans-serif;
}

.form-group {
    margin-bottom: 20px;
}

.form-control {
    width: 100%;
    padding: 12px;
    border: 2px solid #ced4da;
    border-radius: 8px;
    font-size: 16px; /* 防止iOS缩放 */
    transition: border-color 0.3s ease;
}

.form-control:focus {
    outline: none;
    border-color: #007bff;
    box-shadow: 0 0 0 0.2rem rgba(0, 123, 255, 0.25);
}

/* 响应式断点 */
@media (max-width: 768px) {
    .container {
        padding: 10px;
    }

    .btn {
        width: 100%;
        padding: 15px;
        font-size: 18px;
    }

    .status-message {
        font-size: 14px;
        word-break: break-word;
    }
}

@media (max-width: 480px) {
    .form-control {
        font-size: 16px;
        padding: 15px;
    }

    h1 {
        font-size: 24px;
        text-align: center;
    }
}

/* 状态反馈的视觉设计 */
.status-info {
    color: #17a2b8;
    background-color: #d1ecf1;
    border: 1px solid #bee5eb;
    padding: 12px;
    border-radius: 6px;
    margin: 10px 0;
}

.status-success {
    color: #155724;
    background-color: #d4edda;
    border: 1px solid #c3e6cb;
    padding: 12px;
    border-radius: 6px;
    margin: 10px 0;
}

.status-error {
    color: #721c24;
    background-color: #f8d7da;
    border: 1px solid #f5c6cb;
    padding: 12px;
    border-radius: 6px;
    margin: 10px 0;
}

/* 上传区域的交互效果 */
.upload-zone {
    border: 2px dashed #ced4da;
    border-radius: 8px;
    padding: 40px 20px;
    text-align: center;
    transition: all 0.3s ease;
    cursor: pointer;
}

.upload-zone:hover {
    border-color: #007bff;
    background-color: #f8f9fa;
}

.upload-zone.drag-over {
    border-color: #28a745;
    background-color: #d4edda;
    transform: scale(1.02);
}

#!/bin/bash
# start.sh - 智能环境检测和启动

echo "🚀 HebeCrawl 智能启动中..."

# 检测Docker环境
if command -v docker-compose &> /dev/null || command -v docker &> /dev/null; then
    echo "🐳 检测到Docker环境，使用容器化部署"

    # 检查docker-compose文件
    if [ -f "docker-compose.yml" ]; then
        echo "📋 使用 docker-compose 启动..."
        docker-compose down 2>/dev/null  # 清理旧容器
        docker-compose up --build -d

        if [ $? -eq 0 ]; then
            echo "✅ 服务启动成功!"
            echo "🌐 访问地址: http://localhost:3080"
            echo "📖 API文档: http://localhost:3080/docs"
        else
            echo "❌ Docker启动失败，尝试Python启动..."
            python_start
        fi
    else
        echo "❌ 未找到 docker-compose.yml 文件"
        python_start
    fi
else
    echo "🐍 未检测到Docker环境，使用Python启动"
    python_start
fi

# Python启动函数
python_start() {
    echo "🔍 检查Python环境..."

    if command -v python3 &> /dev/null; then
        PYTHON_CMD="python3"
    elif command -v python &> /dev/null; then
        PYTHON_CMD="python"
    else
        echo "❌ 未找到Python环境，请先安装Python"
        exit 1
    fi

    echo "📦 安装依赖..."
    $PYTHON_CMD -m pip install -r requirements.txt

    if [ $? -eq 0 ]; then
        echo "🎯 启动应用..."
        $PYTHON_CMD run.py
    else
        echo "❌ 依赖安装失败"
        exit 1
    fi
}

# 信号处理
trap 'echo "🛑 正在停止服务..."; docker-compose down 2>/dev/null; exit 0' INT TERM

# 多阶段构建优化镜像大小
FROM python:3.9-slim as builder

# 安装构建依赖
RUN apt-get update && apt-get install -y \
    gcc \
    curl \
    && rm -rf /var/lib/apt/lists/*

# 设置工作目录
WORKDIR /app

# 复制依赖文件
COPY requirements.txt .

# 安装Python依赖
RUN pip install --no-cache-dir --user -r requirements.txt

# 运行阶段
FROM python:3.9-slim

# 安装运行时依赖
RUN apt-get update && apt-get install -y \
    curl \
    && rm -rf /var/lib/apt/lists/* \
    && useradd --create-home --shell /bin/bash appuser

# 从构建阶段复制安装的包
COPY --from=builder /root/.local /home/appuser/.local

# 设置工作目录和用户
WORKDIR /app
USER appuser

# 复制应用代码
COPY --chown=appuser:appuser . .

# 创建必要的目录
RUN mkdir -p mark note photo static

# 设置环境变量
ENV PATH=/home/appuser/.local/bin:$PATH
ENV PYTHONPATH=/app

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=40s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["python", "main.py"]

# config.py 中的环境验证
def validate_config():
    """验证必要的配置项"""
    required_vars = [
        'AI_API_KEY',
        'AI_BACKUP_API_KEY',
        'FIRECRAWL_API_KEY'
    ]

    missing_vars = []
    for var in required_vars:
        if not os.getenv(var):
            missing_vars.append(var)

    if missing_vars:
        print(f"❌ 缺少必要的环境变量: {', '.join(missing_vars)}")
        print("请检查 .env 文件或环境变量设置")
        return False

    return True

# 在应用启动时调用
if not validate_config():
    exit(1)

import logging
import json
from datetime import datetime

class StructuredLogger:
    def __init__(self, name="hebecrawl"):
        self.logger = logging.getLogger(name)
        handler = logging.StreamHandler()

        # JSON格式的日志输出
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_ai_call(self, operation, content_preview, result, duration):
        """记录AI调用"""
        log_data = {
            "type": "ai_call",
            "operation": operation,
            "content_length": len(content_preview),
            "content_preview": content_preview[:100],
            "success": result.get("success", False),
            "duration_ms": duration,
            "timestamp": datetime.now().isoformat()
        }
        self.logger.info(json.dumps(log_data, ensure_ascii=False))

    def log_file_operation(self, operation, file_path, size=None):
        """记录文件操作"""
        log_data = {
            "type": "file_operation",
            "operation": operation,
            "file_path": file_path,
            "file_size": size,
            "timestamp": datetime.now().isoformat()
        }
        self.logger.info(json.dumps(log_data, ensure_ascii=False))

# 使用示例
logger = StructuredLogger()

import time
from contextlib import asynccontextmanager

@asynccontextmanager
async def monitor_performance(operation_name):
    """性能监控装饰器"""
    start_time = time.time()
    try:
        yield
    finally:
        duration = (time.time() - start_time) * 1000
        print(f"⏱️ {operation_name} 耗时: {duration:.2f}ms")

        # 记录慢查询
        if duration > 5000:  # 超过5秒
            logger.warning(f"慢操作检测: {operation_name} 耗时 {duration:.2f}ms")

# 使用示例
async def process_content_with_monitoring(content):
    async with monitor_performance("内容处理"):
        result = await ai_service.judge_content_type(content)
        return result

# 传统简单提示词
"判断这段内容是URL还是笔记"

# HebeCrawl的结构化提示词
"""
你是一个内容类型分析和URL提取助手。

判断规则：
1. **URL类型**：如果内容主要是URL链接（即使包含少量其他文字）
   - 单个或多个URL链接
   - URL + 简短描述文字
   - 复制粘贴带来的少量无关文字 + URL
   - Markdown格式的链接和图片

2. **笔记类型**：如果内容主要是文字信息、笔记、想法等
   - 大段文字内容
   - 学习笔记、工作记录
   - 即使包含URL，但文字内容占主导

URL提取要求：
- 仔细分析内容，提取所有完整的、有效的URL
- 支持各种格式：直接URL、Markdown链接[text](url)、Markdown图片![alt](url)
- 确保提取的URL是完整的，以http://或https://开头
- 移除URL中的多余字符（如末尾的括号、引号等）

请返回JSON格式：
{
  "type": "url" 或 "note",
  "urls": ["提取的所有完整URL列表"],
  "reason": "判断理由"
}

示例：
输入：[![](https://example.com/image.png)](https://example.com)
输出：{"type": "url", "urls": ["https://example.com/image.png", "https://example.com"], "reason": "检测到Markdown格式的图片和链接"}
"""

第一层：LLM智能分析
    ↓（JSON解析失败）
第二层：正则表达式备用
    ↓（AI调用失败）
第三层：纯规则兜底

# 统一的处理接口
async def process_content(content, content_type):
    """统一的内容处理接口"""

    # 第一步：内容分析
    analysis = await analyze_content(content, content_type)

    # 第二步：智能处理
    processed = await apply_intelligence(content, analysis)

    # 第三步：结果封装
    result = await package_result(processed, analysis)

    return result