Project

General

Profile

Download (42.4 KB)

Feature #98 » eap_sez_automation.py

Code to upload the pdf to respective bill of entry document - Rajesh Gotur, 09/04/2026 04:38 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://wud.canofy.tech/"

# API_KEY = "9ed2cd4a81647e9"
# API_SECRET = "8f177698daa2370"
# ------------------------------------------eap-----------------
# 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
ocr_reader = easyocr.Reader(["en"], gpu=False)


# ============================================================
# 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"
# Check if base path can be written to
if base_path.exists():
try:
with open(base_path, "a+b"):
pass
return base_path
except (PermissionError, IOError):
pass
# If locked or unwritable, append timestamp/counter
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 resolve_captcha(image_path: Path) -> str:
print("Processing CAPTCHA image via OCR Engine...")
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)
filtered = cv2.bilateralFilter(gray, 9, 75, 75)
processed_img = cv2.adaptiveThreshold(
filtered, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 11, 2
)

processed_path = SCREENSHOT_DIR / "processed_captcha.png"
cv2.imwrite(str(processed_path), processed_img)

results = ocr_reader.readtext(str(processed_path), detail=0)
raw_text = "".join(results).strip()
cleaned_token = re.sub(r"[^A-Za-z0-9]", "", raw_text)
print(f"Extracted CAPTCHA Token: '{cleaned_token}'")
return cleaned_token


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

// 2. Continuous DOM guard to delete any tour artifacts instantly
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):
clean_popups_now(page)
loc = locator_or_selector if hasattr(locator_or_selector, "click") else page.locator(locator_or_selector)
try:
loc.click(timeout=3000, force=True)
except Exception:
loc.dispatch_event("click")


# ============================================================
# LOGIN FLOW
# ============================================================
def login(page):
print()
print("=" * 60)
print("SEZ ONLINE LOGIN (AUTOMATED OCR)")
print("=" * 60)

print("Opening login page...")
page.goto(BASE_URL, wait_until="domcontentloaded")
page.wait_for_load_state("networkidle", timeout=15000)
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)

raw_captcha_path = SCREENSHOT_DIR / "raw_captcha.png"
captcha_image.wait_for(state="visible", timeout=30000)
captcha_image.screenshot(path=str(raw_captcha_path))

captcha_token = resolve_captcha(raw_captcha_path)
print("Entering extracted CAPTCHA...")
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)
page.wait_for_load_state("networkidle", timeout=10000)
except Exception:
pass

page.wait_for_timeout(2000)
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("networkidle", timeout=10000)
except Exception:
pass


# ============================================================
# NAVIGATE TO PRINT & EXPORT PDF
# ============================================================
def navigate_to_print(page, req_id: str, target_dir: Path) -> str:
print()
print("=" * 60)
print("NAVIGATE TO PRINT")
print("=" * 60)
print("Target Request ID:", req_id)

page.wait_for_load_state("networkidle", timeout=10000)
clean_popups_now(page)

# 1. Search Request Link
print("Opening Search Request...")
search_request_button = page.get_by_role("link", name="Search Request")
search_request_button.wait_for(state="attached", timeout=30000)
safe_click(page, search_request_button)

page.wait_for_load_state("networkidle", timeout=10000)
clean_popups_now(page)

# 2. Clear Date Fields via JS
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.evaluate("(el) => el.value = ''")
to_date_input.evaluate("(el) => el.value = ''")

# 3. Enter Request ID
print(f"Entering Request ID: {req_id}...")
request_id_textfield = page.locator("#ctl00_DefaultContent_txtRequestID")
request_id_textfield.fill(req_id)

# 4. Search
print("Searching...")
search_button = page.get_by_role("button", name="Search")
safe_click(page, search_button)

page.wait_for_timeout(2000)
clean_popups_now(page)

# 5. Open Search Result Link
print("Opening request details...")
req_id_link = page.locator(f"xpath=//a[text()='{req_id}']")
req_id_link.wait_for(state="attached", timeout=30000)
safe_click(page, req_id_link)

page.wait_for_timeout(1500)
clean_popups_now(page)

# 6. Click View/Print Button
print("Clicking View / Print...")
view_receipt = page.locator("#ctl00_DefaultContent_btnPrintBOE")
view_receipt.wait_for(state="attached", timeout=30000)
safe_click(page, view_receipt)

# 7. Interact inside Print Dialog Iframe
print("Handling Print Modal iframe...")
print_frame = page.frame_locator("#dialog-body")
original_checkbox = print_frame.locator("#cbNo_0")
original_checkbox.wait_for(state="attached", timeout=30000)
original_checkbox.dispatch_event("click")
print("Selected 'Original' document option.")

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

# 8. Intercept New Window Tab for PDF Export
print("Triggering print preview window...")
context = page.context

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

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

print_page.wait_for_load_state("networkidle", timeout=30000)
print_page.wait_for_timeout(2000)

# Resolve safe destination path (avoiding file lock collisions)
resolved_pdf_path = get_safe_pdf_path(target_dir, req_id)

# 9. Save as PDF
print(f"Generating PDF at: {resolved_pdf_path}...")
print_page.pdf(
path=str(resolved_pdf_path),
print_background=True,
landscape=False
)

# 10. Verify File Creation
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")

print_page.close()
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)

# Exact logout control used by the original framework
logout_button = page.locator("#ctl00_LoginStatus")

logout_button.wait_for(
state="visible",
timeout=30000
)

print("Clicking Logout...")

safe_click(page, logout_button)

# Give the portal time to complete logout
try:
page.wait_for_load_state(
"domcontentloaded",
timeout=15000
)
except Exception:
pass

page.wait_for_timeout(1500)

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

# Read header row
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 "
f"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']} | "
f"Request ID: {item['request_id']} | "
f"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)

# ========================================================
# READ EXCEL
# ========================================================

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"
)
)

# ====================================================
# POPUP PROTECTION
# ====================================================

immunize_page_against_popups(context)

def auto_accept_dialogs(dialog):

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

dialog.accept()

context.on(
"dialog",
auto_accept_dialogs
)

page = context.new_page()

try:

# =================================================
# STEP 1
# LOGIN ONCE
# =================================================

login(page)

# =================================================
# STEP 2
# DOWNLOAD ALL PDFs
# =================================================

print()
print("=" * 60)
print("STARTING BULK PDF DOWNLOAD")
print("=" * 60)

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:

saved_pdf = navigate_to_print(
page,
request_id,
DOWNLOAD_DIR
)

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
})

print()
print(
f"PDF download successful "
f"for {request_id}"
)

print(
"PDF:",
saved_pdf
)

except Exception as exc:

print()
print(
f"PDF DOWNLOAD FAILED "
f"for {request_id}"
)

print(
f"{type(exc).__name__}: {exc}"
)

# Download failed.
# Frappe upload is not attempted.
update_excel_row(
sheet,
headers,
row_number,
download_status="Failed",
timestamp=datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
)

workbook.save(EXCEL_FILE)

# Continue with next request.
continue

# =================================================
# STEP 3
# LOGOUT ONLY AFTER ALL PDFs ARE DOWNLOADED
# =================================================

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
#
# IMPORTANT:
# SEZ browser is already logged out and closed.
# ========================================================

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
)

# ================================================
# UPLOAD / ATTACHMENT RESULT
# ================================================

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 | "
f"Document: "
f"{frappe_result['frappe_docname']} | "
f"System Doc No: "
f"{frappe_result['system_doc_no']} | "
f"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 | "
f"Document: "
f"{frappe_result['frappe_docname']} | "
f"System Doc No: "
f"{frappe_result['system_doc_no']} | "
f"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"
)

# Save Excel after every Frappe request
workbook.save(EXCEL_FILE)

except Exception as exc:

# ================================================
# VALIDATION / UPLOAD FAILURE
# ================================================

print()
print(
f"FRAPPE PROCESSING FAILED "
f"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 with next request.
continue

# ========================================================
# STEP 5
# SAVE UPDATED EXCEL
# ========================================================

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)

# --------------------------------------------------------
# SEARCH THE EXACT DOCTYPE FROM EXCEL
# --------------------------------------------------------

encoded_doctype = quote(
expected_doctype,
safe=""
)

url = (
f"{FRAPPE_URL.rstrip('/')}"
f"/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) == 0:
raise RuntimeError(
"Frappe document mismatch"
)

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"
)

# --------------------------------------------------------
# GET COMPLETE FRAPPE DOCUMENT
# --------------------------------------------------------

document_url = (
f"{FRAPPE_URL.rstrip('/')}"
f"/api/resource/{encoded_doctype}/"
f"{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", {})
)

# ========================================================
# VALIDATION 1
# Excel Request ID == Frappe request_id
# ========================================================

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")

# ========================================================
# VALIDATION 2
# Excel Doctype == Frappe document doctype
# ========================================================

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")

# ========================================================
# BOTH VALIDATIONS 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")
)


# ============================================================
# CHECK WHETHER PDF IS ALREADY ATTACHED
# ============================================================

def is_pdf_already_attached(
pdf_path: str,
frappe_docname: str,
doctype_name: str,
request_id: str
) -> bool:

pdf_path = Path(pdf_path)

# Always use the Request ID as the canonical PDF name.
# This prevents files such as 12345_1.pdf from bypassing
# the "Already Attached" check.
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('/')}"
f"/api/resource/{encoded_file_doctype}"
)

# Check using the canonical Request-ID filename,
# NOT the local filename that may contain _1, _2, etc.
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
# ============================================================
# ATTACH PDF TO FRAPPE DOCUMENT
# ============================================================

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}"
)

# --------------------------------------------------------
# CHECK IF THIS EXACT PDF IS ALREADY ATTACHED
# --------------------------------------------------------

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"
)

# --------------------------------------------------------
# Frappe standard upload_file endpoint
# --------------------------------------------------------

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

headers = get_frappe_headers()

# --------------------------------------------------------
# Multipart upload
# --------------------------------------------------------

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 = None

if isinstance(
file_data,
dict
):

file_url = file_data.get(
"file_url"
)

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

# ============================================================
# COMPLETE FRAPPE POST-PROCESSING
# ============================================================

def process_pdf_in_frappe(
request_id: str,
doctype_name: str,
pdf_path: str
):

print()
print()
print("#" * 60)
print("FRAPPE POST-PROCESSING")
print("#" * 60)

# --------------------------------------------------------
# STEP 1
# Find existing Frappe document using
# Request ID + Doctype from Excel
# --------------------------------------------------------

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

# --------------------------------------------------------
# STEP 2
# Do not attach/submit a cancelled document
# --------------------------------------------------------

if docstatus == 2:

raise RuntimeError(
f"{doctype_name} '{frappe_docname}' "
"is cancelled."
)

# --------------------------------------------------------
# STEP 3
# If already submitted, stop
# --------------------------------------------------------

if docstatus == 1:

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

# --------------------------------------------------------
# STEP 4
# Attach PDF
# --------------------------------------------------------

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)
if attachment_result.get("already_attached", False):
print("Attachment :", "Already Attached")
else:
print("Attachment :", "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()
(1-1/2)