Live clients¶

async def case_status(
    self,
    court: Court,
    *,
    case_type: str,
    case_number: str,
    year: str,
    bench_code: str = "1",
) -> list[CaseInfo]:
    """Look up case status by case number.

    Args:
        court: Court object (use get_court() to obtain).
        case_type: Numeric case type code (e.g. "134" for W.P.(C) in Delhi).
            Use :meth:`list_case_types` to discover available codes.
        case_number: Case number without type/year.
        year: Registration year (e.g. "2024").
        bench_code: Bench code from :meth:`list_benches` (default "1").

    Returns:
        List of matching CaseInfo objects.
    """

    def build_form(captcha: str) -> dict:
        return endpoints.case_status_form(
            state_code=court.state_code,
            court_code=bench_code,
            case_type=case_type,
            case_number=case_number,
            year=year,
            captcha=captcha,
        )

    resp = await self._post_with_captcha_retry(endpoints.SHOW_RECORDS_URL, build_form)
    results = parse_case_status(resp.text)
    for r in results:
        r.court_name = court.name
    return results

async def case_status_by_party(
    self,
    court: Court,
    *,
    party_name: str,
    year: str,
    bench_code: str = "1",
    status_filter: str = "Both",
) -> list[CaseInfo]:
    """Search cases by party name.

    Args:
        court: Court object.
        party_name: Petitioner or respondent name (min 3 chars).
        year: Registration year (**mandatory**, e.g. "2024").
        bench_code: Bench code from :meth:`list_benches` (default "1").
        status_filter: "Pending", "Disposed", or "Both".

    Returns:
        List of matching CaseInfo objects.
    """

    def build_form(captcha: str) -> dict:
        return endpoints.case_status_by_party_form(
            state_code=court.state_code,
            court_code=bench_code,
            petres_name=party_name,
            rgyear=year,
            captcha=captcha,
            status_filter=status_filter,
        )

    resp = await self._post_with_captcha_retry(endpoints.SHOW_RECORDS_URL, build_form)
    results = parse_case_status(resp.text)
    for r in results:
        r.court_name = court.name
    return results

async def court_orders(
    self,
    court: Court,
    *,
    case_type: str,
    case_number: str,
    year: str,
    bench_code: str = "1",
) -> list[CaseOrder]:
    """Get court orders for a case.

    Uses a case number search to get the encrypted order URL path,
    then constructs the PDF download URL from display_pdf.php.

    Args:
        court: Court object.
        case_type: Numeric case type code (e.g. "134").
        case_number: Case number.
        year: Registration year.
        bench_code: Bench code from :meth:`list_benches` (default "1").

    Returns:
        List of CaseOrder objects with PDF URLs.
    """

    def build_form(captcha: str) -> dict:
        return endpoints.case_status_form(
            state_code=court.state_code,
            court_code=bench_code,
            case_type=case_type,
            case_number=case_number,
            year=year,
            captcha=captcha,
        )

    resp = await self._post_with_captcha_retry(endpoints.SHOW_RECORDS_URL, build_form)
    return parse_orders(
        resp.text,
        base_url=endpoints.BASE_URL,
        bench_code=bench_code,
        state_code=court.state_code,
    )

async def cause_list(
    self,
    court: Court,
    *,
    civil: bool = True,
    bench_code: str = "1",
    causelist_date: str = "",
) -> list[CauseListPDF]:
    """Get cause list PDFs for a court.

    The HC Services portal returns a table of PDF links, one per bench/judge.
    Each entry contains the bench name, cause list type, and PDF URL.

    Args:
        court: Court object.
        civil: True for civil cases, False for criminal.
        bench_code: Bench code from list_benches() (default "1" = principal).
        causelist_date: Date in DD-MM-YYYY format (defaults to today).

    Returns:
        List of CauseListPDF objects with bench info and PDF URLs.
    """
    # Determine selprevdays: "1" if date is in the past, "0" otherwise
    selprevdays = "0"
    if causelist_date:
        from datetime import date, datetime

        try:
            sel = datetime.strptime(causelist_date, "%d-%m-%Y").date()
            if sel < date.today():
                selprevdays = "1"
        except ValueError:
            pass

    def build_form(captcha: str) -> dict:
        return endpoints.cause_list_form(
            state_code=court.state_code,
            court_code=bench_code,
            captcha=captcha,
            causelist_date=causelist_date,
            flag="civ_t" if civil else "cri_t",
            selprevdays=selprevdays,
        )

    resp = await self._post_with_captcha_retry(endpoints.INDEX_QRY_URL, build_form)
    return parse_cause_list(resp.text, base_url=endpoints.BASE_URL)

async def list_benches(self, court: Court) -> dict[str, str]:
    """Get available benches for a High Court.

    Returns:
        Dict mapping bench code to bench name, e.g.
        {"1": "Principal Bench at Delhi", "2": "Lucknow Bench"}.
    """
    await self._init_session()
    form = endpoints.fill_bench_form(state_code=court.state_code)
    resp = await self._http.post(endpoints.INDEX_QRY_URL, data=form)
    benches = {}
    for entry in resp.text.split("#"):
        entry = entry.strip()
        if "~" in entry:
            code, name = entry.split("~", 1)
            # Strip BOM (\ufeff) and whitespace from portal response
            code = code.strip().strip("\ufeff")
            name = name.strip().strip("\ufeff")
            if code and code != "0" and name and "select" not in name.lower():
                benches[code] = name
    return benches

async def list_case_types(self, court: Court, *, bench_code: str = "1") -> dict[str, str]:
    """Get available case types for a High Court bench.

    Returns:
        Dict mapping case type code to name, e.g.
        {"134": "W.P.(C)(CIVIL WRITS)-134", "27": "W.P.(CRL)..."}.
    """
    await self._init_session()
    form = endpoints.fill_case_type_form(
        state_code=court.state_code,
        court_code=bench_code,
    )
    resp = await self._http.post(endpoints.FILL_CASE_TYPE_URL, data=form)
    case_types = {}
    for entry in resp.text.split("#"):
        entry = entry.strip().strip("\ufeff")
        if "~" in entry:
            code, name = entry.split("~", 1)
            code = code.strip()
            name = name.strip()
            if code and code != "0" and name and "select" not in name.lower():
                case_types[code] = name
    return case_types

async def download_order_pdf(self, pdf_url: str) -> bytes:
    """Download an order/judgment PDF.

    The display_pdf.php endpoint requires a valid Referer header
    and session cookies from the same client that performed the search.

    Args:
        pdf_url: URL from CaseOrder.pdf_url.

    Returns:
        Raw PDF bytes.
    """
    resp = await self._http.get(
        pdf_url,
        headers={
            "Referer": endpoints.MAIN_PAGE_URL,
            "Accept": "application/pdf,*/*",
        },
    )
    content = resp.content
    if content[:4] != b"%PDF":
        raise RuntimeError(
            f"PDF download did not return a valid PDF "
            f"(got {len(content)} bytes; head={content[:64]!r})"
        )
    return content

async def list_states(self) -> dict[str, str]:
    """Get available states/UTs.

    Returns:
        Dict mapping state code to state name.
    """
    # States are static, from the portal dropdown
    return {v: k for k, v in endpoints.DISTRICT_STATES.items()}

async def list_districts(self, state_code: str) -> dict[str, str]:
    """Get districts for a state.

    Args:
        state_code: State code (e.g. "8" for Bihar).

    Returns:
        Dict mapping district code to district name.
    """
    await self._init_session()
    form = endpoints.fill_district_form(state_code=state_code)
    result = await self._post_ajax("casestatus/fillDistrict", form)
    dist_html = result.get("dist_list", "")
    return parse_option_tags(dist_html)

async def list_complexes(self, state_code: str, dist_code: str) -> dict[str, str]:
    """Get court complexes for a district.

    Args:
        state_code: State code.
        dist_code: District code.

    Returns:
        Dict mapping complex value (``code@ests@flag``) to complex name.
        Use :func:`parse_complex_value` to extract the complex code
        and determine if establishment selection is needed.
    """
    await self._init_session()
    # Ensure district is filled first
    await self._post_ajax(
        "casestatus/fillDistrict",
        endpoints.fill_district_form(state_code=state_code),
    )
    form = endpoints.fill_complex_form(state_code=state_code, dist_code=dist_code)
    result = await self._post_ajax("casestatus/fillcomplex", form)
    complex_html = result.get("complex_list", "")
    return parse_option_tags(complex_html)

async def list_establishments(
    self,
    state_code: str,
    dist_code: str,
    court_complex_code: str,
) -> dict[str, str]:
    """Get establishments for a court complex.

    Only needed when the complex flag is 'Y'.

    Args:
        state_code: State code.
        dist_code: District code.
        court_complex_code: Raw complex code (without @ests@flag).

    Returns:
        Dict mapping establishment code to name.
    """
    await self._init_session()
    form = endpoints.fill_establishment_form(
        state_code=state_code,
        dist_code=dist_code,
        court_complex_code=court_complex_code,
    )
    result = await self._post_ajax("casestatus/fillCourtEstablishment", form)
    est_html = result.get("est_list", result.get("establishment_list", ""))
    return parse_option_tags(est_html)

async def list_cause_list_courts(
    self,
    state_code: str,
    dist_code: str,
    court_complex_code: str,
    est_code: str = "",
) -> dict[str, str]:
    """Get the courts dropdown for cause-list lookup.

    The cause-list form requires both ``court_no`` (the option's
    value, e.g. ``"1@2"``) and ``court_name`` (the option's display
    text, e.g. ``"District & Sessions Judge - DJ Div. Patna Sadar"``).
    Use this method to discover them; pass either the code through
    directly to :meth:`cause_list`, which will look up the matching
    name automatically.
    """
    await self._init_session()
    form = endpoints.fill_cause_list_form(
        state_code=state_code,
        dist_code=dist_code,
        court_complex_code=court_complex_code,
        est_code=est_code,
    )
    result = await self._post_ajax("cause_list/fillCauseList", form)
    html = result.get("cause_list", "")
    return parse_option_tags(html)

async def list_case_types(
    self,
    state_code: str,
    dist_code: str,
    court_complex_code: str,
    est_code: str = "",
) -> dict[str, str]:
    """Get available case types for a court.

    Args:
        state_code: State code.
        dist_code: District code.
        court_complex_code: Court complex code.
        est_code: Establishment code (if needed).

    Returns:
        Dict mapping case type code to name. Codes are returned in the
        portal's compound ``"<case_type>^<est_code>"`` format (e.g.
        ``"89^2": "ADMINISTRATIVE SUITE"``). Pass the full compound
        string back as ``case_type`` to :meth:`case_status` /
        :meth:`court_orders`; do not strip the suffix.
    """
    await self._init_session()
    await self._setup_court(
        state_code=state_code,
        dist_code=dist_code,
        court_complex_code=court_complex_code,
        est_code=est_code,
    )
    form = endpoints.fill_case_type_form(
        state_code=state_code,
        dist_code=dist_code,
        court_complex_code=court_complex_code,
        est_code=est_code,
        search_type="c_no",
    )
    result = await self._post_ajax("casestatus/fillCaseType", form)
    ct_html = result.get("casetype_list", "")
    return parse_option_tags(ct_html)

async def case_status(
    self,
    *,
    state_code: str,
    dist_code: str,
    court_complex_code: str,
    est_code: str = "",
    case_type: str,
    case_number: str,
    year: str,
) -> list[CaseInfo]:
    """Look up case status by case number.

    Args:
        state_code: State code (e.g. "8" for Bihar).
        dist_code: District code (e.g. "1" for Patna).
        court_complex_code: Court complex code (e.g. "1080010").
        est_code: Establishment code (if needed).
        case_type: Case type code from :meth:`list_case_types`.
        case_number: Case number.
        year: Registration year.

    Returns:
        List of matching CaseInfo objects.
    """

    def build_form(captcha: str) -> dict:
        return endpoints.case_status_by_number_form(
            state_code=state_code,
            dist_code=dist_code,
            court_complex_code=court_complex_code,
            est_code=est_code,
            case_type=case_type,
            case_number=case_number,
            year=year,
            captcha=captcha,
        )

    result = await self._post_with_captcha_retry(
        "casestatus/submitCaseNo",
        build_form,
        state_code=state_code,
        dist_code=dist_code,
        court_complex_code=court_complex_code,
        est_code=est_code,
    )
    html = result.get("case_data", "")
    return parse_case_status_html(html)

async def case_status_by_party(
    self,
    *,
    state_code: str,
    dist_code: str,
    court_complex_code: str,
    est_code: str = "",
    party_name: str,
    year: str,
    status_filter: str = "Both",
) -> list[CaseInfo]:
    """Search cases by party name.

    Args:
        state_code: State code.
        dist_code: District code.
        court_complex_code: Court complex code.
        est_code: Establishment code (if needed).
        party_name: Petitioner/respondent name (min 3 chars).
        year: Registration year (mandatory).
        status_filter: "Pending", "Disposed", or "Both".

    Returns:
        List of matching CaseInfo objects.
    """

    def build_form(captcha: str) -> dict:
        return endpoints.case_status_by_party_form(
            state_code=state_code,
            dist_code=dist_code,
            court_complex_code=court_complex_code,
            est_code=est_code,
            party_name=party_name,
            year=year,
            status_filter=status_filter,
            captcha=captcha,
        )

    result = await self._post_with_captcha_retry(
        "casestatus/submitPartyName",
        build_form,
        state_code=state_code,
        dist_code=dist_code,
        court_complex_code=court_complex_code,
        est_code=est_code,
    )
    html = result.get("party_data", "")
    return parse_case_status_html(html)

async def court_orders(
    self,
    *,
    state_code: str,
    dist_code: str,
    court_complex_code: str,
    est_code: str = "",
    case_type: str,
    case_number: str,
    year: str,
) -> list[CaseOrder]:
    """Get court orders for a case.

    Args:
        state_code: State code.
        dist_code: District code.
        court_complex_code: Court complex code.
        est_code: Establishment code (if needed).
        case_type: Case type code.
        case_number: Case number.
        year: Registration year.

    Returns:
        List of CaseOrder objects.
    """

    def build_form(captcha: str) -> dict:
        return endpoints.court_orders_by_number_form(
            state_code=state_code,
            dist_code=dist_code,
            court_complex_code=court_complex_code,
            est_code=est_code,
            case_type=case_type,
            case_number=case_number,
            year=year,
            captcha=captcha,
        )

    result = await self._post_with_captcha_retry(
        "courtorder/submitCaseNo",
        build_form,
        state_code=state_code,
        dist_code=dist_code,
        court_complex_code=court_complex_code,
        est_code=est_code,
    )
    html = result.get("order_data", result.get("case_data", ""))
    return parse_court_orders_html(html, base_url=endpoints.BASE_URL)

async def cause_list(
    self,
    *,
    state_code: str,
    dist_code: str,
    court_complex_code: str,
    est_code: str = "",
    court_no: str,
    court_name: str = "",
    causelist_date: str = "",
    civil: bool = True,
) -> list[CauseListEntry]:
    """Get cause list for a court.

    Args:
        state_code: State code.
        dist_code: District code.
        court_complex_code: Court complex code.
        est_code: Establishment code (if needed).
        court_no: Court code from :meth:`list_cause_list_courts`
            (the option's ``value``).
        court_name: Court display name (the option's text). The
            portal validates against this — sending an empty
            ``court_name_txt`` triggers ``"Court Name is required"``.
            If left empty, this method calls
            :meth:`list_cause_list_courts` once to look up the
            matching name for ``court_no``.
        causelist_date: Date in DD-MM-YYYY format (defaults to today).
        civil: True for civil, False for criminal.

    Returns:
        List of CauseListEntry objects.
    """
    if not court_name:
        mapping = await self.list_cause_list_courts(
            state_code, dist_code, court_complex_code, est_code
        )
        court_name = mapping.get(court_no, "")
        if not court_name:
            raise ValueError(
                f"court_no={court_no!r} not found in fillCauseList for "
                f"complex={court_complex_code} est={est_code}. "
                f"Available: {list(mapping)[:5]}{'...' if len(mapping) > 5 else ''}"
            )

    def build_form(captcha: str) -> dict:
        return endpoints.cause_list_form(
            state_code=state_code,
            dist_code=dist_code,
            court_complex_code=court_complex_code,
            est_code=est_code,
            court_no=court_no,
            court_name=court_name,
            causelist_date=causelist_date,
            civil=civil,
            captcha=captcha,
        )

    result = await self._post_with_captcha_retry(
        "cause_list/submitCauseList",
        build_form,
        state_code=state_code,
        dist_code=dist_code,
        court_complex_code=court_complex_code,
        est_code=est_code,
    )
    html = result.get("causelist_data", result.get("cause_list_data", ""))
    return parse_cause_list_html(html)

async def search(
    self,
    search_text: str,
    *,
    page: int = 1,
    page_size: int = 10,
    search_opt: str = "PHRASE",
    court_type: str = "2",
    max_captcha_attempts: int = 5,
) -> SearchResult:
    """Search for judgments by keyword.

    Args:
        search_text: Keywords / phrase to search for.
        page: 1-indexed page number.
        page_size: Rows per page (portal supports 10/25/50/100/1000).
        search_opt: ``"PHRASE"``, ``"ANY"``, or ``"ALL"``.
        court_type: ``"2"`` for High Courts, ``"3"`` for SCR.
        max_captcha_attempts: Max CAPTCHA solve retries before giving up.

    Returns:
        ``SearchResult`` of :class:`JudgmentResult` items.

    Raises:
        CaptchaError: if the CAPTCHA solver couldn't produce a valid
            solution within ``max_captcha_attempts`` tries. Empty
            results are now distinguishable from "we gave up": empty
            means the portal returned zero rows.
    """
    captcha_text = await self._authenticate(
        search_text, max_captcha_attempts=max_captcha_attempts
    )
    if captcha_text is None:
        raise CaptchaError(f"Failed to solve CAPTCHA after {max_captcha_attempts} attempts")

    data = await self._post_search(
        search_text=search_text,
        captcha_text=captcha_text,
        search_opt=search_opt,
        court_type=court_type,
        page=page,
        page_size=page_size,
    )
    return parse_search_response(data, page=page, page_size=page_size)