Content Collector

# Content Collector 环境变量配置
# 复制此文件为 .env 并填入实际值

# 飞书多维表格配置
FEISHU_BITABLE_APP_TOKEN=your_app_token_here
FEISHU_BITABLE_TABLE_ID=your_table_id_here

# 用户访问令牌（可选，OpenClaw 会自动管理）
# FEISHU_USER_ACCESS_TOKEN=your_user_token_here

# 本地缓存（去重记录等，环境相关）
.cache/

# 本地配置（飞书 Token、文档 ID 等）
.env
config.local.*

# Python
__pycache__/
*.pyc
*.pyo

# OS
.DS_Store
Thumbs.db

# IDE
.vscode/
.idea/

#!/usr/bin/env python3
"""
追加内容到飞书文档
支持：格式化输出、去重检查、序号自动递增
"""

import json
import sys
import os
import re
from datetime import datetime

def format_content_item(item: dict, index: int) -> str:
    """
    格式化单条内容为飞书文档格式
    
    输出 Markdown 格式，供 feishu_doc append 使用
    """
    platform = item.get('platform', '未知平台')
    author = item.get('author', '未知作者')
    title = item.get('title', '')
    content = item.get('content', '')
    url = item.get('url', '')
    created_at = item.get('created_at', '')
    summary = item.get('summary', '')
    keywords = item.get('keywords', [])
    reason = item.get('reason', '')
    stats = item.get('stats', {})
    
    # 构建互动数据字符串
    stats_str = ''
    if stats:
        parts = []
        if stats.get('likes'):
            parts.append(f"👍 {stats['likes']}")
        if stats.get('retweets'):
            parts.append(f"🔄 {stats['retweets']}")
        if stats.get('bookmarks'):
            parts.append(f"💾 {stats['bookmarks']}")
        if stats.get('views'):
            parts.append(f"👁️ {stats['views']}")
        if stats.get('comments'):
            parts.append(f"💬 {stats['comments']}")
        stats_str = ' | '.join(parts) if parts else '无数据'
    
    # 格式化时间
    time_str = created_at
    if isinstance(created_at, str) and len(created_at) > 10:
        try:
            dt = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
            time_str = dt.strftime('%Y-%m-%d %H:%M')
        except (ValueError, TypeError):
            pass
    
    # 构建关键词标签
    keywords_str = ', '.join(keywords) if keywords else '无'
    
    # 使用标题或内容前 30 字作为标题
    display_title = title if title else (content[:30] + '...' if len(content) > 30 else content)
    
    markdown = f"""### {index}. {display_title}

| 属性 | 内容 |
|:---|:---|
| **作者** | {author} |
| **平台** | {platform} |
| **发布时间** | {time_str} |
| **原文链接** | [查看原帖]({url}) |
| **互动数据** | {stats_str} |

**原文内容**：
> {content}

**AI 摘要**：
{summary if summary else '（待生成）'}

**关键词**：{keywords_str}

**为什么收藏**：
{reason if reason else '（待补充）'}

---

"""
    
    return markdown

def get_next_index(doc_content: str) -> int:
    """从文档内容中获取下一个序号"""
    if not doc_content:
        return 1
    
    # 查找所有 ### {数字}. 标题
    pattern = r'###\s+(\d+)\s*\.'
    matches = re.findall(pattern, doc_content)
    
    if matches:
        return max(int(m) for m in matches) + 1
    
    return 1

def main():
    if len(sys.argv) < 2:
        print("Usage: python3 append_to_feishu.py '<json_content>'", file=sys.stderr)
        print("  json_content: 要追加的内容 JSON 字符串", file=sys.stderr)
        sys.exit(1)
    
    # 解析输入
    try:
        content_json = sys.argv[1]
        item = json.loads(content_json)
    except json.JSONDecodeError as e:
        print(f"JSON 解析错误: {e}", file=sys.stderr)
        sys.exit(1)
    
    # 格式化输出
    # 注意：实际追加需要使用 feishu_doc 工具的 append 操作
    # 这里返回格式化后的 Markdown
    
    formatted = format_content_item(item, item.get('index', 1))
    
    result = {
        'success': True,
        'markdown': formatted,
        'item': item,
        'note': '使用 feishu_doc append 操作将此 Markdown 追加到文档'
    }
    
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == '__main__':
    main()

#!/bin/bash
#
# 内容收藏完整流程脚本
# 强制使用 save_to_bitable.py，禁止直接调用工具
#
# 用法:
#   ./collect_and_save.sh \
#       --url "https://x.com/i/status/..." \
#       --title "文章标题" \
#       --source "X/Twitter" \
#       --category "🔧工具推荐"

set -e

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
CONTENT_FILE="/tmp/collect_content_$$.md"

# 解析参数
URL=""
TITLE=""
SOURCE=""
CATEGORY=""

while [[ $# -gt 0 ]]; do
    case $1 in
        --url)
            URL="$2"
            shift 2
            ;;
        --title)
            TITLE="$2"
            shift 2
            ;;
        --source)
            SOURCE="$2"
            shift 2
            ;;
        --category)
            CATEGORY="$2"
            shift 2
            ;;
        *)
            echo "未知参数: $1"
            exit 1
            ;;
    esac
done

# 参数校验
if [[ -z "$URL" || -z "$TITLE" || -z "$SOURCE" || -z "$CATEGORY" ]]; then
    echo "错误: 缺少必需参数"
    echo "用法: $0 --url <url> --title <title> --source <source> --category <category>"
    exit 1
fi

echo "📝 步骤 1: 提取内容..."
python3 "${SCRIPT_DIR}/extract_content.py" "$URL" > "$CONTENT_FILE"

if [[ ! -s "$CONTENT_FILE" ]]; then
    echo "❌ 内容提取失败"
    rm -f "$CONTENT_FILE"
    exit 1
fi

echo "✅ 内容已保存到 $CONTENT_FILE"

echo "📝 步骤 2: 调用 save_to_bitable.py 写入..."
python3 "${SCRIPT_DIR}/save_to_bitable.py" \
    --title "$TITLE" \
    --source "$SOURCE" \
    --category "$CATEGORY" \
    --url "$URL" \
    --content-file "$CONTENT_FILE"

SAVE_RESULT=$?

# 清理临时文件
rm -f "$CONTENT_FILE"

if [[ $SAVE_RESULT -eq 0 ]]; then
    echo "✅ 收藏完成"
    
    echo "📝 步骤 3: 更新去重缓存..."
    python3 "${SCRIPT_DIR}/deduplicate.py" --add "$URL"
    
    exit 0
else
    echo "❌ 保存失败"
    exit 1
fi

#!/usr/bin/env python3
"""
Deduplication - 检查链接是否已存在
支持：飞书多维表格检查、本地缓存检查
"""

import json
import re
import os
import logging
from typing import Optional
from urllib.parse import urlparse, parse_qs, urlencode
from datetime import datetime, timedelta
from urllib.request import Request, urlopen
from urllib.error import URLError

logger = logging.getLogger(__name__)

# 域名别名映射：将变体域名统一到主域名
DOMAIN_ALIASES = {
    'twitter.com': 'x.com',
    'www.twitter.com': 'x.com',
    'mobile.twitter.com': 'x.com',
    'www.x.com': 'x.com',
    'm.okjike.com': 'okjike.com',
    'web.okjike.com': 'okjike.com',
    'www.okjike.com': 'okjike.com',
    'old.reddit.com': 'reddit.com',
    'www.reddit.com': 'reddit.com',
    'np.reddit.com': 'reddit.com',
    'i.reddit.com': 'reddit.com',
    'amp.reddit.com': 'reddit.com',
    'www.weixin.qq.com': 'mp.weixin.qq.com',
}

# 已知短链接域名
SHORT_URL_DOMAINS = {'t.co', 'bit.ly', 'tinyurl.com', 'goo.gl', 'ow.ly', 'is.gd', 'buff.ly'}

# 缓存文件路径
CACHE_FILE = os.path.join(os.path.dirname(__file__), '..', '.cache', 'collected_urls.json')

# 多维表格配置（从环境变量读取，或使用默认值）
BITABLE_APP_TOKEN = os.environ.get('FEISHU_BITABLE_APP_TOKEN') or os.environ.get('BITABLE_APP_TOKEN', '')
BITABLE_TABLE_ID = os.environ.get('FEISHU_BITABLE_TABLE_ID') or os.environ.get('BITABLE_TABLE_ID', '')

# 缓存策略
CACHE_TTL_DAYS = 30       # 缓存条目 30 天后过期
CACHE_MAX_ENTRIES = 1000   # 最多保留 1000 条，超出按最早添加淘汰

def ensure_cache_dir():
    """确保缓存目录存在"""
    cache_dir = os.path.dirname(CACHE_FILE)
    if not os.path.exists(cache_dir):
        os.makedirs(cache_dir)

def load_cache():
    """加载已收藏的 URL 缓存"""
    ensure_cache_dir()
    if os.path.exists(CACHE_FILE):
        with open(CACHE_FILE, 'r', encoding='utf-8') as f:
            cache = json.load(f)
        return _cleanup_cache(cache)
    return {}

def _cleanup_cache(cache: dict) -> dict:
    now = datetime.now()
    cutoff = now - timedelta(days=CACHE_TTL_DAYS)

    active = {}
    for url, entry in cache.items():
        entry_date = entry.get("date", "")
        try:
            dt = datetime.fromisoformat(entry_date)
            if dt >= cutoff:
                active[url] = entry
        except (ValueError, TypeError):
            active[url] = entry

    if len(active) > CACHE_MAX_ENTRIES:
        sorted_entries = sorted(active.items(), key=lambda x: x[1].get("date", ""), reverse=True)
        active = dict(sorted_entries[:CACHE_MAX_ENTRIES])

    return active

def save_cache(cache: dict):
    """保存 URL 缓存"""
    ensure_cache_dir()
    with open(CACHE_FILE, 'w', encoding='utf-8') as f:
        json.dump(cache, f, ensure_ascii=False, indent=2)

def _resolve_short_url(url: str, timeout: int = 5) -> str:
    """短链接展开：通过 HEAD 请求跟随重定向获取最终 URL"""
    parsed = urlparse(url)
    if parsed.netloc.lower() not in SHORT_URL_DOMAINS:
        return url
    try:
        req = Request(url, method='HEAD')
        req.add_header('User-Agent', 'Mozilla/5.0')
        with urlopen(req, timeout=timeout) as resp:
            return resp.url
    except (URLError, OSError, ValueError) as e:
        logger.debug(f"短链接展开失败 {url}: {e}")
        return url


def normalize_url(url: str) -> str:
    url = _resolve_short_url(url)

    tracking_params = {'utm_source', 'utm_medium', 'utm_campaign', 'utm_term',
                       'utm_content', 'fbclid', 'gclid', 'ref', 'source'}

    parsed = urlparse(url)

    netloc = parsed.netloc.lower()
    netloc = DOMAIN_ALIASES.get(netloc, netloc)

    path = parsed.path.rstrip('/')

    query_params = parse_qs(parsed.query)
    filtered_params = {k: v for k, v in query_params.items() if k not in tracking_params}

    base_url = f"{parsed.scheme}://{netloc}{path}"
    if filtered_params:
        query_string = urlencode(filtered_params, doseq=True)
        return f"{base_url}?{query_string}"
    return base_url

def extract_url_from_text(text: str) -> list:
    """从文本中提取所有 URL"""
    url_pattern = r'https?://[^\s<>"\')\]]+(?:\([^\s]*\))?'
    urls = re.findall(url_pattern, text)
    return [normalize_url(url) for url in urls]

def is_duplicate(url: str, doc_content: Optional[str] = None) -> dict:
    """
    检查 URL 是否已存在
    
    Returns:
        {
            'is_duplicate': bool,
            'source': str,  # 'cache' | 'document' | 'none'
            'normalized_url': str,
            'message': str
        }
    """
    normalized = normalize_url(url)
    
    # 1. 检查本地缓存
    cache = load_cache()
    if normalized in cache:
        return {
            'is_duplicate': True,
            'source': 'cache',
            'normalized_url': normalized,
            'message': f'链接已在缓存中 (收藏于 {cache[normalized].get("date", "未知时间")})'
        }
    
    # 2. 检查飞书文档内容
    if doc_content:
        existing_urls = extract_url_from_text(doc_content)
        if normalized in existing_urls or url in existing_urls:
            return {
                'is_duplicate': True,
                'source': 'document',
                'normalized_url': normalized,
                'message': '链接已在飞书文档中'
            }
    
    return {
        'is_duplicate': False,
        'source': 'none',
        'normalized_url': normalized,
        'message': '新链接，可以收藏'
    }

def add_to_cache(url: str, metadata: Optional[dict] = None):
    """添加 URL 到缓存"""
    cache = load_cache()
    normalized = normalize_url(url)
    
    cache[normalized] = {
        'original_url': url,
        'date': datetime.now().isoformat(),
        'metadata': metadata or {}
    }
    save_cache(cache)

def main():
    """命令行入口"""
    import sys
    
    if len(sys.argv) < 2:
        print("Usage:", file=sys.stderr)
        print("  python3 deduplicate.py <url>              # 检查是否重复", file=sys.stderr)
        print("  python3 deduplicate.py --add <url>        # 添加到缓存", file=sys.stderr)
        sys.exit(1)
    
    # 支持 --add 参数
    if sys.argv[1] == '--add':
        if len(sys.argv) < 3:
            print("Error: --add requires a URL", file=sys.stderr)
            sys.exit(1)
        url = sys.argv[2]
        add_to_cache(url)
        print(json.dumps({'success': True, 'message': f'已添加到缓存: {url}'}, ensure_ascii=False))
        return
    
    url = sys.argv[1]
    doc_content = None
    
    if len(sys.argv) > 2:
        with open(sys.argv[2], 'r', encoding='utf-8') as f:
            doc_content = f.read()
    
    result = is_duplicate(url, doc_content)
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Content Extractor - 检测平台并输出提取指引

设计原则：
- 不硬编码任何外部路径
- 不直接调用其他 skill 的脚本（由 Agent 在运行时通过 skill 调用）
- 只做：平台检测 + 输出推荐的 skill/tool 和提取方式
"""

import sys
import json
from urllib.parse import urlparse


# 平台域名 → 提取策略映射
PLATFORM_RULES = {
    'twitter': {
        'domains': ['x.com', 'twitter.com', 'mobile.twitter.com'],
        'label': 'X/Twitter',
        'skill': 'x-tweet-fetcher',
        'fallback_skills': [],
        'note': '使用 x-tweet-fetcher skill 提取推文/文章内容',
    },
    'weixin': {
        'domains': ['mp.weixin.qq.com'],
        'label': '微信公众号',
        'skill': 'web-content-fetcher',
        'fallback_skills': ['defuddle'],
        'note': '使用 web-content-fetcher (Scrapling) 提取正文，完美绕过微信反爬',
        'scrapling_command': 'python3 ~/.openclaw/workspace/skills/web-content-fetcher/scripts/fetch.py "{url}" 50000',
        'selectors': {
            'title': '#activity-name, .rich_media_title',
            'author': '#js_name, .profile_nickname',
            'content': '#js_content, .rich_media_content',
            'publish_time': '#publish_time, em#publish_time',
        },
    },
    'jike': {
        'domains': ['okjike.com', 'jike.cn', 'm.okjike.com', 'web.okjike.com'],
        'label': '即刻',
        'skill': 'defuddle',
        'fallback_skills': ['baoyu-url-to-markdown'],
        'note': '即刻网页版可能需要登录，优先 defuddle，fallback 到 baoyu-url-to-markdown',
        'selectors': {
            'author': 'meta[property="og:title"], .user-name',
            'content': 'meta[property="og:description"], .content',
            'time': 'time, .time',
        },
    },
    'reddit': {
        'domains': ['reddit.com', 'www.reddit.com', 'old.reddit.com'],
        'label': 'Reddit',
        'skill': 'defuddle',
        'fallback_skills': ['baoyu-url-to-markdown'],
        'note': 'Reddit 支持 JSON API（URL 末尾加 .json），也可用 defuddle',
        'json_api': True,
    },
    'hackernews': {
        'domains': ['news.ycombinator.com'],
        'label': 'Hacker News',
        'skill': 'defuddle',
        'fallback_skills': [],
        'note': 'HN 支持 Firebase API: https://hacker-news.firebaseio.com/v0/item/{id}.json',
    },
    'zhihu': {
        'domains': ['zhihu.com', 'www.zhihu.com', 'zhuanlan.zhihu.com'],
        'label': '知乎',
        'skill': 'defuddle',
        'fallback_skills': ['baoyu-url-to-markdown'],
        'note': '知乎部分内容需要登录，使用 defuddle 或 baoyu-url-to-markdown',
    },
    'bilibili': {
        'domains': ['bilibili.com', 'www.bilibili.com', 'b23.tv'],
        'label': 'Bilibili',
        'skill': 'defuddle',
        'fallback_skills': ['baoyu-url-to-markdown'],
        'note': '视频类内容仅能提取标题和描述',
    },
}


def detect_platform(url: str) -> dict:
    """
    检测链接来源平台，返回平台信息和推荐的提取方式。
    """
    domain = urlparse(url).netloc.lower()
    bare_domain = domain.lstrip('www.')

    for platform_id, rule in PLATFORM_RULES.items():
        for d in rule['domains']:
            if d in domain or d in bare_domain:
                return {
                    'platform_id': platform_id,
                    'platform_label': rule['label'],
                    'url': url,
                    'skill': rule['skill'],
                    'fallback_skills': rule.get('fallback_skills', []),
                    'note': rule['note'],
                    'selectors': rule.get('selectors'),
                    'json_api': rule.get('json_api', False),
                }

    # 未知平台 → 通用提取
    return {
        'platform_id': 'generic',
        'platform_label': 'Web',
        'url': url,
        'skill': 'defuddle',
        'fallback_skills': ['baoyu-url-to-markdown'],
        'note': '未知平台，使用 defuddle 通用提取，fallback 到 baoyu-url-to-markdown',
        'selectors': None,
        'json_api': False,
    }


def main():
    if len(sys.argv) < 2:
        print(json.dumps({
            'error': 'Usage: python3 extract_content.py [--url] <url>',
        }, ensure_ascii=False), file=sys.stderr)
        sys.exit(1)

    # 兼容 --url <url> 和直接 <url> 两种调用方式
    if sys.argv[1] == '--url' and len(sys.argv) >= 3:
        url = sys.argv[2]
    else:
        url = sys.argv[1]

    result = detect_platform(url)
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
标签生成器 v2.0 - 独立生成模式

功能：
1. 根据文章内容生成 5 个标签（对象2 + 场景1 + 类型1 + 方法1）
2. 规范化标签格式
3. 校验标签结构

输出：JSON 格式的标签
"""

import json
import re
import sys
import subprocess
from typing import Dict, List, Any

# 飞书配置
BITABLE_APP_TOKEN = "ND8ObCuSya5Dv3sREZYc03Ilngh"
BITABLE_TABLE_ID = "tblaHDM5kjtikIl9"


def normalize_tag(tag: str) -> str:
    """
    标签规范化规则：
    1. 去掉首尾空格
    2. 英文转小写
    3. 英文多个单词用 `-` 连接
    4. 去重
    """
    # 1. 去掉首尾空格
    tag = tag.strip()
    
    # 2. 英文转小写
    tag = tag.lower()
    
    # 3. 英文多个单词用 `-` 连接
    # 判断是否全是英文单词组成
    if ' ' in tag:
        # 检查是否全是字母和空格
        words = tag.split()
        if all(w.isalpha() for w in words):
            tag = '-'.join(words)
    
    # 4. 去除特殊字符（只保留字母、数字、中文、连字符）
    tag = re.sub(r'[^\w\u4e00-\u9fff-]', '', tag)
    
    return tag


def normalize_tags(tags: Dict[str, List[str]]) -> Dict[str, List[str]]:
    """规范化所有标签"""
    normalized = {}
    for category, tag_list in tags.items():
        normalized[category] = [normalize_tag(t) for t in tag_list]
    return normalized


def validate_tags(tags: Dict[str, List[str]]) -> List[str]:
    """
    校验规则：
    - 对象恰好 2 个
    - 场景恰好 1 个
    - 类型恰好 1 个
    - 方法恰好 1 个
    - 总数恰好 5 个
    - 标签无重复（跨类别也检查）
    """
    errors = []
    
    # 检查各类别数量
    if len(tags.get("对象", [])) != 2:
        errors.append(f"对象需要 2 个，当前 {len(tags.get('对象', []))} 个")
    if len(tags.get("场景", [])) != 1:
        errors.append(f"场景需要 1 个，当前 {len(tags.get('场景', []))} 个")
    if len(tags.get("类型", [])) != 1:
        errors.append(f"类型需要 1 个，当前 {len(tags.get('类型', []))} 个")
    if len(tags.get("方法", [])) != 1:
        errors.append(f"方法需要 1 个，当前 {len(tags.get('方法', []))} 个")
    
    # 检查总数
    total = sum(len(v) for v in tags.values())
    if total != 5:
        errors.append(f"标签总数需要 5 个，当前 {total} 个")
    
    # 检查重复（跨类别）
    all_tags = []
    for v in tags.values():
        all_tags.extend(v)
    if len(all_tags) != len(set(all_tags)):
        errors.append("存在重复标签")
    
    return errors


def flatten_tags(tags: Dict[str, List[str]]) -> List[str]:
    """将标签字典扁平化为列表"""
    result = []
    result.extend(tags.get("对象", []))  # 2个
    result.extend(tags.get("场景", []))   # 1个
    result.extend(tags.get("类型", []))   # 1个
    result.extend(tags.get("方法", []))   # 1个
    return result


def generate_tags_prompt(content: str) -> str:
    """生成标签的 prompt"""
    return f"""请阅读以下内容，并严格按照"对象、场景、类型、方法"四类标签体系输出 5 个标签。

【标签体系】
- 对象（2个）：内容涉及的核心主体/技术/工具，如 openclaude、agent、mcp、prompt、浏览器自动化、记忆系统、量化交易、学习资源、实战案例、产品思考
- 场景（1个）：内容应用的实际场景/用途，如 投资分析、自动化测试、知识管理、代码生成、数据处理
- 类型（1个）：内容的表现形式，如 技术教程、实战案例、产品分析、工具推荐、观点分享
- 方法（1个）：内容涉及的方法论/技巧，如 工作流、评测、prompt优化、架构设计

【规则】
1. 对象 2 个，场景 1 个，类型 1 个，方法 1 个，共 5 个
2. 不参考历史标签，只根据当前文章内容生成
3. 英文标签小写，多个单词用 `-` 连接（如 claude-code）
4. 中文标签使用简洁固定短语（如 投资分析、技术教程）
5. 不输出空泛标签，如 AI、工具、技术、效率
6. 只输出 JSON，不输出解释
7. 标签必须与内容相关，不要编造

【输出格式】
{{
  "tags": {{
    "对象": ["标签1", "标签2"],
    "场景": ["标签3"],
    "类型": ["标签4"],
    "方法": ["标签5"]
  }}
}}

【内容】
{content}
"""


def call_llm(prompt: str) -> Dict[str, List[str]]:
    """
    调用 LLM 生成标签
    这里需要根据实际环境调用 LLM
    返回格式化的标签字典
    """
    # 这里需要调用实际的 LLM
    # 临时返回示例，实际使用时替换为真实调用
    # 
    # 示例调用方式：
    # result = subprocess.run([
    #     'curl', '-X', 'POST', 'http://localhost:11434/api/generate',
    #     '-d', json.dumps({"model": "qwen", "prompt": prompt, "stream": False})
    # ], capture_output=True, text=True)
    # 
    # 这里应该返回实际的 LLM 响应
    pass


def main(content: str, max_retries: int = 2) -> List[str]:
    """
    主函数：生成并校验标签
    
    Args:
        content: 文章内容
        max_retries: 最大重试次数
    
    Returns:
        扁平化的标签列表（5个）
    """
    for attempt in range(max_retries):
        # 生成 prompt
        prompt = generate_tags_prompt(content)
        
        # 调用 LLM（这里需要替换为实际调用）
        # 假设返回的是 JSON 字符串
        # llm_response = call_llm(prompt)
        
        # 模拟 LLM 响应（实际使用时删除）
        # 示例响应
        # llm_response = '''
        # {
        #   "tags": {
        #     "对象": ["openclaude", "agent"],
        #     "场景": ["投资分析"],
        #     "类型": ["实战案例"],
        #     "方法": ["工作流"]
        #   }
        # }
        # '''
        
        # 解析响应
        # tags = json.loads(llm_response)["tags"]
        
        # 规范化
        # tags = normalize_tags(tags)
        
        # 校验
        # errors = validate_tags(tags)
        
        # if not errors:
        #     return flatten_tags(tags)
        # else:
        #     print(f"校验失败: {errors}", file=sys.stderr)
        #     if attempt < max_retries - 1:
        #         print("重试...", file=sys.stderr)
        #         continue
        #     else:
        #         raise ValueError(f"标签校验失败: {{errors}}")
        
        pass
    
    return []


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python3 generate_tags.py <content>", file=sys.stderr)
        sys.exit(1)
    
    content = sys.argv[1]
    tags = main(content)
    print(json.dumps(tags, ensure_ascii=False, indent=2))

#!/usr/bin/env python3
"""
图片 OCR 识别
支持：提取图片中的文字和链接
"""

import json
import re
import sys
import os
from urllib.parse import urlparse

# 检查是否安装了 pytesseract
try:
    import pytesseract
    from PIL import Image
    HAS_TESSERACT = True
except ImportError:
    HAS_TESSERACT = False

def extract_urls_from_text(text: str) -> list:
    """从文本中提取 URL"""
    url_patterns = [
        r'https?://[^\s<>"\')\]]+',
        r'(?:www\.)?[-a-zA-Z0-9@:%._\+~#=]{1,256}\.[a-zA-Z0-9()]{1,6}\b(?:[-a-zA-Z0-9()@:%_\+.~#?&/=]*)',
    ]
    
    urls = []
    for pattern in url_patterns:
        matches = re.findall(pattern, text)
        urls.extend(matches)
    
    # 去重并保持顺序
    seen = set()
    unique_urls = []
    for url in urls:
        url = url.strip()
        if url and url not in seen:
            seen.add(url)
            # 补全协议
            if url.startswith('www.'):
                url = 'https://' + url
            unique_urls.append(url)
    
    return unique_urls

def detect_platform_from_url(url: str) -> str:
    """从 URL 检测平台"""
    domain = urlparse(url).netloc.lower()
    
    if any(x in domain for x in ['x.com', 'twitter.com']):
        return 'twitter'
    elif 'mp.weixin.qq.com' in domain:
        return 'weixin'
    elif any(x in domain for x in ['okjike.com', 'jike.cn']):
        return 'jike'
    elif 'reddit.com' in domain:
        return 'reddit'
    elif any(x in domain for x in ['youtube.com', 'youtu.be']):
        return 'youtube'
    elif 'bilibili.com' in domain:
        return 'bilibili'
    elif 'zhihu.com' in domain:
        return 'zhihu'
    elif any(x in domain for x in ['douyin.com', 'tiktok.com']):
        return 'tiktok'
    else:
        return 'generic'

def ocr_image(image_path: str) -> dict:
    """
    对图片进行 OCR 识别
    
    Returns:
        {
            'success': bool,
            'text': str,  # 识别出的完整文本
            'urls': list,  # 提取的链接
            'platforms': list,  # 检测到的平台
            'method': str  # 使用的识别方法
        }
    """
    if not os.path.exists(image_path):
        return {
            'success': False,
            'error': f'图片不存在: {image_path}',
            'text': '',
            'urls': [],
            'platforms': []
        }
    
    result = {
        'success': False,
        'text': '',
        'urls': [],
        'platforms': [],
        'method': 'none'
    }
    
    # 方法 1: 使用 pytesseract (如果已安装)
    if HAS_TESSERACT:
        try:
            image = Image.open(image_path)
            text = pytesseract.image_to_string(image, lang='chi_sim+eng')
            result['text'] = text
            result['method'] = 'tesseract'
            result['success'] = True
        except Exception as e:
            result['error'] = f'Tesseract OCR 失败: {str(e)}'
    
    # 方法 2: 提示 Agent 可使用外部 OCR 服务
    # 不再硬编码特定路径，由 Agent 根据当前环境选择 OCR 方案
    if not result['success']:
        result['note'] = '本地 Tesseract 不可用，可使用以下外部 OCR 方案'
    
    # 方法 3: 返回图片路径，供 Agent 使用外部 OCR
    if not result['success']:
        result['method'] = 'external'
        result['note'] = '本地 OCR 不可用，请使用外部 OCR 服务'
        result['image_path'] = image_path
        result['recommendations'] = [
            '使用腾讯云数据万象 CI OCR',
            '使用飞书内置 OCR',
            '手动输入链接'
        ]
    
    # 提取 URL
    if result['text']:
        result['urls'] = extract_urls_from_text(result['text'])
        result['platforms'] = [detect_platform_from_url(url) for url in result['urls']]
    
    return result

def main():
    if len(sys.argv) < 2:
        print("Usage: python3 ocr_image.py <image_path>", file=sys.stderr)
        print("\n注意: 如需本地 OCR，请安装依赖:", file=sys.stderr)
        print("  pip install pytesseract pillow", file=sys.stderr)
        print("  apt-get install tesseract-ocr tesseract-ocr-chi-sim", file=sys.stderr)
        sys.exit(1)
    
    image_path = sys.argv[1]
    result = ocr_image(image_path)
    print(json.dumps(result, ensure_ascii=False, indent=2))

if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
保存内容到飞书多维表格

解决长文本写入问题：
- 短文本字段由 LLM 输出（标题、来源、分类、链接）
- 长文本字段通过文件路径传入（原文内容）
- 跳过 LLM 输出长文本的 token 消耗

使用方式：
    python3 save_to_bitable.py \
        --title "标题" \
        --source "来源" \
        --category "分类" \
        --url "https://..." \
        --content-file /path/to/content.md \
        [--app-token TOKEN] \
        [--table-id ID]

环境变量：
    FEISHU_USER_ACCESS_TOKEN: 用户访问令牌（必需）
    或通过 --token 参数传入

多维表格配置（默认值）：
    App Token: ND8ObCuSya5Dv3sREZYc03Ilngh
    Table ID: tblaHDM5kjtikIl9
"""

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Optional
import urllib.request
import urllib.error


# 默认配置（从环境变量读取）
DEFAULT_APP_TOKEN = os.environ.get("FEISHU_BITABLE_APP_TOKEN", "")
DEFAULT_TABLE_ID = os.environ.get("FEISHU_BITABLE_TABLE_ID", "")

# 飞书 API 配置
FEISHU_API_BASE = "https://open.feishu.cn/open-apis"


def get_user_access_token() -> Optional[str]:
    """获取用户访问令牌"""
    # 优先从环境变量获取
    token = os.environ.get("FEISHU_USER_ACCESS_TOKEN")
    if token:
        return token
    
    # 尝试从 OpenClaw token store 读取
    # 路径: ~/.openclaw/tokens/feishu/default/user_access_token
    token_path = Path.home() / ".openclaw" / "tokens" / "feishu" / "default" / "user_access_token"
    if token_path.exists():
        try:
            data = json.loads(token_path.read_text())
            token = data.get("access_token") or data.get("user_access_token")
            if token:
                return token
        except (json.JSONDecodeError, KeyError):
            pass
    
    return None


def call_feishu_api(
    method: str,
    path: str,
    token: str,
    body: Optional[dict] = None,
) -> dict:
    """调用飞书开放平台 API"""
    url = f"{FEISHU_API_BASE}{path}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }
    
    data = None
    if body:
        data = json.dumps(body).encode("utf-8")
    
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    
    try:
        with urllib.request.urlopen(req, timeout=30) as response:
            result = json.loads(response.read().decode("utf-8"))
            if result.get("code") != 0:
                raise Exception(f"API error: {result.get('msg', result)}")
            return result.get("data", {})
    except urllib.error.HTTPError as e:
        error_body = e.read().decode("utf-8")
        raise Exception(f"HTTP {e.code}: {error_body}")


def upload_file_to_feishu(
    file_path: str,
    token: str,
    parent_token: str = "nodcnCk8EyTpvvn9plx0OLTdSlg",
) -> dict:
    """上传文件到飞书云空间"""
    import mimetypes
    
    file_name = os.path.basename(file_path)
    file_size = os.path.getsize(file_path)
    mime_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
    
    # 1. 获取上传凭证
    path = "/drive/v1/files/upload_prepare"
    body = {
        "file_name": file_name,
        "size": file_size,
        "parent_type": "explorer",
        "parent_node": parent_token,
    }
    
    prepare_result = call_feishu_api("POST", path, token, body)
    upload_ticket = prepare_result.get("upload_ticket")
    
    # 2. 上传文件内容
    upload_url = f"{FEISHU_API_BASE}/drive/v1/files/upload_content"
    
    with open(file_path, "rb") as f:
        file_data = f.read()
    
    # 构造 multipart/form-data 请求
    import io
    import uuid
    
    boundary = f"----WebKitFormBoundary{uuid.uuid4().hex}"
    
    # 构建请求体
    body_parts = []
    body_parts.append(f"--{boundary}\r\n".encode())
    body_parts.append(f'Content-Disposition: form-data; name="upload_ticket"\r\n\r\n'.encode())
    body_parts.append(f"{upload_ticket}\r\n".encode())
    
    body_parts.append(f"--{boundary}\r\n".encode())
    body_parts.append(f'Content-Disposition: form-data; name="file"; filename="{file_name}"\r\n'.encode())
    body_parts.append(f"Content-Type: {mime_type}\r\n\r\n".encode())
    body_parts.append(file_data)
    body_parts.append(b"\r\n")
    
    body_parts.append(f"--{boundary}--\r\n".encode())
    
    data = b"".join(body_parts)
    
    req = urllib.request.Request(
        upload_url,
        data=data,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": f"multipart/form-data; boundary={boundary}",
        },
        method="POST",
    )
    
    try:
        with urllib.request.urlopen(req, timeout=60) as response:
            result = json.loads(response.read().decode("utf-8"))
            if result.get("code") != 0:
                raise Exception(f"Upload error: {result.get('msg', result)}")
            return result.get("data", {})
    except urllib.error.HTTPError as e:
        error_body = e.read().decode("utf-8")
        raise Exception(f"Upload HTTP {e.code}: {error_body}")


def save_to_bitable(
    app_token: str,
    table_id: str,
    token: str,
    title: str,
    source: str,
    category: str,
    url: str,
    content: str,
    content_file: str = None,
) -> dict:
    """保存内容到多维表格"""
    
    # 上传文件到云空间（如果提供了文件路径）
    file_url = None
    upload_success = False
    if content_file and os.path.exists(content_file):
        try:
            upload_result = upload_file_to_feishu(content_file, token)
            file_token = upload_result.get("file_token")
            if file_token:
                file_url = f"https://my.feishu.cn/file/{file_token}"
                upload_success = True
                print(f"✅ 文件上传成功: {file_url}", file=sys.stderr)
            else:
                print(f"⚠️ 文件上传返回空token", file=sys.stderr)
        except Exception as e:
            print(f"❌ 文件上传失败: {e}", file=sys.stderr)
    
    # 构建记录字段
    fields = {
        "标题": title,
        "来源": source,
        "分类": category,
        "原文链接": {"text": "查看原文", "link": url},
        "摘要内容": content[:2000] if len(content) > 2000 else content,  # 摘要限制长度
    }
    
    # 只有文件上传成功时，才添加原文文件字段
    if upload_success and file_url:
        fields["原文文件"] = {"text": "查看完整内容", "link": file_url}
    else:
        print(f"⚠️ 原文文件字段未写入（上传失败或未提供文件）", file=sys.stderr)
    
    # 调用 API 创建记录
    path = f"/bitable/v1/apps/{app_token}/tables/{table_id}/records"
    body = {
        "fields": fields,
    }
    
    result = call_feishu_api("POST", path, token, body)
    return result


def update_content_field(
    app_token: str,
    table_id: str,
    token: str,
    record_id: str,
    content: str,
) -> dict:
    """更新多维表格记录的"原文内容"字段
    
    用于两步写入场景：
    1. Agent 先写入短字段（标题、来源、分类、链接）
    2. 脚本更新长字段（原文内容）
    """
    
    fields = {
        "原文内容": content,
    }
    
    path = f"/bitable/v1/apps/{app_token}/tables/{table_id}/records/{record_id}"
    body = {
        "fields": fields,
    }
    
    result = call_feishu_api("PUT", path, token, body)
    return result


def main():
    parser = argparse.ArgumentParser(
        description="保存内容到飞书多维表格（支持文件路径传入长文本）",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例：
  # 创建完整记录
  python3 save_to_bitable.py \
      --title "OpenClaw 发布新版本" \
      --source "Twitter" \
      --category "🔧 工具推荐" \
      --url "https://x.com/..." \
      --content-file /tmp/content.md

  # 更新已有记录的长文本字段（两步写入）
  python3 save_to_bitable.py \
      --record-id "recXXXXXX" \
      --content-file /tmp/content.md

注意：
    需要设置环境变量 FEISHU_USER_ACCESS_TOKEN 或通过 --token 传入
        """
    )
    
    # 模式选择
    parser.add_argument("--record-id", help="更新已有记录（只更新原文内容字段）")
    
    # 创建记录所需参数
    parser.add_argument("--title", help="内容标题（创建模式必需）")
    parser.add_argument("--source", help="来源（创建模式必需）")
    parser.add_argument("--category", help="分类（创建模式必需）")
    parser.add_argument("--url", help="原文链接（创建模式必需）")
    
    # 通用参数
    parser.add_argument("--content-file", required=True, help="原文内容的 .md 文件路径")
    parser.add_argument("--app-token", default=DEFAULT_APP_TOKEN, help="多维表格 App Token")
    parser.add_argument("--table-id", default=DEFAULT_TABLE_ID, help="数据表 ID")
    parser.add_argument("--token", help="用户访问令牌（优先于环境变量）")
    
    args = parser.parse_args()
    
    # 参数校验
    if args.record_id:
        # 更新模式
        mode = "update"
    elif all([args.title, args.source, args.category, args.url]):
        # 创建模式
        mode = "create"
    else:
        print("错误：需要提供 --record-id（更新模式）或 --title/--source/--category/--url（创建模式）", file=sys.stderr)
        sys.exit(1)
    
    # 获取 token
    token = args.token or get_user_access_token()
    if not token:
        print("错误：未找到用户访问令牌", file=sys.stderr)
        print("请设置环境变量 FEISHU_USER_ACCESS_TOKEN 或使用 --token 参数", file=sys.stderr)
        sys.exit(1)
    
    # 读取内容文件
    content_path = Path(args.content_file)
    if not content_path.exists():
        print(f"错误：文件不存在: {content_path}", file=sys.stderr)
        sys.exit(1)
    
    content = content_path.read_text(encoding="utf-8")
    
    # 调用 API
    try:
        if mode == "create":
            result = save_to_bitable(
                app_token=args.app_token,
                table_id=args.table_id,
                token=token,
                title=args.title,
                source=args.source,
                category=args.category,
                url=args.url,
                content=content,
                content_file=str(content_path),
            )
            
            output = {
                "success": True,
                "mode": "create",
                "record_id": result.get("record", {}).get("record_id"),
                "app_token": args.app_token,
                "table_id": args.table_id,
                "title": args.title,
                "content_length": len(content),
                "message": "✅ 内容已保存到多维表格",
                "url": f"https://my.feishu.cn/base/{args.app_token}?table={args.table_id}",
            }
        else:
            # 更新模式
            result = update_content_field(
                app_token=args.app_token,
                table_id=args.table_id,
                token=token,
                record_id=args.record_id,
                content=content,
            )
            
            output = {
                "success": True,
                "mode": "update",
                "record_id": args.record_id,
                "content_length": len(content),
                "message": "✅ 原文内容已更新",
                "url": f"https://my.feishu.cn/base/{args.app_token}?table={args.table_id}",
            }
        
        print(json.dumps(output, ensure_ascii=False, indent=2))
        
    except Exception as e:
        output = {
            "success": False,
            "error": str(e),
            "message": f"❌ 保存失败: {e}",
        }
        print(json.dumps(output, ensure_ascii=False, indent=2))
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Content Collector v1.2.0 - 主流程测试用例
测试完整工作流程：平台检测 → 去重检查 → 内容提取 → 格式化 → 飞书输出
"""

import json
import subprocess
import sys
import tempfile
import os
from pathlib import Path

# 测试配置
TEST_URL = "https://x.com/i/status/2032275457987539209"  # xiaoerzhan 的 OpenClaw Skill 榜单
SCRIPTS_DIR = Path(__file__).parent / "scripts"
CACHE_DIR = Path(__file__).parent / ".cache"


def run_script(script_name: str, *args) -> tuple[int, str, str]:
    """运行脚本并返回结果"""
    script_path = SCRIPTS_DIR / script_name
    cmd = [sys.executable, str(script_path)] + list(args)
    
    print(f"\n{'='*60}")
    print(f"▶ 运行: {script_name}")
    print(f"  命令: {' '.join(cmd)}")
    print(f"{'='*60}")
    
    result = subprocess.run(cmd, capture_output=True, text=True)
    
    if result.stdout:
        print(f"📤 STDOUT:\n{result.stdout}")
    if result.stderr:
        print(f"📥 STDERR:\n{result.stderr}")
    
    print(f"✅ 返回码: {result.returncode}")
    return result.returncode, result.stdout, result.stderr


def test_step1_platform_detection():
    """Step 1: 平台检测"""
    print("\n" + "🧪"*30)
    print("STEP 1: 平台检测 (extract_content.py)")
    print("🧪"*30)
    
    code, stdout, stderr = run_script("extract_content.py", TEST_URL)
    
    if code != 0:
        print(f"❌ 平台检测失败: {stderr}")
        return None
    
    try:
        result = json.loads(stdout)
        print(f"\n✅ 平台检测成功:")
        print(f"   平台: {result.get('platform_label', 'N/A')}")
        print(f"   推荐 Skill: {result.get('skill', 'N/A')}")
        print(f"   备注: {result.get('note', 'N/A')}")
        return result
    except json.JSONDecodeError as e:
        print(f"❌ JSON 解析失败: {e}")
        print(f"   原始输出: {stdout[:500]}")
        return None


def test_step2_deduplication():
    """Step 2: 去重检查"""
    print("\n" + "🧪"*30)
    print("STEP 2: 去重检查 (deduplicate.py)")
    print("🧪"*30)
    
    code, stdout, stderr = run_script("deduplicate.py", TEST_URL)
    
    if code != 0:
        print(f"❌ 去重检查失败: {stderr}")
        return False
    
    try:
        result = json.loads(stdout)
        is_duplicate = result.get('is_duplicate', False)
        
        if is_duplicate:
            print(f"⚠️  检测到重复内容:")
            print(f"   原因: {result.get('reason', 'N/A')}")
            print(f"   首次收藏时间: {result.get('first_seen', 'N/A')}")
        else:
            print(f"✅ 新内容，未重复")
            print(f"   标准化 URL: {result.get('normalized_url', 'N/A')[:80]}...")
        
        return not is_duplicate  # 返回是否应该继续处理
    except json.JSONDecodeError as e:
        print(f"❌ JSON 解析失败: {e}")
        return False


def test_step3_content_extraction(platform_info: dict):
    """Step 3: 内容提取（模拟调用 x-tweet-fetcher）"""
    print("\n" + "🧪"*30)
    print("STEP 3: 内容提取")
    print("🧪"*30)
    
    skill = platform_info.get('skill', '')
    
    if skill == 'x-tweet-fetcher':
        print(f"📋 推荐调用 skill: {skill}")
        print(f"   命令示例: python3 ~/.openclaw/workspace/skills/x-tweet-fetcher/scripts/fetch_tweet.py --url \"{TEST_URL}\" --text-only")
        
        # 实际调用 x-tweet-fetcher
        x_script = Path.home() / ".openclaw/workspace/skills/x-tweet-fetcher/scripts/fetch_tweet.py"
        if x_script.exists():
            print(f"\n▶ 实际调用 x-tweet-fetcher...")
            code, stdout, stderr = run_script(str(x_script), "--url", TEST_URL, "--text-only")
            
            if code == 0:
                print(f"✅ 内容提取成功")
                # 构造模拟的提取结果
                return {
                    "platform": "X/Twitter",
                    "author": "@xiaoerzhan",
                    "title": "OpenClaw 神级 Skill 榜单",
                    "content": stdout[:2000] if len(stdout) > 2000 else stdout,  # 截断避免过长
                    "url": TEST_URL,
                    "created_at": "2026-03-14T10:00:00",
                    "summary": "分享了20个评分3.4+的OpenClaw必装Skill，分四类整理",
                    "keywords": ["OpenClaw", "Skill", "自动化", "AI工具"],
                    "reason": "整理了高质量的OpenClaw生态插件清单",
                    "stats": {"likes": 419, "retweets": 122, "bookmarks": 0, "views": 29960}
                }
            else:
                print(f"⚠️  x-tweet-fetcher 调用失败，使用模拟数据")
        else:
            print(f"⚠️  x-tweet-fetcher 未安装，使用模拟数据")
    else:
        print(f"📋 推荐调用 skill: {skill}")
    
    # 模拟数据
    return {
        "platform": "X/Twitter",
        "author": "@xiaoerzhan",
        "title": "OpenClaw 神级 Skill 榜单",
        "content": "🦞 特别需要的来了... [内容截断]",
        "url": TEST_URL,
        "created_at": "2026-03-14T10:00:00",
        "summary": "分享了20个评分3.4+的OpenClaw必装Skill",
        "keywords": ["OpenClaw", "Skill"],
        "reason": "高质量插件清单",
        "stats": {"likes": 419, "retweets": 122, "bookmarks": 0, "views": 29960}
    }


def test_step4_formatting(content_data: dict):
    """Step 4: 格式化输出"""
    print("\n" + "🧪"*30)
    print("STEP 4: 格式化 (append_to_feishu.py)")
    print("🧪"*30)
    
    # 构造输入 JSON
    input_json = json.dumps(content_data, ensure_ascii=False)
    
    code, stdout, stderr = run_script("append_to_feishu.py", input_json)
    
    if code != 0:
        print(f"❌ 格式化失败: {stderr}")
        return None
    
    print(f"✅ 格式化成功")
    print(f"\n📄 输出预览（前1000字符）:\n{'-'*60}")
    print(stdout[:1000])
    print(f"{'-'*60}")
    
    return stdout


def test_step5_document_simulation(formatted_content: str):
    """Step 5: 模拟飞书文档追加"""
    print("\n" + "🧪"*30)
    print("STEP 5: 飞书文档追加（模拟）")
    print("🧪"*30)
    
    print(f"📋 实际使用时，Agent 应调用 feishu-doc skill:")
    print(f"   feishu_doc(action='append', doc_token='...', content='...')")
    print(f"\n✅ 格式化内容已准备就绪，可以追加到飞书文档")
    
    # 保存到临时文件供查看
    with tempfile.NamedTemporaryFile(mode='w', suffix='.md', delete=False, encoding='utf-8') as f:
        f.write(formatted_content)
        print(f"\n💾 完整内容已保存到: {f.name}")
        return f.name


def run_full_test():
    """运行完整测试流程"""
    print("\n" + "🚀"*40)
    print("Content Collector v1.2.0 - 主流程测试")
    print("测试 URL:", TEST_URL)
    print("🚀"*40)
    
    results = {
        "step1_platform": False,
        "step2_dedup": False,
        "step3_extraction": False,
        "step4_formatting": False,
        "step5_document": False,
    }
    
    # Step 1: 平台检测
    platform_info = test_step1_platform_detection()
    if platform_info:
        results["step1_platform"] = True
    else:
        print("\n❌ 测试中止：平台检测失败")
        return results
    
    # Step 2: 去重检查
    should_continue = test_step2_deduplication()
    if should_continue:
        results["step2_dedup"] = True
    else:
        print("\n⚠️  内容已存在，跳过后续步骤")
        return results
    
    # Step 3: 内容提取
    content_data = test_step3_content_extraction(platform_info)
    if content_data:
        results["step3_extraction"] = True
    else:
        print("\n❌ 测试中止：内容提取失败")
        return results
    
    # Step 4: 格式化
    formatted = test_step4_formatting(content_data)
    if formatted:
        results["step4_formatting"] = True
    else:
        print("\n❌ 测试中止：格式化失败")
        return results
    
    # Step 5: 文档追加（模拟）
    temp_file = test_step5_document_simulation(formatted)
    results["step5_document"] = True
    
    # 打印总结
    print("\n" + "📊"*40)
    print("测试总结")
    print("📊"*40)
    for step, passed in results.items():
        status = "✅ 通过" if passed else "❌ 失败"
        print(f"   {step}: {status}")
    
    all_passed = all(results.values())
    print(f"\n{'='*60}")
    if all_passed:
        print("🎉 所有测试通过！")
    else:
        print("⚠️  部分测试未通过，请检查日志")
    print(f"{'='*60}\n")
    
    return results


if __name__ == "__main__":
    run_full_test()