#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ inquiry_local_agent.py —— 设备询价「本机代理」 ===================================================================== 运行在你自己的电脑上（Windows / macOS / Linux 均可），用本机干净宽带 IP （或住宅/动态代理）自动按设备名上网询价、截图，把结果回传网页端。 为什么放本机而不是服务器？ —— 云服务器(49.235.177.123)的机房 IP 已被政府采购网等站点反爬拦截； 本机家庭/企业宽带 IP 通常可正常访问，因此「实时询价」在本地跑最稳。 合规提醒： —— 抓取到的价格仅作「市场参考价」，不等于中标成交价，生成对比表时会标注； —— 请遵守目标站点 robots 与速率限制，仅用于内部询价参考，勿高频刷接口； —— 如需更高成功率可设置住宅/动态代理：export PROCURE_SCRAPE_PROXY=http://user:pass@host:port 接口： GET http://127.0.0.1:8765/ 健康检查 POST http://127.0.0.1:8765/inquiry 入参 {list:[{seq,name,spec}]} 返回 {quotes:{ "": {q1:{price,shot,source,param}, q2:{...}, q3:{...}} }} 运行： pip install playwright playwright install chromium python inquiry_local_agent.py # 默认端口 8765 PROCURE_SCRAPE_PROXY=http://... python inquiry_local_agent.py """ import base64 import hashlib import json import os import re import sys import threading import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer PORT = int(os.environ.get("INQUIRY_AGENT_PORT", "8765")) PROXY = os.environ.get("PROCURE_SCRAPE_PROXY", "") UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36") PRICE_RE = re.compile(r'([¥￥]\s?[\d,]+(?:\.\d+)?\s*(?:万?元|万元|w|W)?)|' r'([\d,]+(?:\.\d+)?\s*(?:万?元|万元))', re.I) SEARCH_ENGINES = [ ("bing", "https://cn.bing.com/search?q={q}"), ("baidu", "https://www.baidu.com/s?wd={q}"), ] def launch_browser(pw): args = ["--no-sandbox", "--disable-dev-shm-usage"] kwargs = dict(headless=True, args=args) if PROXY: kwargs["proxy"] = {"server": PROXY} return pw.chromium.launch(**kwargs) def extract_price(text): if not text: return "" for m in PRICE_RE.finditer(text): val = (m.group(1) or m.group(2) or "").strip() if val: return val return "" def first_result(page, engine): """取搜索结果第一条的链接与摘要。""" if engine == "bing": try: a = page.locator("li.b_algo h2 a").first href = a.get_attribute("href") snippet = page.locator("li.b_algo p").first.inner_text() return href, snippet except Exception: return None, "" else: # baidu try: a = page.locator("div.result h3 a, .c-container h3 a").first href = a.get_attribute("href") snippet = page.locator("div.result .c-abstract, .c-container .c-abstract").first.inner_text() return href, snippet except Exception: return None, "" def inquire_one(page, name, intent): """对一个设备做一次搜索意图询价，返回 (price, shot_b64, source, param)。""" q = f"{name} {intent}" last_err = "" for engine, tpl in SEARCH_ENGINES: try: url = tpl.format(q=urllib.parse.quote(q)) page.goto(url, wait_until="domcontentloaded", timeout=30000) page.wait_for_timeout(1200) href, snippet = first_result(page, engine) if not href: continue page.goto(href, wait_until="domcontentloaded", timeout=30000) page.wait_for_timeout(1500) shot = page.screenshot(full_page=False) text = page.evaluate("() => document.body.innerText || ''") price = extract_price(text) or extract_price(snippet) param = ("网络参考价（非成交价） | " + (snippet or ""))[:120] return price, base64.b64encode(shot).decode("utf-8"), href, param except Exception as e: last_err = str(e) continue return "", "", "", "未取到（" + last_err[:60] + "）" def run_inquiry(payload): from playwright.sync_api import sync_playwright items = payload.get("list") or [] quotes = {} with sync_playwright() as pw: browser = launch_browser(pw) ctx = browser.new_context(user_agent=UA, locale="zh-CN", viewport={"width": 1280, "height": 900}) page = ctx.new_page() intents = ["报价 价格", "询价 采购", "中标 公告"] for it in items: seq = str(it.get("seq") if it.get("seq") is not None else it.get("序号", "")) name = (it.get("name") or it.get("设备名称") or "").strip() if not name: continue print(f"[agent] 询价: {name} (seq={seq})", flush=True) q = {} for i, intent in enumerate(intents, start=1): price, shot, src, param = inquire_one(page, name, intent) q[f"q{i}"] = {"price": price, "shot": shot, "source": src, "param": param} if shot: break # 取到截图即止，避免重复 quotes[seq] = q browser.close() return {"quotes": quotes} class Handler(BaseHTTPRequestHandler): def _cors(self): self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") def do_OPTIONS(self): self.send_response(204) self._cors() self.end_headers() def do_GET(self): self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self._cors() self.end_headers() self.wfile.write(json.dumps({"ok": True, "service": "inquiry-local-agent", "proxy": bool(PROXY)}).encode("utf-8")) def do_POST(self): if self.path.rstrip("/") != "/inquiry": self._json(404, {"error": "unknown path"}) return try: length = int(self.headers.get("Content-Length", 0)) raw = self.rfile.read(length) if length else b"{}" payload = json.loads(raw.decode("utf-8") or "{}") except Exception as e: self._json(400, {"error": "bad json: " + str(e)}) return try: result = run_inquiry(payload) self._json(200, result) except Exception as e: self._json(500, {"error": "inquiry failed: " + str(e)}) def _json(self, code, obj): body = json.dumps(obj, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self._cors() self.end_headers() self.wfile.write(body) def log_message(self, *args): pass # 静默访问日志 def main(): server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler) mode = f"代理 {PROXY}" if PROXY else "本机宽带 IP" print(f"[agent] 设备询价本机代理已启动: http://127.0.0.1:{PORT}/inquiry (使用{mode})", flush=True) print("[agent] 按 Ctrl+C 停止。", flush=True) try: server.serve_forever() except KeyboardInterrupt: print("\n[agent] 已停止。", flush=True) finally: server.server_close() if __name__ == "__main__": main()