Files
cmshoppe/docs/api.md
T
chengma 26dac2a718 feat: 抽取 Shopee 编辑器模块
新增 app/editor.py,封装商品页打开、登录检测、标题读取写入、封面读取下载、封面上传拖拽、更新按钮点击和 apply_task。

扩展 app/cdp.py 的 host 参数,支持后续按账号端口连接 CDP。

同步任务看板、模块合约、当前状态和进度记录,标记 T-001 完成;记录测试商品标题读写、封面读取与 1_TY030.jpg 上传拖拽验证结果。
2026-06-26 18:02:57 +08:00

12 KiB
Raw Blame History

模块 / CLI 合约

本项目无后端 API,全部为本地 Python 模块合约。实现前可细化,但不要另起一套不兼容接口。

通用约定

  • 形态:本地函数 + 子进程(Chrome)+ CDP(127.0.0.1:<port>)+ SQLite + openpyxl + AI 服务调用。
  • 编码:UTF-8;传 Chrome / setFileInputFiles 的路径为 Windows 绝对路径。
  • 凭证:登录态在 user-data-dir;密码、AI Key 本地明文存于 config/DB;config.json、config/ai_models.json、cmshopee.db、chrome_user_data_dir/、images/ 必须 gitignore;UI 打码显示,不出现在日志/导出。
  • 失败处理:抛带中文说明的异常或返回状态字段;GUI 负责提示,不静默吞错。

appconfig 模块(app/appconfig.py,待建)

读写 config.json(schema 见 架构 5.1)。

load_config(path="config.json") -> dict      # 不存在则写默认
chrome_path() -> str
user_data_root() -> str
image_dir() -> str
db_path() -> str
ai_config() -> dict                          # default_text_model/default_image_model/
                                             # title_concurrency/image_concurrency/retry/jpg_quality/
                                             # resolution/resolution_timeouts
response_timeout() -> int                     # = resolution_timeouts[resolution](返回超时,随分辨率)

# AI 模型清单 config/ai_models.json(含本地明文密钥;CRUD 由 ⑤ 设置)
list_ai_models(category=None) -> list[dict]   # category=text/image 过滤;含 connect_timeout_seconds 等
add_ai_model(model) -> None                   # name 唯一校验
update_ai_model(name, **fields) -> None
delete_ai_model(name) -> None                 # 至少各留一个 text+image;删到剩一禁用
test_ai_model(name) -> dict                   # 「测试连接」:用 key/url/model 发最小请求 -> {ok, error}
get_model(name) -> dict                        # 返回模型定义,含 api_key(调用方不得写日志)

db 模块(app/db.py,待建)

SQLite 读写,表见 架构 5.2。

connect(path=None) -> sqlite3.Connection      # 设置 foreign_keys/WAL/busy_timeout/synchronous/row_factory
init_db(path)
create_batch(file_paths, note=None) -> str
get_batch(batch_id) -> Batch|None
list_batches(status=None) -> list[Batch]
update_batch(batch_id, **fields) -> None
# 账号
list_accounts() -> list[Account]
get_account_by_alias(alias) -> Account|None
add_account(account_name, alias, region_host, debug_port, password=None, note=None) -> Account
update_account(alias, **fields) -> None
delete_account(alias) -> None
# 任务 / 各阶段结果
insert_tasks(batch_id, rows) -> int          # 写输入列;rows 含 source_file_abs/source_sheet/source_row/row_key
list_tasks(batch_id=None, stage=None, status=None, alias=None) -> list[Task]
mark_running(task_id, phase) -> None
mark_failed(task_id, phase, error) -> None
mark_skipped(task_id, reason) -> None
set_collected(task_id, old_title, old_cover_path) -> None      # stage=collected,status=success, attempts+1
set_generated(task_id, new_title, new_cover_path) -> None      # stage=generated,status=success, attempts+1
set_applied(task_id, committed, error=None) -> None            # stage=applied,status=success/failed, attempts+1

stage 表示最后成功业务阶段(imported→collected→generated→applied);status 表示当前处理结果(pending/running/success/failed/skipped/cancelled)。任意步失败写 last_error 且 status=failed,stage 不前进。无 confirmed 阶段。

SQLite 连接规则:

  • 每个 worker/线程使用自己的 connect();禁止跨线程共享 connection。
  • connect() 必须设置 PRAGMA foreign_keys=ON、journal_mode=WAL、busy_timeout=5000、synchronous=NORMAL。
  • DB 写入短事务、单条提交;Excel 回写失败不回滚 DB。

excel 模块(app/excel.py,待建,依赖 openpyxl)

import_tasks(file_paths: list[str]) -> dict
# 只解析【输入列】:账号名、别名、商品id;并记录 source_file/source_file_abs/source_sheet/source_row/row_key
# -> {"rows": [...], "stats": {"files": int, "total": int, "valid": int, "invalid": int, "file_errors": [...]}}
#    必需列缺失(别名/商品id)= 整个文件拒绝并记录到 file_errors,不导入该文件任何行
#    invalid = 单行缺别名/商品id 或商品id格式错误等脏数据;逐行跳过,不阻塞同文件其他有效行

match_summary(rows: list[dict], accounts: list) -> dict
# 用 accounts 的别名对 rows 做匹配统计(导入汇总栏用)
# -> {"matched": int, "unmatched": int, "by_account": {别名: 行数}, "unmatched_aliases": [..]}

write_back(batch_id, excel_path=None) -> dict
# 把【旧标题/旧封面/新标题/新封面/更新状态】批量回写到【原 Excel】
# excel_path 为空则按 source_file_abs 分组回写本批次涉及的所有原文件
# 原文件被占用(锁) → 抛错,调用方提示“请关闭后重试”,或改用 export_copy
export_copy(batch_id, out_dir_or_path) -> dict       # 退路:另存新结果文件,不动原文件

列模板见 架构 5.3;别名以“别名”列为权威。

config 模块(app/config.py,待建)

make_slug(alias) -> str                  # 别名→唯一 slug [a-z0-9_]
ensure_user_data_dir(slug) -> str        # chrome_user_data_dir/<slug> 绝对路径,按需创建

chrome 模块(app/chrome.py,待建)

build_launch_args(account) -> list[str]  # chrome + --remote-debugging-port + --remote-allow-origins=* + --user-data-dir
launch_chrome(account) -> subprocess.Popen
wait_debug_ready(port, timeout=60) -> bool
is_running(port) -> bool
create_shortcut(account, dest_dir=None) -> str   # 可选 .lnk,PowerShell WScript.Shell

cdp 模块(app/cdp.py,T-000 由根目录 cdp.py 迁入)

CDP_HOST: str
http_get(path, host=None); find_product_tab(item_id, host=None); create_tab(url, host=None)
class CDP: send/ev/val/object_id/drag/close   # suppress_origin、trust_env=False

editor 模块(app/editor.py,已建,重构自现有脚本)

is_logged_in(account) -> bool                # 重定向登录页或缺 SPC_ST → False
open_product(account, item_id) -> CDP        # 连端口、导航商品页、等就绪

# 采集(只读)
read_title(cdp) -> str
read_cover_src(cdp) -> str                   # 第一张 itembox 的 img.src
download_cover(src, out_path) -> str         # 下载旧封面到本地
collect(account, task) -> dict               # -> {old_title, old_cover_path}

# 应用
change_title(cdp, new_title) -> dict         # {ok, value, modelvalue},要求三者相等
replace_cover(cdp, image_win_path) -> dict   # 上传→等CDN→拖第一位;满9张暂返回 FULL_IMAGE_SLOTS,删除流程留给 T-502
click_update(cdp) -> dict                    # {clicked, reason};禁用则记失败
apply_task(account, task) -> dict            # 对已生成任务:换标题+换封面+点「更新」提交(调用前必须已做批量确认)
                                             # -> {committed, error}

ai 模块(app/ai.py,待建,外部 AI,服务商待定)

gen_title(title_prompt, old_title, retry=2) -> str
# 文本生成:提示词 + 旧标题 → 新标题

gen_cover(cover_prompt, old_cover_path, out_path, resolution, jpg_quality, retry=2) -> str
# 图像生成(image-to-image):提示词 + 旧封面 → 新封面,按 resolution 生成、jpg_quality 存盘,返回路径

generate_batch(tasks, prompts, ai_cfg, on_progress, should_stop) -> None
# 编排:先以 title_concurrency 线程池并发跑 gen_title,再以 image_concurrency 并发跑 gen_cover
# 每条完成即 db.set_generated(实时落库);should_stop() 为真则取消未开始项
# on_progress(标题完成数, 封面完成数, 失败数) 回调刷新进度

要点:

  • 标题用 default_text_model、封面用 default_image_model(appconfig.get_model 取定义,含 url/key/api_type)。
  • 连接超时 = 模型 connect_timeout_seconds;返回超时 = appconfig.response_timeout()(随分辨率:512/1k/2k/4k → 180/240/360/600)。
  • 并发数/重试/分辨率/jpg 质量来自 appconfig.ai_config();Key 本地明文存储,但不入日志、不导出。
  • 标题快、图片慢:分两段、各用各自并发数;失败按 retry 重试,仍失败记 error 不阻塞其余。
  • 调用有成本与失败可能:超时、限流、内容安全拒绝都要返回明确错误。
  • 生成结果直接进入 ③ 更新候选;③ 点击「开始更新」后弹窗批量确认,确认后提交线上。本地留档 + 回写 Excel 供追溯。

prompts 模块(app/prompts.py,待建)

# 标题提示词:单文件
load_title_prompt(path="title_prompt.txt") -> str        # 启动回显;缺失返回 ""
save_title_prompt(text, path="title_prompt.txt") -> None # 「保存」按钮

# 封面提示词:多模板(prompts/cover/<名称>.txt)
list_cover_templates() -> list[str]                      # 模板名列表(下拉用)
load_cover_template(name) -> str
save_cover_template(name, text) -> None                  # 保存 / 另存为
rename_cover_template(old, new) -> None                  # 重名校验,重复则报错
delete_cover_template(name) -> None                      # 删除(二次确认由 GUI 负责)

# 变量替换
render_prompt(template_text, task) -> str
# 占位符 {旧标题}/{新标题}/{商品id}/{店铺} → 该任务真实值;预览与生成时调用

要点:

  • 「插入标题」在封面提示词光标处插入 {新标题};「预览」对选中任务调用 render_prompt 后展示。
  • 生成封面时 gen_cover 的 prompt = render_prompt(当前封面模板, task)。
  • 模板与 title_prompt.txt 均为可手改的纯文本文件。

gui / workers 模块(app/gui.py / app/workers.py,待建,PySide6)

# GUI 入口
main() -> int                              # 创建 QApplication + MainWindow
class MainWindow(QMainWindow)              # QTabWidget: ①②③④⑤

# worker 约定
class BaseWorker(QObject):
    progress = Signal(dict)                # {"done": int, "total": int, ...}
    row_updated = Signal(int, dict)         # task_id, changed fields
    log = Signal(str)
    failed = Signal(int, str)               # task_id, error
    finished = Signal(dict)                 # summary
    cancelled = Signal(dict)
    def cancel(self) -> None: ...

run_worker(worker: BaseWorker) -> QThread   # 绑定 signals、启动、收尾 deleteLater

要点:

  • GUI 线程只操作 Qt widget;后台 worker 不直接访问 QWidget。
  • 采集、AI 生成、更新、Excel 回写都通过 worker 执行,用 signal 回传进度。
  • 每个 worker/线程按需创建自己的 SQLite connection,不跨线程共享连接。
  • ③ 的批量确认弹窗在 GUI 主线程完成;用户确认后才创建 ApplyWorker。

启动入口

# main.py
from app.gui import main

raise SystemExit(main())

# app/__main__.py
from .gui import main

raise SystemExit(main())

正式运行入口:python main.py;开发/包入口:python -m app。

CLI / 触发合约(现有脚本,过渡期保留)

python prototypes/demo.py                 # 单账号闭环:改标题+换封面,不提交
set AUTO=1 && python prototypes/demo.py
set UPDATE=1 && python prototypes/demo.py # 走完点击「更新」提交

环境变量:CDP_HOST、ITEM_ID、IMG_WIN、NEW_TITLE、AUTO、UPDATE。

待实现时确认

  • AI 服务商/模型/计费;图像 image-to-image 能力与合规。
  • 满 9 张删除封面的确认框选择器(需实测)。
  • 旧封面下载的图片格式/扩展名处理。