# feishu-org-sync

**Repository Path**: jasonfjs/feishu-org-sync

## Basic Information

- **Project Name**: feishu-org-sync
- **Description**: 飞书组织架构同步工具 - 通用配置化解决方案
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-03-11
- **Last Updated**: 2026-03-11

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# 🏢 飞书组织架构同步工具

[![Tests](https://img.shields.io/badge/Tests-53%20passed-green)](tests/)
[![Python 3.8+](https://img.shields.io/badge/Python-3.8+-blue)](https://www.python.org/)
[![License](https://img.shields.io/badge/License-MIT-yellow)](LICENSE)

一个**通用、零代码**的飞书组织架构同步工具。只需要修改配置文件，即可快速部署运行！

## ✨ 特性

- ✅ **多数据库支持**: MySQL、PostgreSQL、Oracle、SQL Server、SQLite
- ✅ **零代码部署**: 只需修改配置文件，不需要写代码
- ✅ **灵活映射**: 支持自定义 SQL 和字段映射
- ✅ **增量同步**: 支持只同步变更数据
- ✅ **定时任务**: 支持自动定时同步
- ✅ **完整日志**: 记录同步全过程
- ✅ **单元测试**: 53 个测试用例，全部通过

## 🚀 快速开始

### 1. 安装依赖

```bash
pip install -r requirements.txt
```

### 2. 配置同步

```bash
# 复制配置文件
cp config.example.yaml config.yaml

# 编辑配置
vim config.yaml
```

### 3. 运行同步

```bash
# 测试连接
python main.py --test

# 执行同步
python main.py

# 全量同步
python main.py --full
```

---

## 📁 项目结构

```
feishu-org-sync/
├── config.example.yaml      # 配置文件模板
├── config.yaml             # 实际配置文件（需要修改）
├── requirements.txt        # 依赖包
├── main.py                 # 主入口
├── sync_engine.py          # 同步引擎
├── scheduler.py            # 定时任务
├── field_transformer.py    # 字段转换
├── README.md               # 本文档
├── db/                     # 数据库适配器
│   ├── __init__.py
│   ├── base.py            # 基类
│   ├── mysql.py           # MySQL
│   ├── postgresql.py      # PostgreSQL
│   ├── oracle.py          # Oracle
│   ├── mssql.py           # SQL Server
│   └── sqlite.py          # SQLite
├── feishu/                # 飞书 API
│   ├── __init__.py
│   └── client.py          # 飞书客户端
├── tests/                 # 单元测试
│   ├── __init__.py
│   ├── test_db_adapters.py
│   ├── test_field_transformer.py
│   └── test_sync_engine.py
└── logs/                   # 日志目录
    └── sync.log
```

---

## 📝 配置说明

### 1. 飞书应用配置

```yaml
feishu:
  app_id: "your-app-id"
  app_secret: "your-app-secret"
  tenant_key: ""  # 可选
```

### 2. 数据源配置

**MySQL:**
```yaml
datasource:
  type: "mysql"
  connection:
    host: "192.168.1.100"
    port: 3306
    username: "root"
    password: "password"
    database: "hr_db"
    charset: "utf8mb4"
```

**PostgreSQL:**
```yaml
datasource:
  type: "postgresql"
  connection:
    host: "192.168.1.100"
    port: 5432
    username: "postgres"
    password: "password"
    database: "hr_db"
```

**Oracle:**
```yaml
datasource:
  type: "oracle"
  connection:
    host: "192.168.1.100"
    port: 1521
    username: "HR_USER"
    password: "password"
    service_name: "ORCL"
```

**SQL Server:**
```yaml
datasource:
  type: "mssql"
  connection:
    host: "192.168.1.100"
    port: 1433
    username: "sa"
    password: "YourPassword"
    database: "HR_DB"
```

### 3. SQL 查询

```yaml
sql:
  departments: |
    SELECT
      id AS external_id,
      name,
      parent_id,
      code,
      leader_id,
      sort_order AS `order`
    FROM departments
    WHERE is_active = 1

  users: |
    SELECT
      id AS external_id,
      name,
      mobile,
      email,
      department_id,
      employee_no,
      UNIX_TIMESTAMP(join_date) AS join_time
    FROM employees
    WHERE status IN ('正式', '实习')
```

### 4. 字段映射

```yaml
mapping:
  departments:
    external_id: "external_id"   # 外部唯一标识
    name: "name"                  # 部门名称
    parent_id: "parent_id"       # 父部门 ID
    code: "code"                 # 部门编码
    leader_id: "leader_id"       # 负责人

  users:
    external_id: "external_id"   # 外部唯一标识
    name: "name"                # 用户名
    mobile: "mobile"            # 手机号（必填）
    email: "email"              # 邮箱
    department_ids: "department_id"  # 部门 ID
    employee_no: "employee_no"   # 工号
    employee_type: "employee_type"  # 员工类型
    join_time: "join_time"      # 入职时间
    leader_id: "leader_id"      # 上级 ID
    work_station: "work_station"  # 工位
    gender: "gender"            # 性别
```

### 5. 同步策略

```yaml
sync:
  mode: "incremental"         # full / incremental
  delete_missing: false        # 是否删除缺失的数据
  batch_size: 100            # 批量处理大小
  rate_limit: 10             # 每秒最大请求数
  retry_times: 3             # 重试次数
  retry_delay: 5              # 重试间隔（秒）
```

---

## 💡 使用示例

### 示例 1：基础同步

```bash
# 测试数据库和飞书连接
python main.py --test

# 执行增量同步（默认）
python main.py

# 执行全量同步
python main.py --full

# 仅同步部门
python main.py --departments

# 仅同步用户
python main.py --users

# 删除缺失的数据
python main.py --delete-missing
```

### 示例 2：定时任务

```yaml
# config.yaml
schedule:
  enabled: true
  cron: "0 2 * * *"   # 每天凌晨 2 点
```

```bash
# 启动定时任务
python scheduler.py
```

---

## 🧪 测试

```bash
# 运行所有测试
python -m pytest tests/ -v

# 运行并生成覆盖率报告
python -m pytest tests/ --cov=. --cov-report=html

# 查看覆盖率报告
open htmlcov/index.html
```

**测试结果:**
- 总测试数: 53
- 通过: 53
- 失败: 0

---

## 📖 API 参考

### 命令行参数

| 参数 | 说明 |
|-----|------|
| `-c, --config` | 配置文件路径（默认: config.yaml） |
| `--full` | 全量同步 |
| `--test` | 仅测试连接 |
| `--departments` | 仅同步部门 |
| `--users` | 仅同步用户 |
| `--delete-missing` | 删除缺失的数据 |

### 同步统计

同步完成后会输出以下统计信息：

```text
同步完成! 耗时: 12.34秒
部门: 新增 5, 更新 3, 删除 0, 跳过 2
用户: 新增 20, 更新 10, 删除 5, 跳过 15
```

---

## ⚠️ 注意事项

1. **权限要求**: 飞书应用需要配置 `contact:*` 相关权限
2. **手机号**: 必须是 86 开头的中国大陆手机号
3. **频率限制**: 默认 100次/分钟/应用
4. **数据校验**: 同步前请确保外部数据格式正确
5. **备份**: 首次全量同步前建议备份飞书通讯录

---

## 🔧 常见问题

### Q: 连接失败怎么办？

```bash
# 检查配置文件
python main.py --test
```

### Q: 同步很慢怎么办？

```yaml
sync:
  rate_limit: 5      # 降低请求频率
  batch_size: 50     # 减小批次大小
```

### Q: 如何查看日志？

```bash
# 实时查看日志
tail -f logs/sync.log

# 查看错误日志
grep "ERROR" logs/sync.log
```

### Q: 想同步多个数据源怎么办？

目前支持单个数据源，如需多个数据源，可以合并 SQL 查询或在数据库层创建视图。

---

## 🐛 调试技巧

### 1. 启用详细日志

```yaml
logging:
  level: "DEBUG"
```

### 2. 测试单个功能

```python
from db import get_adapter

# 测试数据库连接
adapter = get_adapter('mysql', {...})
adapter.test_connection()
```

### 3. 检查字段映射

```python
from field_transformer import FieldTransformer

t = FieldTransformer({})
print(t.transform('mobile', '13800138000'))
```

---

## 📚 参考链接

- [飞书开放平台](https://open.feishu.cn/)
- [SDK GitHub](https://github.com/larksuite/oapi-sdk-python)
- [API 文档](https://open.feishu.cn/document/)
- [飞书开发者后台](https://open.feishu.cn/app/)

---

## 📄 许可证

MIT License

---

## 🤝 贡献

欢迎提交 Issue 和 Pull Request！

1. Fork 本仓库
2. 创建分支 (`git checkout -b feature/amazing-feature`)
3. 提交更改 (`git commit -m 'Add amazing feature'`)
4. 推送分支 (`git push origin feature/amazing-feature`)
5. 创建 Pull Request