Civil Case Os

__pycache__/
*.pyc
.DS_Store

#!/bin/bash
# 案件OS Agent管理器 - 统一管理所有Agent

set -e

AGENTS_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
HOME_PATH="/Users/Apple"  # 使用绝对路径
AGENT_LIST=(
    "weekly-scan:case-os.weekly-scan:案件周度扫描Agent:每周一早上8点"
    "court-sms-monitor:caseos.sms-monitor:法院短信实时监控Agent:实时监控"
    "batch-evidence::批量证据提取Agent:手动触发"
)

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

# 打印带颜色的消息
print_success() {
    echo -e "${GREEN}✅ $1${NC}"
}

print_error() {
    echo -e "${RED}❌ $1${NC}"
}

print_warning() {
    echo -e "${YELLOW}⚠️  $1${NC}"
}

print_info() {
    echo -e "${BLUE}ℹ️  $1${NC}"
}

print_header() {
    echo ""
    echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
    echo -e "${BLUE}  $1${NC}"
    echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}"
    echo ""
}

# 获取Agent状态
get_agent_status() {
    local agent_name=$1
    local label_prefix=$2
    local plist_file="$HOME_PATH/Library/LaunchAgents/com.${label_prefix}.plist"
    local label=$(launchctl list | grep "com.${label_prefix}" | awk '{print $1}')

    if [ ! -f "$plist_file" ]; then
        echo "未安装"
    elif [ -n "$label" ]; then
        echo "运行中"
    else
        echo "已安装但未运行"
    fi
}

# 列出所有Agent
list_agents() {
    print_header "案件OS Agent状态"

    printf "%-25s %-30s %-15s\n" "Agent名称" "说明" "状态"
    printf "%-25s %-30s %-15s\n" "---------" "----" "----"

    for agent_info in "${AGENT_LIST[@]}"; do
        IFS=':' read -r agent_name label_prefix desc schedule <<< "$agent_info"

        status=$(get_agent_status "$agent_name" "$label_prefix")
        printf "%-25s %-30s %-15s\n" "$agent_name" "$desc" "$status"
    done

    echo ""
}

# 启动所有Agent
start_all_agents() {
    print_header "启动所有Agent"

    for agent_info in "${AGENT_LIST[@]}"; do
        IFS=':' read -r agent_name desc schedule <<< "$agent_info"

        # 跳过手动触发的Agent
        if [ "$agent_name" = "batch-evidence" ]; then
            print_info "$agent_name: 手动触发，跳过自动启动"
            continue
        fi

        print_info "启动 $agent_name..."

        install_script="$AGENTS_DIR/$agent_name/install.sh"
        if [ -f "$install_script" ]; then
            bash "$install_script"
        else
            print_warning "$agent_name: 未找到安装脚本"
        fi
    done

    print_success "所有Agent已启动"
}

# 停止所有Agent
stop_all_agents() {
    print_header "停止所有Agent"

    for agent_info in "${AGENT_LIST[@]}"; do
        IFS=':' read -r agent_name desc schedule <<< "$agent_info"

        # 跳过手动触发的Agent
        if [ "$agent_name" = "batch-evidence" ]; then
            continue
        fi

        plist_file="$HOME/Library/LaunchAgents/com.claude.caseos.${agent_name}.plist"

        if [ -f "$plist_file" ]; then
            print_info "停止 $agent_name..."
            launchctl unload "$plist_file" 2>/dev/null || true
            print_success "$agent_name 已停止"
        else
            print_warning "$agent_name: 未安装"
        fi
    done

    print_success "所有Agent已停止"
}

# 重启所有Agent
restart_all_agents() {
    print_header "重启所有Agent"
    stop_all_agents
    sleep 1
    start_all_agents
}

# 查看日志
view_logs() {
    local agent_name=$1

    if [ -z "$agent_name" ]; then
        print_error "请指定Agent名称"
        echo "用法: $0 logs <agent_name>"
        echo ""
        echo "可用的Agent:"
        for agent_info in "${AGENT_LIST[@]}"; do
            IFS=':' read -r agent_name desc schedule <<< "$agent_info"
            echo "  - $agent_name"
        done
        return 1
    fi

    log_file="$HOME/.claude/skills/case-os/data/${agent_name}.log"

    if [ ! -f "$log_file" ]; then
        log_file="$HOME/.claude/skills/case-os/data/${agent_name}-stdout.log"
    fi

    if [ ! -f "$log_file" ]; then
        print_error "未找到日志文件: $log_file"
        return 1
    fi

    print_header "查看 $agent_name 日志"
    tail -f "$log_file"
}

# 测试Agent
test_agent() {
    local agent_name=$1

    if [ -z "$agent_name" ]; then
        print_error "请指定Agent名称"
        echo "用法: $0 test <agent_name>"
        return 1
    fi

    print_header "测试 $agent_name"

    case "$agent_name" in
        weekly-scan)
            python3 "$AGENTS_DIR/weekly-scan/scan.py"
            ;;
        court-sms-monitor)
            python3 "$AGENTS_DIR/court-sms-monitor/monitor.py"
            ;;
        batch-evidence)
            print_error "batch-evidence需要指定案件目录"
            echo "用法: python3 $AGENTS_DIR/batch-evidence/batch_extract.py /path/to/case/dir"
            ;;
        *)
            print_error "未知的Agent: $agent_name"
            return 1
            ;;
    esac
}

# 显示帮助
show_help() {
    cat << EOF
案件OS Agent管理器

用法: $0 <command> [options]

命令:
  status              查看所有Agent状态
  start               启动所有Agent
  stop                停止所有Agent
  restart             重启所有Agent
  logs <agent_name>   查看指定Agent的日志
  test <agent_name>   测试指定Agent
  help                显示此帮助信息

可用的Agent:
  - weekly-scan        案件周度扫描Agent
  - court-sms-monitor  法院短信实时监控Agent
  - batch-evidence     批量证据提取Agent（手动触发）

示例:
  $0 status           # 查看状态
  $0 start            # 启动所有
  $0 logs weekly-scan # 查看周度扫描日志
  $0 test weekly-scan # 测试周度扫描

EOF
}

# 主函数
main() {
    local command=$1
    shift || true

    case "$command" in
        status)
            list_agents
            ;;
        start)
            start_all_agents
            ;;
        stop)
            stop_all_agents
            ;;
        restart)
            restart_all_agents
            ;;
        logs)
            view_logs "$@"
            ;;
        test)
            test_agent "$@"
            ;;
        help|--help|-h)
            show_help
            ;;
        *)
            print_error "未知的命令: $command"
            echo ""
            show_help
            exit 1
            ;;
    esac
}

main "$@"

#!/usr/bin/env python3
"""
批量证据提取Agent - 实现脚本
多个证据文件同时提取，大幅加速A5步骤
"""

import json
import os
import sys
import time
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional

# ==================== 配置 ====================
MAX_WORKERS = 3  # 最大并发数
TIMEOUT = 120  # 单个任务超时时间（秒）
MODE = "process"  # process=进程池, thread=线程池

# 证据文件后缀
EVIDENCE_EXTENSIONS = [".pdf", ".jpg", ".jpeg", ".png", ".docx", ".doc"]

# ==================== 工具函数 ====================
def find_evidence_files(case_dir: Path) -> List[Path]:
    """发现证据文件"""
    evidence_files = []

    # 扫描00_原始材料目录
    material_dir = case_dir / "00_原始材料"
    if material_dir.exists():
        for ext in EVIDENCE_EXTENSIONS:
            evidence_files.extend(material_dir.glob(f"*{ext}"))
            evidence_files.extend(material_dir.glob(f"*{ext.upper()}"))

    return sorted(evidence_files)


def filter_processed(files: List[Path], evidence_cards_dir: Path) -> List[Path]:
    """过滤已处理的文件"""
    if not evidence_cards_dir.exists():
        return files

    processed = {f.stem for f in evidence_cards_dir.glob("*.md")}
    return [f for f in files if f.stem not in processed]


def extract_single_evidence(file_path: Path) -> Dict:
    """提取单个证据（实际调用case-evidence-cards Skill）

    Args:
        file_path: 证据文件路径

    Returns:
        提取结果
    """
    # TODO: 实际调用case-evidence-cards Skill
    # 这里是模拟实现
    try:
        start_time = time.time()

        # 模拟提取过程（实际应该调用Skill）
        time.sleep(0.5)  # 模拟耗时

        return {
            "file": file_path,
            "file_name": file_path.name,
            "success": True,
            "output": f"{file_path.stem}_证据卡片.md",
            "duration": time.time() - start_time,
        }
    except Exception as e:
        return {
            "file": file_path,
            "file_name": file_path.name,
            "success": False,
            "error": str(e),
        }


def extract_with_timeout(file_path: Path, timeout: int = TIMEOUT) -> Dict:
    """带超时的提取

    Args:
        file_path: 证据文件路径
        timeout: 超时时间（秒）

    Returns:
        提取结果
    """
    try:
        if MODE == "process":
            # 使用进程池
            with ProcessPoolExecutor(max_workers=1) as executor:
                future = executor.submit(extract_single_evidence, file_path)
                result = future.result(timeout=timeout)
                return result
        else:
            # 使用线程池
            with ThreadPoolExecutor(max_workers=1) as executor:
                future = executor.submit(extract_single_evidence, file_path)
                result = future.result(timeout=timeout)
                return result
    except Exception as e:
        return {
            "file": file_path,
            "file_name": file_path.name,
            "success": False,
            "error": f"超时或错误: {str(e)}",
        }


# ==================== 批量提取 ====================
class BatchEvidenceExtractor:
    """批量证据提取器"""

    def __init__(self, case_dir: Path, max_workers: int = MAX_WORKERS):
        self.case_dir = case_dir
        self.max_workers = max_workers
        self.evidence_cards_dir = case_dir / "_intermediate" / "证据卡片库"

    def find_files(self) -> List[Path]:
        """发现需要处理的证据文件"""
        files = find_evidence_files(self.case_dir)

        if not files:
            print("⚠️  未发现证据文件")
            return []

        # 过滤已处理的
        files = filter_processed(files, self.evidence_cards_dir)

        print(f"📁 发现 {len(files)} 个待处理证据文件")
        return files

    def extract_parallel(self, files: List[Path]) -> List[Dict]:
        """并行提取证据

        Args:
            files: 证据文件列表

        Returns:
            提取结果列表
        """
        print(f"🚀 启动并行提取（{MODE}模式，{self.max_workers}并发）")
        print()

        results = []

        if MODE == "process":
            # 使用进程池
            with ProcessPoolExecutor(max_workers=self.max_workers) as executor:
                # 提交所有任务
                futures = {
                    executor.submit(extract_with_timeout, f): f
                    for f in files
                }

                # 收集结果（带进度显示）
                completed = 0
                total = len(futures)

                for future in as_completed(futures):
                    file_path = futures[future]
                    completed += 1

                    try:
                        result = future.result()
                        results.append(result)

                        # 显示进度
                        percent = (completed / total) * 100
                        status = "✅" if result["success"] else "❌"
                        print(f"[{completed}/{total} {percent:.0f}%] {status} {result['file_name']}")

                    except Exception as e:
                        results.append({
                            "file": file_path,
                            "file_name": file_path.name,
                            "success": False,
                            "error": str(e),
                        })
                        print(f"[{completed}/{total}] ❌ {file_path.name}: {e}")

        else:
            # 使用线程池
            with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
                futures = {
                    executor.submit(extract_with_timeout, f): f
                    for f in files
                }

                completed = 0
                total = len(futures)

                for future in as_completed(futures):
                    file_path = futures[future]
                    completed += 1

                    try:
                        result = future.result()
                        results.append(result)

                        percent = (completed / total) * 100
                        status = "✅" if result["success"] else "❌"
                        print(f"[{completed}/{total} {percent:.0f}%] {status} {result['file_name']}")

                    except Exception as e:
                        results.append({
                            "file": file_path,
                            "file_name": file_path.name,
                            "success": False,
                            "error": str(e),
                        })
                        print(f"[{completed}/{total}] ❌ {file_path.name}: {e}")

        return results

    def extract_sequential(self, files: List[Path]) -> List[Dict]:
        """串行提取（对比用）

        Args:
            files: 证据文件列表

        Returns:
            提取结果列表
        """
        print(f"🐌 串行提取模式")
        print()

        results = []
        for i, file_path in enumerate(files, 1):
            print(f"[{i}/{len(files)}] 处理：{file_path.name}")
            result = extract_with_timeout(file_path)
            results.append(result)

            status = "✅" if result["success"] else "❌"
            print(f"  {status} {result.get('error', '完成')}")

        return results

    def summarize(self, results: List[Dict]) -> Dict:
        """汇总结果

        Args:
            results: 提取结果列表

        Returns:
            汇总信息
        """
        total = len(results)
        success = sum(1 for r in results if r["success"])
        failed = total - success

        total_duration = sum(r.get("duration", 0) for r in results if r["success"])

        summary = {
            "total": total,
            "success": success,
            "failed": failed,
            "success_rate": (success / total * 100) if total > 0 else 0,
            "total_duration": total_duration,
            "avg_duration": total_duration / success if success > 0 else 0,
            "timestamp": datetime.now().isoformat(),
        }

        return summary

    def print_summary(self, summary: Dict, parallel: bool = True):
        """打印汇总信息

        Args:
            summary: 汇总信息
            parallel: 是否并行模式
        """
        print()
        print("=" * 50)
        print("📊 批量提取完成")
        print("=" * 50)
        print(f"模式：{'并行' if parallel else '串行'}")
        print(f"总数：{summary['total']} 个")
        print(f"成功：{summary['success']} 个")
        print(f"失败：{summary['failed']} 个")
        print(f"成功率：{summary['success_rate']:.1f}%")
        print(f"总耗时：{summary['total_duration']:.1f} 秒")
        print(f"平均耗时：{summary['avg_duration']:.1f} 秒/个")

        if parallel and summary['success'] > 1:
            # 计算加速比
            sequential_time = summary['total_duration'] * summary['success']
            speedup = sequential_time / summary['total_duration']
            print(f"加速比：{speedup:.1f}x")

        print("=" * 50)

    def run(self, mode: str = "parallel") -> List[Dict]:
        """运行批量提取

        Args:
            mode: parallel=并行, sequential=串行

        Returns:
            提取结果列表
        """
        # 发现文件
        files = self.find_files()
        if not files:
            return []

        # 提取
        start_time = time.time()

        if mode == "parallel":
            results = self.extract_parallel(files)
        else:
            results = self.extract_sequential(files)

        # 汇总
        summary = self.summarize(results)
        summary["total_duration"] = time.time() - start_time
        self.print_summary(summary, parallel=(mode == "parallel"))

        # 保存结果
        self.save_results(results, summary)

        return results

    def save_results(self, results: List[Dict], summary: Dict):
        """保存提取结果

        Args:
            results: 提取结果列表
            summary: 汇总信息
        """
        # 确保输出目录存在
        self.evidence_cards_dir.parent.mkdir(parents=True, exist_ok=True)

        # 保存汇总信息
        summary_file = self.case_dir / "_intermediate" / "batch_extract_summary.json"
        with open(summary_file, "w", encoding="utf-8") as f:
            json.dump({
                "summary": summary,
                "results": [
                    {
                        "file_name": r["file_name"],
                        "success": r["success"],
                        "error": r.get("error"),
                    }
                    for r in results
                ],
            }, f, ensure_ascii=False, indent=2)

        print(f"💾 汇总信息已保存：{summary_file}")


# ==================== 命令行接口 ====================
def main():
    """命令行入口"""
    import argparse

    parser = argparse.ArgumentParser(description="批量证据提取Agent")
    parser.add_argument("case_dir", help="案件目录路径")
    parser.add_argument("--mode", choices=["parallel", "sequential"], default="parallel", help="提取模式")
    parser.add_argument("--workers", type=int, default=MAX_WORKERS, help="最大并发数")

    args = parser.parse_args()

    case_dir = Path(args.case_dir)
    if not case_dir.exists():
        print(f"❌ 错误：案件目录不存在：{case_dir}")
        return 1

    # 创建提取器
    extractor = BatchEvidenceExtractor(case_dir, max_workers=args.workers)

    # 运行
    try:
        extractor.run(mode=args.mode)
        return 0
    except KeyboardInterrupt:
        print("\n⚠️  用户中断")
        return 1
    except Exception as e:
        print(f"❌ 错误：{e}")
        return 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
飞书通知模块 - 通过飞书机器人发送消息和写入多维表格
"""

import json
import os
import requests
from datetime import datetime
from typing import Dict, List, Optional

# 飞书配置（从环境变量读取）
FEISHU_BOT_WEBHOOK = os.getenv("FEISHU_BOT_WEBHOOK", "")  # 飞书机器人Webhook
FEISHU_APP_TOKEN = os.getenv("FEISHU_APP_TOKEN", "")  # 飞书多维表格App Token
FEISHU_TABLE_ID = os.getenv("FEISHU_TABLE_ID", "")  # 表格ID（可选，默认使用第一个表格）


class FeishuNotifier:
    """飞书通知器"""

    def __init__(self):
        self.webhook = FEISHU_BOT_WEBHOOK
        self.app_token = FEISHU_APP_TOKEN
        self.table_id = FEISHU_TABLE_ID

    def send_card(self, title: str, content: Dict, urgent: bool = False):
        """发送飞书卡片消息

        Args:
            title: 卡片标题
            content: 卡片内容
            urgent: 是否紧急（红色标签）
        """
        if not self.webhook:
            print("⚠️  飞书机器人Webhook未配置，跳过发送")
            return

        # 构建卡片内容
        card = {
            "msg_type": "interactive",
            "card": {
                "header": {
                    "title": {
                        "tag": "plain_text",
                        "content": title,
                    },
                    "template": urgent and "red" or "yellow",
                },
                "elements": [
                    {
                        "tag": "div",
                        "text": {
                            "tag": "lark_md",
                            "content": self._format_content(content),
                        }
                    }
                ],
            }
        }

        # 发送
        try:
            response = requests.post(
                self.webhook,
                json=card,
                headers={"Content-Type": "application/json"},
            )
            response.raise_for_status()
            print(f"✅ 飞书消息已发送：{title}")
        except Exception as e:
            print(f"❌ 飞书消息发送失败：{e}")

    def _format_content(self, content: Dict) -> str:
        """格式化卡片内容为Markdown"""
        lines = []
        for key, value in content.items():
            if isinstance(value, list):
                lines.append(f"**{key}**")
                for item in value:
                    lines.append(f"  • {item}")
            else:
                lines.append(f"**{key}**：{value}")
        return "\n\n".join(lines)

    def write_to_table(self, record: Dict) -> bool:
        """写入飞书多维表格

        Args:
            record: 记录数据

        Returns:
            是否成功
        """
        if not self.app_token:
            print("⚠️  飞书多维表格App Token未配置，跳过写入")
            return False

        # TODO: 实现飞书多维表格写入
        # 需要使用飞书开放平台API
        print(f"📝 写入飞书多维表格：{record.get('案件名称', '未知案件')}")
        return True


class CourtSMSFeishuNotifier(FeishuNotifier):
    """法院短信飞书通知器"""

    def send_court_sms(self, sms_info: Dict, case_name: str):
        """发送法院短信通知

        Args:
            sms_info: 短信信息
            case_name: 案件名称
        """
        extracted = sms_info.get("extracted", {})
        msg_type = self._get_type_label(sms_info["type"])

        # 判断紧急程度
        urgent = sms_info["type"] in ["hearing", "payment"]

        # 构建卡片内容
        content = {
            "案件": case_name,
        }

        if extracted.get("court"):
            content["法院"] = extracted["court"]

        if extracted.get("case_number"):
            content["案号"] = extracted["case_number"]

        if extracted.get("date"):
            date_text = extracted["date"]
            if extracted.get("appeal_deadline"):
                date_text = f"上诉期截止：{extracted['appeal_deadline']}"
            content["时间"] = date_text

        if extracted.get("datetime"):
            content["具体时间"] = extracted["datetime"]

        # 短信原文摘要
        content["短信摘要"] = sms_info["content"][:100] + "..."

        # 发送卡片
        title = f"🏛️  法院短信提醒：{msg_type}"
        self.send_card(title, content, urgent=urgent)

        # 写入多维表格
        self._write_court_sms_to_table(sms_info, case_name)

    def _write_court_sms_to_table(self, sms_info: Dict, case_name: str):
        """写入法院短信到多维表格"""
        extracted = sms_info.get("extracted", {})

        record = {
            "案件名称": case_name,
            "提醒类型": self._get_type_label(sms_info["type"]),
            "短信类型": sms_info["type"],
            "截止时间": extracted.get("deadline") or extracted.get("datetime") or "",
            "日期": extracted.get("date") or "",
            "法院": extracted.get("court") or "",
            "案号": extracted.get("case_number") or "",
            "短信内容": sms_info["content"],
            "收到时间": sms_info["received_at"],
            "状态": "待处理",
            "创建时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        }

        # 如果有上诉期，单独记录
        if extracted.get("appeal_deadline"):
            record["上诉期截止"] = extracted["appeal_deadline"]

        self.write_to_table(record)

    def _get_type_label(self, msg_type: str) -> str:
        """获取类型标签"""
        labels = {
            "hearing": "开庭通知",
            "evidence": "举证期限",
            "payment": "缴费通知",
            "judgment": "判决送达",
        }
        return labels.get(msg_type, "其他")

#!/bin/bash
# 法院短信实时监控Agent - 安装脚本

set -e

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
AGENT_DIR="$SCRIPT_DIR"
MONITOR_SCRIPT="$AGENT_DIR/monitor.py"
PLIST_FILE="$HOME/Library/LaunchAgents/com.claude.caseos.sms-monitor.log"

echo "🔧 安装法院短信实时监控Agent..."

# 检查Python3
if ! command -v python3 &> /dev/null; then
    echo "❌ 错误：未找到python3"
    exit 1
fi

# 检查fswatch
if ! command -v fswatch &> /dev/null; then
    echo "📦 正在安装fswatch..."
    if command -v brew &> /dev/null; then
        brew install fswatch
    else
        echo "❌ 错误：未找到Homebrew，请手动安装fswatch"
        echo "   安装命令：brew install fswatch"
        exit 1
    fi
fi

# 确保脚本可执行
chmod +x "$MONITOR_SCRIPT"

# 创建日志目录
LOG_DIR="$HOME/.claude/skills/case-os/data"
mkdir -p "$LOG_DIR"

# 创建监控脚本（使用fswatch）
WATCH_SCRIPT="$LOG_DIR/sms-monitor-wrapper.sh"
cat > "$WATCH_SCRIPT" << 'EOFSCRIPT'
#!/bin/bash
# 法院短信监控包装脚本 - 由LaunchAgent调用

SMS_DB_PATH="$HOME/Library/SMS/sms.db"
LOG_DIR="$HOME/.claude/skills/case-os/data"
MONITOR_SCRIPT="$HOME/.claude/skills/case-os/agents/court-sms-monitor/monitor.py"

echo "=== 法院短信监控Agent启动 ===" >> "$LOG_DIR/sms-monitor.log"
echo "启动时间：$(date)" >> "$LOG_DIR/sms-monitor.log"
echo "监控目标：$SMS_DB_PATH" >> "$LOG_DIR/sms-monitor.log"

# 监控短信数据库变化
fswatch -o "$SMS_DB_PATH" | while read event; do
    echo "=== 检测到短信数据库变化 ===" >> "$LOG_DIR/sms-monitor.log"
    echo "时间：$(date)" >> "$LOG_DIR/sms-monitor.log"

    # 调用监控脚本
    python3 "$MONITOR_SCRIPT" >> "$LOG_DIR/sms-monitor.log" 2>&1

    echo "=== 处理完成 ===" >> "$LOG_DIR/sms-monitor.log"
done
EOFSCRIPT

chmod +x "$WATCH_SCRIPT"

# 创建LaunchAgent配置文件
PLIST_FILE="$HOME/Library/LaunchAgents/com.claude.caseos.sms-monitor.plist"
cat > "$PLIST_FILE" << EOF
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.claude.caseos.sms-monitor</string>

    <key>ProgramArguments</key>
    <array>
        <string>$WATCH_SCRIPT</string>
    </array>

    <key>RunAtLoad</key>
    <true/>

    <key>KeepAlive</key>
    <true/>

    <key>StandardOutPath</key>
    <string>$LOG_DIR/sms-monitor-stdout.log</string>

    <key>StandardErrorPath</key>
    <string>$LOG_DIR/sms-monitor-stderr.log</string>
</dict>
</plist>
EOF

echo "✅ LaunchAgent配置文件已创建：$PLIST_FILE"

# 加载LaunchAgent
echo "📝 正在加载LaunchAgent..."
launchctl unload "$PLIST_FILE" 2>/dev/null || true
launchctl load "$PLIST_FILE"

echo "✅ LaunchAgent已加载"
echo ""
echo "📋 调度信息："
echo "   - 运行模式：实时监控（fswatch）"
echo "   - 监控目标：~/Library/SMS/sms.db"
echo "   - 日志路径：~/.claude/skills/case-os/data/sms-monitor*.log"
echo ""
echo "🧪 手动测试："
echo "   python3 $MONITOR_SCRIPT"
echo ""
echo "📝 卸载命令："
echo "   launchctl unload $PLIST_FILE"
echo "   rm $PLIST_FILE"
echo "   rm $WATCH_SCRIPT"

#!/usr/bin/env python3
"""
法院短信实时监控Agent - 实现脚本
监控iOS短信数据库，发现法院相关短信后实时提醒并同步到飞书
"""

import json
import os
import re
import sqlite3
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Any

# 导入飞书通知器
try:
    from feishu_notifier import CourtSMSFeishuNotifier
except ImportError:
    print("⚠️  警告：未找到飞书通知器，飞书功能将不可用")
    CourtSMSFeishuNotifier = None

# ==================== 配置 ====================
SMS_DB_PATH = Path.home() / "Library" / "SMS" / "sms.db"
SCAN_ROOTS = [
    Path.home() / "Documents" / Documents / cases / "案件",
]
ARCHIVE_DIR = "_archive"
COURT_SMS_FILE = "court-sms.json"
STATE_FILE = Path.home() / ".claude" / "skills" / "case-os" / "data" / "sms-monitor-state.json"

# 飞书配置（从环境变量读取）
FEISHU_APP_TOKEN = os.getenv("FEISHU_APP_TOKEN", "")

# 关键词配置
KEYWORDS = {
    "hearing": ["开庭", "庭审", "传票", "出庭", "听证"],
    "evidence": ["举证", "证据交换", "证据材料", "提交证据"],
    "payment": ["缴费", "诉讼费", "受理费", "预交"],
    "judgment": ["判决", "裁定", "送达", "裁判文书"],
}

IGNORE_KEYWORDS = ["验证码", "优惠", "促销", "活动", "中奖"]

# ==================== 短信解析 ====================
class SMSParser:
    """短信解析器"""

    @staticmethod
    def extract_case_number(text: str) -> Optional[str]:
        """提取案号"""
        patterns = [
            r'[(\（]?\d{4}[)\）]?[^\d]+民[一-龥]{1,3}?\d+号',
            r'\(\d{4}\)[^\d]+民[一-龥]{1,3}?\d+号',
        ]
        for pattern in patterns:
            match = re.search(pattern, text)
            if match:
                return match.group(0)
        return None

    @staticmethod
    def extract_date(text: str) -> Optional[str]:
        """提取日期"""
        patterns = [
            r'(\d{4})年(\d{1,2})月(\d{1,2})日',
            r'(\d{1,2})月(\d{1,2})日',
            r'(\d{4})-(\d{1,2})-(\d{1,2})',
        ]
        for pattern in patterns:
            matches = re.findall(pattern, text)
            if matches:
                match = matches[0]
                if len(match) == 3:
                    year, month, day = match
                    if len(year) == 4:
                        return f"{year}-{month.zfill(2)}-{day.zfill(2)}"
                    else:
                        current_year = datetime.now().year
                        return f"{current_year}-{month.zfill(2)}-{day.zfill(2)}"
        return None

    @staticmethod
    def extract_time(text: str) -> Optional[str]:
        """提取时间"""
        patterns = [
            r'(\d{1,2}):(\d{2})',
            r'(\d{1,2})点(\d{1,2})分',
        ]
        for pattern in patterns:
            match = re.search(pattern, text)
            if match:
                hour, minute = match.groups()
                return f"{hour.zfill(2)}:{minute.zfill(2)}"
        return None

    @staticmethod
    def extract_court(text: str) -> Optional[str]:
        """提取法院名称"""
        court_patterns = [
            r'[^，。]+?法院',
            r'[^，。]+?法庭',
        ]
        for pattern in court_patterns:
            match = re.search(pattern, text)
            if match:
                return match.group(0)
        return None

    @staticmethod
    def classify_sms(text: str) -> Optional[str]:
        """分类短信类型"""
        for msg_type, keywords in KEYWORDS.items():
            if any(keyword in text for keyword in keywords):
                return msg_type
        return None


class CourtSMSMonitor:
    """法院短信监控器"""

    def __init__(self):
        self.parser = SMSParser()
        self.state = self.load_state()
        self.feishu_notifier = CourtSMSFeishuNotifier() if CourtSMSFeishuNotifier else None

    def load_state(self) -> Dict:
        """加载监控状态"""
        if STATE_FILE.exists():
            try:
                return json.loads(STATE_FILE.read_text(encoding="utf-8"))
            except (json.JSONDecodeError, IOError):
                pass
        return {"last_sms_id": 0, "processed_ids": []}

    def save_state(self):
        """保存监控状态"""
        STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
        STATE_FILE.write_text(json.dumps(self.state, ensure_ascii=False, indent=2), encoding="utf-8")

    def get_new_messages(self) -> List[Dict]:
        """获取新短信"""
        if not SMS_DB_PATH.exists():
            print(f"⚠️  短信数据库不存在：{SMS_DB_PATH}")
            return []

        try:
            conn = sqlite3.connect(str(SMS_DB_PATH))
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()

            # 查询新短信
            query = """
                SELECT rowid, date, address, text
                FROM message
                WHERE rowid > ?
                ORDER BY date DESC
                LIMIT 50
            """
            cursor.execute(query, (self.state["last_sms_id"],))
            messages = [dict(row) for row in cursor.fetchall()]
            conn.close()

            return messages
        except sqlite3.Error as e:
            print(f"❌ 数据库错误：{e}")
            return []

    def is_court_sms(self, text: str) -> bool:
        """判断是否为法院短信"""
        # 检查忽略关键词
        if any(keyword in text for keyword in IGNORE_KEYWORDS):
            return False

        # 检查法院关键词
        msg_type = self.parser.classify_sms(text)
        return msg_type is not None

    def parse_court_sms(self, sms: Dict) -> Optional[Dict]:
        """解析法院短信"""
        text = sms.get("text", "")
        if not self.is_court_sms(text):
            return None

        msg_type = self.parser.classify_sms(text)

        parsed = {
            "id": f"msg_{sms['rowid']}",
            "type": msg_type,
            "content": text,
            "sender": sms.get("address", "未知"),
            "received_at": datetime.fromtimestamp(sms["date"] / 1000000000).isoformat(),
            "extracted": {},
        }

        # 提取关键信息
        case_number = self.parser.extract_case_number(text)
        if case_number:
            parsed["extracted"]["case_number"] = case_number

        court = self.parser.extract_court(text)
        if court:
            parsed["extracted"]["court"] = court

        date_str = self.parser.extract_date(text)
        time_str = self.parser.extract_time(text)

        if date_str:
            if time_str:
                parsed["extracted"]["datetime"] = f"{date_str} {time_str}"
            else:
                parsed["extracted"]["date"] = date_str

            # 计算截止时间
            try:
                if msg_type == "judgment":
                    # 判决书：上诉期15天
                    delivery_date = datetime.strptime(date_str, "%Y-%m-%d")
                    deadline = delivery_date + timedelta(days=15)
                    parsed["extracted"]["appeal_deadline"] = deadline.strftime("%Y-%m-%d")
                    parsed["deadline"] = deadline.strftime("%Y-%m-%d")
                else:
                    # 其他：使用提取的日期
                    parsed["deadline"] = date_str
            except ValueError:
                pass

        return parsed

    def identify_case(self, sms_info: Dict) -> Optional[Path]:
        """识别短信属于哪个案件"""
        extracted = sms_info.get("extracted", {})
        case_number = extracted.get("case_number")
        court = extracted.get("court")

        # 遍历案件文件夹
        for root in SCAN_ROOTS:
            if not root.exists():
                continue
            for case_dir in root.rglob("CLAUDE.md"):
                case_path = case_dir.parent

                # 读取案件信息
                claude_md = case_path / "CLAUDE.md"
                if claude_md.exists():
                    content = claude_md.read_text(encoding="utf-8")

                    # 匹配案号
                    if case_number and case_number in content:
                        return case_path

                    # 匹配法院
                    if court and court in content:
                        return case_path

        return None

    def save_to_archive(self, case_path: Path, sms_info: Dict):
        """保存到案件档案"""
        archive_dir = case_path / ARCHIVE_DIR
        archive_dir.mkdir(parents=True, exist_ok=True)

        court_sms_file = archive_dir / COURT_SMS_FILE

        # 读取现有数据
        if court_sms_file.exists():
            try:
                data = json.loads(court_sms_file.read_text(encoding="utf-8"))
            except (json.JSONDecodeError, IOError):
                data = {"messages": []}
        else:
            data = {"messages": []}

        # 添加新短信
        data["messages"].append(sms_info)

        # 保存
        court_sms_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")

    def sync_to_feishu(self, sms_info: Dict, case_name: str):
        """同步到飞书（机器人消息 + 多维表格）"""
        if not self.feishu_notifier:
            print("⚠️  飞书通知器未配置，跳过同步")
            return

        try:
            # 发送飞书机器人消息（卡片形式）
            # 并写入多维表格
            self.feishu_notifier.send_court_sms(sms_info, case_name)
        except Exception as e:
            print(f"❌ 飞书同步失败：{e}")

    def get_type_label(self, msg_type: str) -> str:
        """获取类型标签"""
        labels = {
            "hearing": "开庭通知",
            "evidence": "举证期限",
            "payment": "缴费通知",
            "judgment": "判决送达",
        }
        return labels.get(msg_type, "其他")

    def send_notification(self, sms_info: Dict, case_name: str):
        """发送通知"""
        msg_type = self.get_type_label(sms_info["type"])
        extracted = sms_info.get("extracted", {})

        # 构建通知内容
        title = f"🔔 法院短信提醒：{msg_type}"
        message_parts = [
            f"案件：{case_name}",
        ]

        if extracted.get("court"):
            message_parts.append(f"法院：{extracted['court']}")

        if extracted.get("date"):
            deadline_str = extracted["date"]
            if extracted.get("appeal_deadline"):
                deadline_str = f"上诉期截止：{extracted['appeal_deadline']}"
            message_parts.append(f"时间：{deadline_str}")

        if extracted.get("case_number"):
            message_parts.append(f"案号：{extracted['case_number']}")

        message = "\n".join(message_parts)

        # 发送macOS桌面通知
        os.system(f'osascript -e \'display notification "{message}" with title "{title}"\'')

        print(f"\n🔔 {title}")
        print(message)
        print()

    def process(self):
        """处理新短信"""
        messages = self.get_new_messages()

        if not messages:
            print("✅ 无新短信")
            return

        print(f"📱 发现 {len(messages)} 条新短信")

        court_sms_count = 0
        for sms in messages:
            sms_info = self.parse_court_sms(sms)
            if not sms_info:
                continue

            court_sms_count += 1
            print(f"\n{'='*50}")
            print(f"🏛️  发现法院短信：{self.get_type_label(sms_info['type'])}")
            print(f"内容：{sms_info['content'][:100]}...")
            print(f"{'='*50}")

            # 识别案件
            case_path = self.identify_case(sms_info)
            if case_path:
                case_name = case_path.name
                print(f"✅ 匹配案件：{case_name}")

                # 保存到档案
                self.save_to_archive(case_path, sms_info)

                # 同步到飞书
                self.sync_to_feishu(sms_info, case_name)

                # 发送通知
                self.send_notification(sms_info, case_name)
            else:
                print(f"⚠️  未找到匹配案件")
                # TODO: 保存到未分类记录

            # 标记为已处理
            self.state["processed_ids"].append(sms["rowid"])

        # 更新状态
        if messages:
            self.state["last_sms_id"] = max(msg["rowid"] for msg in messages)
            self.save_state()

        if court_sms_count > 0:
            print(f"\n✅ 处理完成：发现 {court_sms_count} 条法院短信")
        else:
            print(f"\n✅ 处理完成：无法院短信")


def main():
    """主函数"""
    print("🔍 法院短信监控Agent启动...")
    print(f"📂 监控数据库：{SMS_DB_PATH}")
    print(f"📂 扫描案件根目录：{SCAN_ROOTS}")
    print()

    monitor = CourtSMSMonitor()
    monitor.process()


if __name__ == "__main__":
    sys.exit(main())

display_name: "案件OS"
short_description: "全局案件管理层：自动初始化案件三文件、恢复上下文并在任务后写回结论。"
default_prompt: |
  你是“案件OS（case-os）”，用于中国律师办案流程中的全局案件管理。

  你的固定职责：
  1) 初始化案件工作区并补建核心文件；
  2) 在任务前恢复案件上下文；
  3) 在任务后写回关键结论；
  4) 维护可直接执行的任务面板。

  进入任一案件目录后，必须先检查并确保以下文件存在：
  - 00_案件概览.md
  - 01_任务面板.md
  - 99_工作日志.md

  如文件缺失，立即补建并写入基础内容。

  初始化触发语包括但不限于：
  - 初始化本案件
  - 初始化这个案子
  - 开始整理本案
  - 建立案件工作区
  - 启用案件OS

  初始化必须执行：
  1. 扫描当前案件目录及子目录；
  2. 识别基础案情与关键材料；
  3. 生成或更新三核心文件；
  4. 追加第一条或最新一条工作日志。

  材料扫描规则：
  - 文件类型优先：PDF > DOCX > TXT > Markdown > 图片
  - 目录优先：立案材料/ > materials/ > evidence/ > 根目录 > 其他子目录
  - 关键文件优先：起诉状、判决书/裁定书、合同、聊天记录、证据目录、借条、转账记录、开庭传票/举证通知/执行通知、会见笔录/阅卷笔录

  输出约束：
  - 不得编造案件事实；
  - 不得跨案件复用事实；
  - 推测必须标注“待核实”；
  - 默认不做长篇实体法律分析与完整文书起草；
  - 若材料不足，必须明确列出缺失材料与下一步可执行动作。

  协同规则：
  - 执行型任务转交 ai-law-firm 与 legal-aid-lawyer；
  - 执行完成后回写案件概览、任务面板（必要时）与工作日志（默认追加）。

#!/bin/bash
# 测试get_agent_status函数

HOME_PATH="/Users/Apple"

get_agent_status() {
    local agent_name=$1
    local plist_file="$HOME_PATH/Library/LaunchAgents/com.claude.caseos.${agent_name}.plist"
    local label=$(launchctl list | grep "com.claude.caseos.${agent_name}" | awk '{print $1}')

    echo "DEBUG: agent_name=$agent_name"
    echo "DEBUG: plist_file=$plist_file"
    echo "DEBUG: label=$label"
    echo "DEBUG: 文件存在测试: $( [ -f "$plist_file" ] && echo "存在" || echo "不存在" )"

    if [ ! -f "$plist_file" ]; then
        echo "未安装"
    elif [ -n "$label" ]; then
        echo "运行中"
    else
        echo "已安装但未运行"
    fi
}

echo "=== 测试 court-sms-monitor ==="
get_agent_status "court-sms-monitor"

echo ""
echo "=== 测试 weekly-scan ==="
get_agent_status "weekly-scan"

#!/bin/bash
# 案件周度扫描Agent - 安装脚本

set -e

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
AGENT_DIR="$SCRIPT_DIR"
SCAN_SCRIPT="$AGENT_DIR/scan.py"
PLIST_FILE="$HOME/Library/LaunchAgents/com.claude.caseos.weekly-scan.plist"

echo "🔧 安装案件周度扫描Agent..."

# 检查Python3
if ! command -v python3 &> /dev/null; then
    echo "❌ 错误：未找到python3"
    exit 1
fi

# 确保脚本可执行
chmod +x "$SCAN_SCRIPT"

# 创建LaunchAgent配置文件
cat > "$PLIST_FILE" << EOF
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
    <key>Label</key>
    <string>com.claude.caseos.weekly-scan</string>

    <key>ProgramArguments</key>
    <array>
        <string>/usr/bin/python3</string>
        <string>$SCAN_SCRIPT</string>
    </array>

    <key>StartCalendarInterval</key>
    <dict>
        <key>Weekday</key>
        <integer>1</integer>
        <key>Hour</key>
        <integer>8</integer>
        <key>Minute</key>
        <integer>0</integer>
    </dict>

    <key>StandardOutPath</key>
    <string>$HOME/.claude/skills/case-os/data/launchd-stdout.log</string>

    <key>StandardErrorPath</key>
    <string>$HOME/.claude/skills/case-os/data/launchd-stderr.log</string>

    <key>RunAtLoad</key>
    <false/>

    <key>KeepAlive</key>
    <false/>
</dict>
</plist>
EOF

echo "✅ LaunchAgent配置文件已创建：$PLIST_FILE"

# 加载LaunchAgent
echo "📝 正在加载LaunchAgent..."
launchctl unload "$PLIST_FILE" 2>/dev/null || true
launchctl load "$PLIST_FILE"

echo "✅ LaunchAgent已加载"
echo ""
echo "📋 调度信息："
echo "   - 运行时间：每周一早上8点"
echo "   - 日志路径：~/.claude/skills/case-os/data/launchd-*.log"
echo ""
echo "🧪 手动测试："
echo "   python3 $SCAN_SCRIPT"
echo ""
echo "📝 卸载命令："
echo "   launchctl unload $PLIST_FILE"
echo "   rm $PLIST_FILE"

#!/usr/bin/env python3
"""
案件周度扫描Agent - 实现脚本
安全设计：只读扫描，无写入权限，无网络访问
"""

import json
import os
import re
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Any


# ==================== 配置 ====================
SCAN_ROOTS = [
    Path.home() / "Documents" / Documents / cases / "案件",
]

STATE_DIR = Path.home() / ".claude" / "skills" / "case-os" / "data"
LAST_SCAN_FILE = STATE_DIR / "last-scan-result.txt"
PENDING_QUEUE_FILE = STATE_DIR / "pending-files.json"
OUTPUT_DIR = Path.home() / "Desktop"

# 忽略的文件/目录
IGNORE_PATTERNS = [
    r"^\.",           # 以.开头的隐藏文件
    r"^\.DS_Store$",
    r"^~.*",          # ~开头的临时文件
    r"^_archive",     # 归档目录
]

# ==================== 数据结构 ====================
class ScanResult:
    def __init__(self):
        self.urgent = []      # 紧急待办
        self.warning = []     # 警告
        self.info = []        # 信息
        self.errors = []      # 错误
        self.scan_time = datetime.now()


class CaseItem:
    def __init__(self, name: str, path: Path):
        self.name = name
        self.path = path
        self.claude_md = path / "CLAUDE.md"
        self.tracking_md = path / "阶段追踪.md"
        self.memo_md = path / "案件备忘录.md"
        self.court_sms = path / "_archive" / "court-sms.json"


# ==================== 核心扫描函数 ====================
def find_case_folders() -> List[CaseItem]:
    """扫描所有案件文件夹（查找CLAUDE.md）"""
    cases = []
    for root in SCAN_ROOTS:
        if not root.exists():
            continue
        for item in root.rglob("CLAUDE.md"):
            case_dir = item.parent
            # 跳归档目录
            if "_archive" in str(case_dir):
                continue
            cases.append(CaseItem(case_dir.name, case_dir))
    return cases


def check_court_reminder(case: CaseItem) -> Dict[str, Any]:
    """检查开庭提醒（从court-sms.json）"""
    if not case.court_sms.exists():
        return None

    try:
        data = json.loads(case.court_sms.read_text(encoding="utf-8"))
        if not data.get("messages"):
            return None

        # 找7天内的开庭
        upcoming = []
        now = datetime.now()
        for msg in data["messages"]:
            if msg.get("type") == "开庭通知":
                try:
                    court_date = datetime.strptime(msg.get("date", ""), "%Y-%m-%d")
                    days_until = (court_date - now).days
                    if 0 <= days_until <= 7:
                        upcoming.append({
                            "case": case.name,
                            "date": msg.get("date"),
                            "court": msg.get("court", "未知法院"),
                            "days_until": days_until,
                        })
                except ValueError:
                    pass

        return upcoming[0] if upcoming else None
    except (json.JSONDecodeError, KeyError):
        return None


def check_evidence_gaps(case: CaseItem) -> List[str]:
    """检查证据缺口（从证据卡片库）"""
    evidence_dir = case.path / "_intermediate" / "证据卡片库"
    if not evidence_dir.exists():
        return []

    gaps = []
    for file in evidence_dir.rglob("*.md"):
        content = file.read_text(encoding="utf-8")
        # 检查是否标记为待补充
        if re.search(r'\[待补充\]|\[缺失\]|\[待提供\]', content):
            gaps.append(file.stem)

    return gaps


def check_pending_tasks(case: CaseItem) -> List[str]:
    """检查待办事项（从阶段追踪.md）"""
    if not case.tracking_md.exists():
        return []

    content = case.tracking_md.read_text(encoding="utf-8")
    pending = []

    # 查找未完成的checkbox
    for line in content.split('\n'):
        if re.match(r'^- \[ \]', line):
            pending.append(line.strip()[6:])  # 去掉"- [ ] "

    return pending


def check_last_update(case: CaseItem) -> int:
    """检查最后更新时间（CLAUDE.md修改时间）"""
    if not case.claude_md.exists():
        return None

    mtime = case.claude_md.stat().st_mtime
    last_update = datetime.fromtimestamp(mtime)
    days_since = (datetime.now() - last_update).days
    return days_since


def detect_new_files(case: CaseItem, known_files: set) -> List[str]:
    """检测新增文件"""
    new_files = []
    for file in case.path.iterdir():
        if should_ignore_file(file.name):
            continue
        if file.is_file():
            key = f"{case.name}/{file.name}"
            if key not in known_files:
                new_files.append(file.name)
    return new_files


def should_ignore_file(name: str) -> bool:
    """判断是否应忽略该文件"""
    for pattern in IGNORE_PATTERNS:
        if re.match(pattern, name):
            return True
    return False


# ==================== 报告生成 ====================
def generate_report(results: ScanResult, case_count: int) -> str:
    """生成Markdown报告"""
    lines = [
        "# 案件周度扫描报告",
        "",
        f"**扫描时间**：{results.scan_time.strftime('%Y-%m-%d %H:%M')}",
        f"**扫描范围**：{case_count}个案件",
        "",
    ]

    # 紧急待办
    if results.urgent:
        lines.extend([
            "## 🚨 紧急待办（需要立即处理）",
            "",
        ])
        for item in results.urgent:
            lines.extend(format_urgent_item(item))
            lines.append("")

    # 警告
    if results.warning:
        lines.extend([
            "## ⚠️ 本周待办（建议本周处理）",
            "",
        ])
        for item in results.warning:
            lines.extend(format_warning_item(item))
            lines.append("")

    # 新增材料
    if results.info:
        lines.extend([
            "## 📄 新增材料（本周）",
            "",
        ])
        for item in results.info:
            lines.extend(format_info_item(item))
            lines.append("")

    # 错误
    if results.errors:
        lines.extend([
            "## ⚠️ 扫描错误",
            "",
        ])
        for error in results.errors:
            lines.append(f"- {error}")
        lines.append("")

    # 下次扫描时间
    next_scan = results.scan_time + timedelta(days=7)
    lines.extend([
        "---",
        "",
        f"**下次扫描时间**：{next_scan.strftime('%Y-%m-%d %H:%M')}",
    ])

    return "\n".join(lines)


def format_urgent_item(item: Dict) -> List[str]:
    """格式化紧急项"""
    if item["type"] == "court_reminder":
        return [
            f"### {item['case']} - 开庭提醒",
            f"- 开庭时间：{item['date']}",
            f"- 法院：{item['court']}",
            f"- 距离：{item['days_until']}天",
        ]
    return []


def format_warning_item(item: Dict) -> List[str]:
    """格式化警告项"""
    if item["type"] == "evidence_gap":
        lines = [f"### {item['case']} - 证据缺口"]
        for gap in item["gaps"]:
            lines.append(f"- 缺失证据：{gap}")
        return lines
    elif item["type"] == "stale":
        return [
            f"### {item['case']} - 久未更新",
            f"- 最后更新：{item['last_update']}（{item['days_since']}天前）",
            "- 建议：检查是否有新进展",
        ]
    elif item["type"] == "pending_task":
        lines = [f"### {item['case']} - 待办事项"]
        for task in item["tasks"][:3]:  # 最多显示3条
            lines.append(f"- {task}")
        if len(item["tasks"]) > 3:
            lines.append(f"- ...还有{len(item['tasks'])-3}项待办")
        return lines
    return []


def format_info_item(item: Dict) -> List[str]:
    """格式化信息项"""
    if item["type"] == "new_files":
        lines = [f"### {item['case']}"]
        for file in item["files"][:5]:  # 最多显示5个
            lines.append(f"- 新增：{file}")
        if len(item["files"]) > 5:
            lines.append(f"- ...还有{len(item['files'])-5}个文件")
        return lines
    return []


# ==================== 主流程 ====================
def load_known_files() -> set:
    """加载已知文件列表"""
    if PENDING_QUEUE_FILE.exists():
        try:
            data = json.loads(PENDING_QUEUE_FILE.read_text(encoding="utf-8"))
            return {f"{item['case']}/{item['file']}" for item in data.get("pending", [])}
        except (json.JSONDecodeError, KeyError):
            pass
    return set()


def save_scan_result(report: str):
    """保存扫描结果"""
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    LAST_SCAN_FILE.write_text(report, encoding="utf-8")


def save_report_to_desktop(report: str):
    """保存报告到桌面"""
    filename = f"案件周度扫描报告-{datetime.now().strftime('%Y-%m-%d')}.md"
    output_path = OUTPUT_DIR / filename
    output_path.write_text(report, encoding="utf-8")
    return output_path


def main():
    """主扫描流程"""
    print("🔍 开始扫描案件...")
    results = ScanResult()
    known_files = load_known_files()

    # 扫描所有案件
    cases = find_case_folders()
    print(f"发现 {len(cases)} 个案件文件夹")

    for case in cases:
        print(f"  扫描：{case.name}")

        # 检查开庭提醒（紧急）
        court_info = check_court_reminder(case)
        if court_info:
            results.urgent.append({
                "type": "court_reminder",
                **court_info,
            })

        # 检查证据缺口（警告）
        gaps = check_evidence_gaps(case)
        if gaps:
            results.warning.append({
                "type": "evidence_gap",
                "case": case.name,
                "gaps": gaps,
            })

        # 检查待办事项（警告）
        pending = check_pending_tasks(case)
        if pending:
            results.warning.append({
                "type": "pending_task",
                "case": case.name,
                "tasks": pending,
            })

        # 检查久未更新（警告）
        days_since = check_last_update(case)
        if days_since and days_since >= 7:
            mtime = datetime.fromtimestamp(case.claude_md.stat().st_mtime)
            results.warning.append({
                "type": "stale",
                "case": case.name,
                "days_since": days_since,
                "last_update": mtime.strftime("%Y-%m-%d"),
            })

        # 检测新增文件（信息）
        new_files = detect_new_files(case, known_files)
        if new_files:
            results.info.append({
                "type": "new_files",
                "case": case.name,
                "files": new_files,
            })

    # 生成报告
    report = generate_report(results, len(cases))

    # 保存报告
    save_scan_result(report)
    output_path = save_report_to_desktop(report)

    # 输出摘要
    print("\n" + "="*50)
    print(f"✅ 扫描完成")
    print(f"📊 紧急待办：{len(results.urgent)}项")
    print(f"⚠️  本周待办：{len(results.warning)}项")
    print(f"📄 新增材料：{sum(len(i.get('files', [])) for i in results.info)}个文件")
    print(f"💾 报告已保存：{output_path}")
    print("="*50)

    # 如果有紧急待办，单独提醒
    if results.urgent:
        print("\n🚨 紧急待办提醒：")
        for item in results.urgent:
            print(f"  - {item['case']}: {item['date']} 开庭")

    print("\n⏸️  扫描完成，等待用户确认...")
    return 0


if __name__ == "__main__":
    sys.exit(main())