# -*- coding: utf-8 -*- """ 京东物流 · 本地理赔单自动填充脚本（在本机运行） ============================================== 本脚本由「售后工作台」下载到你的电脑上运行，**复用你本机浏览器里已经登录的 京东物流商家工作台会话**，无需导出 Cookie。 它会： 1. 从售后工作台（云端）拉取某个理赔单的材料与三类照片； 2. 用 Playwright 打开京东「新建理赔单」页面并填表、分类上传照片； 3. **默认停在提交按钮前**，由你人工核对后点击「提交」。 字段映射（来自售后工作台理赔单）： 单票号 <- order_no 问题类型 <- problem_type（下拉选择） 联系人 <- contact 联系电话 <- contact_phone 可接听电话时间 <- call_time 商品名称 <- product_name 品牌型号 <- brand_model 丢损数量 <- lost_qty 丢损金额 <- lost_amount 货物总价值 <- goods_value 问题描述 <- problem_desc 面单照片 <- img_waybill（上传区 1） 包装照片 <- img_package（上传区 2） 物品照片 <- img_item（上传区 3） ---------------------------------------------------------------------- 使用方法（首次）： 1) 安装依赖（只需一次）： pip install playwright playwright install chromium 2) 把本脚本放到任意目录，双击或在命令行运行： python jd_autofill_local.py --server http://你的服务器地址:8001 --ticket-id 12 注意： - 运行前请**关闭本机 Chrome / Edge**（否则浏览器用户目录被占用，脚本无法复用登录态）。 - 脚本会自动探测本机 Chrome 用户目录；若你用 Edge 或目录不同，加参数： --browser edge --profile "C:\\Users\\你\\AppData\\Local\\Google\\Chrome\\User Data" - 京东页面结构可能随版本变化，下方 SEL 选择器需按实际页面用 DevTools 核对调整。 - 默认 headless=False（有窗口，便于你提交前人工核对）；加 --headless 可无窗口运行。 参数： --server 售后工作台地址，如 http://139.155.96.56:8001 --ticket-id 要填充的理赔单 ID（工作台详情页标题上的 #编号） --browser chrome（默认）或 edge --profile 浏览器用户数据目录（不填则自动探测） --headless 无窗口运行（服务器/后台用） --submit 填充后自动点击提交（跳过人工确认，慎用） """ import argparse import json import os import sys import tempfile import urllib.request import urllib.parse def _fix_win_console(): """Windows 控制台默认 GBK，脚本按 UTF-8 输出会乱码 → 切到 65001 并重置流编码""" if sys.platform != "win32": return try: import ctypes ctypes.windll.kernel32.SetConsoleOutputCP(65001) ctypes.windll.kernel32.SetConsoleCP(65001) except Exception: pass for stream in ("stdout", "stderr"): try: getattr(sys, stream).reconfigure(encoding="utf-8") except Exception: pass _fix_win_console() # 京东物流商家工作台「新建理赔单」页面 CLAIM_URL = os.environ.get( "JD_CLAIM_URL", "https://ebusiness.jdl.com/claimcenter/newClaim" ) # ---- 京东页面选择器（⚠️ 需按实际页面用 DevTools 核对调整）---- # 通用策略：优先按 label/占位文字定位，其次按 id/name。 SEL = { "order_no": "#waybillNo, input[name*='waybill'], input[placeholder*='单票'], input[placeholder*='运单']", "problem_type": "select[name*='problemType'], select[name*='type'], #problemType", "contact": "input[name*='contactName'], input[placeholder*='联系人']", "contact_phone":"input[name*='contactPhone'], input[placeholder*='电话']", "call_time": "input[name*='callTime'], input[placeholder*='接听']", "product_name": "input[name*='goodsName'], input[placeholder*='商品名称']", "brand_model": "input[name*='brand'], input[placeholder*='品牌']", "lost_qty": "input[name*='lostNum'], input[placeholder*='丢失数量'], input[placeholder*='丢损数量']", "lost_amount": "input[name*='lostAmount'], input[placeholder*='丢损金额']", "goods_value": "input[name*='goodsValue'], input[placeholder*='货物总价值']", "problem_desc": "textarea[name*='desc'], textarea[placeholder*='描述']", # 三类照片上传：按出现顺序的三个 file input（面单/包装/物品） "uploads": "input[type=file]", "submit": "button:has-text('提交'), button[type=submit]", } # 命令行 --browser 取值 → Playwright channel 名（注意 Edge 是 msedge） CHANNELS = {"chrome": "chrome", "edge": "msedge", "msedge": "msedge", "chromium": "chromium"} def _known_browser_paths(channel): """常见安装路径兜底：channel 找不到时直接用 executable_path 启动""" if channel == "msedge": paths = [ r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe", r"C:\Program Files\Microsoft\Edge\Application\msedge.exe", os.path.expandvars(r"%LOCALAPPDATA%\Microsoft\Edge\Application\msedge.exe"), ] else: paths = [ r"C:\Program Files\Google\Chrome\Application\chrome.exe", r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe", os.path.expandvars(r"%LOCALAPPDATA%\Google\Chrome\Application\chrome.exe"), ] return [p for p in paths if p and os.path.exists(p)] def _win_chrome_profile(): base = os.path.expandvars(r"%LOCALAPPDATA%") return os.path.join(base, "Google", "Chrome", "User Data") def _win_edge_profile(): base = os.path.expandvars(r"%LOCALAPPDATA%") return os.path.join(base, "Microsoft", "Edge", "User Data") def detect_profile(browser): if browser == "edge": p = _win_edge_profile() else: p = _win_chrome_profile() return p if os.path.isdir(p) else None def fetch_ticket(server, ticket_id): url = "%s/api/tickets/%s" % (server.rstrip("/"), ticket_id) try: req = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(req, timeout=20) as resp: data = json.loads(resp.read().decode("utf-8")) if "error" in data: raise SystemExit("拉取理赔单失败：" + data["error"]) return data except SystemExit: raise except Exception as e: raise SystemExit("无法连接售后工作台（%s）：%s" % (url, e)) def download_images(server, ticket, tmpdir): """把三类照片从云端下载到本地临时目录，返回 cat->本地路径列表""" result = {} server = server.rstrip("/") for cat in ("img_waybill", "img_package", "img_item"): result[cat] = [] for rel in (ticket.get(cat) or []): if not rel: continue img_url = rel if rel.startswith("http") else server + rel try: ext = os.path.splitext(urllib.parse.urlparse(img_url).path)[1] or ".png" dest = os.path.join(tmpdir, "%s_%d%s" % (cat, len(result[cat]), ext)) urllib.request.urlretrieve(img_url, dest) result[cat].append(dest) except Exception as e: print(" ⚠️ 下载照片失败 %s: %s" % (img_url, e)) return result def field_text(t, key): v = t.get(key) return "" if v is None else str(v) def _run(server, t, browser, profile, headless, do_submit, explicit_profile=False): try: from playwright.sync_api import sync_playwright except ImportError: print("缺少 playwright，请先安装：pip install playwright && playwright install chromium") return 2 # 下载照片到临时目录 tmpdir = tempfile.mkdtemp(prefix="jdclaim_") local_imgs = download_images(server, t, tmpdir) total = sum(len(v) for v in local_imgs.values()) print("已下载照片 %d 张到临时目录：%s" % (total, tmpdir)) with sync_playwright() as pw: # 复用本机已登录的浏览器用户目录（无需 Cookie 文件） preferred = CHANNELS.get((browser or "chrome").lower(), "chrome") alt = "msedge" if preferred == "chrome" else "chrome" tmp_profile = None if profile: print("复用浏览器用户目录：%s" % profile) else: # launch_persistent_context 的 user_data_dir 是必填项，未探测到时退回临时目录 tmp_profile = tempfile.mkdtemp(prefix="jdprofile_") print("⚠️ 未探测到本机 %s 用户目录，已使用临时目录：%s" % (browser or "chrome", tmp_profile)) print(" 该目录没有京东登录态，请先在弹出的窗口里登录一次；") print(" 或用参数指定目录，例如：--profile \"C:\\\\Users\\\\你\\\\AppData\\\\Local\\\\Google\\\\Chrome\\\\User Data\"") # 依次尝试：首选浏览器 channel → 首选的已知安装路径 → 备用浏览器 → 备用的已知安装路径 main_udd = profile or tmp_profile attempts = [({"channel": preferred}, main_udd)] for p in _known_browser_paths(preferred): attempts.append(({"executable_path": p}, main_udd)) if not explicit_profile: # 两个浏览器的用户目录不通用，换浏览器时改用临时目录 alt_udd = tempfile.mkdtemp(prefix="jdprofile_") attempts.append(({"channel": alt}, alt_udd)) for p in _known_browser_paths(alt): attempts.append(({"executable_path": p}, alt_udd)) ctx, last_err = None, None for extra, udd in attempts: label = extra.get("channel") or extra.get("executable_path") try: ctx = pw.chromium.launch_persistent_context( user_data_dir=udd, headless=headless, args=["--disable-blink-features=AutomationControlled"], **extra) print("已启动浏览器：%s" % label) break except Exception as e: last_err = e if (extra, udd) != attempts[-1]: print(" ⚠️ 用 %s 启动失败，换下一个…" % label) if ctx is None: print("\n❌ 浏览器启动失败：%s" % last_err) print(" 常见原因：1) 本机未安装 Chrome / Edge；2) 浏览器正在运行占用了用户目录（请先关闭）；") print(" 3) --profile 路径写错。可用 --browser edge 或指定 --profile 重试。") return 2 page = ctx.new_page() page.goto(CLAIM_URL, wait_until="networkidle", timeout=60000) def fill(sel, val): if not val: return try: el = page.locator(sel).first el.wait_for(state="visible", timeout=10000) tag = el.first_element_handle().evaluate("e => e.tagName") if tag == "SELECT": el.select_option(label=val) else: el.fill(val) print(" 填充成功:", sel, "=", val) except Exception as e: print(" ⚠️ 填充失败:", sel, "->", e) fill(SEL["order_no"], field_text(t, "order_no")) fill(SEL["problem_type"], field_text(t, "problem_type")) fill(SEL["contact"], field_text(t, "contact")) fill(SEL["contact_phone"], field_text(t, "contact_phone")) fill(SEL["call_time"], field_text(t, "call_time")) fill(SEL["product_name"], field_text(t, "product_name")) fill(SEL["brand_model"], field_text(t, "brand_model")) fill(SEL["lost_qty"], field_text(t, "lost_qty")) fill(SEL["lost_amount"], field_text(t, "lost_amount")) fill(SEL["goods_value"], field_text(t, "goods_value")) fill(SEL["problem_desc"], field_text(t, "problem_desc")) # 三类照片上传 file_inputs = page.locator(SEL["uploads"]) n = file_inputs.count() cats = [("img_waybill", "面单"), ("img_package", "包装"), ("img_item", "物品")] for i, (cat, label) in enumerate(cats): paths = local_imgs.get(cat) or [] if not paths: continue if i < n: try: file_inputs.nth(i).set_input_files(paths) print(" 上传%s照片 %d 张" % (label, len(paths))) except Exception as e: print(" ⚠️ 上传%s照片失败: %s" % (label, e)) else: print(" ⚠️ 第 %d 个上传控件不存在，无法上传%s照片" % (i + 1, label)) shot = os.path.join(tmpdir, "jd_fill_%s.png" % t.get("id", "x")) try: page.screenshot(path=shot) print(" 已截图:", shot) except Exception: pass if do_submit: try: page.locator(SEL["submit"]).first.click() print(" 已点击提交") except Exception as e: print(" ⚠️ 提交失败:", e) else: print(" ✅ 已填充完成，停在提交前——请人工核对后点击「提交」。") if not headless: input(" 按回车退出（退出后浏览器关闭，临时照片已清理）...") ctx.close() # 退出时清理临时照片 try: for f in os.listdir(tmpdir): os.remove(os.path.join(tmpdir, f)) os.rmdir(tmpdir) except Exception: pass return 0 def main(): ap = argparse.ArgumentParser(description="京东理赔单本地自动填充（复用本机登录态）") ap.add_argument("--server", default="http://localhost:8001", help="售后工作台地址，如 http://139.155.96.56:8001") ap.add_argument("--ticket-id", type=int, required=True, help="理赔单 ID（详情页 #编号）") ap.add_argument("--browser", default="chrome", help="chrome（默认）或 edge") ap.add_argument("--profile", default=None, help="浏览器用户数据目录（不填自动探测）") ap.add_argument("--headless", action="store_true", help="无窗口运行") ap.add_argument("--submit", action="store_true", help="填充后自动点击提交（慎用）") args = ap.parse_args() profile = args.profile or detect_profile(args.browser) t = fetch_ticket(args.server, args.ticket_id) print("=== 理赔单 #%s | 单号 %s | 问题类型 %s ===" % ( t.get("id"), field_text(t, "order_no"), field_text(t, "problem_type") or "-")) return _run(args.server, t, args.browser, profile, args.headless, args.submit, explicit_profile=bool(args.profile)) if __name__ == "__main__": sys.exit(main())