#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 报表/文档下载器 (独立后台进程)。 从 SQLite 选出 35 个报表/文档类页面, 逐个: 随机 5-15s 延时 -> 导航 -> 等待加载 -> 隐藏浮层 -> save_as_pdf 按结构归档 -> 若存在"明确的导出按钮且点击后未弹出需选参数的弹窗", 顺带把导出的 Excel 归集 所有结果写入 SQLite downloads 表; 进度写 db/dl_progress.log。支持断点续跑(PDF已存在则跳过)。 用法: python3 scripts/dl_reports.py """ import os, sys, json, time, random, shutil, sqlite3, urllib.request, glob D = "http://127.0.0.1:10086/command" S = "2haohr" BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DB = os.path.join(BASE, "db", "research.db") LOG = os.path.join(BASE, "db", "dl_progress.log") PDF_TMP = "/tmp/kimi-webbridge-pdfs" DOWNLOADS = os.path.expanduser("~/Downloads") KW = ['报表', '分析', '仪表盘', '统计', '档案', '模板', '文档', '制度', '公告', '证明'] EXPORT_TEXTS = ('导出', '导出Excel', '导出报表', '下载', '批量导出', '导出全部') def call(action, args, timeout=90): body = json.dumps({"action": action, "args": args, "session": S}).encode() req = urllib.request.Request(D, data=body, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode()) def ev(code): return call("evaluate", {"code": code}).get("data", {}).get("value") def log(msg): line = time.strftime("%H:%M:%S ") + msg print(line, flush=True) with open(LOG, "a") as f: f.write(line + "\n") def reconnect(): r = call("find_tab", {"url": "2haohr.com", "active": True}) return r.get("ok") def wait_ready(max_wait=10): t0 = time.time() while time.time() - t0 < max_wait: try: if ev("document.readyState") == "complete": time.sleep(1.5) return True except Exception: pass time.sleep(0.6) return False HIDE_JS = r"""(()=>{const sels=['[class*="helpBox"]','[class*="aiAssistant"]','[class*="product_"]','[class*="feedback"]','[class*="Feedback"]','[class*="floatBar"]','[class*="robot"]','[class*="advert"]'];let n=0;sels.forEach(s=>document.querySelectorAll(s).forEach(el=>{const r=el.getBoundingClientRect();if(r.width<600&&getComputedStyle(el).position==='fixed'){el.style.setProperty('display','none','important');n++;}}));return n;})()""" # 找一个"明确的"导出按钮(可见、文案精确、非下拉触发器) FIND_EXPORT_JS = r"""(()=>{const texts=%s;const els=[...document.querySelectorAll('button,a,[role=button]')];for(const el of els){const t=(el.innerText||'').trim();if(t.length>0&&t.length<=8&&texts.includes(t)){const r=el.getBoundingClientRect();if(r.width>0&&r.height>0&&getComputedStyle(el).display!=='none'){el.setAttribute('data-dlmark','1');return t;}}}return '';})()""" % json.dumps(list(EXPORT_TEXTS)) # 点击已标记的导出按钮 CLICK_EXPORT_JS = r"""(()=>{const el=document.querySelector('[data-dlmark="1"]');if(!el)return false;el.click();el.removeAttribute('data-dlmark');return true;})()""" # 检测是否弹出了需要选参数的弹窗(modal/dialog) MODAL_JS = r"""(()=>{const ms=[...document.querySelectorAll('[class*="modal"],[class*="Modal"],[role="dialog"],[class*="dialog"],[class*="Dialog"]')];for(const m of ms){const r=m.getBoundingClientRect();if(r.width>200&&r.height>120&&getComputedStyle(m).display!=='none'&&getComputedStyle(m).visibility!=='hidden')return true;}return false;})()""" CLOSE_MODAL_JS = r"""document.activeElement&&document.activeElement.dispatchEvent(new KeyboardEvent('keydown',{key:'Escape',bubbles:true}));document.dispatchEvent(new KeyboardEvent('keydown',{key:'Escape',bubbles:true}));""" def snapshot_downloads(): """当前 ~/Downloads 下的文件集合(忽略 .crdownload 临时文件)""" return set(p for p in glob.glob(os.path.join(DOWNLOADS, "*")) if not p.endswith(".crdownload")) def wait_new_download(before, max_wait=12): """等待新下载文件出现且完成(无 .crdownload)。返回新文件路径或 None""" t0 = time.time() while time.time() - t0 < max_wait: time.sleep(1.0) # 仍有下载中的临时文件则继续等 if glob.glob(os.path.join(DOWNLOADS, "*.crdownload")): continue now = snapshot_downloads() new = now - before if new: # 取最新的一个 return max(new, key=lambda p: os.path.getmtime(p)) return None def rec(conn, node_key, full_path, url, kind, file_path, status, size, note): conn.execute( "INSERT INTO downloads(node_key,full_path,url,kind,file_path,status,size,note,ts) " "VALUES(?,?,?,?,?,?,?,?,?)", (node_key, full_path, url, kind, file_path, status, size, note, time.strftime("%Y-%m-%d %H:%M:%S"))) conn.commit() def main(): conn = sqlite3.connect(DB) rows = conn.execute( "SELECT node_key,full_path,url,screenshot_path FROM features WHERE level=2").fetchall() targets = [r for r in rows if any(k in (r[1] or "") for k in KW)] targets.sort(key=lambda r: r[3] or "") total = len(targets) log(f"=== 报表/文档下载开始, 目标 {total} 页 ===") if not reconnect(): log("!! 无法连接浏览器标签, 退出"); return okp = okx = fail = skip = 0 for i, (nk, fp, url, sp) in enumerate(targets): rel_pdf = (sp or "").replace("screenshots/", "downloads/").rsplit(".", 1)[0] + ".pdf" out_pdf = os.path.join(BASE, rel_pdf) if os.path.exists(out_pdf) and os.path.getsize(out_pdf) > 5000: skip += 1 continue os.makedirs(os.path.dirname(out_pdf), exist_ok=True) delay = random.uniform(5, 15) log(f"[{i+1}/{total}] wait {delay:.1f}s -> {fp}") time.sleep(delay) try: call("navigate", {"url": url, "newTab": False}) wait_ready() time.sleep(random.uniform(1.0, 2.5)) try: ev(HIDE_JS) except Exception: pass # --- 存 PDF --- fname = os.path.basename(rel_pdf)[:-4] r = call("save_as_pdf", {"paper_format": "a4", "print_background": True, "file_name": fname}, timeout=120) tmp = r.get("data", {}).get("path") size = r.get("data", {}).get("sizeBytes", 0) if tmp and os.path.exists(tmp): shutil.move(tmp, out_pdf) okp += 1 rec(conn, nk, fp, url, "pdf", rel_pdf, "ok", size, "") log(f" PDF ok {size//1024}KB -> {rel_pdf}") else: fail += 1 rec(conn, nk, fp, url, "pdf", rel_pdf, "fail", 0, json.dumps(r)[:150]) log(f" PDF FAIL -> {json.dumps(r)[:150]}") # --- 尝试导出 Excel(仅当有明确按钮且不弹参数窗) --- try: btn = ev(FIND_EXPORT_JS) if btn: before = snapshot_downloads() ev(CLICK_EXPORT_JS) time.sleep(2.0) if ev(MODAL_JS): ev(CLOSE_MODAL_JS) rec(conn, nk, fp, url, "xlsx", "", "skip", 0, f"按钮[{btn}]触发弹窗,跳过") log(f" Excel skip (弹窗需选参数, 按钮={btn})") else: newf = wait_new_download(before) if newf: ext = os.path.splitext(newf)[1] or ".xlsx" dst = out_pdf[:-4] + ext shutil.move(newf, dst) okx += 1 rel_x = os.path.relpath(dst, BASE) rec(conn, nk, fp, url, "xlsx", rel_x, "ok", os.path.getsize(dst), f"按钮={btn}") log(f" Excel ok -> {rel_x}") else: rec(conn, nk, fp, url, "xlsx", "", "skip", 0, f"按钮[{btn}]点击无文件") log(f" Excel none (点击无下载, 按钮={btn})") except Exception as e: log(f" Excel err {str(e)[:120]}") except Exception as e: fail += 1 rec(conn, nk, fp, url, "pdf", rel_pdf, "fail", 0, str(e)[:150]) log(f" FAIL {fp} err={str(e)[:150]}") try: reconnect() except Exception: pass log(f"=== DONE pdf_ok={okp} xlsx_ok={okx} skip={skip} fail={fail} total={total} ===") conn.close() if __name__ == "__main__": main()