从视图导出为表格但是图片列图片消失的一种解决办法,基于python脚本的处理

之前遇到过一个痛点,就是导出表格时,图片列有时候不会随表格导出(即便导出时选择一并导出图片。)。

为什么会出现这种情况呢?调查发现:如果A-base的表复制了一张图片,直接粘贴B-base的表里,虽然B-base的表里图片可以正常显示,但导出时不会一并导出。只有B-base图片列手动上传的图片才可以随表格导出。这是因为图片来自不同的workspace。

我写了一个脚本来解决这个问题,本地python脚本,原理是把原图片下载后再上传一遍。注意,这里为了容错不会直接修改表格里的图片列,而是建议你新增一个图片2的列来保存上传后的图片。

# -*- coding: utf-8 -*-
"""
SeaTable 图片下载替换工具 - 直接使用 Account Token 下载图片

使用方法:
1. 确保配置正确
2. 运行:python SeaTable 图片下载替换工具.py
"""

import requests
import os
import re
from urllib.parse import unquote

# ==================== 配置区域 ====================

# SeaTable 配置
SERVER_URL = 'https://cloud.seatable.cn'

# Account Token(从账号后台获取)- 直接用于下载图片
ACCOUNT_TOKEN = '填入你的ACCOUNT_TOKEN'

# API Token - 用于读取表格数据
API_TOKEN = '填入表格API'

# 表格配置
TABLE_NAME = '修改成子表名称'  # 子表名称
IMAGE_COL_NAME = '修改成图片列名称'  # 图片列名称(源)
UPLOAD_COL_NAME = '修改成目标图片列名称,建议不要和原图片列名称一样'  # 上传目标列名称(留空则不上传)
VIEW_NAME = '修改成你的视图名称'  # 视图名称(留空处理所有数据)
BATCH_SIZE = 10  # 批量处理大小
REQUEST_DELAY = 0.5  # 请求间隔(秒)
RETRY_TIMES = 2  # 下载/上传重试次数

# 图片保存目录
IMAGE_DIR = './downloaded_images'

# ==================== 下载函数 ====================

def download_image_with_account_token(img_url, account_token, save_path):
    """
    直接使用 Account Token 下载图片
    
    认证格式:Authorization: Bearer {Account_Token}
    """
    headers = {
        'Authorization': f'Bearer {account_token}'
    }
    
    try:
        print(f"  请求头:Authorization: Bearer {account_token[:20]}...")
        
        response = requests.get(img_url, headers=headers, timeout=15)
        print(f"  HTTP 状态码:{response.status_code}")
        print(f"  Content-Type:{response.headers.get('Content-Type', 'N/A')}")
        
        response.raise_for_status()
        
        # 检查是否是图片
        content_type = response.headers.get('Content-Type', '')
        
        if 'image' in content_type or response.content[:4] == b'\x89PNG' or response.content[:2] == b'\xff\xd8':
            # 保存图片
            with open(save_path, 'wb') as f:
                f.write(response.content)
            
            print(f"    ✓ 下载成功,大小:{len(response.content)} bytes")
            return True
        else:
            print(f"    ✗ 非图片内容")
            # 显示前 300 字符看看是什么
            try:
                text = response.content[:300].decode('utf-8', errors='ignore')
                print(f"    内容预览:{text}")
            except:
                pass
            return False
            
    except Exception as e:
        print(f"    ✗ 下载失败:{e}")
        return False


def extract_workspace_info(img_url):
    """从图片 URL 中提取 workspace_id"""
    match = re.search(r'/workspace/(\d+)/asset/([^/]+)/(.+)', img_url)
    if match:
        return {
            'workspace_id': match.group(1),
            'asset_id': match.group(2),
            'file_path': match.group(3)
        }
    return None


def upload_image_to_row(base, table_name, row_id, col_name, file_path):
    """
    上传图片到指定行的图片列

    步骤:
    1. 使用 upload_local_file 上传图片到表格所在 workspace
    2. 使用 update_row 更新行的图片列

    返回:成功返回 True,失败返回 False
    """
    try:
        # 1. 上传图片文件
        # file_type='image' 表示上传为图片类型
        upload_result = base.upload_local_file(file_path, file_type='image')

        if not upload_result:
            print(f"    ↑ 上传文件失败")
            return False

        img_url = upload_result.get('url')
        print(f"    ↑ 上传成功:{img_url[:60]}...")

        # 2. 更新行的图片列
        # 图片列是字符串数组格式:[url_string]
        row_data = {
            col_name: [img_url]
        }

        base.update_row(table_name, row_id, row_data)
        print(f"    ✓ 行数据已更新")
        return True

    except Exception as e:
        print(f"    ↑ 上传失败:{e}")
        return False


# ==================== 主流程 ====================

def main():
    print("=" * 60)
    print("SeaTable 图片下载工具")
    print("=" * 60)
    
    # 1. 连接到 SeaTable
    print("\n[1] 连接 SeaTable...")
    from seatable_api import Base
    
    base = Base(API_TOKEN, SERVER_URL)
    base.auth()
    print("✓ 连接成功")
    
    # 2. 读取表格数据
    print("\n[2] 读取表格数据...")
    
    # 使用视图筛选
    if VIEW_NAME:
        print(f"  使用视图:{VIEW_NAME}")
        rows = base.list_rows(TABLE_NAME, view_name=VIEW_NAME)
    else:
        rows = base.list_rows(TABLE_NAME)
    
    print(f"✓ 共 {len(rows)} 行数据")
    
    # 3. 创建保存目录
    if not os.path.exists(IMAGE_DIR):
        os.makedirs(IMAGE_DIR)
    
    # 4. 下载图片
    print("\n[3] 下载图片...")
    print("=" * 60)
    
    success_count = 0
    fail_count = 0
    
    for idx, row in enumerate(rows, 1):
        images = row.get(IMAGE_COL_NAME, [])
        
        if not images:
            continue
        
        print(f"\n[{idx}/{len(rows)}] 行 {row['_id'][:8]}... | 图片数:{len(images)}")
        
        # 处理第一张图片
        img = images[0]
        img_url = img if isinstance(img, str) else img.get('url', '')
        
        # 生成文件名
        img_name = img_url.split('/')[-1].split('?')[0]
        img_name = unquote(img_name)
        
        if len(img_name) > 50:
            ext = os.path.splitext(img_name)[1]
            img_name = f"image_{idx}{ext}"
        
        print(f"  图片:{img_name}")
        print(f"  URL:{img_url[:60]}...")
        
        # 提取 workspace 信息
        workspace_info = extract_workspace_info(img_url)
        if workspace_info:
            print(f"  Workspace:{workspace_info['workspace_id']}")
        
        # 下载图片(直接使用 Account Token)
        save_path = os.path.join(IMAGE_DIR, img_name)

        if download_image_with_account_token(img_url, ACCOUNT_TOKEN, save_path):
            success_count += 1

            # 上传到目标列(如已配置)
            if UPLOAD_COL_NAME:
                print(f"  → 上传到列:{UPLOAD_COL_NAME}")
                upload_result = upload_image_to_row(
                    base, TABLE_NAME, row['_id'], UPLOAD_COL_NAME, save_path
                )
                if upload_result:
                    print(f"    ✓ 上传完成")
                else:
                    print(f"    ✗ 上传失败")
        else:
            fail_count += 1

    # 5. 输出统计
    print("\n" + "=" * 60)
    print("下载完成")
    print("=" * 60)
    print(f"成功:{success_count} 张")
    print(f"失败:{fail_count} 张")

    if success_count > 0:
        print(f"\n✓ 图片已保存到:{IMAGE_DIR}/")

    if UPLOAD_COL_NAME:
        print(f"✓ 图片已上传到列:{UPLOAD_COL_NAME}")


if __name__ == '__main__':
    main()