本文含工具体验分享

外贸业务中 WhatsApp 传输的文件量有多大？一个中型团队的手机拿出来，聊天记录里至少塞了几百个 PDF 报价单、产品图片、技术规格书甚至装箱视频。大部分人靠手动长按→另存为→拖到电脑→上传网盘，每天至少浪费 20-30 分钟在文件搬运上。我把这套流程自动化了之后，整个团队的文件管理效率提升了至少 3 倍。这篇文章把完整方案和代码分享出来。

一、WhatsApp 文件管理的痛点分析

1.1 典型的混乱场景

客户A发来报价单(PDF) → 手动保存 → 命名随意"报价.pdf" → 半年后找不到
客户B发来产品图片(10张) → 一张张保存 → 忘记分类 → 和抖音截图混在一起
客户C发来装箱视频(50MB) → 手机空间不足 → 先清缓存再下载 → 操作链断裂
客户D发来验厂报告(DOCX) → WhatsApp 7天过期删除 → 客户问起来才发现已无法下载

1.2 自动化目标

我们要实现的是一个闭环管道：

WhatsApp 文件接收 → 自动检测 → 本地缓存 → 分类归档 → 云存储上传 → 删除本地副本（可选）
                          ↑                     ↑                ↑
                     文件类型识别          按客户/日期/类型      S3/OSS/百度网盘

二、文件类型自动检测

2.1 检测方案

WhatsApp 的文件没有可靠的 MIME 信令，我们需要根据文件内容（魔数/Magic Number）和扩展名组合判断：

"""
WhatsApp 文件类型自动检测模块
支持: 图片、PDF、文档(Word/Excel)、视频、压缩包、CAD图等
"""
import hashlib
from pathlib import Path
from typing import Tuple, Optional
from enum import Enum


class FileCategory(Enum):
    """文件分类枚举"""
    IMAGE = "图片"
    PDF = "PDF文档"
    DOCUMENT = "办公文档"
    VIDEO = "视频"
    AUDIO = "音频"
    ARCHIVE = "压缩包"
    CAD = "工程图纸"
    SPREADSHEET = "电子表格"
    UNKNOWN = "未知类型"


# 文件头魔数签名（文件类型识别）
MAGIC_SIGNATURES = {
    FileCategory.PDF: [
        b"%PDF-",
    ],
    FileCategory.IMAGE: [
        b"\xFF\xD8\xFF",         # JPEG
        b"\x89PNG\r\n\x1A\n",    # PNG
        b"GIF87a",               # GIF87
        b"GIF89a",               # GIF89
        b"RIFF",                 # WEBP
        b"\x00\x00\x01\x00",     # ICO
    ],
    FileCategory.VIDEO: [
        b"\x00\x00\x00\x18ftypmp42",  # MP4
        b"\x00\x00\x00\x20ftyp",       # QuickTime
        b"\x1AE\xDF\xA3",              # MKV/WebM
    ],
    FileCategory.AUDIO: [
        b"OggS",     # OGG/OPUS (WhatsApp 语音)
        b"ID3",      # MP3
        b"fLaC",     # FLAC
    ],
    FileCategory.DOCUMENT: [
        b"\xD0\xCF\x11\xE0\xA1\xB1\x1A\xE1",  # DOC (OLE2)
        b"PK\x03\x04",                         # DOCX/XLSX/PPTX (ZIP-based)
    ],
    FileCategory.ARCHIVE: [
        b"PK\x03\x04",         # ZIP (与 DOCX 共用，需结合扩展名)
        b"Rar!\x1A\x07\x00",   # RAR
        b"\x1F\x8B\x08",       # GZ
    ],
    FileCategory.CAD: [
        b"AC10",    # DWG
    ],
}

# 扩展名到分类的映射（兜底策略）
EXTENSION_MAP = {
    ".jpg":  FileCategory.IMAGE,   ".jpeg": FileCategory.IMAGE,
    ".png":  FileCategory.IMAGE,   ".gif":  FileCategory.IMAGE,
    ".webp": FileCategory.IMAGE,   ".bmp":  FileCategory.IMAGE,
    ".pdf":  FileCategory.PDF,
    ".doc":  FileCategory.DOCUMENT,  ".docx": FileCategory.DOCUMENT,
    ".xls":  FileCategory.SPREADSHEET, ".xlsx": FileCategory.SPREADSHEET,
    ".csv":  FileCategory.SPREADSHEET,
    ".ppt":  FileCategory.DOCUMENT,  ".pptx": FileCategory.DOCUMENT,
    ".txt":  FileCategory.DOCUMENT,
    ".mp4":  FileCategory.VIDEO,    ".mov":  FileCategory.VIDEO,
    ".avi":  FileCategory.VIDEO,    ".mkv":  FileCategory.VIDEO,
    ".mp3":  FileCategory.AUDIO,    ".ogg":  FileCategory.AUDIO,
    ".opus": FileCategory.AUDIO,    ".wav":  FileCategory.AUDIO,
    ".zip":  FileCategory.ARCHIVE,  ".rar":  FileCategory.ARCHIVE,
    ".7z":   FileCategory.ARCHIVE,  ".gz":   FileCategory.ARCHIVE,
    ".dwg":  FileCategory.CAD,      ".dxf":  FileCategory.CAD,
}


def detect_file_type(file_path: str) -> FileCategory:
    """通过文件头魔数 + 扩展名综合检测文件类型"""
    path = Path(file_path)

    # 1. 扩展名预判断
    ext_category = EXTENSION_MAP.get(path.suffix.lower())

    # 2. 读取文件头（前 16 字节）
    try:
        with open(file_path, "rb") as f:
            header = f.read(16)
    except (IOError, PermissionError):
        return ext_category or FileCategory.UNKNOWN

    # 3. 魔数匹配（排除 ZIP/DOCX 二义性）
    for category, signatures in MAGIC_SIGNATURES.items():
        for sig in signatures:
            if header.startswith(sig):
                # ZIP 格式的二义性处理
                if sig == b"PK\x03\x04":
                    if path.suffix.lower() in (".docx", ".xlsx", ".pptx"):
                        return FileCategory.DOCUMENT if path.suffix == ".docx" \
                            else FileCategory.SPREADSHEET
                    return FileCategory.ARCHIVE
                return category

    # 4. 兜底：扩展名判断
    return ext_category or FileCategory.UNKNOWN


def safe_filename(original_name: str, customer_id: str, timestamp: str) -> str:
    """生成安全、可追溯的文件名"""
    path = Path(original_name)
    safe_name = re.sub(r"[<>:\"/\\|?*\s]", "_", path.stem)  # 替换非法字符
    clean_suffix = path.suffix.lower()

    # 格式: 客户ID_原始文件名_HASH前8位.扩展名
    content_hash = hashlib.md5(f"{customer_id}_{timestamp}".encode()).hexdigest()[:8]
    return f"{customer_id}_{safe_name}_{content_hash}{clean_suffix}"

三、自动下载与本地缓存

"""
WhatsApp 文件自动下载与分类本地缓存
利用 Playwright 监听 WhatsApp Web 的文件接收事件
"""
import asyncio
import shutil
from datetime import datetime
from collections import defaultdict
from dataclasses import dataclass
from typing import List

ARCHIVE_BASE = Path("./whatsapp_archive")
for subdir in ["incoming", "by_customer", "by_type", "staging"]:
    (ARCHIVE_BASE / subdir).mkdir(parents=True, exist_ok=True)


@dataclass
class ArchivedFile:
    """已归档文件的信息"""
    original_name: str
    archived_name: str
    file_category: FileCategory
    file_size: int
    customer_id: str
    received_at: str
    local_path: str
    cloud_url: str = ""


class FileDownloadManager:
    """文件下载与本地缓存管理器"""

    MAX_LOCAL_SIZE_MB = 2000  # 本地存的最大容量，超了触发自动清理
    SIZE_THRESHOLD_WARN = 0.8  # 使用量达到80%时告警

    def __init__(self, download_dir: Path = DOWNLOAD_DIR):
        self.download_dir = download_dir
        self.download_dir.mkdir(exist_ok=True)
        self.archived_files: List[ArchivedFile] = []
        self._stats = defaultdict(lambda: {"count": 0, "size": 0})

    def download_file(self, url: str, filename: str) -> Optional[str]:
        """下载 WhatsApp Web 中的文件到本地"""
        import requests

        local_path = self.download_dir / filename

        try:
            resp = requests.get(url, stream=True, timeout=120)
            resp.raise_for_status()

            with open(local_path, "wb") as f:
                for chunk in resp.iter_content(chunk_size=8192):
                    f.write(chunk)

            return str(local_path)
        except Exception as e:
            print(f"下载失败 {filename}: {e}")
            return None

    def archive_file(
        self, local_path: str, customer_id: str, received_at: str = None
    ) -> Optional[ArchivedFile]:
        """将下载的文件归档到分类目录"""
        path = Path(local_path)
        if not path.exists():
            return None

        # 检测文件类型
        file_type = detect_file_type(local_path)
        file_size = path.stat().st_size
        timestamp = received_at or datetime.now().isoformat()

        # 生成归档文件名
        archived_name = safe_filename(path.name, customer_id, timestamp)

        # 按客户分类的目录
        customer_dir = ARCHIVE_BASE / "by_customer" / customer_id
        customer_dir.mkdir(parents=True, exist_ok=True)

        # 按类型分类的目录
        type_dir = ARCHIVE_BASE / "by_type" / file_type.value.replace("/", "_")
        type_dir.mkdir(parents=True, exist_ok=True)

        # 复制到两个分类目录（硬链接更省空间）
        try:
            # 按客户归档
            cust_path = customer_dir / archived_name
            shutil.copy2(local_path, cust_path)
            # 按类型归档
            type_path = type_dir / archived_name
            shutil.copy2(local_path, type_path)

        except OSError as e:
            print(f"文件复制失败: {e}")
            return None

        # 更新统计
        self._stats[customer_id]["count"] += 1
        self._stats[customer_id]["size"] += file_size
        self._stats[file_type.value]["count"] += 1
        self._stats[file_type.value]["size"] += file_size

        archived = ArchivedFile(
            original_name=path.name,
            archived_name=archived_name,
            file_category=file_type,
            file_size=file_size,
            customer_id=customer_id,
            received_at=timestamp,
            local_path=str(cust_path),
        )
        self.archived_files.append(archived)
        return archived

    def get_local_usage(self) -> float:
        """获取本地归档目录总大小（MB）"""
        total = 0
        for f in ARCHIVE_BASE.rglob("*"):
            if f.is_file():
                total += f.stat().st_size
        return total / (1024 * 1024)

    def is_storage_warning(self) -> bool:
        """检查本地空间是否达到告警阈值"""
        usage_mb = self.get_local_usage()
        return usage_mb > self.MAX_LOCAL_SIZE_MB * self.SIZE_THRESHOLD_WARN

    def get_oldest_files(self, count: int = 10) -> List[Path]:
        """获取最旧的文件列表（用于清理）"""
        files = []
        for f in ARCHIVE_BASE.rglob("*"):
            if f.is_file():
                files.append(f)
        files.sort(key=lambda p: p.stat().st_mtime)
        return files[:count]

四、云存储自动上传

4.1 多云适配方案

外贸团队通常使用阿里云 OSS（国内）、AWS S3（海外）或者 Google Cloud Storage。我们需要一个统一的适配层：

"""
多云存储适配器: 支持阿里云OSS / AWS S3 / 腾讯云COS
pip install boto3 oss2 cos-python-sdk-v5
"""
import boto3
from botocore.config import Config as BotoConfig
from typing import Optional
from enum import Enum


class CloudProvider(Enum):
    """云厂商枚举"""
    AWS_S3 = "aws_s3"
    ALIYUN_OSS = "aliyun_oss"
    TENCENT_COS = "tencent_cos"


class CloudUploader:
    """多云上传器"""

    def __init__(self, provider: CloudProvider, config: dict):
        self.provider = provider
        self.client = self._init_client(config)

    def _init_client(self, config: dict):
        """根据提供商初始化对应的客户端"""
        if self.provider == CloudProvider.AWS_S3:
            return boto3.client(
                "s3",
                aws_access_key_id=config["access_key"],
                aws_secret_access_key=config["secret_key"],
                region_name=config.get("region", "us-east-1"),
                config=BotoConfig(
                    connect_timeout=30,
                    read_timeout=60,
                    retries={"max_attempts": 3},
                ),
            )
        elif self.provider == CloudProvider.ALIYUN_OSS:
            import oss2
            return oss2.Bucket(
                oss2.Auth(config["access_key"], config["secret_key"]),
                config["endpoint"],
                config["bucket"],
            )
        elif self.provider == CloudProvider.TENCENT_COS:
            from qcloud_cos import CosConfig, CosS3Client
            cos_config = CosConfig(
                Region=config.get("region", "ap-guangzhou"),
                SecretId=config["access_key"],
                SecretKey=config["secret_key"],
            )
            return CosS3Client(cos_config)

    def upload(self, local_path: str, remote_key: str,
               bucket: str = None) -> Optional[str]:
        """
        上传文件到云存储并返回访问 URL
        remote_key: 云端路径（如 "customers/cus_uk_0042/quotes/报价_202506.pdf"）
        """
        headers = {"Content-Disposition": "attachment"}  # 强制下载，不在浏览器打开

        if self.provider == CloudProvider.AWS_S3:
            self.client.upload_file(
                local_path,
                bucket,
                remote_key,
                ExtraArgs={"ContentDisposition": "attachment"},
            )
            return f"https://{bucket}.s3.amazonaws.com/{remote_key}"

        elif self.provider == CloudProvider.ALIYUN_OSS:
            self.client.put_object_from_file(
                remote_key, local_path, headers=headers
            )
            return f"https://{self.client.bucket_name}.oss-cn-shanghai.aliyuncs.com/{remote_key}"

        elif self.provider == CloudProvider.TENCENT_COS:
            self.client.upload_file(
                Bucket=bucket,
                Key=remote_key,
                LocalFilePath=local_path,
            )
            return f"https://{bucket}.cos.ap-guangzhou.myqcloud.com/{remote_key}"

        return None

    def list_files(self, prefix: str, bucket: str = None,
                   max_keys: int = 100) -> list:
        """列出云端指定前缀下的文件"""
        if self.provider == CloudProvider.AWS_S3:
            resp = self.client.list_objects_v2(Bucket=bucket, Prefix=prefix, MaxKeys=max_keys)
            return resp.get("Contents", [])
        elif self.provider == CloudProvider.ALIYUN_OSS:
            result = []
            for obj in self.client.list_objects_v2(prefix=prefix, max_keys=max_keys).object_list:
                result.append({"Key": obj.key, "Size": obj.size, "LastModified": obj.last_modified})
            return result
        return []

4.2 全流程自动化

"""
完整的文件管理自动化流水线
监听新文件 → 检测类型 → 归档到本地 → 上传云端 → 清理本地
"""
import time
import json
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class WhatsAppFileHandler(FileSystemEventHandler):
    """监听下载目录，自动处理新文件"""

    def __init__(self, download_mgr: FileDownloadManager, uploader: CloudUploader,
                 customer_map: dict = None):
        self.download_mgr = download_mgr
        self.uploader = uploader
        self.customer_map = customer_map or {}  # 文件名→客户ID映射
        self.bucket_name = "whatsapp-files-archive"

    def on_created(self, event):
        """当新文件出现在下载目录时触发"""
        if event.is_directory:
            return

        local_path = event.src_path
        filename = Path(local_path).name

        # WhatsApp 下载文件的命名规则: 客户名_时间戳.扩展名
        # 或 PTT-日期-WA编号.opus（语音）
        customer_id = self._infer_customer(filename)

        print(f"\n检测到新文件: {filename}")
        print(f"  客户: {customer_id}")

        # 等待文件完全写入（避免拷贝未完成的下载）
        time.sleep(2)

        # 步骤1: 本地归档
        archived = self.download_mgr.archive_file(
            local_path, customer_id
        )
        if not archived:
            print(f"  归档失败: {filename}")
            return

        # 步骤2: 生成云端路径
        file_date = datetime.now().strftime("%Y%m%d")
        remote_key = (
            f"customer/{customer_id}/"
            f"{archived.file_category.value}/"
            f"{file_date}/"
            f"{archived.archived_name}"
        )

        # 步骤3: 上传到云端
        cloud_url = self.uploader.upload(local_path, remote_key, self.bucket_name)
        if cloud_url:
            archived.cloud_url = cloud_url
            print(f"  已上传云端: {cloud_url}")
        else:
            print(f"  云端上传失败，仅保留本地备份")

        # 步骤4: 空间检查与自动清理
        if self.download_mgr.is_storage_warning():
            print(f"  本地空间告警，触发自动清理...")
            old_files = self.download_mgr.get_oldest_files(20)
            for old in old_files:
                # 只删除已经上传到云端的
                old_path = str(old)
                if any(a.local_path == old_path and a.cloud_url
                       for a in self.download_mgr.archived_files):
                    old.unlink(missing_ok=True)
                    print(f"  清理: {old.name}")

    def _infer_customer(self, filename: str) -> str:
        """从文件名推断客户ID（业务规则，按实际情况自定义）"""
        # 尝试从文件名提取客户信息
        for keyword, cid in self.customer_map.items():
            if keyword.lower() in filename.lower():
                return cid
        # 默认: 按日期分组
        return f"unknown_{datetime.now():%Y%m%d}"


def start_file_watch(download_dir: str, uploader: CloudUploader):
    """启动文件监听服务"""
    manager = FileDownloadManager(Path(download_dir))
    handler = WhatsAppFileHandler(manager, uploader)
    observer = Observer()
    observer.schedule(handler, download_dir, recursive=False)
    observer.start()

    print(f"文件监控已启动: {download_dir}")
    print("按 Ctrl+C 停止...")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

五、空间监控与过期处理

WhatsApp 的文件有 7 天过期机制（媒体文件），所以必须在文件过期前完成归档。这里用一个简单的定时任务：

def check_expiring_files(archive_dir: Path, warning_days: int = 5):
    """
    检查即将过期的WhatsApp文件
    WhatsApp 媒体文件下载后 7 天过期，语音 30 天
    """
    now = datetime.now()
    status = {"ok": 0, "warning": 0, "expired": 0, "archived": 0}

    for f in archive_dir.rglob("*"):
        if not f.is_file():
            continue

        file_age_days = (now - datetime.fromtimestamp(f.stat().st_mtime)).days

        # 检查是否已归档
        cloud_url = _find_cloud_url(f)  # 从数据库或索引查询

        if cloud_url:
            status["archived"] += 1
            # 已归档文件，删除本地副本（可选）
            if file_age_days > warning_days:
                f.unlink(missing_ok=True)
        elif file_age_days > 7:
            status["expired"] += 1
            print(f"文件已过期（未归档）: {f.name}")
        elif file_age_days > warning_days:
            status["warning"] += 1
            print(f"文件即将过期 ({file_age_days}天): {f.name}")
        else:
            status["ok"] += 1

    return status

六、方案对比总结

方案	文件检测	自动下载	本地归档	云存储	空间管理	过期预警	适合场景
手动操作	肉眼	逐条点	手动拖拽	手动上传	不管	不关心	个位数文件/天
本文 Python 脚本	魔数+扩展名	Playwright监听	按客户/类型	S3/OSS/COS	自动清理	定时检查	有技术团队
WADesk 文件管理	自动	同步时自动	按会话归档	内置云存储或对接第三方	自动	内置	5-50人团队
企业网盘(Dropbox等)	无	需手动	手动	原生	套餐制	无	仅做存储

WADesk 的文件管理能力比本文脚本多了一层"聊天上下文关联"——文件归档的同时保留了它在哪条对话中出现的上下文信息，方便回溯查找。本文脚本更灵活，可以对接任意云存储，但没有上下文关联。选哪个取决于你是需要灵活性还是便捷性。

核心原则不变：WhatsApp 里收到的每一个业务文件，都应该在 24 小时内被归档到云端，而不是烂在手机聊天记录里等 7 天后自动消失。

---

8年跨境私域技术落地，踩过30+次WhatsApp封号坑。有问题可以在评论区交流实操细节。