diff --git a/app/ai.py b/app/ai.py new file mode 100644 index 0000000..03ebffc --- /dev/null +++ b/app/ai.py @@ -0,0 +1,416 @@ +"""AI generation helpers backed by configurable HTTP model endpoints.""" + +import base64 +import copy +import json +import mimetypes +import os +import time +import urllib.error +import urllib.request +import uuid + +from . import appconfig + + +class AIError(RuntimeError): + """Raised when AI generation cannot complete.""" + + +_RESOLUTION_SIZES = { + "512": (512, 512), + "1k": (1024, 1024), + "2k": (2048, 2048), + "4k": (4096, 4096), +} + + +def gen_title( + title_prompt, + old_title, + retry=None, + config=None, + models_path=appconfig.AI_MODELS_PATH, +): + """Generate a new product title from a prompt and the old title.""" + + cfg = appconfig.load_config() if config is None else config + ai_cfg = appconfig.ai_config(cfg) + model = _role_model("text", ai_cfg.get("default_text_model"), models_path) + payload = _chat_payload( + model, + [ + {"role": "system", "content": str(title_prompt or "").strip()}, + { + "role": "user", + "content": "旧标题:\n%s\n\n请只返回新标题,不要解释。" % str(old_title or ""), + }, + ], + ) + attempts = _attempt_count(ai_cfg, retry) + data = _call_with_retry( + model, + payload, + cfg, + attempts, + request_kind="json", + ) + text = _extract_text(data).strip() + if not text: + raise AIError("AI 返回为空标题") + return text + + +def gen_cover( + cover_prompt, + old_cover_path, + out_path, + resolution=None, + jpg_quality=None, + retry=None, + config=None, + models_path=appconfig.AI_MODELS_PATH, +): + """Generate a new cover image and save it as a JPEG file.""" + + old_cover_path = os.path.abspath(str(old_cover_path)) + if not os.path.exists(old_cover_path): + raise FileNotFoundError("旧封面图片不存在: %s" % old_cover_path) + if not out_path: + raise AIError("缺少新封面输出路径") + + cfg = appconfig.load_config() if config is None else config + ai_cfg = appconfig.ai_config(cfg) + model = _role_model("image", ai_cfg.get("default_image_model"), models_path) + resolution = str(resolution or ai_cfg.get("resolution", "1k")) + quality = _jpg_quality(jpg_quality if jpg_quality is not None else ai_cfg.get("jpg_quality", 90)) + attempts = _attempt_count(ai_cfg, retry) + + api_type = model.get("api_type", "auto") + if api_type == "images_edits": + body, content_type = _image_edit_body(model, cover_prompt, old_cover_path, resolution) + data = _call_with_retry( + model, + body, + cfg, + attempts, + request_kind="multipart", + content_type=content_type, + ) + else: + payload = _image_chat_payload(model, cover_prompt, old_cover_path, resolution) + data = _call_with_retry(model, payload, cfg, attempts, request_kind="json") + + image_bytes = _extract_image_bytes(data, model, cfg) + return _save_jpeg(image_bytes, out_path, resolution, quality) + + +def _role_model(category, name, models_path): + if not name: + raise AIError("未配置默认 %s 模型" % category) + model = appconfig.get_model(name, path=models_path) + if model.get("category") != category: + raise AIError("模型 %s 不是 %s 类别" % (name, category)) + if not model.get("enabled", True): + raise AIError("模型已禁用: %s" % name) + missing = [ + field + for field in ("url", "model", "api_key") + if not str(model.get(field, "")).strip() + ] + if missing: + raise AIError("模型 %s 缺少字段: %s" % (name, ", ".join(missing))) + return model + + +def _attempt_count(ai_cfg, retry): + retry_count = ai_cfg.get("retry", 2) if retry is None else retry + return max(1, int(retry_count) + 1) + + +def _read_timeout(model, config): + return int(model.get("timeout_seconds") or appconfig.response_timeout(config)) + + +def _connect_timeout(model): + return int(model.get("connect_timeout_seconds") or 30) + + +def _headers(model, content_type): + return { + "Authorization": "Bearer " + model["api_key"], + "Content-Type": content_type, + } + + +def _call_with_retry(model, body, config, attempts, request_kind, content_type=None): + last_exc = None + for index in range(attempts): + try: + return _call_once(model, body, config, request_kind, content_type=content_type) + except Exception as exc: + last_exc = exc + if index + 1 >= attempts: + break + time.sleep(min(2.0, 0.4 * (index + 1))) + raise AIError( + "AI 调用失败(已尝试 %s 次): %s" + % (attempts, _redact(str(last_exc), model)) + ) from last_exc + + +def _call_once(model, body, config, request_kind, content_type=None): + if request_kind == "json": + data = json.dumps(body, ensure_ascii=False).encode("utf-8") + content_type = "application/json" + else: + data = body + request = urllib.request.Request( + model["url"], + data=data, + headers=_headers(model, content_type), + method="POST", + ) + timeout = max(_connect_timeout(model), _read_timeout(model, config)) + try: + with urllib.request.urlopen(request, timeout=timeout) as response: + raw = response.read() + except urllib.error.HTTPError as exc: + detail = _safe_http_error(exc, model) + raise AIError("HTTP %s: %s" % (exc.code, detail)) from exc + except urllib.error.URLError as exc: + raise AIError(_redact(str(exc.reason), model)) from exc + try: + return json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise AIError("AI 返回不是有效 JSON") from exc + + +def _safe_http_error(exc, model): + try: + body = exc.read(2048).decode("utf-8", errors="replace") + except Exception: + body = "" + return _redact(body or str(exc), model) + + +def _chat_payload(model, messages): + payload = { + "model": model["model"], + "messages": messages, + } + payload.update(copy.deepcopy(model.get("extra_body", {}))) + return payload + + +def _image_chat_payload(model, cover_prompt, old_cover_path, resolution): + prompt = "%s\n\n目标分辨率:%s。" % (str(cover_prompt or "").strip(), resolution) + content = [ + {"type": "text", "text": prompt.strip()}, + { + "type": "image_url", + "image_url": {"url": _image_data_url(old_cover_path)}, + }, + ] + return _chat_payload(model, [{"role": "user", "content": content}]) + + +def _image_edit_body(model, cover_prompt, old_cover_path, resolution): + fields = { + "model": model["model"], + "prompt": str(cover_prompt or ""), + "size": _resolution_size_text(resolution), + } + fields.update(copy.deepcopy(model.get("extra_body", {}))) + files = { + "image": ( + os.path.basename(old_cover_path), + open(old_cover_path, "rb").read(), + mimetypes.guess_type(old_cover_path)[0] or "application/octet-stream", + ) + } + return _multipart_body(fields, files) + + +def _multipart_body(fields, files): + boundary = "----cmshopee-%s" % uuid.uuid4().hex + chunks = [] + for name, value in fields.items(): + chunks.extend( + [ + ("--%s\r\n" % boundary).encode("utf-8"), + ('Content-Disposition: form-data; name="%s"\r\n\r\n' % name).encode("utf-8"), + str(value).encode("utf-8"), + b"\r\n", + ] + ) + for name, file_info in files.items(): + filename, data, content_type = file_info + chunks.extend( + [ + ("--%s\r\n" % boundary).encode("utf-8"), + ( + 'Content-Disposition: form-data; name="%s"; filename="%s"\r\n' + % (name, filename) + ).encode("utf-8"), + ("Content-Type: %s\r\n\r\n" % content_type).encode("utf-8"), + data, + b"\r\n", + ] + ) + chunks.append(("--%s--\r\n" % boundary).encode("utf-8")) + return b"".join(chunks), "multipart/form-data; boundary=%s" % boundary + + +def _image_data_url(path): + mime = mimetypes.guess_type(path)[0] or "image/jpeg" + with open(path, "rb") as fh: + encoded = base64.b64encode(fh.read()).decode("ascii") + return "data:%s;base64,%s" % (mime, encoded) + + +def _extract_text(data): + if isinstance(data, dict): + for key in ("output_text", "text", "content"): + value = data.get(key) + if isinstance(value, str): + return value + choices = data.get("choices") + if isinstance(choices, list) and choices: + first = choices[0] + if isinstance(first, dict): + if isinstance(first.get("text"), str): + return first["text"] + message = first.get("message") or {} + content = message.get("content") + return _content_text(content) + return "" + + +def _content_text(content): + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [] + for item in content: + if not isinstance(item, dict): + continue + value = item.get("text") or item.get("content") + if isinstance(value, str): + parts.append(value) + return "".join(parts) + return "" + + +def _extract_image_bytes(data, model, config): + image_ref = _find_image_ref(data) + if not image_ref: + raise AIError("AI 返回中没有图片数据") + if image_ref.startswith("data:"): + return _decode_data_url(image_ref) + if _looks_base64(image_ref): + return base64.b64decode(image_ref) + return _download_image(image_ref, model, config) + + +def _find_image_ref(value): + if isinstance(value, dict): + for key in ("b64_json", "base64", "image_base64", "image", "url"): + candidate = value.get(key) + if isinstance(candidate, str) and candidate.strip(): + return candidate.strip() + image_url = value.get("image_url") + if isinstance(image_url, str): + return image_url + if isinstance(image_url, dict): + candidate = image_url.get("url") + if isinstance(candidate, str): + return candidate + for key in ("data", "choices", "output", "content", "images"): + candidate = _find_image_ref(value.get(key)) + if candidate: + return candidate + message = value.get("message") + if message is not None: + candidate = _find_image_ref(message) + if candidate: + return candidate + elif isinstance(value, list): + for item in value: + candidate = _find_image_ref(item) + if candidate: + return candidate + return None + + +def _decode_data_url(value): + if "," not in value: + raise AIError("图片 data URL 格式错误") + return base64.b64decode(value.split(",", 1)[1]) + + +def _looks_base64(value): + compact = value.strip() + if compact.startswith(("http://", "https://")): + return False + if len(compact) < 32: + return False + allowed = set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=\n\r") + return all(char in allowed for char in compact) + + +def _download_image(url, model, config): + request = urllib.request.Request(url, method="GET") + try: + with urllib.request.urlopen(request, timeout=_read_timeout(model, config)) as response: + return response.read() + except Exception as exc: + raise AIError("下载 AI 图片失败: %s" % _redact(str(exc), model)) from exc + + +def _save_jpeg(image_bytes, out_path, resolution, jpg_quality): + try: + from PIL import Image + except ImportError as exc: + raise AIError("缺少 Pillow,无法保存 AI 图片") from exc + + import io + + out_path = os.path.abspath(str(out_path)) + directory = os.path.dirname(out_path) + if directory: + os.makedirs(directory, exist_ok=True) + size = _resolution_size(resolution) + try: + with Image.open(io.BytesIO(image_bytes)) as image: + image = image.convert("RGB") + if size: + image = image.resize(size, Image.LANCZOS) + image.save(out_path, "JPEG", quality=jpg_quality, optimize=True) + except Exception as exc: + raise AIError("AI 图片保存失败: %s" % exc) from exc + return out_path + + +def _resolution_size(resolution): + return _RESOLUTION_SIZES.get(str(resolution)) + + +def _resolution_size_text(resolution): + size = _resolution_size(resolution) + if not size: + return str(resolution) + return "%sx%s" % size + + +def _jpg_quality(value): + value = int(value) + return min(100, max(1, value)) + + +def _redact(text, model): + result = str(text) + for secret in (model.get("api_key"),): + if secret: + result = result.replace(str(secret), "***") + return result diff --git a/docs/00-ai-start-here.md b/docs/00-ai-start-here.md index a10d228..6fd3570 100644 --- a/docs/00-ai-start-here.md +++ b/docs/00-ai-start-here.md @@ -11,7 +11,7 @@ cmshopee 是一个 Windows 本地桌面自动化工具(PySide6,5 Tab), 目标闭环:④ 配账号并登录 → ① 导入 Excel(按“别名”列关联账号)、采集旧标题/旧封面并回写 → ② 用提示词 AI 生成新标题/新封面(不设逐条确认阶段)→ ③ 对已生成任务点击「开始更新」,弹窗确认后批量改标题+换封面并点「更新」提交 → 结果实时存 SQLite、批量回写原 Excel。 -存储:应用设置 `config.json` + AI 模型清单 `config/ai_models.json`(含本地明文 AI Key,必须 gitignore)+ 业务数据 SQLite `cmshopee.db` + Excel 用 openpyxl + 图片存本地 `images/`。AI 服务商待定。 +存储:应用设置 `config.json` + AI 模型清单 `config/ai_models.json`(含本地明文 AI Key,必须 gitignore)+ 业务数据 SQLite `cmshopee.db` + Excel 用 openpyxl + 图片存本地 `images/`。AI 服务商/模型由 `config/ai_models.json` 配置。 ## 必读顺序 diff --git a/docs/03-tech-stack.md b/docs/03-tech-stack.md index 06b299a..3200106 100644 --- a/docs/03-tech-stack.md +++ b/docs/03-tech-stack.md @@ -18,8 +18,8 @@ | 业务数据 | SQLite(stdlib `sqlite3`,`cmshopee.db`) | 已定 | 账号、任务、结果:成行增长、要查询/统计/导出 | | Excel 读写 | `openpyxl` | 已定 | 导入任务、回写结果;stdlib 读不了 .xlsx | | AI 模型注册 | `config/ai_models.json` 多模型清单(HTTP 调用) | 已定(结构) | 每模型 name/category(text/image)/url/model/key/api_type/连接超时;⑤ 设置可增删改+测试连接 | -| AI 文本生成 | 选 `default_text_model`(category=text) | 选型在配置 | 提示词+旧标题→新标题 | -| AI 图像生成 | 选 `default_image_model`(category=image,image-to-image) | 选型在配置 | 提示词+旧封面→新封面;分辨率 512/1k/2k/4k,返回超时随分辨率 | +| AI 文本生成 | `app/ai.py` 读取 `default_text_model`(category=text),通用 chat JSON HTTP | 已接入 | 提示词+旧标题→新标题;失败重试,错误脱敏 | +| AI 图像生成 | `app/ai.py` 读取 `default_image_model`(category=image),支持 chat 多模态 JSON / images_edits multipart | 已接入 | 提示词+旧封面→新封面;分辨率 512/1k/2k/4k,jpg_quality 存盘,返回超时随分辨率 | | 并发 | 标准库 `concurrent.futures.ThreadPoolExecutor` | 已定 | 标题/图片分别按并发数并行;可停止、可重试 | | 图片处理 | `requests`(下载)+ `Pillow`(按分辨率/jpg质量存盘) | 部分待定 | 下载旧封面;新封面按 resolution 生成、jpg_quality 存盘 | | 测试 | `python -m compileall app main.py` + `unittest` + 手动 CDP/AI 验证 | 已定(分层) | 配置/DB/Excel/prompts 用单测;CDP/Shopee 与真实 AI 属集成验证或 mock | @@ -32,7 +32,7 @@ - **Excel 用 openpyxl**:运营用真实 .xlsx;stdlib 无法读写 xlsx,引入一个轻依赖比改用 CSV 更贴合用户习惯。 - **多账号隔离用独立 user-data-dir,不用 Chrome profile**:profile 共享同一 user-data-dir/进程/调试端口,无法每账号独立 CDP 与并行;独立 user-data-dir 才契合自动化。详见 [架构 3.0](04-architecture.md)。 - **快捷方式生成用 PowerShell(无额外依赖)**:用 `WScript.Shell.CreateShortcut` 生成 `.lnk`,不引入 `pywin32` 等依赖。 -- **AI 服务商待定**:需选支持文本生成 + 图像 image-to-image 的服务;选型要权衡能力、合规(电商主图)、计费、Key 管理。**未确认前不在代码里写死某家 SDK**,先在 `app/ai.py` 留稳定接口(`gen_title`/`gen_cover`)。 +- **AI 服务商不写死在代码里**:T-301 已采用 `config/ai_models.json` 的通用 HTTP 接入,当前支持 OpenAI-compatible chat JSON 与 images_edits multipart;具体服务商/模型/Key 由⑤设置维护。 - **AI 产出无逐条审核**:生成的新标题/新封面经 ③ 批量确认后提交线上;无常驻提交开关,本地留档 + 回写 Excel 供追溯。 - **不引入数据库(指外部 DB)**:用 stdlib SQLite 足够;不引入 Postgres/MySQL 等。 - **生产在 Windows 直跑**:开发期我们用过 WSL→Windows 的 `netsh portproxy`(9333→9222)连 CDP;但 GUI 与 Chrome 都在 Windows 时,直接连 `127.0.0.1:9222`,无需 portproxy。 diff --git a/docs/04-architecture.md b/docs/04-architecture.md index c6c01dd..a89ec8a 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -38,7 +38,7 @@ Shopee 卖家中心页面 / 本地图片目录 - GUI 入口:根目录 `main.py` 调用 `app/gui.py`(待建,PySide6 + `QMainWindow` + `QTabWidget`,5 Tab);也支持 `python -m app`。 - 核心模块统一放在正式代码包 `app/`:`appconfig.py`、`db.py`、`excel.py`、`config.py`、`accounts.py`、`chrome.py`、`editor.py`、`workers.py`、`ai.py`、`prompts.py`;CDP 底座迁入 `app/cdp.py`(当前根目录 `cdp.py` 为已验证来源)。 - 已验证脚本(重构进模块):`prototypes/demo.py`、`prototypes/set_title.py`、`prototypes/set_cover.py`、`prototypes/get_title.py`、`prototypes/cookies.py`、`prototypes/inspect_images.py`、`prototypes/grab.py`。 -- 外部依赖:本机 Google Chrome;Shopee;AI 服务(文本+图像,服务商待定);`openpyxl`。 +- 外部依赖:本机 Google Chrome;Shopee;AI 服务(文本+图像,服务商/模型由 `config/ai_models.json` 配置);`openpyxl`。 ## 二、流水线(核心) @@ -71,7 +71,7 @@ imported → collected → generated → applied - `chrome`:拼接启动命令、启动、探测端口、(可选)生成快捷方式。 - `cdp`:连接调试端口、找/开 tab、执行 JS、拖拽、注入文件。 - `editor`:登录检测、**采集**(读旧标题、下载旧封面)、改标题、换封面、点更新。 -- `ai`:`gen_title(prompt, old_title)`、`gen_cover(prompt, old_cover_path)`(外部 AI;服务商待定)。 +- `ai`:`gen_title(prompt, old_title)`、`gen_cover(prompt, old_cover_path)`(外部 AI;模型清单配置;通用 HTTP)。 **存储(同一事实只存一处)** diff --git a/docs/06-tasks.md b/docs/06-tasks.md index 594c4cb..db7297e 100644 --- a/docs/06-tasks.md +++ b/docs/06-tasks.md @@ -59,7 +59,7 @@ | ID | 任务 | 依赖 | 验收要点 | 状态 | | --- | --- | --- | --- | --- | -| T-301 | 确定 AI 服务商/模型并接入 `app/ai.py`(`gen_title`/`gen_cover`,带重试/分辨率/jpg质量) | T-005 | 从 `config/ai_models.json` 读取模型与本地明文 Key;`gen_cover` 支持 resolution+jpg_quality;失败按 retry 重试;错误明确;日志脱敏 | TODO | +| T-301 | 确定 AI 服务商/模型并接入 `app/ai.py`(`gen_title`/`gen_cover`,带重试/分辨率/jpg质量) | T-005 | 从 `config/ai_models.json` 读取模型与本地明文 Key;`gen_cover` 支持 resolution+jpg_quality;失败按 retry 重试;错误明确;日志脱敏 | DONE | | T-302 | Tab② 左右布局:左提示词(标题/封面),右按批次/店铺/状态筛选 + 任务列表 | T-301, T-203 | 左 ~1/4 提示词多行;右筛选+列表(店铺/商品id/旧标题/新标题/状态) | TODO | | T-302p | `app/prompts.py` + Tab② 提示词管理 | T-302 | 标题保存/启动回显 title_prompt.txt;封面多模板(下拉+新建/保存/另存为/重命名/删除,存 prompts/cover/);插入 `{新标题}`;预览变量替换;render_prompt 接入生成 | TODO | | T-303 | Tab② 开始生成(单按钮)+ 停止 + 进度:**先并发标题再并发图片** | T-302, T-104b | `generate_batch` 先 title_concurrency 并发标题、再 image_concurrency 并发图片;worker/signal 回传进度;每条 set_generated 立即写库;停止可取消未开始项;进度 标题/封面/失败 计数;双击弹窗看新旧封面 | TODO | diff --git a/docs/api.md b/docs/api.md index acf9800..cfae69e 100644 --- a/docs/api.md +++ b/docs/api.md @@ -218,25 +218,30 @@ apply_task(account, task) -> dict # 对已生成任务:换标题+ `login_status()` 不自动登录;无 Shopee tab 时打开卖家中心根地址 `https:///`(默认 `https://seller.shopee.tw/`)用于检测/人工登录。判断规则:最终 URL 是登录页 → `LOGIN_PAGE`;缺少 `SPC_ST`/`SPC_U` → `NO_SESSION_COOKIE`;有会话 Cookie → 已登录。 -## ai 模块(`app/ai.py`,待建,外部 AI,服务商待定) +## ai 模块(`app/ai.py`,已建,外部 AI,通用 HTTP) ```python -gen_title(title_prompt, old_title, retry=2) -> str -# 文本生成:提示词 + 旧标题 → 新标题 +class AIError(RuntimeError): ... -gen_cover(cover_prompt, old_cover_path, out_path, resolution, jpg_quality, retry=2) -> str -# 图像生成(image-to-image):提示词 + 旧封面 → 新封面,按 resolution 生成、jpg_quality 存盘,返回路径 +gen_title(title_prompt, old_title, retry=None, config=None, models_path="config/ai_models.json") -> str +# 文本生成:读取 default_text_model,chat JSON 请求;提示词 + 旧标题 → 新标题 + +gen_cover(cover_prompt, old_cover_path, out_path, resolution=None, jpg_quality=None, retry=None, config=None, models_path="config/ai_models.json") -> str +# 图像生成(image-to-image):读取 default_image_model;chat 多模态 JSON 或 images_edits multipart; +# 支持返回 url / data URL / b64_json,按 resolution resize 并以 jpg_quality 保存 JPEG,返回路径 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(标题完成数, 封面完成数, 失败数) 回调刷新进度 +# T-303 实现 ``` 要点: - 标题用 `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)**。 +- `api_type=chat/auto` 走 OpenAI-compatible chat JSON;`api_type=images_edits` 走 multipart form。 +- 连接超时参考模型 `connect_timeout_seconds`;**返回超时 = 模型 `timeout_seconds` 或 `appconfig.response_timeout()`(随分辨率:512/1k/2k/4k → 180/240/360/600)**。 - 并发数/重试/分辨率/jpg 质量来自 `appconfig.ai_config()`;Key 本地明文存储,但不入日志、不导出。 - 标题快、图片慢:分两段、各用各自并发数;失败按 `retry` 重试,仍失败记 error 不阻塞其余。 - 调用有成本与失败可能:超时、限流、内容安全拒绝都要返回明确错误。 diff --git a/docs/current-state.md b/docs/current-state.md index defa135..741efb4 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -6,10 +6,10 @@ ## 当前快照 - 日期:2026-06-27 -- 阶段:V0 单账号 CDP 流程已验证;V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端、T-006 单元测试基座、T-101 账号 user-data-dir 工具、T-102 Chrome 启动器、T-103 登录保活与检测、T-104 PySide6 主窗口骨架、T-104b PySide6 worker 基座、T-105 Tab④ 账号管理、T-106 账号快捷方式、T-201 Excel 导入入库、T-202 Tab① 任务列表与导入按钮、T-202b Tab① 导入汇总栏、T-203 采集旧标题旧封面、T-204 回写旧字段到原 Excel、T-204b 采集完成自动回写、T-205 首次未配账号/Chrome 未启动/未登录引导保护。 -- 技术栈:Python 3.10+,自研 CDP(websocket-client + requests),SQLite(sqlite3)+ `config.json` + openpyxl + AI(服务商待定),GUI PySide6 5 Tab(已定)。 -- 生产代码:已建立 `app/` 包 + 根入口 `main.py`;`app/cdp.py` 为已验证 CDP 底座;`app/editor.py` 已封装登录状态检测、标题/封面/采集/更新按钮能力;`app/appconfig.py` 已实现 `config.json` 默认值/读写/更新、AI 参数与端口读取,以及 `config/ai_models.json` 模型清单 CRUD/过滤/打码/测试连接;`app/db.py` 已实现 SQLite schema、连接 PRAGMA、批次/账号/任务与阶段写库函数;`app/excel.py` 已实现多 Excel 输入列解析、整文件列校验、脏行统计跳过、导入批次与任务入库、别名匹配统计、旧标题/旧封面路径回写原 Excel 与另存副本;`app/config.py` 已实现账号 slug 与 user-data-dir 创建;`app/accounts.py` 已实现账号 CRUD 服务、端口默认分配、启动登录、检测登录、生成快捷方式;`app/chrome.py` 已实现 Chrome 参数拼装、启动、CDP 端口探测、PowerShell `.lnk` 快捷方式生成;`app/gui.py` 已实现 PySide6 `MainWindow`、五 Tab、顶部 Tab 栏防误点样式、① 导入采集的 Excel 导入按钮/导入汇总栏/QTableView 任务列表/未匹配筛选与略过标记/采集旧标题旧封面 worker/采集前账号就绪预检与④引导/采集完成自动回写/旧数据回写重试按钮与 worker、④ 账号管理表格/弹窗/按钮/快捷方式与状态栏;`app/workers.py` 已实现 `BaseWorker`、通用 signals、取消标记和 `QThread` 启动包装。 -- 测试:`tests/` 已建立;T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`,当前覆盖 appconfig/db/config/accounts/chrome 启动与快捷方式/editor 登录检测/excel 导入/gui ① 导入采集/gui ④ 账号管理/worker signal 与线程包装,并对尚未实现的 app.prompts 做契约占位 skip;CDP/Shopee 改动仍需测试商品手动验证。 +- 阶段:V0 单账号 CDP 流程已验证;V1 已完成 T-000 正式代码包结构、T-001 `app/editor.py` 模块化、T-002 `app/appconfig.py` 应用配置、T-003 SQLite 持久化地基、T-004 本地数据忽略规则、T-005 AI 模型清单后端、T-006 单元测试基座、T-101 账号 user-data-dir 工具、T-102 Chrome 启动器、T-103 登录保活与检测、T-104 PySide6 主窗口骨架、T-104b PySide6 worker 基座、T-105 Tab④ 账号管理、T-106 账号快捷方式、T-201 Excel 导入入库、T-202 Tab① 任务列表与导入按钮、T-202b Tab① 导入汇总栏、T-203 采集旧标题旧封面、T-204 回写旧字段到原 Excel、T-204b 采集完成自动回写、T-205 首次未配账号/Chrome 未启动/未登录引导保护、T-301 AI 生成接口。 +- 技术栈:Python 3.10+,自研 CDP(websocket-client + requests),SQLite(sqlite3)+ `config.json` + openpyxl + AI(`config/ai_models.json` 通用 HTTP,chat JSON / images_edits),GUI PySide6 5 Tab(已定)。 +- 生产代码:已建立 `app/` 包 + 根入口 `main.py`;`app/cdp.py` 为已验证 CDP 底座;`app/editor.py` 已封装登录状态检测、标题/封面/采集/更新按钮能力;`app/appconfig.py` 已实现 `config.json` 默认值/读写/更新、AI 参数与端口读取,以及 `config/ai_models.json` 模型清单 CRUD/过滤/打码/测试连接;`app/ai.py` 已实现 `gen_title()`/`gen_cover()`,按默认文本/图片模型发起通用 HTTP 调用,支持重试、错误脱敏、图片 URL/base64 解析、resolution resize 与 jpg_quality 保存;`app/db.py` 已实现 SQLite schema、连接 PRAGMA、批次/账号/任务与阶段写库函数;`app/excel.py` 已实现多 Excel 输入列解析、整文件列校验、脏行统计跳过、导入批次与任务入库、别名匹配统计、旧标题/旧封面路径回写原 Excel 与另存副本;`app/config.py` 已实现账号 slug 与 user-data-dir 创建;`app/accounts.py` 已实现账号 CRUD 服务、端口默认分配、启动登录、检测登录、生成快捷方式;`app/chrome.py` 已实现 Chrome 参数拼装、启动、CDP 端口探测、PowerShell `.lnk` 快捷方式生成;`app/gui.py` 已实现 PySide6 `MainWindow`、五 Tab、顶部 Tab 栏防误点样式、① 导入采集的 Excel 导入按钮/导入汇总栏/QTableView 任务列表/未匹配筛选与略过标记/采集旧标题旧封面 worker/采集前账号就绪预检与④引导/采集完成自动回写/旧数据回写重试按钮与 worker、④ 账号管理表格/弹窗/按钮/快捷方式与状态栏;`app/workers.py` 已实现 `BaseWorker`、通用 signals、取消标记和 `QThread` 启动包装。 +- 测试:`tests/` 已建立;T-006 后纯逻辑改动必须运行 `python -m unittest discover -s tests`,当前覆盖 appconfig/db/config/accounts/chrome 启动与快捷方式/editor 登录检测/excel 导入/ai 标题与封面 HTTP 解析/gui ① 导入采集/gui ④ 账号管理/worker signal 与线程包装,并对尚未实现的 app.prompts 做契约占位 skip;CDP/Shopee 改动仍需测试商品手动验证。 - 数据:`config.json`、`config/ai_models.json`、`cmshopee.db`、`chrome_user_data_dir/`、`images/` 已由 `.gitignore` 排除;运营填写后的 Excel 业务文件默认忽略,标准空模板 `shopee待处理任务模板.xlsx` 可提交;`app/appconfig.py` 首次读取缺失的 `config.json` 时会在本地写默认配置,`app/db.py` 调用 `init_db()` 时会在本地创建 SQLite DB。 ## 既定设计要点(文档已定) @@ -20,7 +20,7 @@ - 多账号隔离:每账号独立 user-data-dir(非 profile)。 - 账号↔任务绑定:以 Excel“别名”列为权威;未匹配略过,结束弹窗汇总。 - 执行:多账号串行、单条失败继续;③ 点击「开始更新」后弹窗确认当前筛选范围和任务数量,确认后逐条点「更新」提交线上。 -- AI:服务商待定;生成内容直接用于更新,本地留档+回写 Excel 供追溯。 +- AI:服务商/模型/Key 由 `config/ai_models.json` 配置;`app/ai.py` 支持 chat JSON 与 images_edits multipart;生成内容直接用于更新,本地留档+回写 Excel 供追溯。 - 登录:人工登录 + 程序检测,不自动登录;无 Shopee tab 时检测入口为 `https:///`(默认 `https://seller.shopee.tw/`);首次未配账号、对应账号 Chrome 未启动或未登录时,① ③ 应禁用或执行前预检提示,并引导去④。 ## 当前目录要点 @@ -37,13 +37,14 @@ | `app/accounts.py` | 已有 | T-105/T-106 产出:账号 CRUD 服务、目录创建、端口分配、启动登录、检测登录、快捷方式 | | `app/editor.py` | 已有 | T-001/T-103 产出:登录状态检测、打开商品页、读/写标题、读/下载封面、上传拖封面、更新按钮、apply_task | | `app/appconfig.py` | 已有 | T-002 产出:`config.json` 默认值、读写、更新、路径/端口/AI 参数读取;拒绝敏感字段写入 | +| `app/ai.py` | 已有 | T-301 产出:`gen_title()`/`gen_cover()`;读取默认模型;通用 HTTP 调用;失败重试;错误脱敏;封面按 resolution/jpg_quality 保存 | | `app/db.py` | 已有 | T-003 产出:batches/accounts/tasks schema;WAL/busy_timeout/foreign_keys;账号/批次/任务与 set_* 阶段写库 | | `app/config.py` | 已有 | T-101 产出:别名→稳定 slug;创建并返回绝对 user-data-dir | | `app/chrome.py` | 已有 | T-102/T-106 产出:Chrome 启动参数、`subprocess.Popen` 启动、`/json/version` 端口探测、PowerShell `.lnk` 快捷方式 | -| `tests/` | 已有 | T-006/T-201/T-202/T-202b/T-203/T-204/T-204b/T-205 产出:stdlib unittest 基座;覆盖 appconfig/db/config/accounts/chrome/editor/excel/gui/workers;prompts 模块契约占位测试 | +| `tests/` | 已有 | T-006/T-201/T-202/T-202b/T-203/T-204/T-204b/T-205/T-301 产出:stdlib unittest 基座;覆盖 appconfig/db/config/accounts/chrome/editor/excel/ai/gui/workers;prompts 模块契约占位测试 | | `app/excel.py` | 已有 | T-201/T-204 产出:多文件 Excel 输入列解析、必需列整文件拒绝、脏行逐行跳过、批次/任务入库、匹配统计;按源文件/工作表/行号回写旧标题与旧封面路径;支持原文件被占用时另存副本 | -| `shopee待处理任务模板.xlsx` | 已有,待提交 | 标准空 Excel 模板;单工作表 `待处理任务`,表头 `账号名 | 别名 | 商品id | 旧标题 | 旧封面图片路径 | 新标题 | 新封面图片路径 | 更新状态`;运营复制后填写,填写副本不提交 | -| `config.json` / `config/ai_models.json` / `cmshopee.db` / `chrome_user_data_dir/` / `images/` | 本地待建,已忽略 | 含配置、密钥、业务、登录态、图片,不提交版本库 | +| `shopee待处理任务模板.xlsx` | 已有,已提交 | 标准空 Excel 模板;单工作表 `待处理任务`,表头 `账号名 | 别名 | 商品id | 旧标题 | 旧封面图片路径 | 新标题 | 新封面图片路径 | 更新状态`;运营复制后填写,填写副本不提交 | +| `config.json` / `config/ai_models.json` / `cmshopee.db` / `chrome_user_data_dir/` / `images/` | 本地存在或按需生成,已忽略 | 含配置、密钥、业务、登录态、图片,不提交版本库 | ## 已验证能力(单账号) @@ -52,18 +53,20 @@ - 改标题:原生 setter + 派发事件,`value` 与 `modelvalue` 双等于新值。 - 换封面:`setFileInputFiles` 上传(`.shopee-image-manager__upload input[type=file]`)→ 等 CDN 链接 → 等 1 秒 → `Input.dispatchMouseEvent` 拖到第一位(落点 `first.left - 0.30*w`);使用 `1_TY030.jpg` 在测试商品验证通过。 - 「更新」按钮:可点才点,禁用态识别;③ 未批量确认前不提交。 +- Tab① 真机冒烟(2026-06-27):账号 `papa`,导入临时 Excel 5 行,采集 5 条旧标题/旧封面,下载 5 张旧封面并回写临时 Excel 成功。 ## 任务看板状态 任务状态以 [`06-tasks.md`](06-tasks.md) 为准,历史记录见 [`../progress.md`](../progress.md)。 -- 已完成:T-000(正式代码包结构)、T-001(`app/editor.py` 模块化)、T-002(`app/appconfig.py` + `config.json`)、T-003(SQLite 建表)、T-004(本地数据 gitignore)、T-005(AI 模型清单后端)、T-006(单元测试基座)、T-101(账号 slug/user-data-dir)、T-102(Chrome 启动器)、T-103(登录保活与检测)、T-104(PySide6 五 Tab 主窗口骨架)、T-104b(PySide6 worker 基座)、T-105(Tab④ 账号管理)、T-106(账号快捷方式)、T-201(Excel 导入:解析多文件输入列入库)、T-202(Tab① 任务列表 + 导入按钮 + 别名匹配标记)、T-202b(Tab① 导入汇总栏)、T-203(采集旧标题+旧封面)、T-204(回写旧字段到原 Excel)、T-204b(采集完成自动回写旧字段)、T-205(首次未配账号 / Chrome 未启动 / 未登录引导保护)。 +- 已完成:T-000(正式代码包结构)、T-001(`app/editor.py` 模块化)、T-002(`app/appconfig.py` + `config.json`)、T-003(SQLite 建表)、T-004(本地数据 gitignore)、T-005(AI 模型清单后端)、T-006(单元测试基座)、T-101(账号 slug/user-data-dir)、T-102(Chrome 启动器)、T-103(登录保活与检测)、T-104(PySide6 五 Tab 主窗口骨架)、T-104b(PySide6 worker 基座)、T-105(Tab④ 账号管理)、T-106(账号快捷方式)、T-201(Excel 导入:解析多文件输入列入库)、T-202(Tab① 任务列表 + 导入按钮 + 别名匹配标记)、T-202b(Tab① 导入汇总栏)、T-203(采集旧标题+旧封面)、T-204(回写旧字段到原 Excel)、T-204b(采集完成自动回写旧字段)、T-205(首次未配账号 / Chrome 未启动 / 未登录引导保护)、T-301(AI 生成接口)。 - 正在进行:无。 -- 下一个可领取任务:**T-301(确定 AI 服务商/模型并接入 `app/ai.py`)**。 +- 下一个可领取任务:**T-302(Tab② 左右布局:左提示词,右筛选 + 任务列表)**。 ## 当前已知限制 - ① 采集依赖对应账号 Chrome 已用专属 user-data-dir 和 CDP 端口启动并登录;T-205 已在采集前拦截未配置账号、Chrome 未启动、未登录,并引导去④账号管理,但不会无提示批量启动所有账号 Chrome。 +- T-301 已完成通用 HTTP AI 接口和 mock 单测;真实 AI 生成还需要在 `config/ai_models.json` 填入可用 url/model/api_key 后做一次成本可控的实测。 ## 当前可运行内容 @@ -83,6 +86,9 @@ git check-ignore -v -- config.json config/ai_models.json cmshopee.db chrome_user # ai_models 临时清单读写与打码检查 py -3 -c "import os,tempfile; from app import appconfig; d=tempfile.TemporaryDirectory(dir='.'); p=os.path.join(d.name,'ai_models.json'); print([m['category'] for m in appconfig.list_ai_models(path=p)])" +# AI 接口单测(不连真实服务) +python -m unittest discover -s tests -p "test_ai.py" + # config slug 与 user-data-dir 临时目录检查 py -3 -c "import tempfile; from app import config; d=tempfile.TemporaryDirectory(dir='.'); print(config.ensure_user_data_dir(config.make_slug('alias'), root=d.name))" diff --git a/progress.md b/progress.md index aeec422..e68c3b6 100644 --- a/progress.md +++ b/progress.md @@ -465,3 +465,19 @@ - 状态:DONE - 变更:修正评审报告中的 GUI 用例数量为 18;确认 T-205「预检拦截 + 弹窗说明 + 自动跳④」为最终方案,不再作为待决事项;同步 `docs/api.md`、`docs/06-tasks.md`、`docs/current-state.md`,明确匹配账号未登录由 T-205 预检 `blocked=True` 阻断,不进入逐条采集、不写 skipped/failed,别名未匹配仍按 T-203 `mark_skipped`。 - 测试补强:`tests/test_gui.py` 为未登录预检分支增加断言,确认任务保持 `stage=imported/status=pending/last_error=None`,防止后续误写 skipped。 + +## 【2026-06-27】冒烟 · Tab① 真实采集回写 + +- 状态:DONE +- 前置:账号 `papa` 使用专属 Chrome user-data-dir 启动,CDP 端口 `9224` 就绪,`accounts.detect_login()` 返回已登录。 +- 验证:因原业务 Excel 正被 Excel 锁定,改用数据库现有 5 条任务生成 gitignore 的临时 Excel;执行 `excel.import_tasks()` 导入 5 行,`CollectWorker` 真连 Shopee/CDP 采集 5 条旧标题/旧封面,下载 5 张旧封面到 `images/`,随后 `excel.write_back()` 成功回写临时 Excel。 +- 结果:`collect_summary {'ok': True, 'total': 5, 'done': 5, 'collected': 5, 'skipped': 0, 'failed': 0}`;旧标题长度非空,旧封面路径均存在。 + +## 【2026-06-27】T-301 AI 生成接口 + +- 状态:DONE +- 变更:新增 `app/ai.py`,实现 `AIError`、`gen_title()`、`gen_cover()`;按 `config.json` 的默认文本/图片模型名读取 `config/ai_models.json` 明文模型定义,调用时不记录 Key;支持 retry、模型启用/字段校验、错误脱敏、OpenAI-compatible chat JSON、images_edits multipart、图片 URL/data URL/base64/b64_json 解析、按 `resolution` resize 并用 `jpg_quality` 保存 JPEG。 +- 文档:`docs/06-tasks.md` 将 T-301 标为 DONE;同步 `docs/api.md`、`docs/03-tech-stack.md`、`docs/04-architecture.md`、`docs/current-state.md`,下一个可领取任务更新为 T-302。 +- 测试:新增 `tests/test_ai.py`,覆盖文本生成重试与 extra_body、缺字段错误不泄露 Key、图片生成保存 JPEG。 +- 验证:`python -m unittest discover -s tests -p "test_ai.py"` 通过(3 tests);`python -m compileall app main.py tests` 通过;`python -m unittest discover -s tests` 通过(63 tests,skipped=1);`py -3 -m compileall app main.py tests` 通过;`py -3 -m unittest discover -s tests -p "test_ai.py"` 通过(3 tests,skipped=1,py -3 环境缺 Pillow);`py -3 -m unittest discover -s tests` 通过(37 tests,skipped=5,py -3 环境缺 openpyxl/PySide6/Pillow,相关测试按设计跳过)。 +- 注意:T-301 只做通用 AI 接口与 mock 单测;真实 AI 调用还需要在 `config/ai_models.json` 填入可用 url/model/api_key 后做一次小样本实测。生成批次编排和 GUI 接入留给 T-303/T-302。 diff --git a/tests/test_ai.py b/tests/test_ai.py new file mode 100644 index 0000000..23e4c40 --- /dev/null +++ b/tests/test_ai.py @@ -0,0 +1,164 @@ +import base64 +import io +import json +import os +import sys +import unittest +from unittest import mock + +sys.path.insert(0, os.path.dirname(__file__)) + +from _helpers import TempDirMixin + +from app import ai, appconfig + + +class _Response: + def __init__(self, payload): + self.payload = payload + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def read(self, size=-1): + return json.dumps(self.payload).encode("utf-8") + + +class AITests(TempDirMixin, unittest.TestCase): + def _write_models(self, path, text=None, image=None): + text = text or { + "name": "Text", + "category": "text", + "enabled": True, + "url": "https://example.invalid/v1/chat/completions", + "model": "text-model", + "api_key": "sk-text-secret", + "api_type": "chat", + "connect_timeout_seconds": 1, + "timeout_seconds": 1, + "extra_body": {"temperature": 0}, + } + image = image or { + "name": "Image", + "category": "image", + "enabled": True, + "url": "https://example.invalid/v1/chat/completions", + "model": "image-model", + "api_key": "sk-image-secret", + "api_type": "auto", + "connect_timeout_seconds": 1, + "timeout_seconds": 1, + "extra_body": {}, + } + appconfig.save_ai_models_config({"models": [text, image]}, path=path) + + def _config(self): + cfg = appconfig.default_config() + cfg["ai"]["default_text_model"] = "Text" + cfg["ai"]["default_image_model"] = "Image" + cfg["ai"]["retry"] = 1 + cfg["ai"]["resolution"] = "512" + cfg["ai"]["jpg_quality"] = 80 + return cfg + + def test_gen_title_uses_configured_model_and_retries(self): + with self.make_temp_dir() as temp_dir: + models_path = os.path.join(temp_dir, "ai_models.json") + self._write_models(models_path) + calls = [] + + def fake_urlopen(request, timeout=None): + calls.append((request, timeout)) + if len(calls) == 1: + raise ai.urllib.error.URLError("temporary") + return _Response({"choices": [{"message": {"content": " 新标题 "}}]}) + + with mock.patch("app.ai.urllib.request.urlopen", side_effect=fake_urlopen): + title = ai.gen_title( + "优化标题", + "旧标题", + config=self._config(), + models_path=models_path, + ) + + self.assertEqual("新标题", title) + self.assertEqual(2, len(calls)) + body = json.loads(calls[-1][0].data.decode("utf-8")) + self.assertEqual("text-model", body["model"]) + self.assertEqual(0, body["temperature"]) + self.assertNotIn("sk-text-secret", body["messages"][1]["content"]) + + self.assert_removed(temp_dir) + + def test_missing_model_fields_raise_clear_error_without_secret(self): + with self.make_temp_dir() as temp_dir: + models_path = os.path.join(temp_dir, "ai_models.json") + self._write_models( + models_path, + text={ + "name": "Text", + "category": "text", + "enabled": True, + "url": "", + "model": "text-model", + "api_key": "sk-text-secret", + "api_type": "chat", + "connect_timeout_seconds": 1, + "timeout_seconds": 1, + "extra_body": {}, + }, + ) + + with self.assertRaises(ai.AIError) as raised: + ai.gen_title("prompt", "old", config=self._config(), models_path=models_path) + + message = str(raised.exception) + self.assertIn("缺少字段: url", message) + self.assertNotIn("sk-text-secret", message) + + self.assert_removed(temp_dir) + + def test_gen_cover_saves_jpeg_with_resolution_and_quality(self): + try: + from PIL import Image + except ImportError: + self.skipTest("Pillow not installed") + + with self.make_temp_dir() as temp_dir: + models_path = os.path.join(temp_dir, "ai_models.json") + self._write_models(models_path) + old_cover = os.path.join(temp_dir, "old.jpg") + output = os.path.join(temp_dir, "new.jpg") + Image.new("RGB", (16, 16), (20, 30, 40)).save(old_cover, "JPEG") + generated = io.BytesIO() + Image.new("RGB", (8, 8), (200, 120, 80)).save(generated, "PNG") + b64_image = base64.b64encode(generated.getvalue()).decode("ascii") + + def fake_urlopen(request, timeout=None): + body = json.loads(request.data.decode("utf-8")) + self.assertEqual("image-model", body["model"]) + self.assertIn("目标分辨率:512", body["messages"][0]["content"][0]["text"]) + return _Response({"data": [{"b64_json": b64_image}]}) + + with mock.patch("app.ai.urllib.request.urlopen", side_effect=fake_urlopen): + result = ai.gen_cover( + "生成封面", + old_cover, + output, + config=self._config(), + models_path=models_path, + ) + + self.assertEqual(os.path.abspath(output), result) + with Image.open(output) as saved: + self.assertEqual((512, 512), saved.size) + self.assertEqual("JPEG", saved.format) + + self.assert_removed(temp_dir) + + +if __name__ == "__main__": + unittest.main()