from flask import Flask, jsonify, render_template, request, send_from_directory from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError from pathlib import Path import base64 import json import os import subprocess import threading import time BASE_DIR = Path.cwd().resolve() STATIC_DIR = BASE_DIR / "static" TEMPLATE_DIR = BASE_DIR / "templates" DATA_DIR = BASE_DIR / "data" STATIC_DIR.mkdir(parents=True, exist_ok=True) TEMPLATE_DIR.mkdir(parents=True, exist_ok=True) DATA_DIR.mkdir(parents=True, exist_ok=True) TARGET_PAGE = "https://ynsdfz.aliwork.com/APP_DKQG7TSUK6ZNU3RDFGVO/workbench" TARGET_REQUEST_KEYWORD = "/dingtalk/web/APP_DKQG7TSUK6ZNU3RDFGVO/v1/form/searchFormDatas.json" DINGTALK_CHALLENGE_URL_KEYWORD = "https://login.dingtalk.com/oauth2/challenge.htm" DINGTALK_LOGIN_HOST_KEYWORD = "login.dingtalk.com" CORP_TEXT = "云南师范大学附属中学" QUERY_VALUE = "53010320080506463X" LOGIN_WAIT_SECONDS = 120 COOKIE_FILE = BASE_DIR / "cookies.txt" HEADERS_FILE = DATA_DIR / "request_headers_debug.json" STORAGE_STATE_FILE = DATA_DIR / "playwright_state.json" QR_IMAGE_FILE = STATIC_DIR / "dingtalk_qr.png" SCREENSHOT_FILE = STATIC_DIR / "browser_screenshot.png" SHOW_COOKIE_IN_FRONTEND = False state_lock = threading.Lock() worker_lock = threading.Lock() worker_thread = None worker_runtime = { "playwright": None, "browser": None, "context": None, "page": None, } STATE = { "running": False, "status": "idle", "current_url": "", "cookie": "", "cookie_source": "", "request_url": "", "error": "", "last_updated": 0, "logs": [], "has_qr": False, "has_screenshot": False, } def set_state(**kwargs): with state_lock: STATE.update(kwargs) STATE["last_updated"] = int(time.time()) def add_log(message, level="info"): import datetime ts = datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3] with state_lock: STATE["logs"].append({"time": ts, "message": message, "level": level}) STATE["logs"] = STATE["logs"][-300:] STATE["last_updated"] = int(time.time()) def get_state_with_options(include_sensitive=True): with state_lock: data = dict(STATE) data["logs"] = list(STATE["logs"]) data["has_qr"] = QR_IMAGE_FILE.exists() data["has_screenshot"] = SCREENSHOT_FILE.exists() data["show_cookie_in_frontend"] = SHOW_COOKIE_IN_FRONTEND if not include_sensitive: data["cookie"] = "" return data def get_state(include_sensitive=True): return get_state_with_options(include_sensitive=include_sensitive) def build_cookie_header(cookies): parts = [] for c in cookies: name = c.get("name") value = c.get("value") if name and value is not None: parts.append(f"{name}={value}") return "; ".join(parts) def save_text(path: Path, content: str): temp_path = path.with_name(f"{path.name}.tmp") temp_path.write_text(content, encoding="utf-8") temp_path.replace(path) def save_json(path: Path, data): save_text(path, json.dumps(data, ensure_ascii=False, indent=2)) def reset_runtime_artifacts(): for path in (QR_IMAGE_FILE, SCREENSHOT_FILE): try: if path.exists(): path.unlink() except Exception: pass def click_corp_if_present(page): try: if DINGTALK_CHALLENGE_URL_KEYWORD not in page.url: return False corp_item = page.locator("div.module-corp-sel-listitem", has_text=CORP_TEXT).first if corp_item.count() > 0 and corp_item.is_visible(timeout=1000): add_log(f"检测到企业选择项：{CORP_TEXT}，正在点击") corp_item.click() return True except Exception: pass return False def ensure_auto_login_checked(page): try: if DINGTALK_LOGIN_HOST_KEYWORD not in page.url: return False op_item = page.locator("div.module-qrcode-op-item", has_text="自动登录").first if op_item.count() == 0 or not op_item.is_visible(timeout=1000): return False icon = op_item.locator("div.base-comp-check-box-rememberme-box").first icon_class = icon.get_attribute("class") or "" if "checkbox-undone" in icon_class: add_log("检测到“自动登录”未勾选，正在勾选") op_item.click() time.sleep(0.3) return True except Exception: pass return False def export_login_qr_if_present(page): try: if DINGTALK_LOGIN_HOST_KEYWORD not in page.url: return False canvas = page.locator("div.module-qrcode-code canvas").first if canvas.count() == 0 or not canvas.is_visible(timeout=1000): return False data_url = canvas.evaluate("el => el.toDataURL('image/png')") if not data_url or not data_url.startswith("data:image/png;base64,"): return False png_base64 = data_url.split(",", 1)[1] png_bytes = base64.b64decode(png_base64) if len(png_bytes) < 500: add_log("检测到二维码图片可能为空白或未明确加载", "warning") return "blank" previous = QR_IMAGE_FILE.read_bytes() if QR_IMAGE_FILE.exists() else None if previous == png_bytes: return True QR_IMAGE_FILE.write_bytes(png_bytes) set_state(has_qr=True) add_log(f"已提取登录二维码：{QR_IMAGE_FILE.name}", "success") return True except Exception as e: add_log(f"提取登录二维码异常：{e}", "error") return False def click_immediate_login_if_present(page): try: # 检测是否出现快速登录提示 notice = page.get_by_text("此账号已在使用，可直接登录宜搭").first if notice.count() > 0 and notice.is_visible(timeout=500): # 找到立即登录按钮 btn = page.get_by_role("button", name="立即登录").first if btn.count() == 0: # 备用定位方式 btn = page.locator("text=立即登录").first if btn.count() > 0 and btn.is_visible(timeout=500): add_log("检测到快速登录提示，正在点击“立即登录”") btn.click() time.sleep(1) return True except Exception as e: pass return False def is_query_page_ready(page): try: input_box = page.get_by_placeholder("请输入").first if input_box.count() > 0 and input_box.is_visible(timeout=500): query_button = page.get_by_role("button", name="查 询") return query_button.count() > 0 and query_button.first.is_visible(timeout=500) except Exception: return False return False def click_qr_login_tab_if_present(page): try: if DINGTALK_LOGIN_HOST_KEYWORD not in page.url: return False tab = page.locator("div.dtm-segment-item", has_text="扫码登录").first if tab.count() > 0 and tab.is_visible(timeout=500): if tab.get_attribute("aria-selected") != "true": add_log("检测到未处于扫码登录状态，正在切换") tab.click() time.sleep(0.5) return True except Exception: pass return False def wait_until_query_page_ready(page, start_time=None, timeout_seconds=300): if start_time is None: start_time = time.time() deadline = start_time + timeout_seconds last_url = None while time.time() < deadline: try: current_url = page.url except Exception: current_url = "" set_state(current_url=current_url) if current_url != last_url: add_log(f"当前页面：{current_url}") last_url = current_url click_immediate_login_if_present(page) click_corp_if_present(page) click_qr_login_tab_if_present(page) ensure_auto_login_checked(page) qr_status = export_login_qr_if_present(page) if qr_status == "blank": return "blank_qr" if is_query_page_ready(page): return "ready" try: page.wait_for_load_state("domcontentloaded", timeout=1000) except Exception: pass time.sleep(1) return "timeout" def cleanup_runtime(): runtime = worker_runtime for key in ["page", "context", "browser", "playwright"]: obj = runtime.get(key) if obj is None: continue try: if key == "playwright": obj.stop() else: obj.close() except Exception: pass runtime[key] = None def automation_worker(): start_time = time.time() with worker_lock: max_attempts = 2 for attempt in range(max_attempts): if time.time() - start_time > 300: add_log("任务总耗时超过5分钟，自动终止此轮任务", "error") set_state(status="timeout", error="任务总耗时超过5分钟", running=False) break if attempt == 0: reset_runtime_artifacts() set_state( running=True, status="starting", error="", current_url="", cookie="", cookie_source="", request_url="" ) add_log(f"自动化任务启动（第 {attempt + 1} 次尝试）") success = False try: pw = sync_playwright().start() worker_runtime["playwright"] = pw browser = pw.chromium.launch(headless=True) worker_runtime["browser"] = browser if STORAGE_STATE_FILE.exists(): if attempt == 0: context = browser.new_context(storage_state=str(STORAGE_STATE_FILE)) add_log(f"已加载登录状态：{STORAGE_STATE_FILE.name}") else: try: STORAGE_STATE_FILE.unlink() except Exception: pass add_log("已尝试清空旧的登录状态数据") context = browser.new_context() add_log("使用新会话（未加载已有状态）") else: context = browser.new_context() add_log("未发现已保存登录状态，使用新会话") worker_runtime["context"] = context page = context.new_page() worker_runtime["page"] = page set_state(status="opening") add_log("正在打开目标页面") page.goto(TARGET_PAGE, wait_until="load") set_state(status="waiting_login") add_log("正在自动检测登录完成状态") time_elapsed = time.time() - start_time remaining_time = max(10, 300 - time_elapsed) ready_status = wait_until_query_page_ready(page, start_time=time.time(), timeout_seconds=remaining_time) if ready_status == "timeout": add_log("等待登录完成超时", "warning" if attempt == 0 else "error") if attempt == 0: continue else: set_state(status="login_timeout", error="在限定时间内未进入查询页面") break elif ready_status == "blank_qr": add_log("二维码持续空白或异常", "warning" if attempt == 0 else "error") if attempt == 0: continue else: set_state(status="login_timeout", error="二维码为空白") break elif not ready_status or ready_status == False: if attempt == 0: continue else: set_state(status="login_timeout", error="未能进入查询页面") break context.storage_state(path=str(STORAGE_STATE_FILE)) add_log(f"已保存登录状态：{STORAGE_STATE_FILE.name}", "success") set_state(status="querying") input_box = page.get_by_placeholder("请输入").first input_box.click() input_box.fill(QUERY_VALUE) add_log(f"已填入查询值：{QUERY_VALUE}") add_log("准备点击查询，并同步等待目标请求") with page.expect_request(lambda req: TARGET_REQUEST_KEYWORD in req.url, timeout=30000) as req_info: query_button = page.get_by_role("button", name="查 询") query_button.click() request_obj = req_info.value request_url = request_obj.url headers = {} try: headers = request_obj.all_headers() except Exception: try: headers = request_obj.headers except Exception: headers = {} debug_headers = dict(headers or {}) if "cookie" in debug_headers: debug_headers["cookie"] = "[saved-to-cookies.txt]" save_json( HEADERS_FILE, { "url": request_url, "headers": debug_headers, "cookie_header_captured": bool((headers or {}).get("cookie")) } ) cookie = headers.get("cookie") if headers else None source = "直接从目标请求头读取" if not cookie: try: cookies = context.cookies([request_url]) except Exception: cookies = context.cookies() cookie = build_cookie_header(cookies) source = "按目标请求可发送规则重建" if cookie: save_text(COOKIE_FILE, cookie) add_log(f"已覆盖保存本地 Cookie 文件：{COOKIE_FILE.name}", "success") set_state(cookie=cookie, cookie_source=source, request_url=request_url) add_log(f"已成功获取 cookie，来源：{source}", "success") set_state(status="done") else: set_state(status="cookie_missing", request_url=request_url, error="未能拿到 cookie") add_log("已捕获目标请求，但未能拿到 cookie", "error") success = True break # Successful, exit the loop except PlaywrightTimeoutError: set_state(status="request_timeout", error="30 秒内未捕获到目标请求") add_log("等待目标请求超时", "error") if attempt == 0: add_log("准备重试...", "warning") except Exception as e: set_state(status="error", error=str(e)) add_log(f"发生异常：{e}", "error") if attempt == 0: add_log("准备重试...", "warning") finally: cleanup_runtime() if success or attempt == max_attempts - 1: set_state(running=False) add_log("自动化任务结束") def start_worker_if_needed(): global worker_thread if worker_thread and worker_thread.is_alive(): return False worker_thread = threading.Thread(target=automation_worker, daemon=True) worker_thread.start() return True