ft-dev 1.0.1
Modern file transfer tool with CLI and jRPC API - Development Edition
Requires Python: >=3.10
Keywords: file-transfer, jrpc, cli, cloudflare, r2, cdn, ft-dev

# `filetrans` (ft) 全栈开发指导手册 v1.0

## 目录

- [1. 项目定位](#1-项目定位)
- [2. 核心技术栈](#2-核心技术栈)
- [3. 系统架构](#3-系统架构)
- [4. 数据库 Schema](#4-数据库-schema)
- [5. 服务端详细设计](#5-服务端详细设计)
- [6. API 完全规格 (jRPC)](#6-api-完全规格-jrpc)
- [7. CLI 详细设计](#7-cli-详细设计)
- [8. Web 管理端设计](#8-web-管理端设计)
- [9. 自动化清理机制](#9-自动化清理机制)
- [10. 工程化发布指导](#10-工程化发布指导)
- [11. 增强版功能矩阵](#11-增强版功能矩阵)
- [12. 下一步开发方向](#12-下一步开发方向)

---

## 1. 项目定位

### 1.1 核心理念

**`filetrans` (简称 `ft`)** 是一个现代化的文件传输工具，采用"哑客户端 + 智能服务端 + 灵活存储后端"的架构设计。

### 1.2 工作流程

```
┌─────────────┐      jRPC       ┌─────────────┐     Presigned URL     ┌─────────────┐
│   CLI 客户端  │ ──────────────> │   服务端      │ ──────────────────> │  存储后端     │
│  (ft -p)    │ <────────────── │  (FastAPI)  │ <────────────────── │  (S3/Local) │
└─────────────┘   Code + URL   └─────────────┘    Upload Success    └─────────────┘
       │                             │                                    │
       │                             ├────────────────────────────────────┤
       │                             │  元数据维护 (SQLite)                │
       ▼                             ▼                                    ▼
  ┌─────────────┐              ┌─────────────┐                    ┌─────────────┐
  │ 打印 Code   │              │  Web 预览    │                    │  物理文件     │
  │ 分享链接    │              │  /g/<code>  │                    │  按需清理     │
  └─────────────┘              └─────────────┘                    └─────────────┘
```

### 1.3 设计目标

| 目标 | 说明 |
|------|------|
| **简单性** | 客户端只需一个命令完成上传/下载 |
| **安全性** | 支持 Token 认证、密码保护、端到端加密 |
| **可扩展性** | 支持多种存储后端（S3、Local、未来可扩展） |
| **可管理性** | 提供完整的 Web 管理后台和 API |
| **共享性** | 支持服务器内文件共享（Public/Private 模式） |

### 1.4 共享模式说明

| 模式 | 说明 |
|------|------|
| **`-public`** (默认) | 同服务器内所有用户可见（通过 `ft ls` 可列出） |
| **`-private`** | 仅上传者可见 |

> **注意**：无论 Public/Private，所有文件都可以通过 Code 下载。Public 模式只是让其他用户能在 `ft ls` 中看到。

---

## 2. 核心技术栈

### 2.1 技术选型总览

```
┌─────────────────────────────────────────────────────────────────────┐
│                         filetrans 技术栈                            │
├─────────────┬─────────────┬─────────────┬───────────────────────────┤
│    CLI      │   Backend   │  Database   │       Storage            │
├─────────────┼─────────────┼─────────────┼───────────────────────────┤
│  Typer      │  FastAPI    │  SQLite     │  S3 (via Boto3)          │
│  Rich       │  Uvicorn    │  SQLAlchemy │  Local Filesystem        │
│  Requests   │  jRPC 2.0   │  Alembic?   │  (未来: MinIO, Azure)    │
│  Pyclip     │  Pydantic   │             │                           │
└─────────────┴─────────────┴─────────────┴───────────────────────────┘
```

### 2.2 详细依赖说明

#### 2.2.1 CLI 客户端

| 依赖 | 版本要求 | 用途 |
|------|---------|------|
| `typer[all]` | ≥0.9.0 | 命令行参数解析、子命令管理 |
| `rich` | ≥13.0.0 | 进度条、表格输出、终端美化 |
| `requests` | ≥2.28.0 | HTTP 客户端，与 jRPC 服务端通信 |
| `pyclip` | ≥0.7.0 | 跨平台剪贴板操作 |

#### 2.2.2 服务端

| 依赖 | 版本要求 | 用途 |
|------|---------|------|
| `fastapi` | ≥0.100.0 | Web 框架，提供 jRPC 路由 |
| `uvicorn[standard]` | ≥0.23.0 | ASGI 服务器 |
| `pydantic` | ≥2.0.0 | 数据验证与序列化 |
| `sqlalchemy` | ≥2.0.0 | ORM，数据库抽象层 |
| `boto3` | ≥1.28.0 | AWS S3 SDK |

#### 2.2.3 Web 前端

| 技术 | 说明 |
|------|------|
| 原生 HTML/CSS/JS | 无构建工具依赖 |
| Tailwind CSS (CDN) | 快速样式开发 |
| Axios (CDN) | AJAX 请求 |
| Chart.js (CDN) | Dashboard 图表 |

---

## 3. 系统架构

### 3.1 目录结构

```
filetrans/
├── filetrans/
│   ├── __init__.py
│   ├── cli.py                 # Typer CLI 入口
│   ├── client/
│   │   ├── __init__.py
│   │   ├── api.py             # jRPC 客户端封装
│   │   ├── upload.py          # 上传逻辑
│   │   ├── download.py        # 下载逻辑
│   │   └── clipboard.py       # 剪贴板处理
│   ├── server/
│   │   ├── __init__.py
│   │   ├── main.py            # FastAPI 应用入口
│   │   ├── jrpc.py            # jRPC 方法注册与处理
│   │   ├── storage/
│   │   │   ├── __init__.py
│   │   │   ├── base.py        # Storage 抽象基类
│   │   │   ├── s3.py          # S3 实现
│   │   │   └── local.py       # Local 实现
│   │   ├── models.py          # SQLAlchemy 模型
│   │   ├── database.py        # 数据库连接管理
│   │   ├── janitor.py         # 后台清理任务
│   │   └── config.py          # 配置管理
│   └── web/
│       ├── __init__.py
│       ├── templates/
│       │   ├── base.html
│       │   ├── download.html  # /g/<code> 下载页面
│       │   └── admin.html     # 管理后台 SPA
│       └── static/
│           └── css/
│           └── js/
├── tests/
│   ├── __init__.py
│   ├── test_client.py
│   ├── test_server.py
│   └── conftest.py
├── migrations/                # Alembic 迁移文件（如使用）
├── pyproject.toml
├── README.md
└── .env.example
```

### 3.2 配置文件

#### 3.2.1 CLI 客户端配置 (`~/.config/filetrans/config.json`)

```json
{
  "server_url": "https://filetrans.example.com",
  "api_token": "ft_xxxxxxxxxxxxxxxxxxxx",
  "default_ttl": "24h",
  "default_group": "default"
}
```

#### 3.2.2 服务端配置 (`config.yaml` 或环境变量)

```yaml
server:
  host: "0.0.0.0"
  port: 8866
  workers: 4

database:
  url: "sqlite:///./filetrans.db"

storage:
  backend: "s3"  # or "local"
  local:
    root_dir: "./data/files"
  s3:
    bucket: "filetrans-bucket"
    region: "us-east-1"
    endpoint_url: "https://s3.amazonaws.com"

security:
  jwt_secret: "CHANGE_ME_IN_PRODUCTION"
  token_length: 32

janitor:
  interval_seconds: 60
  disk_threshold_percent: 10
```

---

## 4. 数据库 Schema

### 4.1 完整表结构

```sql
-- ==================================================
-- filetrans Database Schema v1.1
-- ==================================================

-- 系统配置表 (单行)
CREATE TABLE system_config (
    id INTEGER PRIMARY KEY CHECK (id = 1),
    server_quota_bytes INTEGER DEFAULT 107374182400,  -- 100GB 服务器默认配额
    server_current_bytes INTEGER DEFAULT 0,            -- 当前使用量
    storage_backend VARCHAR(10) DEFAULT 'LOCAL',        -- 'S3' or 'LOCAL'
    janitor_interval_seconds INTEGER DEFAULT 60,
    max_file_size INTEGER DEFAULT 536870912,           -- 单文件最大 512MB
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
);

-- 用户/API Token 表
CREATE TABLE tokens (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    token VARCHAR(64) UNIQUE NOT NULL,           -- API Token (ft_xxx...)
    user_id VARCHAR(50) NOT NULL,                -- 用户标识
    label VARCHAR(100),                          -- Token 描述 (如 "iPhone Token")
    permissions TEXT,                            -- JSON: ["read", "write", "admin"]
    quota_max_size INTEGER DEFAULT 10737418240,  -- 10GB 默认配额
    quota_current_size INTEGER DEFAULT 0,        -- 当前使用量
    max_file_size INTEGER DEFAULT 536870912,     -- 单文件最大 512MB
    rate_limit_rps INTEGER DEFAULT 10,           -- 每秒请求限制
    is_active BOOLEAN DEFAULT TRUE,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    expires_at DATETIME,
    INDEX idx_token (token),
    INDEX idx_user_id (user_id)
);

-- 文件传输记录表 (核心表)
CREATE TABLE transfers (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    code VARCHAR(10) UNIQUE NOT NULL,            -- 4-6 位短码 (下载提取码)
    owner_id VARCHAR(50) NOT NULL,               -- 关联 tokens.user_id
    group_name VARCHAR(50) DEFAULT 'default',    -- 命名空间/分组

    -- 文件信息
    file_name TEXT NOT NULL,
    file_size INTEGER NOT NULL,
    mime_type VARCHAR(100),
    file_hash VARCHAR(64),                       -- SHA-256

    -- 存储信息
    storage_mode VARCHAR(10) NOT NULL,           -- 'S3' or 'LOCAL'
    storage_key TEXT NOT NULL,                   -- S3 Key 或本地相对路径

    -- 安全信息
    pwd_hash TEXT,                               -- bcrypt 密码哈希 (可为 NULL)
    is_e2ee BOOLEAN DEFAULT FALSE,               -- 端到端加密标记
    is_public BOOLEAN DEFAULT TRUE,              -- 共享模式: public=服务器内可见, private=仅自己可见

    -- 生命周期控制
    status VARCHAR(20) DEFAULT 'pending',        -- pending, active, expired, revoked
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    expire_at DATETIME NOT NULL,
    max_downloads INTEGER DEFAULT -1,            -- -1 = 无限制
    current_downloads INTEGER DEFAULT 0,

    -- 元数据
    upload_ip VARCHAR(45),                       -- 支持 IPv6
    user_agent TEXT,

    INDEX idx_code (code),
    INDEX idx_owner (owner_id),
    INDEX idx_status (status),
    INDEX idx_expire (expire_at),
    INDEX idx_group (group_name),
    INDEX idx_public (is_public),                 -- 用于快速查询公开文件
    FOREIGN KEY (owner_id) REFERENCES tokens(user_id) ON DELETE CASCADE
);

-- 下载日志表
CREATE TABLE download_logs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    transfer_code VARCHAR(10) NOT NULL,
    downloaded_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    ip_address VARCHAR(45),
    user_agent TEXT,
    success BOOLEAN DEFAULT TRUE,
    INDEX idx_code (transfer_code),
    INDEX idx_time (downloaded_at),
    FOREIGN KEY (transfer_code) REFERENCES transfers(code) ON DELETE CASCADE
);

-- Webhook 配置表
CREATE TABLE webhooks (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    owner_id VARCHAR(50) NOT NULL,
    event_type VARCHAR(50) NOT NULL,             -- 'download', 'expire', 'upload'
    url TEXT NOT NULL,
    secret VARCHAR(64),                          -- HMAC 签名密钥
    is_active BOOLEAN DEFAULT TRUE,
    INDEX idx_owner (owner_id),
    INDEX idx_event (event_type)
);
```

### 4.2 SQLAlchemy 模型定义

```python
# filetrans/server/models.py

from sqlalchemy import Column, Integer, String, Boolean, DateTime, BigInteger, Text, ForeignKey, Index, CheckConstraint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func
from datetime import datetime

Base = declarative_base()

class SystemConfig(Base):
    """系统配置表（单行）"""
    __tablename__ = 'system_config'

    id = Column(Integer, primary_key=True, default=1)
    server_quota_bytes = Column(BigInteger, default=100 * 1024 * 1024 * 1024)  # 100GB
    server_current_bytes = Column(BigInteger, default=0)
    storage_backend = Column(String(10), default='LOCAL')
    janitor_interval_seconds = Column(Integer, default=60)
    max_file_size = Column(BigInteger, default=512 * 1024 * 1024)  # 512MB
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())

    __table_args__ = (
        CheckConstraint('id = 1', name='check_single_row'),
    )

class Token(Base):
    __tablename__ = 'tokens'

    id = Column(Integer, primary_key=True, autoincrement=True)
    token = Column(String(64), unique=True, nullable=False, index=True)
    user_id = Column(String(50), nullable=False, index=True)
    label = Column(String(100))
    permissions = Column(Text)  # JSON 存储
    quota_max_size = Column(BigInteger, default=10 * 1024 * 1024 * 1024)  # 10GB
    quota_current_size = Column(BigInteger, default=0)
    max_file_size = Column(BigInteger, default=512 * 1024 * 1024)  # 512MB
    rate_limit_rps = Column(Integer, default=10)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=func.now())
    expires_at = Column(DateTime, nullable=True)

class Transfer(Base):
    __tablename__ = 'transfers'

    id = Column(Integer, primary_key=True, autoincrement=True)
    code = Column(String(10), unique=True, nullable=False, index=True)
    owner_id = Column(String(50), nullable=False, index=True)
    group_name = Column(String(50), default='default', index=True)

    # 文件信息
    file_name = Column(Text, nullable=False)
    file_size = Column(BigInteger, nullable=False)
    mime_type = Column(String(100))
    file_hash = Column(String(64))

    # 存储信息
    storage_mode = Column(String(10), nullable=False)  # 'S3' or 'LOCAL'
    storage_key = Column(Text, nullable=False)

    # 安全信息
    pwd_hash = Column(String)
    is_e2ee = Column(Boolean, default=False)
    is_public = Column(Boolean, default=True, index=True)  # 共享模式

    # 生命周期
    status = Column(String(20), default='pending')  # pending, active, expired, revoked
    created_at = Column(DateTime, default=func.now())
    expire_at = Column(DateTime, nullable=False, index=True)
    max_downloads = Column(Integer, default=-1)
    current_downloads = Column(Integer, default=0)

    # 元数据
    upload_ip = Column(String(45))
    user_agent = Column(Text)

class DownloadLog(Base):
    __tablename__ = 'download_logs'

    id = Column(Integer, primary_key=True, autoincrement=True)
    transfer_code = Column(String(10), nullable=False, index=True)
    downloaded_at = Column(DateTime, default=func.now(), index=True)
    ip_address = Column(String(45))
    user_agent = Column(Text)
    success = Column(Boolean, default=True)

class Webhook(Base):
    __tablename__ = 'webhooks'

    id = Column(Integer, primary_key=True, autoincrement=True)
    owner_id = Column(String(50), nullable=False, index=True)
    event_type = Column(String(50), nullable=False, index=True)  # download, expire, upload
    url = Column(Text, nullable=False)
    secret = Column(String(64))
    is_active = Column(Boolean, default=True)
```

---

## 5. 服务端详细设计

### 5.1 存储目录隔离策略

#### 5.1.1 设计原则

| 原则 | 说明 |
|------|------|
| **用户隔离** | 每个 Token 的文件存储在独立的用户目录下 |
| **不可变路径** | 存储路径生成后不再更改，便于追踪 |
| **可预测性** | 支持从 Code 反推存储路径（便于调试） |

#### 5.1.2 路径生成规则

**S3 模式**:
```
s3://{bucket}/{user_id}/{code}_{uuid[:8]}/{filename}
```

**Local 模式**:
```
{STORAGE_ROOT}/{user_id}/{code}/{filename}
```

#### 5.1.3 代码实现

```python
# filetrans/server/storage/base.py

from abc import ABC, abstractmethod
from typing import Tuple, Optional
import uuid

class StorageBackend(ABC):
    """存储后端抽象基类"""

    def __init__(self, config: dict):
        self.config = config

    @abstractmethod
    async def generate_upload_url(
        self,
        user_id: str,
        code: str,
        filename: str,
        ttl_seconds: int
    ) -> Tuple[str, str]:  # (upload_url, storage_key)
        """生成预签名上传 URL"""
        pass

    @abstractmethod
    async def generate_download_url(
        self,
        storage_key: str,
        ttl_seconds: int
    ) -> str:
        """生成预签名下载 URL"""
        pass

    @abstractmethod
    async def delete_file(self, storage_key: str) -> bool:
        """物理删除文件"""
        pass

    @abstractmethod
    async def file_exists(self, storage_key: str) -> bool:
        """检查文件是否存在"""
        pass

    def generate_storage_key(self, user_id: str, code: str, filename: str) -> str:
        """生成标准化的存储 Key"""
        uuid_suffix = uuid.uuid4().hex[:8]
        return f"{user_id}/{code}_{uuid_suffix}/{filename}"
```

### 5.2 jRPC 方法注册

```python
# filetrans/server/jrpc.py

from fastapi import APIRouter, Header, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import Optional, List
import bcrypt

from .database import get_db
from .models import Transfer, Token
from .storage.base import StorageBackend

router = APIRouter()

# ==================================================
# Pydantic 模型定义
# ==================================================

class JSONRPCRequest(BaseModel):
    jsonrpc: str = "2.0"
    method: str
    params: dict = {}
    id: Optional[str] = None

class JSONRPCResponse(BaseModel):
    jsonrpc: str = "2.0"
    result: Optional[dict] = None
    error: Optional[dict] = None
    id: Optional[str] = None

class InitUploadParams(BaseModel):
    filename: str = Field(..., description="原始文件名")
    filesize: int = Field(..., gt=0, description="文件大小（字节）")
    ttl: int = Field(default=86400, ge=60, le=2592000, description="过期秒数")
    download_limit: int = Field(default=-1, ge=-1, description="下载次数限制")
    has_password: bool = Field(default=False, description="是否设置下载密码")
    e2ee: bool = Field(default=False, description="是否端到端加密")
    password: Optional[str] = Field(None, description="下载密码（如果 has_password=True）")
    is_public: bool = Field(default=True, description="是否公开（同服务器内可见）")
    group: str = Field(default="default", description="命名空间")
    file_hash: Optional[str] = Field(None, description="文件 SHA-256 哈希")

class VerifyAndGetLinkParams(BaseModel):
    code: str = Field(..., min_length=4, max_length=10)
    password: Optional[str] = None

# ==================================================
# 依赖项：Token 认证
# ==================================================

async def verify_token(authorization: str = Header(...), db = Depends(get_db)) -> Token:
    """验证 API Token 并返回 Token 对象"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid authorization header")

    token_str = authorization[7:]  # Remove "Bearer "
    token = db.query(Token).filter(Token.token == token_str, Token.is_active == True).first()

    if not token:
        raise HTTPException(status_code=401, detail="Invalid or inactive token")

    if token.expires_at and token.expires_at < datetime.now():
        raise HTTPException(status_code=401, detail="Token expired")

    return token

# ==================================================
# jRPC 方法实现
# ==================================================

@router.post("/jrpc")
async def jsonrpc_handler(
    request: JSONRPCRequest,
    token: Token = Depends(verify_token),
    storage: StorageBackend = Depends(get_storage),
    db = Depends(get_db)
):
    """jRPC 2.0 请求分发器"""

    methods = {
        "ft.init_upload": handle_init_upload,
        "ft.confirm_upload": handle_confirm_upload,
        "ft.fetch_metadata": handle_fetch_metadata,
        "ft.verify_and_get_link": handle_verify_and_get_link,
        "ft.list_my_files": handle_list_my_files,
        "admin.list_all": handle_admin_list_all,
        "admin.revoke_code": handle_admin_revoke_code,
        "admin.manage_token": handle_admin_manage_token,
        "admin.get_stats": handle_admin_get_stats,
    }

    handler = methods.get(request.method)
    if not handler:
        return JSONRPCResponse(
            id=request.id,
            error={"code": -32601, "message": "Method not found"}
        )

    try:
        result = await handler(request.params, token, storage, db)
        return JSONRPCResponse(id=request.id, result=result)
    except Exception as e:
        return JSONRPCResponse(
            id=request.id,
            error={"code": -32603, "message": str(e)}
        )

# ==================================================
# 核心方法实现
# ==================================================

async def handle_init_upload(
    params: dict,
    token: Token,
    storage: StorageBackend,
    db
) -> dict:
    """初始化上传任务"""

    # 1. 参数验证
    p = InitUploadParams(**params)

    # 2. 检查配额
    if token.quota_current_size + p.filesize > token.quota_max_size:
        raise ValueError("Quota exceeded")

    if p.filesize > token.max_file_size:
        raise ValueError(f"File size exceeds limit of {token.max_file_size} bytes")

    # 3. 生成唯一 Code
    import random
    import string
    code = ''.join(random.choices(string.ascii_uppercase + string.digits, k=6))

    # 4. 生成存储 Key 和预签名 URL
    storage_key = storage.generate_storage_key(token.user_id, code, p.filename)
    upload_url, storage_key = await storage.generate_upload_url(
        user_id=token.user_id,
        code=code,
        filename=p.filename,
        ttl_seconds=p.ttl
    )

    # 5. 计算密码哈希（如果需要）
    pwd_hash = None
    if p.has_password and p.password:
        pwd_hash = bcrypt.hashpw(p.password.encode(), bcrypt.gensalt()).decode()

    # 6. 创建数据库记录
    from datetime import datetime, timedelta
    transfer = Transfer(
        code=code,
        owner_id=token.user_id,
        group_name=p.group,
        file_name=p.filename,
        file_size=p.filesize,
        mime_type=_guess_mime_type(p.filename),
        file_hash=p.file_hash,
        storage_mode=storage.config['backend'].upper(),
        storage_key=storage_key,
        pwd_hash=pwd_hash,
        is_e2ee=p.e2ee,
        status='pending',
        expire_at=datetime.now() + timedelta(seconds=p.ttl),
        max_downloads=p.download_limit
    )
    db.add(transfer)
    db.commit()

    return {
        "code": code,
        "mode": storage.config['backend'].upper(),
        "upload_url": upload_url,
        "view_url": f"{storage.config['base_url']}/g/{code}"
    }

async def handle_confirm_upload(params: dict, token: Token, storage: StorageBackend, db) -> dict:
    """客户端上传成功后的回调"""
    code = params['code']
    status = params.get('status', 'success')

    transfer = db.query(Transfer).filter(Transfer.code == code).first()
    if not transfer:
        raise ValueError("Transfer not found")

    if status == 'success':
        transfer.status = 'active'
        # 更新用户配额
        tkn = db.query(Token).filter(Token.user_id == transfer.owner_id).first()
        if tkn:
            tkn.quota_current_size += transfer.file_size
    else:
        transfer.status = 'failed'
        # 清理已上传的文件
        await storage.delete_file(transfer.storage_key)

    db.commit()
    return {"success": True}

async def handle_fetch_metadata(params: dict, token: Token, storage: StorageBackend, db) -> dict:
    """获取文件元数据"""
    code = params['code']
    transfer = db.query(Transfer).filter(Transfer.code == code).first()

    if not transfer:
        raise ValueError("Transfer not found")

    return {
        "filename": transfer.file_name,
        "size": transfer.file_size,
        "has_password": transfer.pwd_hash is not None,
        "e2ee_status": transfer.is_e2ee,
        "mime_type": transfer.mime_type,
        "created_at": transfer.created_at.isoformat(),
        "expire_at": transfer.expire_at.isoformat()
    }

async def handle_verify_and_get_link(params: dict, token: Token, storage: StorageBackend, db) -> dict:
    """验证密码并获取下载链接"""
    p = VerifyAndGetLinkParams(**params)
    transfer = db.query(Transfer).filter(Transfer.code == p.code).first()

    if not transfer:
        raise ValueError("Transfer not found")

    # 检查状态
    if transfer.status != 'active':
        raise ValueError(f"Transfer is {transfer.status}")

    # 检查过期
    if transfer.expire_at < datetime.now():
        transfer.status = 'expired'
        db.commit()
        raise ValueError("Transfer has expired")

    # 检查下载次数
    if transfer.max_downloads != -1 and transfer.current_downloads >= transfer.max_downloads:
        raise ValueError("Download limit exceeded")

    # 验证密码
    if transfer.pwd_hash:
        if not p.password:
            return {"need_password": True}
        if not bcrypt.checkpw(p.password.encode(), transfer.pwd_hash.encode()):
            raise ValueError("Invalid password")

    # 生成下载链接
    download_url = await storage.generate_download_url(
        storage_key=transfer.storage_key,
        ttl_seconds=3600  # 1 小时有效期
    )

    # 更新下载计数
    transfer.current_downloads += 1
    db.commit()

    return {
        "download_url": download_url,
        "filename": transfer.file_name
    }

async def handle_list_my_files(params: dict, token: Token, storage: StorageBackend, db) -> dict:
    """列出文件（支持 public 模式）"""
    page = params.get('page', 1)
    page_size = params.get('page_size', 20)
    include_public = params.get('include_public', True)  # 默认包含公开文件

    # 查询逻辑：自己的所有文件 + 其他用户的公开文件
    from sqlalchemy import or_

    if include_public:
        query = db.query(Transfer).filter(
            or_(
                Transfer.owner_id == token.user_id,
                Transfer.is_public == True
            )
        )
    else:
        query = db.query(Transfer).filter(Transfer.owner_id == token.user_id)

    # 只显示活跃的文件
    query = query.filter(Transfer.status == 'active')

    query = query.order_by(Transfer.created_at.desc())
    total = query.count()
    transfers = query.offset((page - 1) * page_size).limit(page_size).all()

    return {
        "files": [
            {
                "code": t.code,
                "filename": t.file_name,
                "size": t.file_size,
                "owner": t.owner_id,
                "is_mine": t.owner_id == token.user_id,
                "is_public": t.is_public,
                "created_at": t.created_at.isoformat(),
                "expire_at": t.expire_at.isoformat(),
                "downloads": f"{t.current_downloads}/{t.max_downloads if t.max_downloads != -1 else '∞'}",
                "status": t.status
            }
            for t in transfers
        ],
        "total": total,
        "page": page,
        "page_size": page_size
    }

# ==================================================
# 管理员方法
# ==================================================

async def handle_admin_list_all(params: dict, token: Token, storage: StorageBackend, db) -> dict:
    """管理员：列出所有文件"""
    # 检查权限
    if "admin" not in token.permissions:
        raise ValueError("Admin permission required")

    filter_status = params.get('filter_status')
    group = params.get('group')

    query = db.query(Transfer)
    if filter_status:
        query = query.filter(Transfer.status == filter_status)
    if group:
        query = query.filter(Transfer.group_name == group)

    transfers = query.all()
    return {
        "files": [
            {
                "code": t.code,
                "owner": t.owner_id,
                "filename": t.file_name,
                "size": t.file_size,
                "status": t.status,
                "expire_at": t.expire_at.isoformat()
            }
            for t in transfers
        ]
    }

async def handle_admin_revoke_code(params: dict, token: Token, storage: StorageBackend, db) -> dict:
    """管理员：强制撤销 Code"""
    if "admin" not in token.permissions:
        raise ValueError("Admin permission required")

    code = params['code']
    transfer = db.query(Transfer).filter(Transfer.code == code).first()

    if not transfer:
        raise ValueError("Transfer not found")

    # 物理删除
    await storage.delete_file(transfer.storage_key)
    db.delete(transfer)
    db.commit()

    return {"status": "revoked"}

async def handle_admin_manage_token(params: dict, token: Token, storage: StorageBackend, db) -> dict:
    """管理员：管理 API Token"""
    if "admin" not in token.permissions:
        raise ValueError("Admin permission required")

    action = params['action']
    config = params.get('config', {})

    if action == 'add':
        # 生成新 Token
        import secrets
        new_token = f"ft_{secrets.token_urlsafe(32)}"
        db_token = Token(
            token=new_token,
            user_id=config.get('user_id', 'unknown'),
            label=config.get('label', ''),
            permissions=config.get('permissions', ['read', 'write']),
            quota_max_size=config.get('quota_max_size', 10 * 1024 * 1024 * 1024)
        )
        db.add(db_token)
        db.commit()
        return {"token": new_token, "status": "created"}

    elif action == 'revoke':
        token_str = config['token']
        db_token = db.query(Token).filter(Token.token == token_str).first()
        if db_token:
            db_token.is_active = False
            db.commit()
        return {"status": "revoked"}

    return {"status": "unknown_action"}

async def handle_admin_get_stats(params: dict, token: Token, storage: StorageBackend, db) -> dict:
    """管理员：获取统计信息"""
    if "admin" not in token.permissions:
        raise ValueError("Admin permission required")

    from sqlalchemy import func

    total_files = db.query(func.count(Transfer.id)).scalar()
    total_size = db.query(func.sum(Transfer.file_size)).scalar() or 0
    active_files = db.query(func.count(Transfer.id)).filter(Transfer.status == 'active').scalar()

    return {
        "total_files": total_files,
        "active_files": active_files,
        "total_storage_bytes": total_size,
        "storage_backend": storage.config['backend']
    }

# ==================================================
# 辅助函数
# ==================================================

def _guess_mime_type(filename: str) -> str:
    """简单 MIME 类型检测"""
    import mimetypes
    mime_type, _ = mimetypes.guess_type(filename)
    return mime_type or 'application/octet-stream'
```

### 5.3 FastAPI 主入口

```python
# filetrans/server/main.py

from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
import uvicorn

from .jrpc import router as jrpc_router
from .janitor import start_janitor
from .config import load_config

config = load_config()

app = FastAPI(
    title="filetrans",
    description="Modern file transfer service with jRPC API",
    version="1.0.0"
)

# 挂载 jRPC 路由
app.include_router(jrpc_router, prefix="/api")

# 挂载静态文件
app.mount("/static", StaticFiles(directory="filetrans/web/static"), name="static")

# 模板
templates = Jinja2Templates(directory="filetrans/web/templates")

@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
    """首页"""
    return templates.TemplateResponse("index.html", {"request": request})

@app.get("/g/{code}", response_class=HTMLResponse)
async def download_page(request: Request, code: str):
    """下载页面"""
    return templates.TemplateResponse("download.html", {"request": request, "code": code})

@app.get("/admin", response_class=HTMLResponse)
async def admin_page(request: Request):
    """管理后台"""
    return templates.TemplateResponse("admin.html", {"request": request})

@app.on_event("startup")
async def startup_event():
    """启动时初始化"""
    # 启动清理任务
    start_janitor(config)

def serve():
    """启动服务器"""
    uvicorn.run(
        "filetrans.server.main:app",
        host=config['server']['host'],
        port=config['server']['port'],
        workers=config['server'].get('workers', 1)
    )
```

---

## 6. API 完全规格 (jRPC)

### 6.1 通信协议

#### 6.1.1 请求格式

```json
{
  "jsonrpc": "2.0",
  "method": "ft.init_upload",
  "params": {
    "filename": "document.pdf",
    "filesize": 1048576,
    "ttl": 86400,
    "download_limit": 5,
    "has_password": true,
    "password": "secret123",
    "e2ee": false,
    "group": "default"
  },
  "id": "1"
}
```

#### 6.1.2 响应格式（成功）

```json
{
  "jsonrpc": "2.0",
  "result": {
    "code": "A3F9K2",
    "mode": "S3",
    "upload_url": "https://s3.amazonaws.com/bucket/user123/A3F9K2/document.pdf?signature=...",
    "view_url": "https://filetrans.example.com/g/A3F9K2"
  },
  "id": "1"
}
```

#### 6.1.3 响应格式（错误）

```json
{
  "jsonrpc": "2.0",
  "error": {
    "code": -32602,
    "message": "Invalid params: filesize exceeds quota",
    "data": {
      "field": "filesize",
      "limit": 536870912,
      "provided": 1073741824
    }
  },
  "id": "1"
}
```

### 6.2 API 方法完整列表

#### 6.2.1 用户方法

| 方法 | 权限 | 说明 |
|------|------|------|
| `ft.init_upload` | write | 初始化上传任务 |
| `ft.confirm_upload` | write | 确认上传完成 |
| `ft.fetch_metadata` | read | 获取文件元数据 |
| `ft.verify_and_get_link` | read | 验证密码并获取下载链接 |
| `ft.list_my_files` | read | 列出当前用户的文件 |

#### 6.2.2 管理员方法

| 方法 | 权限 | 说明 |
|------|------|------|
| `admin.list_all` | admin | 列出所有文件 |
| `admin.revoke_code` | admin | 撤销指定 Code |
| `admin.manage_token` | admin | 管理 API Token |
| `admin.get_stats` | admin | 获取系统统计 |

### 6.3 错误码规范

| 错误码 | 名称 | 说明 |
|--------|------|------|
| -32700 | Parse error | JSON 解析错误 |
| -32600 | Invalid Request | 无效的 jRPC 请求 |
| -32601 | Method not found | 方法不存在 |
| -32602 | Invalid params | 参数验证失败 |
| -32603 | Internal error | 服务器内部错误 |
| 401 | Unauthorized | Token 无效或过期 |
| 403 | Forbidden | 权限不足 |
| 404 | Not found | 资源不存在 |
| 429 | Too many requests | 超出速率限制 |

### 6.4 jRPC 请求示例

#### 示例 1：初始化上传

```bash
curl -X POST https://filetrans.example.com/api/jrpc \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer ft_your_token_here" \
  -d '{
    "jsonrpc": "2.0",
    "method": "ft.init_upload",
    "params": {
      "filename": "presentation.pptx",
      "filesize": 5242880,
      "ttl": 604800,
      "download_limit": 10,
      "has_password": true,
      "password": "secure456",
      "e2ee": false
    },
    "id": "req-001"
  }'
```

#### 示例 2：获取下载链接

```bash
curl -X POST https://filetrans.example.com/api/jrpc \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer ft_your_token_here" \
  -d '{
    "jsonrpc": "2.0",
    "method": "ft.verify_and_get_link",
    "params": {
      "code": "A3F9K2",
      "password": "secure456"
    },
    "id": "req-002"
  }'
```

---

## 7. CLI 详细设计

### 7.1 命令结构

```
ft
├── init                    # 初始化配置
├── -p, --push [PATH]       # 上传文件
├── -g, --get [CODE]        # 下载文件
├── ls                      # 列出文件
├── serve                   # 启动服务器
├── token                   # Token 管理
└── --version               # 版本信息
```

### 7.2 完整实现

```python
# filetrans/cli.py

import typer
import requests
import rch
from pathlib import Path
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, DownloadColumn, TimeRemainingColumn
from rich.console import Console
from rich.table import Table
import configparser
import os
from .client.api import FiletransClient

app = typer.Typer(name="ft", help="filetrans - Modern file transfer tool")
console = Console()

CONFIG_DIR = Path.home() / ".config" / "filetrans"
CONFIG_FILE = CONFIG_DIR / "config.json"

# ==================================================
# 初始化命令
# ==================================================

@app.command()
def init():
    """初始化 CLI 配置"""

    console.print("[bold blue]filetrans CLI 初始化向导[/bold blue]\n")

    server_url = typer.prompt("服务器 URL", default="https://filetrans.example.com")
    api_token = typer.prompt("API Token", hide_input=True)

    # 验证 Token
    with console.status("[bold yellow]验证 Token..."):
        client = FiletransClient(server_url, api_token)
        try:
            stats = client.call("admin.get_stats", {})
            console.print("[bold green]✓[/bold green] Token 验证成功")
        except Exception as e:
            console.print(f"[bold red]✗[/bold red] Token 验证失败: {e}")
            raise typer.Exit(1)

    # 创建配置目录
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    # 写入配置
    import json
    config = {
        "server_url": server_url,
        "api_token": api_token
    }
    with open(CONFIG_FILE, "w") as f:
        json.dump(config, f, indent=2)

    # 设置权限
    os.chmod(CONFIG_FILE, 0o600)

    console.print(f"\n[bold green]配置已保存到:[/bold green] {CONFIG_FILE}")

# ==================================================
# 上传命令
# ==================================================

@app.command("-p")
@app.command("--push")
def push(
    path: str = typer.Argument(..., help="文件或目录路径"),
    ttl: str = typer.Option("24h", "--ttl", "-t", help="有效期（如 1h, 30m, 1d）"),
    limit: int = typer.Option(-1, "--limit", "-l", help="下载次数限制"),
    password: str = typer.Option(None, "--pass", "-P", help="下载密码"),
    encrypt: bool = typer.Option(False, "--encrypt", "-e", help="端到端加密"),
    public: bool = typer.Option(True, "--public/--private", help="是否公开（同服务器内可见）"),
    copy: bool = typer.Option(False, "--copy", "-c", help="复制到剪贴板"),
    group: str = typer.Option("default", "--group", help="命名空间"),
    silent: bool = typer.Option(False, "--silent", "-s", help="静默模式")
):
    """上传文件"""

    # 加载配置
    config = _load_config()
    client = FiletransClient(config['server_url'], config['api_token'])

    # 解析 TTL
    ttl_seconds = _parse_ttl(ttl)

    # 读