Project

General

Profile

Download (67.9 KB)

Feature #99 » eap_sez_automation.py

The individual code used to implement the step 3 in framework, which is used to extract the request id from the excel template and download it and attach it to the frappe doctype - Rajesh Gotur, 09/12/2026 05:13 PM

 
import os
import re
import time
import json
import requests
import warnings
from pathlib import Path
from datetime import datetime
import cv2
import easyocr
from openpyxl import load_workbook
from urllib.parse import quote
from playwright.sync_api import sync_playwright

# ============================================================
# WARNING SUPPRESSION
# ============================================================
warnings.filterwarnings("ignore", category=UserWarning)
warnings.filterwarnings("ignore", category=DeprecationWarning)
os.environ["PYTHONWARNINGS"] = "ignore"

# ============================================================
# CONFIGURATION
# ============================================================
USERNAME = "canofy01"
PASSWORD = "Siddhartha#2026"

# ============================================================
# BULK INPUT EXCEL
# ============================================================

EXCEL_FILE = Path(__file__).parent / "request_list.xlsx"

BASE_URL = "https://www.sezonline-ndml.co.in/User/Login.aspx"

# ============================================================
# FRAPPE CONFIGURATION
# ============================================================

FRAPPE_URL = "https://eap.canofy.tech/"

API_KEY = "b51750272b9961b"
API_SECRET = "ad42019b8e8a820"
# -------------------------------------------------------------


# ============================================================
# DIRECTORIES & PATHS
# ============================================================
DOWNLOAD_DIR = Path.home() / "Downloads"
SCREENSHOT_DIR = Path(__file__).parent / "screenshots"

DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)
SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True)

# Initialize EasyOCR Reader ONCE (module load, not per captcha)
ocr_reader = easyocr.Reader(["en"], gpu=False)


# ============================================================
# SEZ ONLINE RATE-LIMIT HANDLING
# ============================================================
class RateLimitError(Exception):
"""Raised when SEZ Online reports that the request rate is limited."""
pass


class PrintStuckError(Exception):
"""
Raised when the SEZ Online print-preview interceptor hangs.

This is what was previously happening "near 98/99 downloads": the
portal's DOM/JS state degrades after many print operations in the
same page/tab and the print-preview popup silently stops firing.
Instead of failing the whole run, this is caught by the caller so a
hard page reset can be performed and the SAME request retried.
"""
pass


class CaptchaUnsolvedError(Exception):
"""Raised when the CAPTCHA could not be confidently read after retries."""
pass


class SessionExpiredError(Exception):
"""
Raised when the SEZ Online server-side session has timed out
(redirected to SessionTimeOut.aspx / "Your Session has expired").

This mostly happens during the long, deliberate `time.sleep()` pauses
(the 15-PDF cooldown and the rate-limit backoff) -- the browser page
sits idle for 2 minutes with no activity, which is long enough for
the portal's own inactivity timeout to log it out server-side. The
fix is reactive: detect it wherever it shows up, click "Here" (or
just re-navigate to the login page), run the full login flow again,
and resume the SAME request that was in progress.
"""
pass


RATE_LIMIT_WAIT_SECONDS = 120
PDF_PAUSE_15_SECONDS = 120

# Lowered from 100 -> 60. The portal's DOM/print-JS state was observed to
# degrade and freeze the print interceptor right around the 98-99 mark,
# i.e. comfortably before a 100-download refresh would ever trigger.
# Refreshing every 60 downloads keeps the page state well inside the
# window where printing is still reliable.
PDF_REFRESH_INTERVAL = 60

# How many times navigate_to_print is allowed to hard-reset-and-retry the
# SAME request after a PrintStuckError before that request is given up on
# and marked Failed. Prevents an infinite loop if the portal itself is down.
MAX_STUCK_RETRIES = 3

# How many times a SAME request is allowed to trigger a full re-login
# after a SessionExpiredError before it's given up on and marked Failed.
MAX_SESSION_RETRIES = 5

# ============================================================
# CAPTCHA CONFIGURATION
# ============================================================
# SEZ Online captchas are short alphanumeric strings. Anything shorter than
# this is almost certainly a misread (dropped/merged characters) rather
# than a real captcha value, so it is rejected and re-read/re-rolled
# instead of being submitted and wasting a whole login attempt.
CAPTCHA_MIN_LENGTH = 4
CAPTCHA_MAX_LENGTH = 8

# Number of times we'll re-screenshot + re-OCR + (if possible) refresh the
# captcha image before giving up on THIS login attempt and starting a
# completely fresh one (new page load = new captcha, new everything).
MAX_CAPTCHA_READ_ATTEMPTS = 6

# Number of full login attempts (fresh page load + fresh captcha each time)
# before we give up entirely and raise.
MAX_LOGIN_ATTEMPTS = 8


# ============================================================
# ============================================================
# ACTIVE IDLE (SESSION KEEP-ALIVE) PAUSE
# ============================================================
def idle_keepalive_pause(page, total_seconds: int, chunk_seconds: int = 5):
"""
Wait out a required cooldown (the 15-PDF pause or the rate-limit
backoff) WITHOUT letting the browser tab go completely idle.

A plain time.sleep(120) leaves the page with zero activity for two
full minutes -- which lines up exactly with when the SEZ Online
session was observed expiring mid-run. This performs small,
harmless mouse-wheel scrolls (down for a few seconds, then up,
repeating) so the tab registers ordinary user activity for the
whole cooldown, while still taking the full required duration and
making no extra network requests (so it can't trip the rate limit
itself). This is a best-effort measure, not a guarantee -- the
session-expiry detection/recovery elsewhere still runs immediately
afterward as the real safety net.
"""
print(f"Idling for {total_seconds}s with periodic keep-alive scrolling...")

elapsed = 0.0
direction = 1 # 1 = scroll down, -1 = scroll up

while elapsed < total_seconds:
chunk_start = time.time()
remaining = total_seconds - elapsed
this_chunk = min(chunk_seconds, remaining)

try:
steps = max(1, int(round(this_chunk)))
step_delta = (300 * direction) / steps
step_wait_ms = int((this_chunk / steps) * 1000)

for _ in range(steps):
page.mouse.wheel(0, step_delta)
page.wait_for_timeout(step_wait_ms)
except Exception:
# Page/tab not in a state where scrolling is possible right
# now -- fall back to a plain wait for this chunk. The
# session-expiry check that follows the full pause will
# catch and recover from an actual timeout regardless.
time.sleep(this_chunk)

direction *= -1
elapsed += (time.time() - chunk_start)

print("Keep-alive idling complete.")


# ============================================================
# SAFE FILE PATH RESOLVER (PREVENTS PERMISSIONERROR 13)
# ============================================================
def get_safe_pdf_path(target_dir: Path, req_id: str) -> Path:
"""Returns a writable file path, adding an increment suffix if the file is locked."""
base_path = target_dir / f"{req_id}.pdf"

if base_path.exists():
try:
with open(base_path, "a+b"):
pass
return base_path
except (PermissionError, IOError):
pass

counter = 1
while True:
alt_path = target_dir / f"{req_id}_{counter}.pdf"
if not alt_path.exists():
return alt_path
try:
with open(alt_path, "a+b"):
pass
return alt_path
except (PermissionError, IOError):
counter += 1


# ============================================================
# CAPTCHA ENGINE
# ============================================================
def is_valid_captcha_token(token: str) -> bool:
"""
Cheap sanity filter applied BEFORE a captcha guess is ever submitted.

Root cause of most "captcha failing" reports: a bad OCR read (empty
string, 1-2 stray characters, or a stray symbol) was being submitted
straight to the login form, burning a login attempt and forcing a
full page reload just to try again. Rejecting obviously-wrong reads
here lets us retry the CHEAP part (re-OCR / refresh the captcha image)
before ever touching the login form.
"""
if not token:
return False
if not token.isalnum():
return False
if not (CAPTCHA_MIN_LENGTH <= len(token) <= CAPTCHA_MAX_LENGTH):
return False
return True


def resolve_captcha(image_path: Path) -> str:
"""
Multi-variant OCR with confidence-weighted consensus voting.

Trimmed from 6 preprocessing variants down to the 4 that actually
contribute distinct, useful signal (bilateral-denoised, CLAHE-enhanced,
Otsu threshold, adaptive threshold). The dropped variants ("original"
color image and plain "gray") were consistently the lowest-confidence,
noisiest readings and mostly diluted the consensus vote rather than
improving it -- cutting them roughly halves per-captcha OCR time
without giving up accuracy.
"""
image = cv2.imread(str(image_path))

if image is None:
raise FileNotFoundError(f"Could not load CAPTCHA image from {image_path}")

resized = cv2.resize(image, None, fx=2.5, fy=2.5, interpolation=cv2.INTER_CUBIC)
gray = cv2.cvtColor(resized, cv2.COLOR_BGR2GRAY)

variants = {}

filtered = cv2.bilateralFilter(gray, 9, 75, 75)
variants["bilateral"] = filtered

clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
variants["enhanced"] = clahe.apply(filtered)

_, otsu = cv2.threshold(filtered, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
variants["otsu"] = otsu

variants["adaptive"] = cv2.adaptiveThreshold(
filtered, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 11, 2
)

allowlist = (
"ABCDEFGHIJKLMNOPQRSTUVWXYZ"
"abcdefghijklmnopqrstuvwxyz"
"0123456789"
)

candidates = []

for variant_name, variant_image in variants.items():
try:
results = ocr_reader.readtext(
variant_image, detail=1, paragraph=False, allowlist=allowlist
)
except Exception as exc:
print(f"OCR failed for {variant_name}: {exc}")
continue

if not results:
continue

try:
results = sorted(results, key=lambda item: min(point[0] for point in item[0]))
except Exception:
pass

raw_text = "".join(str(item[1]) for item in results if len(item) >= 2)
cleaned_text = re.sub(r"[^A-Za-z0-9]", "", raw_text)

if not cleaned_text:
continue

confidences = [float(item[2]) for item in results if len(item) >= 3]
avg_confidence = (sum(confidences) / len(confidences)) if confidences else 0.0

candidates.append({
"variant": variant_name,
"text": cleaned_text,
"confidence": avg_confidence,
})

if not candidates:
return ""

# Group exact matches -- strong agreement across preprocessing methods
# is the best evidence of a correct read.
grouped = {}
for candidate in candidates:
token = candidate["text"]
bucket = grouped.setdefault(token, {"text": token, "confidence_sum": 0.0, "count": 0})
bucket["confidence_sum"] += candidate["confidence"]
bucket["count"] += 1

grouped_candidates = list(grouped.values())
for candidate in grouped_candidates:
candidate["average_confidence"] = candidate["confidence_sum"] / candidate["count"]
candidate["score"] = candidate["average_confidence"] + (0.05 * (candidate["count"] - 1))

grouped_candidates.sort(key=lambda item: item["score"], reverse=True)
best = grouped_candidates[0]

# Case-aware, confidence-weighted per-character voting within the
# winning (case-insensitive) group.
case_groups = {}
for candidate in candidates:
key = candidate["text"].lower()
case_groups.setdefault(key, []).append(candidate)

compatible_group = case_groups.get(best["text"].lower(), [])
best_text = best["text"]

if compatible_group:
max_length = max(len(item["text"]) for item in compatible_group)
final_chars = []

for index in range(max_length):
char_votes = {}
for candidate in compatible_group:
text = candidate["text"]
if index >= len(text):
continue
char = text[index]
char_votes[char] = char_votes.get(char, 0.0) + candidate["confidence"]

if char_votes:
final_chars.append(max(char_votes, key=char_votes.get))

consensus_text = "".join(final_chars)
if consensus_text:
best_text = consensus_text

cleaned_token = re.sub(r"[^A-Za-z0-9]", "", best_text)

print(f"CAPTCHA OCR candidates: {[(c['variant'], c['text'], round(c['confidence'], 3)) for c in candidates]}")
print(f"CAPTCHA consensus token: '{cleaned_token}'")

return cleaned_token


def try_refresh_captcha_image(page) -> bool:
"""
Attempt to click a "refresh captcha" control so a NEW captcha image is
generated without losing the username/password already typed in.

SEZ Online (and most ASP.NET captcha widgets) usually expose a small
refresh icon/link next to the captcha image. We probe a handful of
common selector patterns; if none exist we fall back to clicking the
captcha image itself (many widgets rotate on click), and if that also
does nothing the caller will fall back to a full page reload instead.
"""
candidate_selectors = [
"#captchaControlRevamp_btnRefreshCaptcha",
"#captchaControlRevamp_imgRefresh",
"a[id*='Refresh' i][id*='aptcha' i]",
"img[id*='Refresh' i][id*='aptcha' i]",
"a[href*='refresh' i][id*='aptcha' i]",
"[title*='refresh' i][id*='aptcha' i]",
"[title*='reload' i][id*='aptcha' i]",
]

for selector in candidate_selectors:
try:
control = page.locator(selector).first
if control.count() > 0 and control.is_visible():
control.click(timeout=3000)
page.wait_for_timeout(500)
return True
except Exception:
continue

return False


def get_captcha_token(page, captcha_image_locator) -> str:
"""
Screenshot + OCR the captcha with validation and self-refresh retries.

This is the fix for "captcha reading fails and the script doesn't
retry": previously a single OCR pass was trusted no matter what. Now:
1. Screenshot the current captcha image.
2. Run the multi-variant OCR + consensus voting.
3. If the result looks structurally valid (right length, alnum),
use it immediately.
4. If not, try to refresh the captcha image in place and re-OCR,
up to MAX_CAPTCHA_READ_ATTEMPTS times.
5. If nothing valid ever comes back, return the best (highest
confidence / longest) guess we saw rather than an empty string,
so the outer login-retry loop still has something to try.
"""
raw_captcha_path = SCREENSHOT_DIR / "raw_captcha.png"
best_fallback = ""

for attempt in range(1, MAX_CAPTCHA_READ_ATTEMPTS + 1):
try:
captcha_image_locator.wait_for(state="visible", timeout=15000)
captcha_image_locator.screenshot(path=str(raw_captcha_path))
except Exception as exc:
print(f"Captcha screenshot attempt {attempt} failed: {exc}")
continue

token = resolve_captcha(raw_captcha_path)

if token and len(token) > len(best_fallback):
best_fallback = token

if is_valid_captcha_token(token):
print(f"Captcha read accepted on attempt {attempt}: '{token}'")
return token

print(
f"Captcha read attempt {attempt}/{MAX_CAPTCHA_READ_ATTEMPTS} "
f"rejected (raw='{token}'). Trying to get a fresh captcha image..."
)

refreshed = try_refresh_captcha_image(page)
if not refreshed:
# No refresh control found/worked. Re-screenshotting the same
# static image would just repeat the same failure, so stop
# this inner loop early and let the caller reload the whole
# page for a genuinely new captcha.
break

if best_fallback:
print(f"No structurally valid captcha read; using best guess: '{best_fallback}'")
return best_fallback

raise CaptchaUnsolvedError("Could not obtain any usable CAPTCHA reading.")


# ============================================================
# PERMANENT POPUP & OVERLAY IMMUNIZATION
# ============================================================
def immunize_page_against_popups(context):
"""Permanently disables driver.js and injects aggressive style blockades."""
context.add_init_script("""
window.Driver = function() {
return {
start: function() {},
reset: function() {},
destroy: function() {},
defineSteps: function() {},
showNextStep: function() {},
preventMove: function() {}
};
};

const style = document.createElement('style');
style.innerHTML = `
.driver-overlay, .driver-popover, .driver-popover-tip,
.driver-popover-footer, .driver-popover-description,
.driver-popover-title, .driver-stage-no-animation,
.modal-backdrop, .overlay {
display: none !important;
visibility: hidden !important;
pointer-events: none !important;
width: 0 !important;
height: 0 !important;
opacity: 0 !important;
}
body, html {
overflow: auto !important;
pointer-events: auto !important;
}
`;
document.head ? document.head.appendChild(style) : document.addEventListener('DOMContentLoaded', () => document.head.appendChild(style));
""")


def clean_popups_now(page):
"""Runtime cleanup for elements already present in the active DOM."""
try:
page.evaluate("""
document.querySelectorAll('.driver-overlay, .driver-popover, .driver-popover-tip, .driver-popover-footer, .modal-backdrop, .overlay').forEach(el => el.remove());
if (document.body) {
document.body.classList.remove('driver-active', 'driver-fade');
document.body.style.overflow = 'auto';
document.body.style.pointerEvents = 'auto';
}
if (document.documentElement) {
document.documentElement.classList.remove('driver-active', 'driver-fade');
document.documentElement.style.overflow = 'auto';
}
""")
except Exception:
pass


def safe_click(page, locator_or_selector):
"""Click a portal control reliably without synthetic click fallback."""
clean_popups_now(page)

loc = (
locator_or_selector
if hasattr(locator_or_selector, "click")
else page.locator(locator_or_selector)
)

try:
loc.scroll_into_view_if_needed(timeout=5000)
except Exception:
pass

try:
loc.click(timeout=10000, force=True)
except Exception as first_exc:
try:
loc.click(timeout=10000)
except Exception as second_exc:
raise RuntimeError(f"Could not click portal control: {second_exc}") from first_exc


def check_rate_limit(page) -> None:
"""
Detect the SEZ Online server-side rate-limit page/message AND the
server-side session-timeout redirect (SessionTimeOut.aspx / "Your
Session has expired"). Both are checked here, in one place, so every
one of the many call sites throughout navigate_to_print automatically
gets session-expiry detection for free without needing individual edits.
"""
try:
current_url = page.url
except Exception:
current_url = ""

if "SessionTimeOut.aspx" in current_url or "sessionexpired=true" in current_url.lower():
raise SessionExpiredError(
"SEZ Online session has expired (redirected to SessionTimeOut.aspx)."
)

try:
body_text = page.locator("body").inner_text(timeout=2000)
except Exception:
return

normalized = re.sub(r"\s+", " ", body_text or "").strip().lower()

if "session has expired" in normalized or "session expired" in normalized:
raise SessionExpiredError(
"SEZ Online session has expired ('Your Session has expired' banner detected)."
)

if "rate limit exceeded" in normalized or "try again in 120 seconds" in normalized:
raise RateLimitError(
"SEZ Online rate limit exceeded. The portal requested a retry after 120 seconds."
)


def recover_from_session_expiry(page):
"""
Recover from a detected SessionExpiredError: click the "Here" link on
the SessionTimeOut.aspx page if present (matches the exact banner
"Your Session has expired. Click Here to login again"), then run a
full fresh login regardless, since the "Here" link just routes back
to Login.aspx and a new captcha is needed either way.
"""
print()
print("=" * 60)
print("SESSION EXPIRED — RE-AUTHENTICATING")
print("=" * 60)
print("Current URL:", page.url)

try:
here_link = page.get_by_role("link", name="Here")
if here_link.count() > 0:
print("Clicking 'Here' to return to login...")
here_link.first.click(timeout=5000)
page.wait_for_load_state("domcontentloaded", timeout=15000)
page.wait_for_timeout(500)
except Exception as exc:
print(f"Could not click 'Here' link (continuing anyway): {exc}")

# login() always navigates to BASE_URL itself, so it recovers cleanly
# even if the "Here" click above did nothing.
login(page)

print("Re-authentication after session expiry: SUCCESSFUL.")
print("=" * 60)


def login_failed_reason(page) -> str:
"""
Inspect the page after a login submit and return a short reason string
if the login did NOT succeed (still on Login.aspx or an error banner is
present), or an empty string if login looks successful.
"""
if "Login.aspx" in page.url:
try:
body_text = page.locator("body").inner_text(timeout=2000)
except Exception:
body_text = ""

normalized = re.sub(r"\s+", " ", body_text or "").strip().lower()

if "invalid captcha" in normalized or "captcha" in normalized and "invalid" in normalized:
return "invalid captcha"
if "invalid" in normalized and ("username" in normalized or "password" in normalized):
return "invalid credentials"
return "still on login page"

return ""


# ============================================================
# LOGIN FLOW
# ============================================================
def login(page):
"""
Full login flow with automatic captcha + login retries.

Root cause fix: previously a single bad captcha read meant a failed
login with NO retry logic at all -- the script would just proceed
(and every subsequent step would fail against the still-on-Login.aspx
page). Now every login attempt is verified, and on failure we reload
the login page fresh (new captcha, clean state) and try again, up to
MAX_LOGIN_ATTEMPTS times.
"""
print()
print("=" * 60)
print("SEZ ONLINE LOGIN (AUTOMATED OCR)")
print("=" * 60)

for attempt in range(1, MAX_LOGIN_ATTEMPTS + 1):
print(f"\n--- Login attempt {attempt}/{MAX_LOGIN_ATTEMPTS} ---")

print("Opening login page...")
page.goto(BASE_URL, wait_until="domcontentloaded")
clean_popups_now(page)

username_input = page.get_by_role("textbox", name="User Name")
password_input = page.get_by_role("textbox", name="Password")
captcha_input = page.get_by_role("textbox", name="Captcha")
login_button = page.get_by_role("button", name="Login")
captcha_image = page.locator("#captchaControlRevamp_imgCaptcha")

print("Entering username & password...")
username_input.wait_for(state="visible", timeout=30000)
username_input.fill(USERNAME)
password_input.fill(PASSWORD)

try:
captcha_token = get_captcha_token(page, captcha_image)
except CaptchaUnsolvedError as exc:
print(f"Captcha could not be solved this attempt: {exc}")
continue # fresh page load, fresh captcha, try again

print(f"Entering extracted CAPTCHA: '{captcha_token}'")
captcha_input.fill(captcha_token)

print("Submitting login form...")
clean_popups_now(page)
safe_click(page, login_button)

try:
page.wait_for_load_state("domcontentloaded", timeout=15000)
except Exception:
pass

page.wait_for_timeout(1500)
clean_popups_now(page)
print("Login submission completed. Current URL:", page.url)

# Handle ConfirmUserCredentials screen if redirected there
if "ConfirmUserCredentials.aspx" in page.url:
print("Handling ConfirmUserCredentials screen...")
try:
confirm_btn = page.locator(
"input[type='submit'][value*='Continue'], "
"input[type='submit'][value*='OK'], "
"#btnSubmit, #btnConfirm"
)
if confirm_btn.count() > 0:
safe_click(page, confirm_btn.first)
page.wait_for_load_state("domcontentloaded", timeout=10000)
page.wait_for_timeout(1000)
except Exception:
pass

failure_reason = login_failed_reason(page)
if not failure_reason:
print("Login SUCCESSFUL.")
return

print(f"Login attempt {attempt} failed ({failure_reason}). Retrying with a fresh captcha...")

raise RuntimeError(
f"Could not log in to SEZ Online after {MAX_LOGIN_ATTEMPTS} attempts "
"(captcha and/or credential failures)."
)


# ============================================================
# PRINT/PAGE STATE RESET
# ============================================================
def reset_portal_page(context, page):
"""
Hard-reset the authenticated portal page. Closes stray popup/print
pages, reloads, then replaces the Playwright Page object entirely
(new Page in the same BrowserContext => same login session/cookies,
but zero accumulated JS/DOM/print state). Called both on the periodic
PDF_REFRESH_INTERVAL schedule and reactively when a PrintStuckError
is detected.
"""
print()
print("=" * 60)
print("RESETTING SEZ ONLINE PAGE STATE")
print("=" * 60)

for extra_page in list(context.pages):
if extra_page != page:
try:
print("Closing stale popup/print page:", extra_page.url)
except Exception:
pass
try:
extra_page.close()
except Exception:
pass

current_url = page.url
print("Refreshing current portal page...")
try:
page.reload(wait_until="domcontentloaded", timeout=30000)
except Exception as reload_exc:
print("DOM/page reload warning:", reload_exc)

try:
page.wait_for_timeout(2000)
clean_popups_now(page)
except Exception:
pass

print("Creating a fresh Playwright page in the same session...")
fresh_page = context.new_page()

try:
fresh_page.goto(current_url, wait_until="domcontentloaded", timeout=30000)
except Exception as goto_exc:
try:
fresh_page.close()
except Exception:
pass
raise RuntimeError(
f"Could not recreate SEZ Online page after batch refresh: {goto_exc}"
) from goto_exc

try:
fresh_page.wait_for_timeout(2000)
clean_popups_now(fresh_page)
check_rate_limit(fresh_page)
except RateLimitError:
try:
fresh_page.close()
except Exception:
pass
raise
except Exception:
pass

try:
page.close()
except Exception:
pass

for extra_page in list(context.pages):
if extra_page != fresh_page:
try:
extra_page.close()
except Exception:
pass

print("Fresh authenticated SEZ Online page is ready.")
print("Current URL:", fresh_page.url)
print("=" * 60)

return fresh_page


# ============================================================
# ============================================================
# STRAY / LEFTOVER MODAL RECOVERY
# ============================================================
def dismiss_stray_dialog(page) -> bool:
"""
Detect and forcibly close a leftover print-options dialog ("Please
Select The Type Of Copies You Want To Print") or any other jQuery
UI / Bootstrap-style modal that failed to close on a PREVIOUS
request.

Root cause of the "randomly stuck on this screen" symptom: when
something about the print-options dialog interaction fails
unexpectedly (slow render, wrong frame, a stray double-open, etc.),
the exception raised was a plain, unhandled one -- it was NOT a
RateLimitError / PrintStuckError / SessionExpiredError, so it
skipped straight past the retry loop for that request and landed in
the outer per-row handler, which just marked the row Failed and
moved to the next row WITHOUT ever cleaning up the page. The modal
was left open, silently swallowing every click on the next request
(that's why the log shows a timeout waiting for the date-picker
field to be visible -- the "Search Request" click never actually
reached the page underneath the overlay). Every request after that
point failed identically, looking like the automation was "stuck".

This is called proactively at the start of every navigate_to_print
AND reactively from the per-row failure handler in main(), so a
stuck dialog from one request can never permanently break the rest
of the run.
"""
closed_any = False

# Case 1: dialog rendered inside the #dialog-body iframe (what the
# normal print flow expects).
try:
frame = page.frame_locator("#dialog-body")
close_control = frame.locator(
"xpath=//*[self::a or self::img or self::button or self::span]"
"[contains(@class,'close') or contains(@title,'Close') or contains(@aria-label,'Close')]"
).first
if close_control.count() > 0 and close_control.is_visible():
close_control.click(timeout=3000)
closed_any = True
except Exception:
pass

# Case 2: a plain (non-iframe) modal directly on the main page --
# some renders of the "type of copies" prompt aren't in an iframe.
try:
plain_close = page.locator(
"xpath=//div[contains(@class,'ui-dialog') or contains(@class,'modal') or contains(@id,'dialog')]"
"//*[self::a or self::span or self::button]"
"[contains(@class,'close') or contains(@class,'ui-dialog-titlebar-close') or contains(@aria-label,'Close')]"
).first
if plain_close.count() > 0 and plain_close.is_visible():
plain_close.click(timeout=3000)
closed_any = True
except Exception:
pass

# Last resort: Escape closes most jQuery UI / Bootstrap modals
# regardless of the exact markup, without needing to know it.
if not closed_any:
try:
page.keyboard.press("Escape")
except Exception:
pass

try:
page.wait_for_timeout(500)
clean_popups_now(page)
except Exception:
pass

return closed_any


# ============================================================
# NAVIGATE TO PRINT & EXPORT PDF
# ============================================================
def navigate_to_print(page, req_id: str, target_dir: Path) -> str:
for extra_page in list(page.context.pages):
if extra_page != page:
try:
print("Closing stale print/popup page before request:", extra_page.url)
except Exception:
pass
try:
extra_page.close()
except Exception:
pass

# Proactively clear any dialog left open by a previous, failed
# request BEFORE attempting any navigation for this one.
dismiss_stray_dialog(page)

print()
print("=" * 60)
print("NAVIGATE TO PRINT")
print("=" * 60)
print("Target Request ID:", req_id)

check_rate_limit(page)
clean_popups_now(page)

try:
page.wait_for_load_state("domcontentloaded", timeout=15000)
except Exception:
pass

check_rate_limit(page)

print("Opening Search Request...")
search_request_button = page.get_by_role("link", name="Search Request")
search_request_button.wait_for(state="visible", timeout=30000)
check_rate_limit(page)
safe_click(page, search_request_button)

try:
page.wait_for_load_state("domcontentloaded", timeout=15000)
except Exception:
pass
page.wait_for_timeout(800)
clean_popups_now(page)
check_rate_limit(page)

print("Clearing date filters...")
from_date_input = page.locator("#ctl00_DefaultContent_DatePicker_RequestFromDate_txtMaskedDate")
to_date_input = page.locator("#ctl00_DefaultContent_DatePicker_RequestToDate_txtMaskedDate")

from_date_input.wait_for(state="visible", timeout=15000)
to_date_input.wait_for(state="visible", timeout=15000)
check_rate_limit(page)

from_date_input.evaluate("(el) => el.value = ''")
to_date_input.evaluate("(el) => el.value = ''")

print(f"Entering Request ID: {req_id}...")
request_id_textfield = page.locator("#ctl00_DefaultContent_txtRequestID")
request_id_textfield.wait_for(state="visible", timeout=15000)
request_id_textfield.fill(req_id)

print("Searching...")
search_button = page.get_by_role("button", name="Search")
search_button.wait_for(state="visible", timeout=15000)
check_rate_limit(page)
safe_click(page, search_button)

page.wait_for_timeout(1500)
clean_popups_now(page)
check_rate_limit(page)

print("Opening request details...")
req_id_link = page.locator(f"xpath=//a[normalize-space(text())='{req_id}']")
req_id_link.wait_for(state="visible", timeout=30000)
check_rate_limit(page)
safe_click(page, req_id_link)

try:
page.wait_for_load_state("domcontentloaded", timeout=15000)
except Exception:
pass
page.wait_for_timeout(1200)
clean_popups_now(page)
check_rate_limit(page)

print("Clicking View / Print...")
view_receipt = page.locator("#ctl00_DefaultContent_btnPrintBOE")
view_receipt.wait_for(state="visible", timeout=30000)
check_rate_limit(page)
safe_click(page, view_receipt)

page.wait_for_timeout(800)
clean_popups_now(page)
check_rate_limit(page)

print("Handling Print Modal iframe...")
print_frame = page.frame_locator("#dialog-body")

original_checkbox = print_frame.locator("#cbNo_0")
original_checkbox.wait_for(state="visible", timeout=15000)
original_checkbox.click()
print("Selected 'Original' document option.")

ok_button = print_frame.locator("xpath=//input[@value='OK']")
ok_button.wait_for(state="visible", timeout=15000)

check_rate_limit(page)

print("Triggering print preview window...")
context = page.context
print_page = None

pages_before_print = set(context.pages)

try:
with context.expect_page(timeout=30000) as page_info:
ok_button.click()

print_page = page_info.value
print("Print window intercepted successfully.")

except Exception as exc:
new_pages = [c for c in context.pages if c not in pages_before_print]

if new_pages:
print_page = new_pages[-1]
print("Print window found after interceptor timeout; using it.")
else:
try:
check_rate_limit(page)
except RateLimitError:
raise

for candidate in list(context.pages):
if candidate != page:
try:
candidate.close()
except Exception:
pass

# This is the "stuck at 98/99" symptom: the print interceptor
# never fires and no new page ever appears. Raising a
# dedicated exception lets the caller do a hard page reset
# and retry, instead of treating this the same as an
# ordinary one-off failure.
raise PrintStuckError(
f"Print preview window was not opened for request {req_id} "
f"within 30 seconds (portal DOM/print state likely frozen): {exc}"
) from exc

try:
print_page.wait_for_load_state("domcontentloaded", timeout=15000)
except Exception:
print(
"Print page did not reach DOMContentLoaded within the timeout; "
"continuing because the page was already intercepted."
)

print_page.wait_for_timeout(1500)

resolved_pdf_path = get_safe_pdf_path(target_dir, req_id)

print(f"Generating PDF at: {resolved_pdf_path}...")
try:
print_page.pdf(path=str(resolved_pdf_path), print_background=True, landscape=False)
finally:
try:
print_page.close()
except Exception:
pass

if not resolved_pdf_path.exists() or resolved_pdf_path.stat().st_size == 0:
raise RuntimeError(f"PDF creation failed or output file is empty: {resolved_pdf_path}")

print()
print("=" * 60)
print("PDF CREATED SUCCESSFULLY")
print("=" * 60)
print("Location:", resolved_pdf_path)
print("File Size:", resolved_pdf_path.stat().st_size, "bytes")

return str(resolved_pdf_path)


# ============================================================
# SEZ PORTAL LOGOUT
# ============================================================
def logout(page):
print()
print("=" * 60)
print("LOGGING OUT FROM SEZ PORTAL")
print("=" * 60)

clean_popups_now(page)

logout_button = page.locator("#ctl00_LoginStatus")
logout_button.wait_for(state="visible", timeout=30000)

print("Clicking Logout...")
safe_click(page, logout_button)

try:
page.wait_for_load_state("domcontentloaded", timeout=15000)
except Exception:
pass

page.wait_for_timeout(1000)

print("Successfully Logged out")
print("Current URL:", page.url)


def read_request_excel():
print()
print("=" * 60)
print("READING BULK REQUEST EXCEL")
print("=" * 60)

if not EXCEL_FILE.exists():
raise FileNotFoundError(f"Excel file not found:\n{EXCEL_FILE}")

workbook = load_workbook(EXCEL_FILE)
sheet = workbook.active

headers = {}
for cell in sheet[1]:
if cell.value is not None:
headers[str(cell.value).strip()] = cell.column

required_columns = {"Request ID", "Doctype Name"}
missing_columns = required_columns - set(headers.keys())

if missing_columns:
raise RuntimeError("Missing required Excel columns: " + ", ".join(sorted(missing_columns)))

requests_list = []

for row_number in range(2, sheet.max_row + 1):
request_id = sheet.cell(row=row_number, column=headers["Request ID"]).value
doctype_name = sheet.cell(row=row_number, column=headers["Doctype Name"]).value

if request_id is None:
continue

request_id = str(request_id).strip()
if not request_id:
continue

if doctype_name is None:
raise RuntimeError(f"Doctype Name is empty for Excel row {row_number}.")

doctype_name = str(doctype_name).strip()

requests_list.append({
"row_number": row_number,
"request_id": request_id,
"doctype_name": doctype_name,
})

print("Total requests found:", len(requests_list))
for item in requests_list:
print(f"Row {item['row_number']} | Request ID: {item['request_id']} | Doctype: {item['doctype_name']}")

return workbook, sheet, headers, requests_list


def update_excel_row(
sheet, headers, row_number,
download_status=None, timestamp=None, filename=None,
upload_status=None, upload_timestamp=None, upload_response=None,
):
if download_status is not None:
sheet.cell(row=row_number, column=headers["Download Status"]).value = download_status
if timestamp is not None:
sheet.cell(row=row_number, column=headers["Timestamp"]).value = timestamp
if filename is not None:
sheet.cell(row=row_number, column=headers["Filename"]).value = filename
if upload_status is not None:
sheet.cell(row=row_number, column=headers["Upload Status"]).value = upload_status
if upload_timestamp is not None:
sheet.cell(row=row_number, column=headers["Upload Timestamp"]).value = upload_timestamp
if upload_response is not None:
sheet.cell(row=row_number, column=headers["Upload response"]).value = upload_response


# ============================================================
# MAIN ENTRY POINT
# ============================================================
def main():
print()
print("=" * 60)
print("SEZ PRINT AUTOMATION - BULK RUNNER")
print("=" * 60)

workbook, sheet, headers, request_rows = read_request_excel()

downloaded_items = []

with sync_playwright() as p:
browser = p.chromium.launch(
headless=False,
args=["--start-maximized", "--no-sandbox"],
)

context = browser.new_context(
no_viewport=True,
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
),
)

immunize_page_against_popups(context)

def auto_accept_dialogs(dialog):
print(f"\n[Dialog Intercepted] {dialog.message}")
dialog.accept()

context.on("dialog", auto_accept_dialogs)

page = context.new_page()

try:
# STEP 1: LOGIN ONCE (with internal captcha/login retries)
login(page)

# STEP 2: DOWNLOAD ALL PDFs
print()
print("=" * 60)
print("STARTING BULK PDF DOWNLOAD")
print("=" * 60)

# ========================================================
# BULK DOWNLOAD THROTTLING
# ========================================================
# - Every 15 successful PDFs -> wait 120 seconds.
# - Every 60 successful PDFs -> hard-reset the portal
# DOM/page (lowered from 100 to head off the print-
# interceptor freeze observed near 98/99 downloads).
# - A refresh-interval PDF does NOT also get the 15-PDF
# pause on the same download.
# - A frozen print interceptor (PrintStuckError) triggers
# an immediate reactive reset + retry of the SAME
# request, independent of the periodic schedule.
# ========================================================
download_count = 0

for item in request_rows:
row_number = item["row_number"]
request_id = item["request_id"]
doctype_name = item["doctype_name"]

print()
print("#" * 60)
print("PROCESSING REQUEST")
print("#" * 60)
print("Excel Row :", row_number)
print("Request ID :", request_id)
print("Doctype Name :", doctype_name)

try:
stuck_retries = 0
session_retries = 0

while True:
try:
check_rate_limit(page)
saved_pdf = navigate_to_print(page, request_id, DOWNLOAD_DIR)
break

except SessionExpiredError as session_exc:
session_retries += 1
print()
print("=" * 60)
print("SEZ ONLINE SESSION EXPIRED")
print("=" * 60)
print(f"Request ID : {request_id}")
print(f"Reason : {session_exc}")
print(f"Re-login attempt {session_retries}/{MAX_SESSION_RETRIES}")

if session_retries > MAX_SESSION_RETRIES:
raise RuntimeError(
f"Session kept expiring for {request_id} "
f"after {MAX_SESSION_RETRIES} re-login attempts."
) from session_exc

recover_from_session_expiry(page)
print(f"Retrying SAME request after re-login: {request_id}")
continue

except RateLimitError as rate_exc:
print()
print("=" * 60)
print("SEZ ONLINE RATE LIMIT DETECTED")
print("=" * 60)
print(f"Request ID : {request_id}")
print(f"Reason : {rate_exc}")
print(f"Waiting {RATE_LIMIT_WAIT_SECONDS} seconds before retrying the SAME request...")

idle_keepalive_pause(page, RATE_LIMIT_WAIT_SECONDS)

print("Refreshing SEZ Online page...")
try:
page.reload(wait_until="domcontentloaded", timeout=30000)
except Exception as reload_exc:
print("Page reload warning:", reload_exc)

page.wait_for_timeout(1500)
clean_popups_now(page)

print(f"Retrying SAME request from the beginning: {request_id}")
continue

except PrintStuckError as stuck_exc:
stuck_retries += 1
print()
print("=" * 60)
print("PRINT INTERCEPTOR FROZEN — HARD RESET")
print("=" * 60)
print(f"Request ID : {request_id}")
print(f"Reason : {stuck_exc}")
print(f"Reset attempt {stuck_retries}/{MAX_STUCK_RETRIES}")

if stuck_retries > MAX_STUCK_RETRIES:
raise RuntimeError(
f"Print interceptor stayed frozen for {request_id} "
f"after {MAX_STUCK_RETRIES} hard resets."
) from stuck_exc

page = reset_portal_page(page.context, page)
print(f"Retrying SAME request after hard reset: {request_id}")
continue

update_excel_row(
sheet, headers, row_number,
download_status="Success",
timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
filename=Path(saved_pdf).name,
)
workbook.save(EXCEL_FILE)

downloaded_items.append({
"row_number": row_number,
"request_id": request_id,
"doctype_name": doctype_name,
"pdf_path": saved_pdf,
})

download_count += 1

print()
print(f"PDF download successful for {request_id}")
print("PDF:", saved_pdf)
print(f"Successful PDFs downloaded: {download_count}")

# EVERY 60 PDFs: proactive hard reset (no sleep, just
# DOM/state hygiene) to stay ahead of the freeze point.
if download_count % PDF_REFRESH_INTERVAL == 0:
print()
print("=" * 60)
print(f"{PDF_REFRESH_INTERVAL} PDFs DOWNLOADED SUCCESSFULLY")
print("PROACTIVELY REFRESHING SEZ ONLINE DOM + PAGE...")
print("=" * 60)

page = reset_portal_page(page.context, page)

print()
print("=" * 60)
print("PROACTIVE PAGE RESET COMPLETED")
print("Fresh authenticated page is active. Resuming...")
print("=" * 60)

# EVERY 15 PDFs: rate-limit cooldown pause. Skipped on
# the same download that just got the interval reset.
elif download_count % 15 == 0:
print()
print("=" * 60)
print("15 PDFs DOWNLOADED SUCCESSFULLY")
print(f"Pausing for {PDF_PAUSE_15_SECONDS} seconds...")
print("=" * 60)

idle_keepalive_pause(page, PDF_PAUSE_15_SECONDS)

# The browser page sits fully idle for the whole
# 2-minute pause above -- long enough for SEZ
# Online's server-side inactivity timeout to log
# the session out without the browser noticing.
# Proactively reload + check here so an expired
# session is caught and re-authenticated right
# away, rather than surfacing as a confusing
# failure partway through the NEXT request.
try:
page.reload(wait_until="domcontentloaded", timeout=30000)
except Exception as reload_exc:
print("Post-cooldown reload warning:", reload_exc)

page.wait_for_timeout(1000)
clean_popups_now(page)

try:
check_rate_limit(page)
except SessionExpiredError:
recover_from_session_expiry(page)
except RateLimitError:
# Will be handled reactively when the next
# request hits navigate_to_print.
pass

print()
print("=" * 60)
print("15-PDF COOLDOWN COMPLETED. Resuming...")
print("=" * 60)

except Exception as exc:
print()
print(f"PDF DOWNLOAD FAILED for {request_id}")
print(f"{type(exc).__name__}: {exc}")

update_excel_row(
sheet, headers, row_number,
download_status="Failed",
timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
)
workbook.save(EXCEL_FILE)

# ------------------------------------------------
# PAGE RECOVERY BEFORE MOVING ON
# ------------------------------------------------
# This is the fix for the "randomly stuck on this
# screen" symptom: an unexpected failure here (an
# unhandled TimeoutError, a print dialog that never
# closed, etc.) used to just mark the row Failed and
# fall straight through to the next row with the
# page left in whatever broken/blocked state caused
# the failure -- so every remaining row then failed
# identically. Always attempt to leave the page
# clean before continuing: dismiss any leftover
# dialog, check for rate-limit/session-expiry, and
# fall back to a full hard reset if the page still
# isn't usable.
try:
dismiss_stray_dialog(page)
check_rate_limit(page)
except SessionExpiredError:
try:
recover_from_session_expiry(page)
except Exception as recovery_exc:
print(f"Re-login after failure also failed, hard-resetting page: {recovery_exc}")
try:
page = reset_portal_page(page.context, page)
except Exception as reset_exc:
print(f"Page reset also failed: {reset_exc}")
except RateLimitError:
pass
except Exception as cleanup_exc:
print(f"Light recovery after failure didn't confirm a clean page ({cleanup_exc}); hard-resetting...")
try:
page = reset_portal_page(page.context, page)
except Exception as reset_exc:
print(f"Page reset also failed: {reset_exc}")

continue

print()
print("=" * 60)
print("ALL PDF DOWNLOADS COMPLETED")
print("=" * 60)

logout(page)

except Exception as exc:
print()
print("=" * 60)
print("SEZ PORTAL PIPELINE FAILED")
print("=" * 60)
print(f"{type(exc).__name__}: {exc}")

failure_path = SCREENSHOT_DIR / "failure.png"
page.screenshot(path=str(failure_path), full_page=True)
print("Saved failure diagnostic screenshot to:", failure_path)

raise

finally:
context.close()
browser.close()

# ========================================================
# STEP 4: FRAPPE API PROCESSING
# ========================================================
print()
print("=" * 60)
print("STARTING FRAPPE BULK PROCESSING")
print("=" * 60)

for item in downloaded_items:
row_number = item["row_number"]
request_id = item["request_id"]
doctype_name = item["doctype_name"]
pdf_path = item["pdf_path"]

print()
print("#" * 60)
print("FRAPPE PROCESSING")
print("#" * 60)
print("Request ID :", request_id)
print("Doctype :", doctype_name)
print("PDF :", pdf_path)

try:
frappe_result = process_pdf_in_frappe(request_id, doctype_name, pdf_path)

if frappe_result.get("already_attached", False):
update_excel_row(
sheet, headers, row_number,
upload_status="Already Attached",
upload_timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
upload_response=(
f"PDF already attached | Document: {frappe_result['frappe_docname']} | "
f"System Doc No: {frappe_result['system_doc_no']} | PDF: {Path(pdf_path).name}"
),
)
print()
print("=" * 60)
print("PDF ALREADY ATTACHED")
print("=" * 60)
print("Request ID :", request_id)
print("Frappe Document:", frappe_result["frappe_docname"])
print("System Doc No. :", frappe_result["system_doc_no"])
print("Upload Status :", "Already Attached")
else:
update_excel_row(
sheet, headers, row_number,
upload_status="Success",
upload_timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
upload_response=(
f"PDF attached successfully | Document: {frappe_result['frappe_docname']} | "
f"System Doc No: {frappe_result['system_doc_no']} | PDF: {Path(pdf_path).name}"
),
)
print()
print("=" * 60)
print("PDF ATTACHED SUCCESSFULLY")
print("=" * 60)
print("Request ID :", request_id)
print("Frappe Document:", frappe_result["frappe_docname"])
print("System Doc No. :", frappe_result["system_doc_no"])
print("Upload Status :", "Success")

workbook.save(EXCEL_FILE)

except Exception as exc:
print()
print(f"FRAPPE PROCESSING FAILED FOR {request_id}")
print(f"{type(exc).__name__}: {exc}")

update_excel_row(
sheet, headers, row_number,
upload_status="Failed – Frappe document mismatch",
upload_timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
upload_response=str(exc),
)
workbook.save(EXCEL_FILE)
print("Upload Status:", "Failed – Frappe document mismatch")
continue

workbook.save(EXCEL_FILE)

print()
print("=" * 60)
print("BULK PROCESSING COMPLETED")
print("=" * 60)
print("Updated Excel:", EXCEL_FILE)


# ============================================================
# FRAPPE API HEADERS
# ============================================================
def get_frappe_headers():
return {
"Authorization": f"token {API_KEY}:{API_SECRET}",
"Accept": "application/json",
}


def find_frappe_document(request_id: str, expected_doctype: str):
print()
print("=" * 60)
print("FINDING FRAPPE DOCUMENT")
print("=" * 60)

request_id = str(request_id).strip()
expected_doctype = str(expected_doctype).strip()

if not API_KEY or not API_SECRET:
raise RuntimeError("Frappe API_KEY and API_SECRET are not configured.")

print("Excel Request ID:", request_id)
print("Excel Doctype:", expected_doctype)

encoded_doctype = quote(expected_doctype, safe="")
url = f"{FRAPPE_URL.rstrip('/')}/api/resource/{encoded_doctype}"

params = {
"filters": json.dumps([[expected_doctype, "request_id", "=", request_id]]),
"fields": json.dumps(["name", "request_id", "docstatus"]),
"limit_page_length": 100,
}

response = requests.get(url, headers=get_frappe_headers(), params=params, timeout=120)
print("Frappe search status:", response.status_code)

if not response.ok:
print(response.text)
raise RuntimeError("Failed to search Frappe document.")

documents = response.json().get("data", [])
print("Matching documents found:", len(documents))

if len(documents) != 1:
raise RuntimeError("Frappe document mismatch")

document = documents[0]
frappe_docname = document.get("name")

if not frappe_docname:
raise RuntimeError("Frappe document mismatch")

document_url = f"{FRAPPE_URL.rstrip('/')}/api/resource/{encoded_doctype}/{quote(frappe_docname, safe='')}"
document_response = requests.get(document_url, headers=get_frappe_headers(), timeout=120)

if not document_response.ok:
raise RuntimeError("Frappe document mismatch")

frappe_document = document_response.json().get("data", {})

excel_request_id = str(request_id).strip()
frappe_request_id = str(frappe_document.get("request_id", "")).strip()

print()
print("Request ID validation:")
print("Excel :", excel_request_id)
print("Frappe:", frappe_request_id)

if excel_request_id != frappe_request_id:
print("Request ID validation: FAILED")
raise RuntimeError("Frappe document mismatch")

print("Request ID validation: PASSED")

frappe_doctype = str(frappe_document.get("doctype", "")).strip()

print()
print("Doctype validation:")
print("Excel :", expected_doctype)
print("Frappe:", frappe_doctype)

if expected_doctype != frappe_doctype:
print("Doctype validation: FAILED")
raise RuntimeError("Frappe document mismatch")

print("Doctype validation: PASSED")

print()
print("BOTH FRAPPE VALIDATIONS PASSED")
print("Document Name:", frappe_docname)
print("System Doc No.:", frappe_document.get("system_doc_no"))

return (
frappe_docname,
frappe_document.get("system_doc_no"),
frappe_document.get("docstatus"),
)


def is_pdf_already_attached(pdf_path: str, frappe_docname: str, doctype_name: str, request_id: str) -> bool:
pdf_path = Path(pdf_path)
expected_filename = f"{str(request_id).strip()}.pdf"

print()
print("=" * 60)
print("CHECKING EXISTING FRAPPE ATTACHMENT")
print("=" * 60)
print("Doctype :", doctype_name)
print("Document :", frappe_docname)
print("Request ID :", request_id)
print("Local PDF :", pdf_path.name)
print("Expected PDF :", expected_filename)

encoded_file_doctype = quote("File", safe="")
url = f"{FRAPPE_URL.rstrip('/')}/api/resource/{encoded_file_doctype}"

params = {
"filters": json.dumps([
["File", "attached_to_doctype", "=", doctype_name],
["File", "attached_to_name", "=", frappe_docname],
["File", "file_name", "=", expected_filename],
]),
"fields": json.dumps(["name", "file_name", "file_url", "attached_to_doctype", "attached_to_name"]),
"limit_page_length": 10,
}

response = requests.get(url, headers=get_frappe_headers(), params=params, timeout=120)
print("Attachment check status:", response.status_code)

if not response.ok:
print("Attachment check response:", response.text)
raise RuntimeError("Failed to check existing Frappe attachment.")

files = response.json().get("data", [])

if files:
print()
print("=" * 60)
print("PDF ALREADY ATTACHED")
print("=" * 60)
for file_data in files:
print("File Name:", file_data.get("file_name"))
print("File URL:", file_data.get("file_url"))
return True

print("No existing attachment found.")
return False


def attach_pdf_to_frappe(pdf_path: str, frappe_docname: str, doctype_name: str, request_id: str):
print()
print("=" * 60)
print("ATTACHING PDF TO FRAPPE DOCUMENT")
print("=" * 60)

pdf_path = Path(pdf_path)

if not pdf_path.exists():
raise FileNotFoundError(f"PDF file does not exist:\n{pdf_path}")

if pdf_path.stat().st_size == 0:
raise RuntimeError(f"PDF file is empty:\n{pdf_path}")

if is_pdf_already_attached(pdf_path, frappe_docname, doctype_name, request_id):
print()
print("=" * 60)
print("SKIPPING PDF UPLOAD")
print("=" * 60)
print("Reason: PDF is already attached to this Frappe document.")

return {
"already_attached": True,
"file_name": pdf_path.name,
"doctype": doctype_name,
"docname": frappe_docname,
}

print("Document:", frappe_docname)
print("PDF:", pdf_path)
print("Size:", pdf_path.stat().st_size, "bytes")

upload_url = f"{FRAPPE_URL}/api/method/upload_file"
headers = get_frappe_headers()

with open(pdf_path, "rb") as pdf_file:
files = {"file": (pdf_path.name, pdf_file, "application/pdf")}
data = {"doctype": doctype_name, "docname": frappe_docname, "is_private": "0"}

print("Uploading PDF...")
response = requests.post(upload_url, headers=headers, files=files, data=data, timeout=120)

print("Upload status:", response.status_code)

if not response.ok:
print("Upload response:")
print(response.text)
raise RuntimeError("PDF upload/attachment failed.")

try:
result = response.json()
except ValueError:
raise RuntimeError("Frappe returned a non-JSON response after PDF upload.")

print()
print("Frappe attachment response:")
print(json.dumps(result, indent=4, ensure_ascii=False))

file_data = result.get("message", {})
file_url = file_data.get("file_url") if isinstance(file_data, dict) else None

print()
print("=" * 60)
print("PDF ATTACHED SUCCESSFULLY")
print("=" * 60)
print("Frappe Document:", frappe_docname)
print("Attached File:", pdf_path.name)
if file_url:
print("Frappe File URL:", file_url)

return result


def process_pdf_in_frappe(request_id: str, doctype_name: str, pdf_path: str):
print()
print()
print("#" * 60)
print("FRAPPE POST-PROCESSING")
print("#" * 60)

frappe_docname, system_doc_no, docstatus = find_frappe_document(request_id, doctype_name)

if docstatus == 2:
raise RuntimeError(f"{doctype_name} '{frappe_docname}' is cancelled.")

if docstatus == 1:
raise RuntimeError(
f"{doctype_name} '{frappe_docname}' is already submitted. "
"PDF was not attached/submitted again."
)

attachment_result = attach_pdf_to_frappe(pdf_path, frappe_docname, doctype_name, request_id)

print()
print("=" * 60)
print("PROCESS COMPLETED SUCCESSFULLY")
print("=" * 60)
print("Request ID :", request_id)
print(f"{doctype_name} :", frappe_docname)
print("System Doc No. :", system_doc_no)
print("PDF :", Path(pdf_path).name)
print("Attachment :", "Already Attached" if attachment_result.get("already_attached") else "Successfully attached")
print("=" * 60)

return {
"request_id": request_id,
"frappe_docname": frappe_docname,
"system_doc_no": system_doc_no,
"pdf_path": str(pdf_path),
"already_attached": attachment_result.get("already_attached", False),
}


if __name__ == "__main__":
main()
(6-6/10)