用于人工智能法律工作流程的移民案件状态数据提取

from dataclasses import dataclass
from datetime import datetime, timezone

@dataclass
class ImmigrationCaseStatus:
    matter_id: str
    receipt_number_ref: str
    status_code: str
    status_title: str
    status_message: str
    source_url: str
    checked_at: str
    source_updated_at: str | None = None


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()

import re

RECEIPT_PATTERN = re.compile(r"^[A-Z]{3}[0-9]{10}$")


def normalize_receipt_number(value: str) -> str:
    normalized = value.replace("-", "").replace(" ", "").upper()
    if not RECEIPT_PATTERN.fullmatch(normalized):
        raise ValueError("收据编号必须包含三个字母和十个数字")
    return normalized

from playwright.async_api import Page

CASE_STATUS_URL = "https://egov.uscis.gov/"

async def fetch_status_page(page: Page, receipt_number: str) -> dict:
    await page.goto(CASE_STATUS_URL, wait_until="domcontentloaded")

    # 选择器必须针对当前授权页面进行验证。
    await page.get_by_label("Enter a Receipt Number").fill(receipt_number)
    await page.get_by_role("button", name="Check Status").click()
    await page.wait_for_load_state("networkidle")

    title = await page.locator("main h1, main h2").first.inner_text()
    message = await page.locator("main").inner_text()

    return {
        "status_title": title.strip(),
        "status_message": message.strip(),
        "source_url": page.url,
    }

import os
from capsolver_core import Capsolver

cap = Capsolver(
    api_key=os.environ["CAPSOLVER_API_KEY"],
    default_timeout=180,
)

async def resolve_supported_challenge(page: Page) -> dict:
    detected = await cap.detect(page)
    if not detected:
        return {"handled": False, "reason": "未检测到支持的挑战"}

    solution = await cap.solve_on_page(page)
    return {
        "handled": True,
        "solution": solution,
    }

async def authorized_status_check(page: Page, receipt_number: str) -> dict:
    try:
        return await fetch_status_page(page, receipt_number)
    except Exception as first_error:
        recovery = await resolve_supported_challenge(page)
        if not recovery["handled"]:
            raise first_error
        return await fetch_status_page(page, receipt_number)

STATUS_RULES = {
    "case was received": "received",
    "case is being actively reviewed": "under_review",
    "request for evidence": "evidence_requested",
    "case was approved": "approved",
    "card was produced": "document_produced",
    "case was denied": "adverse_decision",
}


def classify_status(title: str, message: str) -> dict:
    combined = f"{title} {message}".lower()
    for phrase, code in STATUS_RULES.items():
        if phrase in combined:
            return {"status_code": code, "confidence": "rule_match"}
    return {"status_code": "unclassified", "confidence": "needs_review"}

import hashlib


def status_fingerprint(status: dict) -> str:
    canonical = "|".join([
        status.get("status_code", ""),
        status.get("status_title", "").strip(),
        status.get("status_message", "").strip(),
    ])
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def has_changed(previous: dict | None, current: dict) -> bool:
    if previous is None:
        return True
    return status_fingerprint(previous) != status_fingerprint(current)

SAFE_EVENT = {
    "matter_id": "MAT-2026-00172",
    "previous_status": "under_review",
    "current_status": "evidence_requested",
    "change_detected_at": "2026-08-25T08:15:00Z",
    "source": "official_case_status_portal",
    "recommended_action": "create_human_review_task",
}