华为招聘状态码更新通知脚本 2025最新版

可以每小时自动查看应聘状态，状态更新自动发送邮件

状态码对照表

Python源代码：

（修改UID 、PASSWORD 、EMAIL_FROM_TO、EMAIL_AUTH_CODE 为自己真实信息即可运行）

"""
Huawei Recruit Monitor – 2025-06-08
• 每小时自动登录华为招聘门户
• 监控字段IV_DATE提前（INTERVIEW_TYPE字段消失）即发 QQ 邮件
"""
import time, json, requests, traceback
from datetime import datetime
from pathlib import Path
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
import smtplib, ssl, contextlib
import selenium
from selenium import webdriver
from selenium.common import NoSuchElementException
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# ─────────────────── 1. 账户 & 常量 ───────────────────
UID               = ""       # 华为账号
PASSWORD          = ""       # 华为密码
EMAIL_FROM_TO     = ""       # QQ邮箱
EMAIL_AUTH_CODE   = ""       # QQ邮箱授权码——>自行搜索如何获取
CHECK_INTERVAL    = 60 * 60                     # 每小时
CACHE_FILE        = Path("last_status.txt")
HEADLESS          = False                      # 改为True可以隐藏浏览器页面
# ─────────────────── 2. 邮件工具 ───────────────────
def send_email(subject: str, text: str) -> None:
    msg = MIMEMultipart()
    msg["Subject"] = Header(subject, "utf-8")
    msg["From"] = msg["To"] = EMAIL_FROM_TO
    msg.attach(MIMEText(text, "plain", "utf-8"))
    ctx = ssl.create_default_context()
    smtp = smtplib.SMTP("smtp.qq.com", 587, timeout=15)
    try:
        smtp.ehlo()
        smtp.starttls(context=ctx)
        smtp.ehlo()
        smtp.login(EMAIL_FROM_TO, EMAIL_AUTH_CODE)
        smtp.sendmail(EMAIL_FROM_TO, [EMAIL_FROM_TO], msg.as_string())
    finally:
        # 任何 quit() 触发的异常（-1, b'\x00\x00\x00'）都处理
        with contextlib.suppress(smtplib.SMTPResponseException,
                                 smtplib.SMTPServerDisconnected):
            smtp.quit()
        smtp.close()

# ─────────────────── 3. Selenium 登录 ───────────────────
def switch_into_iframe_with_username(driver):
    """递归遍历 iframe，切到含有 #username 的那层。"""
    try:
        driver.find_element(By.ID, "username"); return
    except NoSuchElementException:
        pass
    for f in driver.find_elements(By.TAG_NAME, "iframe"):
        driver.switch_to.frame(f)
        try:
            switch_into_iframe_with_username(driver)
            if driver.find_elements(By.ID, "username"):
                return
        except NoSuchElementException:
            pass
        driver.switch_to.parent_frame()

def ensure_portal_session(driver):
    """显式调用 portalLoginPortal5，保证门户会话就绪。"""
    driver.execute_async_script("""
        const done = arguments[0];
        fetch("https://career.huawei.com/reccampportal/services/portal/portaluser/portalLoginPortal5",
              { credentials: "include" }).finally(done);
    """)

def login_and_get_session() -> requests.Session:
    opts = webdriver.EdgeOptions()
    if HEADLESS:
        # ❶ 新版无头 (--headless=new) + 自定义分辨率
        opts.add_argument("--headless=new")
        opts.add_argument("--window-size=1920,1080")
    opts.add_argument("--disable-gpu")

    driver = webdriver.Edge(options=opts)
    try:
        # ① SSO
        driver.get(
            "https://uniportal.huawei.com/uniportal/login.do"
            "?redirect=https://career.huawei.com/reccampportal/portal5/index.html"
        )
        switch_into_iframe_with_username(driver)

        WebDriverWait(driver, 25).until(
            EC.visibility_of_element_located((By.ID, "username"))
        ).send_keys(UID)
        switch_into_iframe_with_username(driver)

        # ① 账号输入框 <input id="username" name="username">
        user_box = WebDriverWait(driver, 25).until(
            EC.visibility_of_element_located((By.ID, "username"))
        )
        user_box.clear()
        user_box.send_keys(UID)

        # ② 密码框：唯一的 <input type="password">
        pwd_box = driver.find_element(By.CSS_SELECTOR, "input[type='password']")
        pwd_box.clear()
        pwd_box.send_keys(PASSWORD)
        login_btn = driver.find_element(By.ID, "login-btn")
        # ③ 点击登录按钮 <button id="login-btn">
        try:
            # ① 首选——官方等待元素可点击
            WebDriverWait(driver, 5).until(EC.element_to_be_clickable(login_btn))
            login_btn.click()
        except (selenium.common.exceptions.ElementClickInterceptedException,
                selenium.common.exceptions.TimeoutException):
            # ② 兜底一：滚动到按钮正中再点
            driver.execute_script(
                "arguments[0].scrollIntoView({block:'center', inline:'center'});",
                login_btn
            )
            time.sleep(0.3)  # 让动画完成
            try:
                login_btn.click()
            except selenium.common.exceptions.ElementClickInterceptedException:
                # ③ 兜底二：JavaScript 强制点击，无视遮挡
                driver.execute_script("arguments[0].click();", login_btn)
        driver.switch_to.default_content()

        # ④ 等待 SSO Cookie 下发，视为登录成功
        WebDriverWait(driver, 30).until(
            lambda d: "career.huawei.com" in d.current_url
        )
        # ⑤ 把 Cookie 复制到 requests.Session
        sess = requests.Session()
        for c in driver.get_cookies():  # 不再判断名字
            sess.cookies.set(c["name"], c["value"], domain=c["domain"])
        # ② 等登录跳转
        WebDriverWait(driver, 30).until(
            lambda d: "career.huawei.com" in d.current_url
        )
        # ③ ★ 点击昵称，进入个人中心
        nick = WebDriverWait(driver, 20).until(
            EC.element_to_be_clickable(
                (By.CSS_SELECTOR, "dt.header-pc-user-nickname"))
        )
        nick.click()
        # 可能在新 tab，切过去
        driver.switch_to.window(driver.window_handles[-1])

        # ④ ★ 点击「申请进展」
        prog_btn = WebDriverWait(driver, 20).until(
            EC.element_to_be_clickable(
                (By.CSS_SELECTOR, "a[href*='job-progress.html']"))
        )
        prog_btn.click()
        # ⑤ ★ 确保门户会话初始化
        ensure_portal_session(driver)

        # ⑥ 把 Cookie 复制到 requests.Session
        sess = requests.Session()
        for c in driver.get_cookies():
            sess.cookies.set(c["name"], c["value"], domain=c["domain"])
        return sess
    except Exception:
        Path("debug.html").write_text(driver.page_source, "utf-8")
        driver.save_screenshot("debug.png")
        raise
    finally:
        driver.quit()
# ─────────────────── 4. 查询接口 ───────────────────
def fetch_status(sess: requests.Session) -> dict:
    ts  = int(time.time() * 1000)
    url = (
        "https://career.huawei.com/reccampportal/services/portal/"
        f"portaluser/queryMyJobInterviewPortal5/newHr?reqTime={ts}"
    )
    csrf_token = (
        sess.cookies.get("portal_csrf_token")
        or sess.cookies.get("x-csrf-token")
        or ""
    )
    headers = {
        "Referer": "https://career.huawei.com/reccampportal/portal5/job-progress.html",
        "X-Requested-With": "XMLHttpRequest",
        "X-Csrf-Token": csrf_token,
        "Accept": "application/json, text/plain, */*",
    }
    r = sess.get(url, headers=headers, timeout=10)
    r.raise_for_status()
    data = r.json()
    if isinstance(data, dict) and data.get("code"):
        raise RuntimeError(f"接口报错：{data}")
    if not data:
        raise ValueError("接口返回空列表")
    return data[0]
# ─────────────────── 5. 本地状态缓存 ───────────────────
def load_last() -> str | None:
    return CACHE_FILE.read_text("utf-8").strip() if CACHE_FILE.exists() else None

def save_last(val: str) -> None:
    CACHE_FILE.write_text(val, "utf-8")
# ─────────────────── 6. 主循环 ───────────────────
def main():
    last_val = load_last()
    print(f"[Init] 之前缓存状态：{last_val}")

    # 首次启动时执行全链路测试
    if last_val is None:  # 通过last_val判断是否首次运行
        print("\n⚠️ 首次启动，执行全链路测试...")
        try:
            # 测试登录和获取状态
            print(f"[{datetime.now():%F %T}] ➜ 测试登录流程")
            test_sess = login_and_get_session()
            test_data = fetch_status(test_sess)
            print(f"[{datetime.now():%F %T}] 测试数据获取成功")

            # 测试邮件发送（使用模拟数据）
            print(f"[{datetime.now():%F %T}] ➜ 测试邮件发送")
            send_email(
                "⚠️ 华为招聘状态监控系统测试邮件",
                ("这是全链路功能测试邮件，系统运行正常！\n\n"
                 f"测试时间: {datetime.now():%F %T}\n"
                 "此邮件仅用于验证系统功能，非真实状态更新通知")
            )
            print("✅ 测试邮件已发送，请检查收件箱")

            # 保存初始状态（避免后续重复测试）
            current_status = test_data.get('INTERVIEW_TYPE', '泡池中')
            last_val = current_status
            save_last(last_val)
            print(f"[Init] 已保存初始状态: {last_val}")
        except Exception as e:
            print("❌ 全链路测试失败:", e)
            traceback.print_exc()
            print("\n⚠️ 系统启动失败，请检查错误信息后重试")
            return  # 测试失败时直接退出
        print("\n✅ 全链路测试通过，开始正式监控...\n")
    # 主监控循环
    while True:
        try:
            print(f"[{datetime.now():%F %T}] ➜ 开始登录")
            sess = login_and_get_session()
            data = fetch_status(sess)
            #测试
            #data = {"IV_DATE": "2025-03-01 00:12:41", "index": "2", "showStatus": "6"}
            # 判断逻辑：检查INTERVIEW_TYPE字段是否存在
            if 'INTERVIEW_TYPE' not in data:
                # 当INTERVIEW_TYPE不存在时，表示审批已通过
                if not last_val == "APPROVED":  # 避免重复通知
                    send_email(
                        "🎉 华为招聘状态已更新！",
                        (f"审批已通过\n\n"
                         f"(当第一个字段由INTERVIEW_TYPE变为IV_DATE时，表示审批已通过)\n\n"
                         f"完整响应：\n{json.dumps(data, ensure_ascii=False, indent=2)}")
                    )
                    print("📧 已发送状态更新邮件")
                    last_val = "APPROVED"  # 更新最后记录的状态
                    save_last(last_val)
                    break
            else:
                # 未变化时，记录当前状态
                current_status = '泡池中'
                print(f"[{datetime.now():%F %T}] 当前状态 ： {current_status}")
                # 保存当前状态但不发送通知
                last_val = current_status
                save_last(last_val)
        except Exception as e:
            print("❌ 出错:", e)
            traceback.print_exc()
        print(f"🌙 休眠 {CHECK_INTERVAL // 60} 分钟 …\n")
        time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
    main()