简介
之前配置了 Seatable- 4.3.8 版本的环境，具体的配置操作参照：https://wiki.waringid.me/x/HQA1AQ以下的操作基于 Seatable-4.4.9。主要实现以下功能：
- 从 4.3.8 版本升级到 4.4.9 ，确保升级后的数据正常
- 升级到5.0版本也正常，需要重新调整 seatable.sh 脚本文件。4.4.9版本后重新整合了 api 接口，在启动脚本中增加了 api gateway 的启动，启动端口 7780。
- 在新版本中配置 python 脚本环境，python 环境注意变量和配置文件的匹配情况
- 通过 python 脚本实现表格文件自动添加条码和二维码
参考
- 部署 SeaTable Python 运行器 https://docs.seatable.cn/published/seatable-manual/docker/seatable-python-runner/4.3/Deploy%20SeaTable%20Python%20Pipeline.md
SeaTable 的 Python 脚本运行包含多个部分，SeaTable, Python scheduler, Python starter 和 Python runner，它们的功能与关系如下
- SeaTable: 新建/保存/修改脚本，发起运行请求等
- Python scheduler: 调度器，主要负责调度 SeaTable 运行脚本请求、保存/统计脚本运行结果等。相当于一个 master 节点
- Python starter: 真正运行脚本，相当于一个 worker 节点。Python starter 在收到一个脚本运行请求后，会下载脚本内容并启动一个 Docker 容器来运行这个脚本。脚本运行结束后，容器自动销毁，以此保证安全性。
- Python runner: 运行脚本的 Docker 容器。
容器和环境配置
配置 .env 环境脚本
| 代码块 | ||
|---|---|---|
| ||
COMPOSE_FILE='seatable-server.yml' COMPOSE_PATH_SEPARATOR=',' # system settings TIME_ZONE='Asia/Shanghai' # seatable server url SEATABLE_SERVER_HOSTNAME='seatable.waringid.me' SEATABLE_SERVER_PROTOCOL='https' # initial web admin SEATABLE_ADMIN_EMAIL='1377777@139.com' SEATABLE_ADMIN_PASSWORD='XXXXXX' # database SEATABLE_MYSQL_ROOT_PASSWORD='XXXXX' SEATABLE_DIR='/data/seatable' SEATABLE_PYTHON_PIPELINE_DB_NAME=scheduler SEATABLE_PYTHON_PIPELINE_MYSQL_ROOT_PASSWORD='XXXXXX' SEATABLE_PYTHON_PIPELINE_MYSQL_USER=python SEATABLE_PYTHON_PIPELINE_MYSQL_PASSWORD='XXXXXXX' PYTHON_TRANSFER_DIRECTORY_PATH=/data/seatable/python-pipeline/tmp PYTHON_SCHEDULER_AUTH_TOKEN=25d46ca953db0899c2ca9 PYTHON_SCHEDULER_URL=http://python-scheduler PYTHON_SCHEDULER_LOGS_DIR=/data/seatable/python-pipeline/logs/scheduler-logs PYTHON_STARTER_USE_ALTERNATIVE_FILE_SERVER_ROOT=True PYTHON_STARTER_ALTERNATIVE_FILE_SERVER_ROOT=http://seatable-server PYTHON_STARTER_LOGS_DIR=/data/seatable/python-pipeline/logs/starter-logs SEATABLE_SCHEDULER_IMAGE=seatable/seatable-python-scheduler:latest SEATABLE_STARTER_IMAGE=seatable/seatable-python-starter:3.1.1.p SEATABLE_RUNNER_IMAGE=seatable/seatable-python-runner:latest |
- PYTHON_STARTER_ALTERNATIVE_FILE_SERVER_ROOT=http://seatable-server 要确保容器内部能正常获取保存的脚本，这里配置内网访问地址，需要同时注意和 nginx 配置文件中的主机头匹配。
- SEATABLE_STARTER_IMAGE=seatable/seatable-python-starter:3.1.1.p 这个是调整后的镜像文件，原有的镜像配置文件会导致 uwsgi 不断 crash
- PYTHON_SCHEDULER_AUTH_TOKEN 为随机字符串（token 可用命令生成 uuidgen）
- PYTHON_SCHEDULER_URL 默认为 http://python-scheduler，其中 python-scheduler 是 docker 环境下 SeaTable 访问 Python scheduler 网络名称，不需要修改这个值。
- Python 运行器共用 SeaTable MySql数据库，并自动读取已有的配置项（SEATABLE_MYSQL_DB_HOST, SEATABLE_MYSQL_ROOT_PASSWORD）
- 时区也会读取 SeaTable 已有的配置项（TIME_ZONE）
- 如果不设置 PYTHON_SCHEDULER_LOGS_DIR 或 PYTHON_STARTER_LOGS_DIR 将会默认使用宿主机的 /tmp 目录
修改 seatable 配置文件 conf/dtable_web_settings.py
| 代码块 | ||
|---|---|---|
| ||
SEATABLE_FAAS_URL = 'http://python-scheduler' SEATABLE_FAAS_AUTH_TOKEN = '***' # 与 .env 文件 PYTHON_SCHEDULER_AUTH_TOKEN 的配置相同 |
seatable-server.yml
| 代码块 | ||
|---|---|---|
| ||
---
services:
seatable-server:
image: ${SEATABLE_IMAGE:-seatable/seatable-enterprise:latest}
restart: unless-stopped
container_name: seatable-server
ports:
- "8006:80"
volumes:
- "${SEATABLE_DIR}/seatable-server:/shared"
- "${SEATABLE_DIR}/deps/seatable.sh:/templates/seatable.sh:rx"
- "${SEATABLE_DIR}/deps/licenseparse.py:/opt/seatable/seatable-server-latest/dtable-web/seahub/utils/licenseparse.py:rw"
- "${SEATABLE_DIR}/deps/seafile-controller:/opt/seatable/seatable-server-latest/seafile/bin/seafile-controller:rw"
- "${SEATABLE_DIR}/deps/seaf-server:/opt/seatable/seatable-server-latest/seafile/bin/seaf-server:rw"
environment:
- DB_HOST=mariadb
- DB_ROOT_PASSWD=${SEATABLE_MYSQL_ROOT_PASSWORD:?Variable is not set or empty}
- SEATABLE_SERVER_HOSTNAME=${SEATABLE_SERVER_HOSTNAME:?Variable is not set or empty}
- SEATABLE_SERVER_PROTOCOL=${SEATABLE_SERVER_PROTOCOL:-https}
- SEATABLE_ADMIN_EMAIL=${SEATABLE_ADMIN_EMAIL:?Variable is not set or empty}
- SEATABLE_ADMIN_PASSWORD=${SEATABLE_ADMIN_PASSWORD:?Variable is not set or empty}
- TIME_ZONE=${TIME_ZONE}
depends_on:
mariadb:
condition: service_healthy
memcached:
condition: service_healthy
redis:
condition: service_healthy
networks:
- frontend-net
- backend-seatable-net
healthcheck:
test: ["CMD-SHELL", "curl --fail http://localhost:8000 || exit 1"]
interval: 20s
retries: 3
start_period: 30s
timeout: 10s
mariadb:
image: ${SEATABLE_DB_IMAGE:-mariadb:10.11.6}
restart: unless-stopped
container_name: mariadb
environment:
- MYSQL_ROOT_PASSWORD=${SEATABLE_MYSQL_ROOT_PASSWORD:?Variable is not set or empty}
- MYSQL_LOG_CONSOLE=true
- MARIADB_AUTO_UPGRADE=1
volumes:
- "${SEATABLE_DIR}/mariadb:/var/lib/mysql"
networks:
- backend-seatable-net
healthcheck:
test:
[
"CMD",
"/usr/local/bin/healthcheck.sh",
"--connect",
"--mariadbupgrade",
"--innodb_initialized",
]
interval: 20s
retries: 3
start_period: 30s
timeout: 10s
memcached:
image: ${SEATABLE_MEMCACHED_IMAGE:-memcached:1.6.22}
restart: unless-stopped
container_name: memcached
entrypoint: memcached -m 256
networks:
- backend-seatable-net
healthcheck:
test: ["CMD-SHELL", "timeout 2 bash -c '</dev/tcp/localhost/11211'"]
interval: 20s
retries: 3
timeout: 5s
redis:
image: ${SEATABLE_REDIS_IMAGE:-redis:7.2.3}
restart: unless-stopped
container_name: redis
networks:
- backend-seatable-net
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 20s
retries: 3
timeout: 5s
python-scheduler:
image: ${SEATABLE_SCHEDULER_IMAGE:-seatable/seatable-python-scheduler:latest}
restart: unless-stopped
container_name: python-scheduler
environment:
- TIME_ZONE=${TIME_ZONE}
- DATABASE_NAME=${SEATABLE_PYTHON_PIPELINE_MYSQL_DB_NAME:-scheduler}
- DB_HOST=mariadb
- DB_ROOT_PASSWD=${SEATABLE_PYTHON_PIPELINE_MYSQL_ROOT_PASSWORD:-}
- DB_USER=${SEATABLE_PYTHON_PIPELINE_MYSQL_USER:-}
- DB_PASSWD=${SEATABLE_PYTHON_PIPELINE_MYSQL_PASSWORD:-}
- PYTHON_SCHEDULER_AUTH_TOKEN=${PYTHON_SCHEDULER_AUTH_TOKEN:?Variable is not set or empty}
- SEATABLE_SERVER_URL=${SEATABLE_SERVER_PROTOCOL}://${SEATABLE_SERVER_HOSTNAME}
- PYTHON_STARTER_URL=${PYTHON_STARTER_URL:-http://python-starter:8080}
- PYTHON_SCHEDULER_LOG_LEVEL=${PYTHON_SCHEDULER_LOG_LEVEL:-WARNING}
- DELETE_LOG_DAYS=${DELETE_LOG_DAYS:-30}
- PYTHON_PROCESS_TIMEOUT=${PYTHON_PROCESS_TIMEOUT:-900}
- PYTHON_SCHEDULER_SCRIPT_WORKERS=${PYTHON_SCHEDULER_SCRIPT_WORKERS:-5}
volumes:
- "${PYTHON_SCHEDULER_LOGS_DIR:-/tmp}:/opt/scheduler/logs"
extra_hosts:
- "host.docker.internal:host-gateway"
networks:
- backend-scheduler-net
- frontend-net
- backend-seatable-net
depends_on:
mariadb:
condition: service_healthy
healthcheck:
test:
[
"CMD-SHELL",
"pgrep -f 'python3 scheduler.py' && pgrep -f 'python3 flask_server.py'",
]
interval: 20s
retries: 3
start_period: 20s
timeout: 10s
python-starter:
image: ${SEATABLE_STARTER_IMAGE:-seatable/seatable-python-starter:latest}
restart: unless-stopped
container_name: python-starter
volumes:
- "/var/run/docker.sock:/var/run/docker.sock"
- "${PYTHON_TRANSFER_DIRECTORY_PATH:-/tmp}:${PYTHON_TRANSFER_DIRECTORY_PATH:-/tmp}"
- "${PYTHON_STARTER_LOGS_DIR:-/tmp}:/opt/seatable-python-starter/logs"
environment:
- TIME_ZONE=${TIME_ZONE}
- PYTHON_SCHEDULER_URL=http://python-scheduler
- PYTHON_TRANSFER_DIRECTORY=${PYTHON_TRANSFER_DIRECTORY_PATH:-/tmp}
- PYTHON_RUNNER_IMAGE=${SEATABLE_RUNNER_IMAGE:-seatable/seatable-python-runner:latest}
- PYTHON_STARTER_LOG_LEVEL=${PYTHON_STARTER_LOG_LEVEL:-WARNING}
- PYTHON_PROCESS_TIMEOUT=${PYTHON_PROCESS_TIMEOUT:-900}
- PYTHON_STARTER_THREAD_COUNT=${PYTHON_STARTER_THREAD_COUNT:-10}
- PYTHON_STARTER_USE_ALTERNATIVE_FILE_SERVER_ROOT=${PYTHON_STARTER_USE_ALTERNATIVE_FILE_SERVER_ROOT:-}
- PYTHON_STARTER_ALTERNATIVE_FILE_SERVER_ROOT=${PYTHON_STARTER_ALTERNATIVE_FILE_SERVER_ROOT:-}
- PYTHON_RUNNER_OUTPUT_LIMIT=${PYTHON_RUNNER_OUTPUT_LIMIT:-1000000}
- PYTHON_RUNNER_CONTAINER_MEMORY=${PYTHON_RUNNER_CONTAINER_MEMORY:-2g}
- PYTHON_RUNNER_CONTAINER_CPUS=${PYTHON_RUNNER_CONTAINER_CPUS:-}
- PYTHON_RUNNER_OTHER_OPTIONS=${PYTHON_RUNNER_OTHER_OPTIONS:-}
networks:
- backend-scheduler-net
- frontend-net
networks:
frontend-net:
name: frontend-net
backend-seatable-net:
name: backend-seatable-net
backend-scheduler-net:
name: backend-scheduler-net |
nginx.conf
- 在主机头增加 seatable-server 的 server_name 配置
- 在配置文件中增加 api gateway 的配置内容
- 本例中的配置文件为 /data/seatable/seatable-server/seatable/conf/nginx.conf
| 代码块 | ||
|---|---|---|
| ||
location /api-gateway/ {
add_header Access-Control-Allow-Origin *;
add_header Access-Control-Allow-Methods GET,POST,PUT,DELETE,OPTIONS;
add_header Access-Control-Allow-Headers "deviceType,token, authorization, content-type";
if ($request_method = 'OPTIONS') {
add_header Access-Control-Allow-Origin *;
add_header Access-Control-Allow-Methods GET,POST,PUT,DELETE,OPTIONS;
add_header Access-Control-Allow-Headers "deviceType,token, authorization, content-type";
return 204;
}
proxy_pass http://127.0.0.1:7780/;
proxy_redirect off;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Host $server_name;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_hide_header Access-Control-Allow-Origin;
proxy_hide_header Access-Control-Allow-Methods;
proxy_hide_header Access-Control-Allow-Headers;
access_log /opt/nginx-logs/api-gateway.access.log seatableformat;
error_log /opt/nginx-logs/api-gateway.error.log;
} |
seatable/seatable-python-starter:3.1.1.p 镜像
在 Photon5 为宿主机的容器环境下原有的镜像会出现 uwsgi crash 的情况，主要原因是因为配置文件设置存在问题导致
解决的办法是修改配置文件，然后将配置文件打包到镜像文件并在后续的应用中使用新的镜像文件。以下是经测试能正常应用 uwsgi.ini 配置文件内容
| 代码块 | ||
|---|---|---|
| ||
docker cp ./uwsgi.ini python-starter:/opt/seatable-python-starter/uwsgi.ini |
| 代码块 | ||
|---|---|---|
| ||
[uwsgi] http = :8080 wsgi-file = /opt/seatable-python-starter/runner.py uid = root gid = root chmod-socket = 660 callable = app max-fd = 1048576 process = 1 threads = 1 vacuum = true buffer-size = 65536 stats = 127.0.0.1:9191 procname-prefix = run-python logto = /opt/seatable-python-starter/logs/uwsgi.log log-level = info |
Python 操作及配置
明确 API 和表格内容
- 进入系统选择对应的表（Base），选中“Advance”-"API Token"
- 在出现的 API 页面 填写对应的描述并选择允许读写 的权限
- 将对应的 API 和表格需要操作的项填入脚本文件
打开对应的表，我们需要在“商品档案1”表中按照“商品条码”（字符串类型，数字编码）自动生成对应的“条码”（图片类型，图片文件）
python 脚本文件
| 代码块 | ||
|---|---|---|
| ||
import os
import time
import barcode
from barcode.writer import ImageWriter
from seatable_api import Base, context
api_token = context.api_token or "f83c266f818f4df4be30d8fdef8be7b5bbdd5bed"
server_url = context.server_url or "https://seatable.waringid.me"
TEXT_COL = "商品条码" # column which is expected to be transferred into barcode
BARCODE_IMAGE_COL = "条码"
TABLE_NAME = '商品档案'
BARCODE_TYPE = 'code128'
CUSTOM_OPTIONS = {
"module_width": 0.2, # width of single stripe of barcode, mm
"module_height": 15.0, # height of barcode, mm
"quiet_zone": 6.5, # padding size of first and last stripe to the image, mm
"font_size": 10, # font size of the text below the barcode,pt
"text_distance": 5.0, # distance between the text and the barcode, mm
}
#print("test",TEXT_COL)
CODE = barcode.get_barcode_class(BARCODE_TYPE)
base = Base(api_token, server_url)
base.auth()
#rows1 = base.list_rows(TABLE_NAME,view_name='default', start=1005, limit=20)
#rows2 = base.get_table_by_name(TABLE_NAME)
def get_time_stamp():
return str(int(time.time()*100000))
#print("rows1",rows1)
#print("rows2",rows2)
for row in base.list_rows(TABLE_NAME,view_name='default', start=1001, limit=200):
# continue if the image is already shown up here
#print("image",row.get(BARCODE_IMAGE_COL))
if row.get(BARCODE_IMAGE_COL):
continue
if not row.get(TEXT_COL):
continue
try:
row_id = row.get('_id')
msg = str(row.get(TEXT_COL))
#print("row_id",row_id)
#print("msg-numer",msg)
# create a barcode object
code_img = CODE(msg, writer=ImageWriter())
save_name = "%s_%s" % (row_id, get_time_stamp())
# temporarily saved as an image
file_name = code_img.save("/tmp/%s" % save_name, options=CUSTOM_OPTIONS)
# upload the barcode image to the base
info_dict = base.upload_local_file(file_name, name=None, file_type='image', replace=True)
img_url = info_dict.get('url')
data = {
"条码": [img_url]
}
#print("img_url",img_url)
#print("row_id",row_id)
print("msg-numer-",msg+"ok")
base.update_row(TABLE_NAME, row_id, data)
# remove the image file which is saved temporarily
os.remove(file_name)
except Exception as error:
print("error occured during barcode generate", error)
continue |
| 信息 | ||
|---|---|---|
| ||
|
改进后的 Pythone 脚本
上述的脚本在 6.0.10 版本中执行时没有错误提示并且脚本提示执行成功，但是无法生成正常的条码图片。主要原因如下：
- barcode.save()方法会自动添加文件扩展名（如.png）
- 代码期望的文件路径是：/tmp/{row_id}_{timestamp}
- 实际生成的文件路径是：/tmp/{row_id}_{timestamp}.png
- 后续的os.remove(file_name)可能会删除不存在的文件，但不会报错
| 代码块 | ||
|---|---|---|
| ||
import os
import time
import barcode
from barcode.writer import ImageWriter
from seatable_api import Base, context
import traceback
import sys
# 配置参数
TEXT_COL = "序列号"
BARCODE_IMAGE_COL = "条码"
TABLE_NAME = "商品"
BARCODE_TYPE = 'code128'
CUSTOM_OPTIONS = {
"module_width": 0.2,
"module_height": 15.0,
"quiet_zone": 6.5,
"font_size": 10,
"text_distance": 5.0,
"write_text": False, # 可选：不显示条码下方文字
}
def setup_logging():
"""设置日志"""
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
return logging.getLogger(__name__)
logger = setup_logging()
def init_base_connection():
"""初始化Seatable连接"""
try:
api_token = context.api_token
server_url = context.server_url
if not api_token or not server_url:
raise ValueError("API Token或Server URL未在上下文中设置")
base = Base(api_token, server_url)
base.auth()
logger.info("Seatable连接成功")
return base
except Exception as e:
logger.error(f"初始化连接失败: {e}")
raise
def generate_barcode(text, row_id, barcode_type, options):
"""生成条码图片并返回完整文件路径"""
try:
# 获取条码类
BarcodeClass = barcode.get_barcode_class(barcode_type)
# 创建条码对象
# 注意：确保text是字符串且不为空
text_str = str(text).strip()
if not text_str:
raise ValueError("条码文本内容为空")
code_img = BarcodeClass(text_str, writer=ImageWriter())
# 创建临时目录
temp_dir = "/tmp/seatable_barcodes"
os.makedirs(temp_dir, exist_ok=True)
# 生成唯一文件名（不含扩展名）
timestamp = str(int(time.time() * 100000))
filename = f"{row_id}_{timestamp}"
temp_filepath_no_ext = os.path.join(temp_dir, filename)
# 保存条码图片
# barcode.save()会自动添加扩展名（如.png）
saved_filepath = code_img.save(temp_filepath_no_ext, options=options)
logger.debug(f"条码文件保存到: {saved_filepath}")
# 验证文件是否存在
if not os.path.exists(saved_filepath):
raise FileNotFoundError(f"条码文件未生成: {saved_filepath}")
return saved_filepath
except Exception as e:
logger.error(f"生成条码失败 - 文本: '{text}', 错误: {e}")
return None
def upload_and_update(base, file_path, table_name, row_id, barcode_col):
"""上传文件并更新行"""
try:
# 验证文件存在
if not os.path.exists(file_path):
raise FileNotFoundError(f"文件不存在: {file_path}")
# 获取文件大小（用于调试）
file_size = os.path.getsize(file_path)
logger.debug(f"准备上传文件: {file_path}, 大小: {file_size} 字节")
# 上传文件到Seatable
# 注意：上传时需要指定正确的文件名和类型
file_name = os.path.basename(file_path)
info_dict = base.upload_local_file(
file_path,
name=file_name, # 指定文件名
file_type='image',
replace=True
)
if not info_dict:
raise ValueError("上传文件返回空响应")
if 'url' not in info_dict:
logger.error(f"上传响应中没有url字段: {info_dict}")
raise ValueError("上传响应中没有url字段")
img_url = info_dict['url']
logger.debug(f"文件上传成功，URL: {img_url}")
# 更新行数据
# 注意：条码列应该是"文件"类型，所以值应该是列表形式
update_data = {barcode_col: [img_url]}
# 使用更详细的方法调用
result = base.update_row(table_name, row_id, update_data)
logger.info(f"行 {row_id} 更新成功")
return True, img_url
except Exception as e:
logger.error(f"上传或更新失败: {e}")
return False, str(e)
def cleanup_file(file_path):
"""清理临时文件"""
try:
if file_path and os.path.exists(file_path):
os.remove(file_path)
logger.debug(f"清理临时文件: {file_path}")
except Exception as e:
logger.warning(f"清理文件失败 {file_path}: {e}")
def main():
"""主函数"""
logger.info("=== 开始处理条码生成任务 ===")
try:
# 初始化连接
base = init_base_connection()
# 验证表格和列是否存在
try:
# 获取表格数据
rows = base.list_rows(TABLE_NAME)
total_rows = len(rows)
logger.info(f"表格 '{TABLE_NAME}' 共有 {total_rows} 行数据")
if total_rows == 0:
logger.warning("表格中没有数据")
return
except Exception as e:
logger.error(f"访问表格失败: {e}")
logger.info("请检查：1. 表格名称是否正确 2. API Token权限")
return
# 处理统计
processed = 0
successful = 0
failed = 0
skipped = 0
# 分批处理（避免内存占用过大）
BATCH_SIZE = 20
for i in range(0, total_rows, BATCH_SIZE):
batch = rows[i:i + BATCH_SIZE]
for row in batch:
processed += 1
row_id = row.get('_id')
text_value = row.get(TEXT_COL)
barcode_exists = row.get(BARCODE_IMAGE_COL)
logger.info(f"[{processed}/{total_rows}] 处理行ID: {row_id}, 序列号: '{text_value}'")
# 检查条件
if barcode_exists:
logger.info(f" 跳过 - 条码已存在")
skipped += 1
continue
if not text_value:
logger.warning(f" 跳过 - 序列号为空")
skipped += 1
continue
# 生成条码
barcode_path = None
try:
barcode_path = generate_barcode(
text_value, row_id, BARCODE_TYPE, CUSTOM_OPTIONS
)
if not barcode_path:
logger.error(f" 失败 - 条码生成返回空路径")
failed += 1
continue
# 上传并更新
success, result = upload_and_update(
base, barcode_path, TABLE_NAME, row_id, BARCODE_IMAGE_COL
)
if success:
logger.info(f" 成功 - 条码已生成并更新")
successful += 1
else:
logger.error(f" 失败 - {result}")
failed += 1
except Exception as e:
logger.error(f" 处理异常: {e}")
logger.debug(traceback.format_exc())
failed += 1
finally:
# 清理临时文件
if barcode_path:
cleanup_file(barcode_path)
# 每批处理后暂停一下，避免API限制
if i + BATCH_SIZE < total_rows:
time.sleep(0.5)
# 最终报告
logger.info("=== 处理完成 ===")
logger.info(f"总计行数: {total_rows}")
logger.info(f"已处理: {processed}")
logger.info(f"成功: {successful}")
logger.info(f"失败: {failed}")
logger.info(f"跳过: {skipped}")
if successful > 0:
logger.info("✅ 部分条码生成成功！")
elif failed == total_rows:
logger.error("❌ 所有处理都失败了，请检查上述错误信息")
except Exception as e:
logger.error(f"脚本执行失败: {e}")
logger.debug(traceback.format_exc())
sys.exit(1)
if __name__ == "__main__":
main() |
参考
| 目录 |
|---|