#!/usr/bin/env python3 # -*- coding: utf-8 -*- # ai_wo_from_one_request.py (3.8-safe) # # AI extraction path. Same command-line contract and same one-line-JSON output as # wo_from_one_request.py, so makeWOP1 can call it in exactly the same way. The ONLY # difference is HOW the fields are read: instead of the deterministic engine, the # request text is sent to the Anthropic API together with the editable instruction # sheet (wo_extraction_prompt.md), and the model returns the WO fields as JSON. # # Everything AROUND the extraction is reused unchanged from the proven engine: # - .doc/.pdf -> .docx conversion (wo_from_one_request._normalize_to_docx) # - full-text read of the request (Create_WO_from_Request.doc_full_text) # - remote detection & template pick (is_remote_job, select_template) # - merged-cell-safe template fill (fill_template) # - safety-net flags (plausibility_flags, missing_value_flags) # - dateless/garbage-date -> "partial" downgrade (same rule as the v2 net) # # Usage: # python3.8 ai_wo_from_one_request.py # [--prompt PATH] [--templates DIR] [--model NAME] [--keyfile PATH] # # Secrets: the Anthropic API key is read from a file you place on the server # (default engine/anthropic.key). This script never contains or logs the key. # # Output: exactly ONE line of JSON on stdout: # {"status","engine_status","form","wo_path","fields","flags","reason", # "request_file","engine_version","model"} # Normalized status (for PHP): generated | partial | refused | error # Exit code: 0=generated 10=partial 20=refused 1=error import sys import os import re import json import shutil import tempfile import traceback import urllib.request import urllib.error # --- NEVER LEAVE A BYTECODE CACHE BEHIND ------------------------------------ # Python caches compiled modules in a __pycache__ FOLDER next to the source. It # decides a cached .pyc is still good by comparing the source's MODIFICATION TIME # -- and this server's clock runs hours ahead of Eastern, while FTP uploads carry # their own timestamps. If an uploaded .py ever lands with an mtime older than its # cached .pyc, Python happily runs the OLD compiled code while the new source sits # right there on disk, and the version stamp in the file becomes a lie about what # is actually executing. # # Rather than relying on anyone remembering to delete the folder after every # upload: (1) tell Python not to write bytecode at all, and (2) remove any cache # that is already there. Set BEFORE the engine modules are imported (they are # imported later, inside functions), so nothing gets cached on the way in. sys.dont_write_bytecode = True def _purge_pycache(): """Delete the __pycache__ folder beside this script. Best-effort and silent: a cache we cannot remove must never stop a Work Order from being produced.""" try: cache = os.path.join(os.path.dirname(os.path.abspath(__file__)), "__pycache__") if os.path.isdir(cache): shutil.rmtree(cache, ignore_errors=True) except Exception: pass _purge_pycache() # ----------------------------------------------------------------------------- # ---- configuration you may change ------------------------------------------- # The model used for extraction. Set this to the current model string from your # Anthropic console/docs. A Sonnet-class model is a good cost/latency balance for # extraction; use a stronger model while stress-testing hard forms if you prefer. DEFAULT_MODEL = "claude-sonnet-5" DEFAULT_KEYFILE = "anthropic.key" # placed by you, next to this script DEFAULT_PROMPT = "wo_extraction_prompt.md" # the editable instruction sheet API_URL = "https://api.anthropic.com/v1/messages" API_VERSION = "2023-06-01" # --------------------------------------------------------------------------- # IS THIS DOCUMENT ACTUALLY A REQUEST? # --------------------------------------------------------------------------- # makeWOP1 built complete Work Orders out of two interpreter INVOICES -- one for # a hearing that was dismissed, one for a meeting that never happened -- because # nothing asked this question. Extraction succeeded beautifully on the wrong # document: an invoice carries a date, a time, a language, an address and a case # number, so every field filled in and two job numbers were burned. # # This rule was derived from 8,228 archived documents, not from intuition. What # the archive showed: # # * The word "invoice" is USELESS on its own -- it is printed on every UTS # request form, in "Billing Information (Where should we send the invoice?)". # * ORDER works. Where both phrases appear (1,005 documents), "interpreter # request" comes first in 1,004. Order also survives a cover message being # stapled to the front, which is what defeated a percentage-of-document # measure: one real request had a message prepended, pushing "invoice" from # its usual ~85% down to 36%, close enough to a percentage threshold to be # misread. # * For documents carrying only the word "invoice", what matters is whether it # is a HEADING -- measured in ABSOLUTE characters, not percent. 1,231 of them # have it inside the first 60 characters (a title); 172 have it past 600 # characters, which is "...Just the invoice." buried in an email. A percentage # cannot tell those apart: 65% of a short email and 0.8% of a long invoice are # the same kind of word in different-sized documents. # # Measured result: of 2,247 archived documents containing "interpreter request", # exactly ONE is called an invoice by this rule -- and that one genuinely is an # invoice. So it effectively never mistakes a request for a bill. # # IT IS A BACKSTOP, NOT THE DECIDER. It has NO OPINION on 27% of the archive, # including confirmed real requests that simply never use the phrase "interpreter # request". The model reads the document and decides; this can only veto TOWARD # "invoice", never toward "request". _RE_INVOICE_WORD = re.compile(r"\binvoice", re.I) _RE_REQUEST_PHRASE = re.compile(r"\b(?:translator\s*/\s*)?interpreter\s+request", re.I) _RE_INV_NUMBER = re.compile(r"\binvoice\s*(?:#|no\.?|nr\.?|num(?:ber)?|id)?\s*[:#\-]?\s*\d", re.I) _RE_AMOUNT_DUE = re.compile(r"\b(?:total\s+)?(?:amount|balance)\s+due\b" r"|\bamount\s+owed\b|\btotal\s+due\b", re.I) # UTS's own Work Order form is titled "Work Order & Invoice", so it would look # like an invoice by every test above. It is our own output, not an incoming doc. _RE_OWN_WO_TITLE = re.compile(r"work\s*order\s*&\s*invoice|work\s*order\s+and\s+invoice", re.I) _HEADING_CHARS = 200 # "invoice" this close to the start is a TITLE def _document_kind(text): """-> (kind, reason) where kind is 'invoice', 'request' or 'unknown'. 'unknown' is a legitimate and common answer -- it means DEFER TO THE MODEL.""" if not text or not text.strip(): return ("unknown", "no text layer to test (vision path)") if _RE_OWN_WO_TITLE.search(text[:400]): return ("unknown", "carries our own 'Work Order & Invoice' title") mi = _RE_INVOICE_WORD.search(text) mr = _RE_REQUEST_PHRASE.search(text) if mi and mr: if mr.start() < mi.start(): return ("request", "'interpreter request' appears before 'invoice'") return ("invoice", "'invoice' appears before 'interpreter request'") if mr: return ("request", "'interpreter request' present, 'invoice' absent") if mi: if mi.start() <= _HEADING_CHARS: return ("invoice", "'invoice' is a heading (char %d)" % mi.start()) if _RE_INV_NUMBER.search(text): return ("invoice", "an invoice number is present") if _RE_AMOUNT_DUE.search(text): return ("invoice", "an amount owed is present") return ("unknown", "'invoice' appears late (char %d) -- probably prose" % mi.start()) return ("unknown", "neither phrase present") MAX_TOKENS = 8000 # reply cap (OUTPUT only; the long prompt is INPUT, separate). # Clean JSON is well under this; 8000 is truncation insurance. AI_PATH_VERSION = "2026-07-19-g" # -g: --dry-run (used by utilities/test_one_request.php) # runs everything except drawing a job number. # -f: a client instruction about WHO should interpret is # written into Section 1's NAME cell; also fixed a # missing `import docx` that had been silently # stopping the job number being written on the # deterministic fallback path. # -e: REVIEW items are now separate from informational # flags and carry codes; an unmarked weekday no # longer demands review; bold/enlarged/highlighted # text is now reported to the model (it was being # stripped before the model ever saw it). # -d: refuses documents that are not REQUESTS (invoices, # confirmations, our own WOs). Model decides; # the keyword rule may only veto toward # "invoice"; a disagreement flags, never refuses. # -c: image-only (scanned/handwritten) sources now always # raise a review flag, so makeWOP1 puts "needs REVIEW" # in the subject line. Deterministic: driven by # use_vision (no text layer), not by the model. # -b: job-number draw on the deterministic fallback; # -b: MAX_TOKENS 8000 + auto-invent org abbreviations # =========================================================================== # NAME THE WORK ORDER # =========================================================================== # The WO used to be named after the REQUEST file it came from: # "University Translators - Panzo 26-0030 SD -- WO.docx" # Staff name them by client, venue, language and date: # "20260806_15th DC_Spanish_44231_WO.docx" # # wo_filename.py reads the WO we have just written and produces that name, using # the abbreviation map mined from 2000 archived jobs (engine/Filename_Map.xlsx). # # THIS MUST NEVER BREAK THE PIPELINE. # A WO that reaches staff with an ugly name is a nuisance. A WO that never reaches # them at all is an outage -- an interpreter is not booked and a client is not # served. So every failure here is caught, reported as a flag, and the WO goes out # under its old name. Renaming is a courtesy; delivery is the job. # DRY RUN. Set by --dry-run, used by utilities/test_one_request.php so a staffer # can try ONE request without side effects. A job number is spent the moment it is # drawn -- the well never re-issues one -- so a test that draws numbers quietly # eats the sequence. Everything else runs exactly as in production: same model, # same prompt, same template fill, same filename logic. Only the well is untouched. DRY_RUN = False def _take_job_number(): if DRY_RUN: return 0, ("DRY RUN: no job number drawn from the shared well. The Work " "Order keeps [JOB_NUMBER] in its name.") """-> (number, note). One number per Work Order, from the same well staff draw from with the bookmark. See engine/job_number_*.py. NEVER FATAL. If the counter cannot be reached, the WO still goes out -- with [job_number] in its name, exactly as it does today, and a flag saying why. A WO that reaches staff with an ugly name is a nuisance; a WO that never reaches them is an outage. THE NUMBER IS SPENT WHETHER OR NOT THE WO IS EVER USED. That is deliberate: a gap in the sequence harms nobody, and a number issued twice is two invoices in dispute. We burn, we never re-use.""" try: import glob as _glob import importlib.util as _il here = os.path.dirname(os.path.abspath(__file__)) cands = sorted(_glob.glob(os.path.join(here, "job_number_*.py"))) if not cands: return 0, ("JOB NUMBER: job_number_*.py not found in engine/ -- the WO " "keeps its [job_number] placeholder") spec = _il.spec_from_file_location("job_number", cands[-1]) mod = _il.module_from_spec(spec) spec.loader.exec_module(mod) n, why = mod.take("pipeline") if not n: return 0, "JOB NUMBER: " + why return n, "Job number %d (taken from the shared counter)" % n except Exception as exc: return 0, ("JOB NUMBER: could not take one (%s) -- the WO keeps its " "[job_number] placeholder" % str(exc)[:70]) def _write_interpreter_note_into_docx(path, note): """Put a client instruction ABOUT THE INTERPRETER into Section 1's NAME cell. Johann's idea, and a good one. The note has to reach the STAFFER who assigns the job, and it must not survive onto the copy the interpreter finally receives. Section 1's NAME cell does both by itself: * the staffer is looking straight at it at the moment they choose someone, so a preferred-interpreter request cannot be missed; * assigning someone means TYPING THEIR NAME THERE, which overwrites the note. The cleanup is guaranteed by the workflow instead of by anyone remembering to do it. Neither template needs changing: both FORM 0a and FORM 0b already hold 'NAME:' as a label cell with an empty cell beside it (table 1, row 1). Best-effort: returns True/False and never raises. A note we fail to write is still in the email; a crash here would cost the whole Work Order. """ txt = (note or "").strip() if not txt: return False try: import docx # imported HERE: this module has no top-level import doc = docx.Document(path) for tbl in doc.tables: for row in tbl.rows: cells = row.cells if len(cells) < 2: continue label = (cells[0].text or "").strip().upper().rstrip(":") if label != "NAME": continue if (cells[1].text or "").strip(): return False # already filled -- never overwrite cell = cells[1] para = cell.paragraphs[0] if cell.paragraphs else cell.add_paragraph() run = para.add_run("CLIENT REQUEST — " + txt) run.bold = True try: from docx.shared import RGBColor run.font.color.rgb = RGBColor(0xC0, 0x00, 0x00) except Exception: pass # colour is a nicety, not the point doc.save(path) return True except Exception: return False return False def _write_job_number_into_docx(path, number): """Write NUMBER into the WO's 'JOB NUMBER:' field. The deterministic engine leaves that field blank (it blanks ${JobNumber}), so on the AI-fallback path the number has to be written in afterwards -- and BEFORE the file is (re)named, so wo_filename reads it. Best-effort: returns True/False and never raises. NOTE (2026-07-19): this function had no `import docx`, and python-docx is NOT imported at the top of this module -- only inside three other functions, where the binding is local. So `docx.Document(path)` raised NameError on every call, the broad `except Exception` below swallowed it, and the function silently returned False every time. The fallback WO's JOB NUMBER field was therefore never filled in, while the number itself was still drawn from the shared well. A best-effort helper that can never succeed is indistinguishable from one that is merely unlucky -- which is why it went unnoticed.""" try: import docx # imported HERE: this module has no top-level import doc = docx.Document(path) def _all_paras(d): for p in d.paragraphs: yield p for tbl in d.tables: for row in tbl.rows: for cell in row.cells: for p in cell.paragraphs: yield p for p in _all_paras(doc): if "JOB NUMBER" in p.text.upper(): after = p.text.split(":", 1)[1] if ":" in p.text else "" if re.search(r"\d", after): return True # already numbered -- leave it sep = "" if p.text.endswith(" ") else " " p.add_run(sep + str(number)) doc.save(path) return True return False except Exception: return False def _name_wo(path, abbrev_inventor=None): """-> (path, notes). The path is the NEW one if the rename worked, the OLD one if anything at all went wrong.""" if not path or not os.path.isfile(path): return path, [] try: import glob as _glob import importlib.util as _il here = os.path.dirname(os.path.abspath(__file__)) cands = sorted(_glob.glob(os.path.join(here, "wo_filename_*.py"))) if not cands: return path, ["FILENAME: wo_filename_*.py not found in engine/ -- " "the WO keeps its old name"] spec = _il.spec_from_file_location("wo_filename", cands[-1]) mod = _il.module_from_spec(spec) spec.loader.exec_module(mod) try: new_path, notes = mod.rename_wo(path, abbrev_inventor=abbrev_inventor) except TypeError: # older wo_filename_*.py without the auto-invent param -- still works, # it just won't invent an abbreviation. Protects a partial install. new_path, notes = mod.rename_wo(path) notes = list(notes or []) # Only claim a filename if one was actually APPLIED. Logging # "Filename: wo2.docx" when the rename FAILED is a lie in the audit trail, # and a misleading log costs more hours than a missing one. if os.path.abspath(new_path) != os.path.abspath(path): notes.insert(0, "Filename: %s (%s)" % (os.path.basename(new_path), mod.WO_FILENAME_VERSION)) return new_path, notes except Exception as exc: return path, ["FILENAME: could not name the WO (%s) -- it keeps its old " "name. The WO itself is fine." % str(exc)[:90]] # Debug log: strategic trace of each request through the AI path. Written to # engine/logs/ai_debug.log in Eastern time. Set DEBUG=False to quiet it. On a # parse failure the FULL model reply is logged here -- that is how we diagnose # "did not return parseable JSON" instead of guessing. DEBUG = True DEBUG_LOG = None # resolved to engine/logs/ai_debug.log at runtime # ----------------------------------------------------------------------------- _HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, _HERE) import datetime try: from dateutil import tz as _tz _ET = _tz.gettz("America/New_York") except Exception: _ET = None def _log(msg, big=None): """Append a timestamped line (Eastern) to the debug log. Never raises.""" if not DEBUG: return try: path = DEBUG_LOG or os.path.join(_HERE, "logs", "ai_debug.log") d = os.path.dirname(path) if d and not os.path.isdir(d): os.makedirs(d) now = datetime.datetime.now(_ET) if _ET else datetime.datetime.utcnow() stamp = now.strftime("%Y-%m-%d %H:%M:%S %Z") or now.strftime("%Y-%m-%d %H:%M:%S") with open(path, "a", encoding="utf-8") as fh: fh.write("[%s] %s\n" % (stamp, msg)) if big is not None: fh.write(" " + str(big)[:2000].replace("\n", "\n ") + "\n") except Exception: pass # logging must never break extraction def _load_prompt(path): """Read the instruction sheet, dropping editor-note lines (a single leading '#') while KEEPING markdown section headers ('##', '###').""" with open(path, "r", encoding="utf-8") as fh: lines = fh.read().splitlines() kept = [ln for ln in lines if not re.match(r"^\s*#(?!#)", ln)] return "\n".join(kept).strip() def _read_key(keyfile): with open(keyfile, "r", encoding="utf-8") as fh: return fh.read().strip() def _call_model_vision(api_key, model, system_prompt, pdf_path): """Send the PDF itself to the model as a document block so Claude reads it VISUALLY -- for scanned or handwritten forms that have no text layer. Same JSON contract as _call_model. No image-render library needed; the API renders the PDF internally and the model's vision reads typed and handwritten text.""" import base64 with open(pdf_path, "rb") as fh: b64 = base64.standard_b64encode(fh.read()).decode("ascii") user_content = [ {"type": "document", "source": {"type": "base64", "media_type": "application/pdf", "data": b64}}, {"type": "text", "text": ("The document above is an interpreter Request form (it may be " "scanned or handwritten). Extract the Work Order fields per your " "instructions.\nReturn ONLY a single JSON object and nothing else: " "no explanation, no markdown fences. Start your reply with '{'.")}, ] body = json.dumps({ "model": model, "max_tokens": MAX_TOKENS, "system": system_prompt, "messages": [{"role": "user", "content": user_content}], }).encode("utf-8") req = urllib.request.Request(API_URL, data=body, method="POST", headers={ "x-api-key": api_key, "anthropic-version": API_VERSION, "content-type": "application/json", }) with urllib.request.urlopen(req, timeout=180) as resp: payload = json.loads(resp.read().decode("utf-8")) parts = [b.get("text", "") for b in payload.get("content", []) if b.get("type") == "text"] if payload.get("stop_reason") == "max_tokens": _log("WARNING vision reply hit max_tokens (%d) -- may be truncated" % MAX_TOKENS) return "".join(parts).strip() def _call_model(api_key, model, system_prompt, request_text): """POST to the Anthropic Messages API via urllib (no SDK). Returns the concatenated text of the reply. We ask hard for a bare JSON object; parsing is made resilient in _extract_json (strips fences/preamble). NOTE: assistant 'prefill' is NOT used -- current models (e.g. claude-sonnet-5) reject a trailing assistant turn with HTTP 400, so the conversation ends on the user message and we lean on the instruction + retry instead.""" user = ("Here is the interpreter Request document (text). Extract the Work " "Order fields per your instructions.\n" "Return ONLY a single JSON object and nothing else: no explanation, " "no commentary, no markdown code fences. Start your reply with '{'.\n\n" "\n" + request_text + "\n") body = json.dumps({ "model": model, "max_tokens": MAX_TOKENS, "system": system_prompt, "messages": [{"role": "user", "content": user}], }).encode("utf-8") req = urllib.request.Request(API_URL, data=body, method="POST", headers={ "x-api-key": api_key, "anthropic-version": API_VERSION, "content-type": "application/json", }) with urllib.request.urlopen(req, timeout=120) as resp: payload = json.loads(resp.read().decode("utf-8")) parts = [b.get("text", "") for b in payload.get("content", []) if b.get("type") == "text"] stop = payload.get("stop_reason") if stop == "max_tokens": _log("WARNING model hit max_tokens (%d) -- reply likely truncated" % MAX_TOKENS) return "".join(parts).strip() def _extract_json(text): """Pull the JSON object out of the model reply. Tolerates stray prose or ```json fences even though the prompt forbids them. Returns a dict or None.""" if not text: return None t = text.strip() t = re.sub(r"^```(?:json)?\s*|\s*```$", "", t, flags=re.I | re.M).strip() # Grab the outermost {...} start = t.find("{") end = t.rfind("}") if start == -1 or end == -1 or end < start: return None try: return json.loads(t[start:end + 1]) except Exception: return None # Fields the WO template expects (the model's JSON is copied into these). Extra # keys the model returns (MeetingContact*, _remote, _flags) are handled below. _WO_FIELDS = [ "Weekday", "CalendarDate", "Time", "Language", "LEPName", "SubjectMatter", "TypeOfMeeting", "RefCaseNumber", "ClientOrganization", "ClientName", "ClientContactName", "ClientContactNumber", "LocationName", "JobSiteFullAddress", "ApartmentOrRoomNumber", "ZoomLink", "MinHours", "OtherParticipants", ] def _api_abbreviate(org_name, model, keyfile): """Dedicated tiny model call: turn an org the map does NOT know into a short filename abbreviation. Used only on the deterministic fallback path, only on a map miss. Guarded -- returns '' on any failure, and the filename module then falls back to a network-free acronym, so this never blocks a WO.""" try: if not org_name or not keyfile: return "" key = _read_key(keyfile) sysp = ("You create SHORT filename abbreviations for organization names. " "Reply with ONLY the abbreviation and nothing else: letters and " "digits, no spaces, no punctuation, 2-8 characters.") raw = _call_model(key, model, sysp, org_name) return re.sub(r"[^A-Za-z0-9]", "", (raw or "").strip())[:8] except Exception: return "" def _parse_ai_opts(argv): """Parse positionals + the AI flags (--prompt/--model/--keyfile/--templates /--matrix). The engine's own parser only knows matrix/templates, so we need our own here.""" positional, opts, i = [], {}, 0 flags = {"--prompt": "prompt", "--model": "model", "--keyfile": "keyfile", "--templates": "templates", "--matrix": "matrix"} while i < len(argv): a = argv[i] if a == "--dry-run": opts["dry_run"] = True i += 1 continue if a in flags and i + 1 < len(argv): opts[flags[a]] = argv[i + 1] i += 2 continue if a.startswith("--"): i += 1 continue positional.append(a) i += 1 return positional, opts def _deterministic_fallback(request_file, docx_path, output_dir, templates_dir, eng, wf, why, model, keyfile=None): """AI step unavailable -> run the STRONG deterministic engine on this same request and return its work order, tagged so the reason is visible. This is the fallback that must catch an AI failure -- never the weak PHP extractor.""" import shutil as _sh parent = tempfile.mkdtemp(prefix="ai_fb_") try: job = os.path.join(parent, "job") os.makedirs(job, exist_ok=True) _sh.copyfile(docx_path, os.path.join(job, "request.docx")) origin_is_doc = os.path.splitext(request_file)[1].lower() == ".doc" res = eng.create_wo_from_folder( job, matrix_path=None, templates_dir=templates_dir, output_dir=output_dir, force_doc_form=origin_is_doc) engine_status = res.get("status", "") if isinstance(res, dict) else "" norm = {"ok": "generated", "partial_no_form": "partial", "refused": "refused", "unreadable_request": "refused"}.get( engine_status, "refused") if norm == "generated" and isinstance(res, dict): cd = str((res.get("values") or {}).get("CalendarDate", "")).strip() date_flagged = any( re.search(r"MISSING:\s*Calendar Date|REVIEW\s+CalendarDate", str(f)) for f in (res.get("flags") or [])) if not cd or date_flagged: norm = "partial" wo_path = res.get("output_path") if isinstance(res, dict) else None if wo_path and not os.path.isfile(wo_path): wo_path = None _fn_notes = [] _fbjob_note = "" if wo_path: desired = os.path.join(output_dir, wf._safe_stem(request_file) + " -- WO.docx") if os.path.abspath(desired) != os.path.abspath(wo_path): k = 2 while os.path.exists(desired): desired = os.path.join(output_dir, wf._safe_stem(request_file) + " (%d) -- WO.docx" % k) k += 1 try: os.replace(wo_path, desired) wo_path = desired except Exception: pass # THE FALLBACK PATH GETS A PROPER NAME TOO. # It is easy to wire the namer into the AI path and forget this one -- # and then, on exactly the days the AI path fails, every WO would arrive # under its old request-derived name. The half that breaks is the half # nobody is watching. # Draw a real JOB NUMBER for the fallback WO too. The AI path takes one # at _take_job_number(); the fallback historically did NOT, so whenever # the AI step failed, the WO went out with a blank JOB NUMBER and # [job_number] in its name. Write it in BEFORE _name_wo so the filename # picks it up -- exactly as the AI path does. _fbjob, _fbjob_note = _take_job_number() if _fbjob: _write_job_number_into_docx(wo_path, str(_fbjob)) # On a map miss, invent an abbreviation via a dedicated API call # (falls back to a network-free acronym inside wo_filename). _det_inv = (lambda o: _api_abbreviate(o, model, keyfile)) if keyfile else None wo_path, _fn_notes = _name_wo(wo_path, abbrev_inventor=_det_inv) flags = list(res.get("flags", []) if isinstance(res, dict) else []) flags.extend(_fn_notes) flags.insert(0, "Extraction: DETERMINISTIC engine (AI step skipped: %s)." % why) if _fbjob_note: flags.append(_fbjob_note) out = { "status": norm, "engine_status": engine_status, "form": res.get("form") if isinstance(res, dict) else None, "wo_path": wo_path, "fields": res.get("values", {}) if isinstance(res, dict) else {}, "flags": flags, # Coded items a human must check. makeWOP1 builds the subject line # from the CODES -- never by pattern-matching the note text. "review": [("%s|%s" % (r["code"], r["note"])) for r in review], "reason": res.get("reason", "") if isinstance(res, dict) else "", "request_file": os.path.basename(request_file), "engine_version": AI_PATH_VERSION, "model": "deterministic-fallback", } code = {"generated": 0, "partial": 10, "refused": 20}.get(norm, 1) _log("DETERMINISTIC FALLBACK RESULT: status=%s wo=%s (why: %s)" % (norm, os.path.basename(wo_path) if wo_path else "(none)", why)) return wf._emit(out, code) finally: _sh.rmtree(parent, ignore_errors=True) def _docx_markings(docx_path): """Return the strings a human MARKED in the form -- bolded, highlighted, or set in a larger font than the document's body text. WHY THIS EXISTS. The text handed to the model is built from w:t nodes, i.e. characters only: bold, highlight and font size are thrown away before the model sees anything. On the 22nd Circuit Court's form the client marks the weekday by BOLDING AND ENLARGING it -- Day: MO TU WE TH F SA SU (circle or bold the day you need) -- and every trace of that is gone by the time the model reads the line. It was never failing to recognise the marking; the marking never arrived. Twenty Work Orders in a row reported "no day circled" on forms where the day was plainly marked. This does not alter the extracted text (the deterministic path shares that function and must not shift underneath it). It collects the marked strings so the AI path can append them as an explicit note. Never raises: a marking we fail to spot is a missed hint, not a broken run. """ marks = [] try: import docx from docx.oxml.ns import qn d = docx.Document(docx_path) # the document's normal body size, to judge "larger than" against base = None try: base = d.styles["Normal"].font.size except Exception: base = None base_pt = base.pt if base is not None else 11.0 def _runs(doc): for p in doc.paragraphs: for r in p.runs: yield r for tbl in doc.tables: for row in tbl.rows: for cell in row.cells: for p in cell.paragraphs: for r in p.runs: yield r for r in _runs(d): txt = (r.text or "").strip() if not txt or len(txt) > 60: continue why = [] try: if r.bold: why.append("bold") except Exception: pass try: if r.font.size is not None and r.font.size.pt > base_pt + 0.5: why.append("larger font (%.0fpt vs %.0fpt)" % (r.font.size.pt, base_pt)) except Exception: pass try: hl = r.font.highlight_color if hl is not None: why.append("highlighted") except Exception: pass try: if r.underline: why.append("underlined") except Exception: pass if why: marks.append((txt, ", ".join(why))) except Exception: return [] # collapse duplicates, keep order seen, out = set(), [] for txt, why in marks: key = txt.lower() if key in seen: continue seen.add(key) out.append((txt, why)) return out[:40] def _markings_note(marks): """Render the markings as a block appended to the text sent to the model.""" if not marks: return "" lines = ["", "", "=== MARKINGS FOUND IN THE ORIGINAL DOCUMENT ===", "The formatting below is NOT visible in the plain text above. On these", "forms a client marks a choice by bolding, enlarging, highlighting or", "underlining it -- most often the weekday. Treat these as filled-in", "VALUES, not as decoration:"] for txt, why in marks: lines.append(' "%s" (%s)' % (txt, why)) lines.append("=== END OF MARKINGS ===") return "\n".join(lines) def _docx_textbox_text(docx_path): """Extract text that lives INSIDE text boxes (v:textbox / w:txbxContent). python-docx's .paragraphs/.tables only read the top-level body, so a form built entirely from text boxes (a common scanned-form recreation) reads as empty and gets refused. Returns "" on any error. Order follows the XML.""" try: import docx from docx.oxml.ns import qn d = docx.Document(docx_path) lines = [] for tb in d.element.body.iter(qn('w:txbxContent')): for p in tb.iter(qn('w:p')): texts = [t.text for t in p.iter(qn('w:t')) if t.text] if texts: lines.append("".join(texts)) return "\n".join(lines) except Exception: return "" def _pdf_linear_text(pdf_path): """Read a PDF as linear, top-to-bottom text (the way a human reads) rather than as a reconstructed grid. Nested-table layouts -- e.g. a web-form email printed to PDF -- defeat grid conversion (labels survive, values are dropped) yet read perfectly as linear text. Returns "" if nothing is extractable (a scanned/image PDF), so callers can fall back. Never raises.""" text = "" try: import pdfplumber parts = [] with pdfplumber.open(pdf_path) as pdf: for pg in pdf.pages: parts.append(pg.extract_text() or "") text = "\n".join(parts).strip() except Exception: text = "" if not text: try: import fitz d = fitz.open(pdf_path) text = "\n".join(pg.get_text() for pg in d).strip() except Exception: text = "" return text def _parse_hours_value(text): """Parse a stated 'hours to reserve' value into float hours, or None. Handles '2', '1.5', '1.5 hrs', '2 hours', '90 minutes', '90 mins'.""" if not text: return None t = str(text).strip().lower() if not t: return None m = re.search(r'(\d+(?:\.\d+)?)\s*(?:minute|minutes|min|mins)\b', t) if m: try: return float(m.group(1)) / 60.0 except ValueError: return None m = re.search(r'(\d+(?:\.\d+)?)', t) # first number = hours ('2', '1.5', '2 hrs') if m: try: return float(m.group(1)) except ValueError: return None return None def _parse_clock_minutes(text): """Parse a clock time into minutes-since-midnight, or None. Handles '1:00 PM', '1 PM', '1:30pm', '13:00', '9:05 AM', 'noon', 'midnight'.""" if not text: return None t = str(text).strip().lower() if not t: return None if "noon" in t: return 12 * 60 if "midnight" in t: return 0 # Accept the compact forms people actually write: '10a', '10p', '10A', # '10 p.', as well as '10am', '10 a.m.', '1:30pm'. The trailing (?![a-z]) # keeps '10 apples' from reading as 10 AM. Previously a bare 'p' was # ignored, so '10p' silently parsed as 10:00 AM -- a wrong answer, not a # flag. Now the meridiem is taken from its FIRST letter. # Compact digit-run forms people actually write: '130p' = 1:30 pm, # '1030a' = 10:30 am. Without this, '130p' fell through to the bare-hour # branch and silently became 13:00 -- wrong by 30 minutes, and no flag. mc = re.search(r'\b(\d{3,4})\s*([ap])\.?\s*(?:m\.?)?(?![a-z])', t) if mc: digits = mc.group(1) hh = int(digits[:-2]) mm = int(digits[-2:]) ap = mc.group(2) + "m" if hh <= 23 and mm <= 59: if ap == "am" and hh == 12: hh = 0 elif ap == "pm" and hh != 12: hh += 12 return hh * 60 + mm m = re.search(r'(\d{1,2})(?::(\d{2}))?\s*([ap])\.?\s*(?:m\.?)?(?![a-z])', t) if not m: m = re.search(r'(\d{1,2})(?::(\d{2}))?', t) if not m: return None hh = int(m.group(1)) mm = int(m.group(2) or 0) ap = "" if hh > 23 or mm > 59: return None return hh * 60 + mm hh = int(m.group(1)) mm = int(m.group(2) or 0) ap = (m.group(3) or "") + ("m" if m.group(3) else "") if hh > 23 or mm > 59: return None if ap == "am" and hh == 12: hh = 0 elif ap == "pm" and hh != 12: hh += 12 return hh * 60 + mm def _ceil_quarter(x): """Round UP to the next 0.25 hour, with fp tolerance so exact quarters stay put.""" import math q = x / 0.25 if abs(q - round(q)) < 1e-9: q = round(q) return math.ceil(q) * 0.25 def _fmt_hours(x): """Trim trailing zeros: 2.0->'2', 3.5->'3.5', 3.75->'3.75', 3.25->'3.25'.""" return ("%.2f" % x).rstrip("0").rstrip(".") _PURE_TIME_RX = re.compile(r'^\d{1,4}(:\d{2})?\s*[ap]\.?\s*(m\.?)?$', re.I) def _normalize_time_display(text): """'10a' -> '10:00 am'. Only rewrites a value that is PURELY a time token; anything carrying extra words is left verbatim. Idempotent ('9:00 am' stays).""" t = (text or "").strip() if not t or not _PURE_TIME_RX.match(t): return text, None mins = _parse_clock_minutes(t) if mins is None: return text, None hh, mm = divmod(mins, 60) ap = "am" if hh < 12 else "pm" h12 = hh % 12 if h12 == 0: h12 = 12 out = "%d:%02d %s" % (h12, mm, ap) if out.lower() == t.lower(): return text, None return out, "Note Time: normalized %r -> %r." % (t, out) _DATE_HAS_YEAR_RX = re.compile(r'\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\b(?:19|20)\d{2}\b') _REF_DATE_RX = re.compile(r'\b(\d{1,2})/(\d{1,2})/((?:19|20)?\d{2})\b') def _infer_calendar_year(cal_date, request_text): """Add the year to a service date that has none (e.g. 'March 17'). The year is taken ONLY from a full date inside the request itself -- normally the submission date ('Submission Date (Today's Date): 3/6/2025'). It is never taken from the processing clock: these requests are often archived, so 'today' would stamp the wrong year. If the service month/day falls before the reference month/day, the job rolls into the next year (a request filed in December for a January job). With no reference date in the request, the value is left exactly as-is and flagged -- we do not guess.""" cd = (cal_date or "").strip() if not cd or _DATE_HAS_YEAR_RX.search(cd): return cal_date, None m = _REF_DATE_RX.search(request_text or "") if not m: return cal_date, ("REVIEW CalendarDate: %r has no year and the request " "contains no dated reference to infer one from." % cd[:40]) ref_y = int(m.group(3)) if ref_y < 100: ref_y += 2000 ref_mo, ref_d = int(m.group(1)), int(m.group(2)) try: from dateutil import parser as _dp import datetime as _dt parsed = _dp.parse(cd, default=_dt.datetime(ref_y, 1, 1), fuzzy=False) except Exception: return cal_date, ("REVIEW CalendarDate: %r has no year and could not be " "parsed to add one." % cd[:40]) year = ref_y if (parsed.month, parsed.day) < (ref_mo, ref_d): year += 1 # filed late in the year for a job early the next one out = "%s, %d" % (cd.rstrip(", "), year) return out, ("Note CalendarDate: year %d inferred from the dated reference " "%d/%d/%d in the request (source gave no year)." % (year, ref_mo, ref_d, ref_y)) _MERIDIEM_RX = re.compile(r'[ap]\.?\s*m?\.?(?![a-z])', re.I) def _infer_start_meridiem(start_text, end_text): """A start time with NO am/pm next to an end time that HAS one. Real case: the form's "Start time" box was left blank and only "End time: 2:30 pm" was filled, so the start came through as bare "1:00". That parsed as 1:00 AM, and 1:00 AM -> 2:30 PM billed as 13.5 HOURS for a 1.5-hour job. The hour is not wrong -- only the half of the day is -- so pick the meridiem that makes the job plausible: a positive duration of at most 8 hours. If neither reading is plausible, change NOTHING and let the caller flag it.""" st = (start_text or "").strip() if not st or _MERIDIEM_RX.search(st): return start_text, None # already unambiguous end_min = _parse_clock_minutes(end_text or "") if end_min is None: return start_text, None # nothing to infer from m = re.match(r'^(\d{1,2})(?::(\d{2}))?\s*$', st) if not m: return start_text, None hh, mm = int(m.group(1)), int(m.group(2) or 0) if hh > 12 or mm > 59: return start_text, None # 24-hour clock: unambiguous best = None for cand_h, ap in ((hh % 12, "am"), ((hh % 12) + 12, "pm")): dur = (end_min - (cand_h * 60 + mm)) / 60.0 if 0 < dur <= 8 and (best is None or dur < best[0]): best = (dur, cand_h, ap) if best is None: return start_text, None _, cand_h, ap = best h12 = cand_h % 12 or 12 new = "%d:%02d %s" % (h12, mm, ap) return new, ("Note Time: start %r had no AM/PM; read as %r because the end time " "%r makes that the only plausible reading." % (st, new, end_text)) _WD_NAMES = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] def _weekday_check(values): """The weekday is ALWAYS derived from the calendar date -- never trusted from the request. When a client writes "4/17/25, Monday" but 4/17/25 is a Thursday, the date and the day contradict each other and NOBODY can tell from the request which one is right. An interpreter could be sent on the wrong day. So: derive the weekday, and raise a REVIEW CalendarDate flag -- which forces the WO out as a partial for a human to resolve, rather than clean and possibly wrong.""" cal = str(values.get("CalendarDate", "") or "").strip() if not cal: return None try: from dateutil import parser as _dp if not re.search(r"\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\b(19|20)\d{2}\b", cal): return None # no year -> weekday is meaningless actual = _WD_NAMES[_dp.parse(cal, fuzzy=False).weekday()] except Exception: return None stated = str(values.get("Weekday", "") or "").strip() values["Weekday"] = actual # derived value always wins if stated and stated.lower() != actual.lower(): return ("REVIEW CalendarDate: the request says %s but %s falls on a %s. " "The date and the weekday CONTRADICT each other -- confirm which is " "correct before assigning an interpreter." % (stated, cal, actual)) return None _DATE_TOKEN_RX = re.compile( r'\b(\d{1,2})/(\d{1,2})(?:/((?:19|20)?\d{2}))?\b' r'|\b(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)[a-z]*\s+(\d{1,2})(?:,?\s*((?:19|20)\d{2}))?\b', re.I) _MONTHS = {"jan":1,"feb":2,"mar":3,"apr":4,"may":5,"jun":6, "jul":7,"aug":8,"sep":9,"oct":10,"nov":11,"dec":12} _SUBMISSION_RX = re.compile( r"(?:submission\s*date|today'?s?\s*date|date\s*(?:of\s*)?request(?:ed)?|date\s*sent)" r"[^\d]{0,40}(\d{1,2})/(\d{1,2})(?:/((?:19|20)?\d{2}))?", re.I) def _dates_on_line(line): """Distinct date tokens on ONE line.""" out = [] for m in _DATE_TOKEN_RX.finditer(line or ""): if m.group(1): mo, d, y = int(m.group(1)), int(m.group(2)), m.group(3) else: mo = _MONTHS.get(m.group(4)[:3].lower()) d, y = int(m.group(5)), m.group(6) if not mo or mo > 12 or d > 31: continue yr = int(y) + 2000 if (y and int(y) < 100) else (int(y) if y else None) if (mo, d, yr) not in out: out.append((mo, d, yr)) return out def _multidate_guard(values, request_text): """A recurring or multi-day request must NEVER go out as a clean Work Order. WHY THIS IS DETERMINISTIC AND NOT LEFT TO THE MODEL: four BYTE-IDENTICAL copies of one recurring request ("Monday 9/8/25, Wednesday 9/10/25, ... 6 sessions") were processed. Three came back correctly flagged. The FOURTH silently kept only the first date, 9/8/25, and shipped a clean WO. Staff would have booked one session and missed five. A guard that depends on the model noticing is a coin flip. HOW IT TELLS A REAL RECURRENCE FROM NOISE: a genuine multi-date request lists its dates TOGETHER -- "Monday 9/8/25, Wednesday 9/10/25, Monday 9/15/25". Stray dates elsewhere in a document are not service dates: a print footer ("...Temp/eudDF.htm 6/5/2025"), a child's date of birth ("B.D. 7/25/23"), an internal note ("5/13 Job in & Shannon sent..."). Counting every date in the document flags all of those and buries the real alarm in noise. So the test is: do TWO OR MORE dates appear on a SINGLE LINE (i.e. in one field)?""" if not request_text: return None # vision path: no text to count hits = [] for line in request_text.splitlines(): if "file:///" in line.lower(): # print footer, not content continue found = _dates_on_line(line) if len(found) >= 2: for f in found: if f not in hits: hits.append(f) # The model may also have put the whole list into the WO's date field. chosen = str(values.get("CalendarDate", "") or "").strip() if len(_dates_on_line(chosen)) >= 2: for f in _dates_on_line(chosen): if f not in hits: hits.append(f) if len(hits) < 2: return None shown = ", ".join("%d/%d%s" % (m, d, ("/%d" % y) if y else "") for m, d, y in hits[:8]) return ("REVIEW CalendarDate: the request lists %d dates together (%s%s) -- this " "looks like a recurring or multi-day job. The WO carries %r. Which " "session(s) it covers is a HUMAN decision; do not assume it covers them " "all." % (len(hits), shown, " ..." if len(hits) > 8 else "", chosen[:50])) def _compute_min_hours(start_text, end_text, reserved_text): """Return (min_hours_string, flag_or_None). UTS bills a 2-hour minimum to ALL clients, so the result is always >= 2. Resolution order: 1) reserved hours stated -> max(2, ceil_quarter(reserved)) 2) else end time present -> max(2, ceil_quarter(end - start)) 3) else -> 2 Durations round UP to the nearest 0.25 hour. The arithmetic is done HERE (never by the model) from the raw values the model reports.""" FLOOR = 2.0 reserved = _parse_hours_value(reserved_text) if reserved is not None and reserved > 0: return _fmt_hours(max(FLOOR, _ceil_quarter(reserved))), None end_min = _parse_clock_minutes(end_text) if end_min is not None: start_min = _parse_clock_minutes(start_text) if start_min is None: return _fmt_hours(FLOOR), ("REVIEW MinHours: an end time was given but the " "start time was unreadable; billed the 2-hour minimum.") dur = (end_min - start_min) / 60.0 if dur <= 0: return _fmt_hours(FLOOR), ("REVIEW MinHours: end time is not after start time " "(%s -> %s); billed the 2-hour minimum." % (start_text, end_text)) out = _fmt_hours(max(FLOOR, _ceil_quarter(dur))) if dur > 8: return out, ("REVIEW MinHours: %s hours is implausibly long for an " "interpreting job (start %r, end %r) -- verify the AM/PM " "on both times." % (out, start_text, end_text)) return out, None return _fmt_hours(FLOOR), None def main(argv): import wo_from_one_request as wf # reuse its helpers (stub, normalize, emit) global DRY_RUN positional, opts = _parse_ai_opts(argv) if opts.get("dry_run"): DRY_RUN = True _log("DRY RUN: the job-number well will NOT be touched.") if len(positional) < 2: return wf._emit({"status": "error", "reason": "usage: ai_wo_from_one_request.py " " [--prompt PATH] [--templates DIR] " "[--model NAME] [--keyfile PATH]"}, 1) request_file = os.path.abspath(positional[0]) output_dir = os.path.abspath(positional[1]) model = opts.get("model", DEFAULT_MODEL) prompt_path = opts.get("prompt", os.path.join(_HERE, DEFAULT_PROMPT)) keyfile = opts.get("keyfile", os.path.join(_HERE, DEFAULT_KEYFILE)) # Template folder: makeWOP1 calls this script WITHOUT --templates, so default # to this script's own directory (engine/), where FORM 0a/0b live. Without # this default, select_template(None) returns None and the whole AI path was # (wrongly) refusing every request -> makeWOP1 fell through to the weak PHP # extractor. This one line is what makes the AI path work under the cron. templates_dir = opts.get("templates") or _HERE _log("=== AI PATH START v%s | model=%s | request=%s | templates=%s" % (AI_PATH_VERSION, model, os.path.basename(request_file), templates_dir)) if not os.path.isfile(request_file): return wf._emit({"status": "error", "reason": "request file not found: %s" % request_file}, 1) try: os.makedirs(output_dir, exist_ok=True) except Exception as e: return wf._emit({"status": "error", "reason": "could not create output dir: %s" % e}, 1) wf._install_tk_stub_if_needed() try: import Create_WO_from_Request as eng import docx except Exception as e: return wf._emit({"status": "error", "reason": "could not import engine: %s" % e}, 1) tmp = tempfile.mkdtemp(prefix="ai_wo_") try: ext = os.path.splitext(request_file)[1].lower() is_pdf = (ext == ".pdf") is_txt = (ext == ".txt") # 1) Convert to a grid .docx for the deterministic FALLBACK's use. Returns # None for a scanned/image-only PDF or a .txt -- that's fine. docx_path, note = wf._normalize_to_docx(request_file, tmp) # 2) Decide what the AI reads. # - PDF with a text layer -> LINEAR text (beats grid on nested tables). # - PDF with NO text layer -> VISION (scanned/handwritten). # - .txt -> plain text as-is (e.g. a web-form email body # routed here by makeWOP1). # - .doc/.docx -> doc_full_text of the converted docx. use_vision = False request_text = "" if is_pdf: request_text = _pdf_linear_text(request_file) if request_text.strip(): _log("PDF read as linear text: %d chars (grid conversion bypassed for AI)" % len(request_text)) else: use_vision = True _log("PDF has NO text layer -> VISION path (scanned or handwritten)") elif is_txt: try: with open(request_file, "r", encoding="utf-8", errors="replace") as fh: request_text = fh.read() except Exception as e: request_text = "" _log("TXT read error: %s" % e) _log("TXT request read as plain text: %d chars" % len(request_text)) else: if docx_path is None: return wf._emit({ "status": "refused", "engine_status": "refused", "form": None, "wo_path": None, "fields": {}, "flags": [note], "reason": note, "request_file": os.path.basename(request_file), "engine_version": AI_PATH_VERSION, "model": model}, 20) request_text = eng.doc_full_text(docx.Document(docx_path)) # Some request forms put ALL content inside text boxes (v:textbox / # w:txbxContent), which doc_full_text (paragraphs + tables) skips, so # the doc reads as empty and is refused. Append any text-box text. # No-op for normal docs, so the working cases are unaffected. _tb = _docx_textbox_text(docx_path) if _tb.strip(): request_text = (request_text + "\n" + _tb).strip() _log("added text-box content: %d chars" % len(_tb)) # Bold / enlarged / highlighted text is stripped out by the w:t-only # extraction above, so a marked weekday reaches the model as ordinary # text among six other days. Append what was marked, explicitly. _marks = _docx_markings(docx_path) if _marks: request_text = request_text + _markings_note(_marks) _log("markings detected (%d): %s" % (len(_marks), "; ".join("%s [%s]" % (t, w) for t, w in _marks[:6]))) else: _log("no bold/enlarged/highlighted runs found in the document") if not use_vision and not request_text.strip(): msg = "request converted but contained no readable text" return wf._emit({ "status": "refused", "engine_status": "unreadable_request", "form": None, "wo_path": None, "fields": {}, "flags": [msg], "reason": msg, "request_file": os.path.basename(request_file), "engine_version": AI_PATH_VERSION, "model": model}, 20) _log("text for AI: %s (source %s)" % ("VISION (PDF sent to model)" if use_vision else "%d chars" % len(request_text), os.path.basename(request_file))) # 2) Call the model with the editable instruction sheet. On ANY failure # of the AI step, fall back to the STRONG deterministic engine (never # let makeWOP1 drop through to the weak PHP extractor). Retry once on # an unparseable reply; capture the raw reply for diagnosis either way. ai_fail = "" data = None raw_reply = "" try: system_prompt = _load_prompt(prompt_path) api_key = _read_key(keyfile) for attempt in (1, 2): if use_vision: raw_reply = _call_model_vision(api_key, model, system_prompt, request_file) else: raw_reply = _call_model(api_key, model, system_prompt, request_text) data = _extract_json(raw_reply) _log("AI call attempt %d (%s): reply %d chars, JSON parse %s" % (attempt, "vision" if use_vision else "text", len(raw_reply), "OK" if isinstance(data, dict) else "FAILED")) if isinstance(data, dict): break data = None if data is None: ai_fail = "model did not return parseable JSON (2 tries)" _log("AI parse FAILED -- raw reply follows:", big=raw_reply) except urllib.error.HTTPError as e: detail = "" try: detail = e.read().decode("utf-8")[:200] except Exception: pass ai_fail = "API HTTP %s: %s" % (e.code, detail) _log("AI HTTPError: " + ai_fail) except Exception as e: ai_fail = "AI step failed: %s" % e _log("AI exception: " + ai_fail) if data is None: # Strong fallback: run the proven deterministic engine on this same # request and return ITS work order, tagged so the AI step's failure # (and a snippet of what it said) is visible in the WO email. But a # scanned/handwritten PDF has NO grid docx (docx_path is None) and no # text the deterministic engine could read -- so there is nothing to # fall back TO; emit a clear refusal instead of crashing on None. why = ai_fail if raw_reply: why += " | model said: " + raw_reply[:180].replace("\n", " ") if docx_path is None: _log("AI (vision) failed and no text layer for fallback -> refused: " + why) return wf._emit({ "status": "refused", "engine_status": "refused", "form": None, "wo_path": None, "fields": {}, "flags": ["AI vision read failed on an image-only PDF: " + why], "reason": why, "request_file": os.path.basename(request_file), "engine_version": AI_PATH_VERSION, "model": model}, 20) return _deterministic_fallback(request_file, docx_path, output_dir, templates_dir, eng, wf, why, model, keyfile=keyfile) _log("AI extraction OK -- proceeding to fill template") # 3) Marshal model output into the engine's values dict (blank-safe). values = {} for k in _WO_FIELDS: v = data.get(k, "") values[k] = "" if v is None else str(v).strip() # ===================================================================== # NOTHING IN, NOTHING OUT. # ===================================================================== # If the model came back with EVERY field empty, there was no request here. # A Work Order with no data is not a partial Work Order -- it is not a Work # Order. Emitting one produces a shell that LOOKS like a booked job: # # [DATE INCONCLUSIVE]_[organization not found]_[language not found]_ # [job_number]_WO.docx # # That is what happened on 2026-07-13, when Raquel mailed a new client the # BLANK request form and CC'd request@. The AI diagnosed it perfectly -- # "Form appears to be a blank/unfilled template; no data present for any # field" -- and the engine built a Work Order out of it anyway. # # WHY THE EXISTING NON-REQUEST GUARD (below, step 5a) DID NOT CATCH IT: # that guard requires the word "request" to be ABSENT from the document. A # blank INTERPRETER REQUEST FORM is full of the word "request" -- it is in # the title, the labels, the instructions. The guard was looking for the # absence of request-language in a document that is nothing BUT # request-language. This check does not care what words are on the page; it # asks only whether any FIELD came back with a value. # # This refusal is the RIGHT outcome for a client's blank form too: staff get # "WORK ORDER COULD NOT BE GENERATED", see that the form arrived empty, and # go back to the client. That is a real thing staff need to know. What they # do not need is a Work Order that is a row of placeholders. _CORE_FIELDS = ("Language", "CalendarDate", "Time", "ClientOrganization", "ClientName", "LocationName", "JobSiteFullAddress", "LEPName", "SubjectMatter", "TypeOfMeeting") if not any(str(values.get(k, "")).strip() for k in _CORE_FIELDS): reason = ("NO DATA AT ALL: the model extracted not one field. This is a " "blank or unfilled form, not a request. No Work Order made.") _log("EMPTY-FORM GUARD -> refused: " + reason) return wf._emit({ "status": "refused", "engine_status": "rejected_no_valid_request", "form": None, "wo_path": None, "fields": {}, "flags": [reason], "reason": reason, "request_file": os.path.basename(request_file), "engine_version": AI_PATH_VERSION, "model": model}, 20) # MinHours: UTS bills EVERY client a 2-hour minimum. Compute it # deterministically here (never trust the model's arithmetic) from the raw # values the model reports: reserved hours if the form states them, else # the end-minus-start duration, else the 2-hour floor. Always >= 2; # durations round UP to the nearest 0.25 hour. # Normalize a compact start time ('10a' -> '10:00 am') and, if the service # date carries no year, infer it from a dated reference in the request. # NOTE: these flags are collected in a LOCAL list. They must NOT be pushed # into model_flags here -- model_flags is not defined until further below, # and doing so raised NameError, killing the AI path and silently dropping # the request into the legacy PHP extractor. _norm_flags = [] _t_new, _t_flag = _normalize_time_display(values.get("Time", "")) if _t_flag: values["Time"] = _t_new _norm_flags.append(_t_flag) _d_new, _d_flag = _infer_calendar_year(values.get("CalendarDate", ""), request_text) if _d_flag: values["CalendarDate"] = _d_new _norm_flags.append(_d_flag) # A start time with no AM/PM, resolved against the end time (this is what # billed a 1.5-hour job as 13.5 hours). _s_new, _s_flag = _infer_start_meridiem(values.get("Time", ""), data.get("EndTime", "")) if _s_flag: values["Time"] = _s_new _norm_flags.append(_s_flag) # The weekday is derived from the date, never trusted from the request. _w_flag = _weekday_check(values) if _w_flag: _norm_flags.append(_w_flag) # DETERMINISTIC multi-date guard -- never left to the model's discretion. _md_flag = _multidate_guard(values, request_text) if _md_flag: _norm_flags.append(_md_flag) _mh, _mh_flag = _compute_min_hours( values.get("Time", ""), str(data.get("EndTime", "") or ""), str(data.get("HoursReserved", "") or ""), ) values["MinHours"] = _mh _log("MinHours computed = %s (start=%r end=%r reserved=%r)" % (_mh, values.get("Time", ""), data.get("EndTime", ""), data.get("HoursReserved", ""))) model_flags = data.get("_flags") or [] if isinstance(model_flags, str): model_flags = [model_flags] # ---- IS THIS A REQUEST AT ALL? ------------------------------------- # THE MODEL DECIDES; the keyword rule may only veto TOWARD "invoice". # Graded on purpose, because the two error directions cost very # different amounts: # refusing a real request -> nobody gets an interpreter, and nobody # is watching for the silence # costing a real invoice -> a phantom WO and a burned job number, # which a human notices (and did) # So we only REFUSE when the model -- which actually read the document -- # says it is not a request. When the model says "request" and the keyword # rule disagrees, we still build the WO and FLAG it; makeWOP1 turns any # flag into "needs REVIEW" in the subject line, so a human looks. model_kind = str(data.get("_doc_type", "") or "").strip().lower() det_kind, det_why = _document_kind(request_text) _log("document type: model says '%s'; keyword rule says '%s' (%s)" % (model_kind or "(nothing)", det_kind, det_why)) if model_kind and model_kind != "request": why = str(data.get("_not_request_reason", "") or "").strip() if not why: why = "the model classified this document as '%s'" % model_kind _log("NOT A REQUEST -> refusing. " + why) return wf._emit({ "status": "not_a_request", "engine_status": "not_a_request", "form": None, "wo_path": None, "fields": {}, "flags": ["Not an interpreter request (%s): %s" % (model_kind, why), "Keyword check agreed." if det_kind == "invoice" else "Keyword check said '%s' (%s)." % (det_kind, det_why)], "reason": why, "doc_type": model_kind, "request_file": os.path.basename(request_file), "engine_version": AI_PATH_VERSION, "model": model}, 20) if det_kind == "invoice": # The model called it a request; the keyword rule sees an invoice. # Do NOT refuse on a disagreement -- build the WO and say so loudly. model_flags = list(model_flags) + [ "MAY NOT BE A REQUEST: the document reads like an invoice (%s). " "The AI judged it a request and the Work Order below was built " "anyway -- confirm before sending." % det_why] _log("DISAGREEMENT: model says request, keyword rule says invoice (%s) " "-> building the WO with a review flag." % det_why) meeting_name = str(data.get("MeetingContactName", "") or "").strip() meeting_num = str(data.get("MeetingContactNumber", "") or "").strip() # 4) Remote decision: trust the model's _remote, cross-check with the # deterministic detector; if they disagree, keep model's answer but flag. model_remote = bool(data.get("_remote", False)) flags = list(model_flags) flags.extend(_norm_flags) # time-normalization / inferred-year notes # ------------------------------------------------------------------ # REVIEW ITEMS ARE NOT THE SAME THING AS FLAGS. # ------------------------------------------------------------------ # _flags are NOTES: "weekday derived from date", "charge and judge paired # by position", "normalized '9:00am' -> '9:00 am'". They describe work the # engine did correctly and belong in the email body. # # For one day makeWOP1 treated ANY flag as a demand for review, and a batch # of 20 Work Orders arrived all saying "needs REVIEW" -- which is the same # as none of them saying it. Worse, the subject reason was guessed by # grepping the note text, so a flag reading "boilerplate ASL language, # unrelated to this request (Arabic)" was labelled "language", and one # reading "No remote/Zoom link present" was labelled "location". Both notes # said the opposite of what the subject claimed. # # So: the model states review items SEPARATELY, each with a short CODE, and # makeWOP1 builds the subject from the codes. No text is ever pattern-matched # to decide severity. review = [] for item in (data.get("_review") or []): if isinstance(item, dict): code = str(item.get("code", "") or "").strip().lower() note = str(item.get("note", "") or "").strip() else: s = str(item) code, _, note = s.partition("|") code, note = code.strip().lower(), note.strip() if not note: continue if not code: code = "check" # WEEKDAY: an UNMARKED day is not a problem. The calendar date implies # the weekday, so the field is obsolete -- deriving it silently is the # correct behaviour, not something to escalate. ONLY a marked day that # CONTRADICTS the date is worth a human's time. if code.startswith("weekday") and "conflict" not in code: flags.append(note) continue review.append({"code": code, "note": note}) if use_vision: flags.append("Read from a scanned image, not text -- every value was " "transcribed by eye from the form. Check against the " "attached original before sending.") review.append({"code": "handwritten", "note": "Every value was transcribed by eye from a scan. " "Check them against the attached original."}) if _mh_flag: flags.append(_mh_flag) # TWO DIFFERENT QUESTIONS -- do not conflate them: # # remote_template : "Is this actually a video job?" -> THE MODEL DECIDES. # The model reads the filled form and knows which box the client # ticked. A keyword scan cannot: many request FORMS carry the words # "Zoom" and "Remote" in their PRINTED BOILERPLATE -- e.g. the label # "REMOTE VIDEO INTERPRETING (RVI) (e.g. Zoom, GoogleMeet etc.)". # Letting the scan force the template put ON-SITE jobs on the remote # template with the street address stuffed into the Zoom placeholder # ("REMOTE VIDEO INTERPRETING: Washtenaw County CMH, 2140 E. # Ellsworth Road") -- an interpreter could read that as a video job # and never show up in person. 7 of 10 WOs in one batch were wrong # this way. The model was right every single time; the scan was not. # # remote_lenient : "Must this WO carry a physical street address?" # -> NO, if either the model OR the keyword scan sees Zoom/Remote. # This is the business rule as stated: the mere MENTION of Zoom or # Remote means we do not insist on an address (the Zoom details often # arrive in a separate message). Being lenient here is safe: the worst # case is that a flag is not raised. Being lenient about the TEMPLATE # is not safe -- it misinforms the interpreter. engine_remote = bool(request_text) and (not use_vision) and eng.is_remote_job(request_text) remote_template = model_remote remote_lenient = model_remote or engine_remote remote = remote_template # template + ZoomLink follow the model if not use_vision and model_remote != engine_remote: flags.append("Note _remote: model=%s, keyword scan=%s. Template follows " "the model (%s); the street-address requirement is relaxed." % (model_remote, engine_remote, "remote" if remote_template else "in-person")) # CONTACT PRIORITY (business rule, option b). The number the interpreter # should call from the field is the ON-SITE contact's, when the form # specifies one -- that is who to reach for an on-site problem. So promote # the on-site/meeting contact into the WO's Client Contact line; the # submitter is only the fallback (used when no on-site contact is given). # The Client Contact NAME follows the number: on-site name if given, else # BLANK -- never pair the submitter's name with the on-site number. The # submitter's contact is preserved in a flag so nothing is lost. onsite_promoted = False if meeting_num: sub_name = values.get("ClientContactName", "").strip() sub_num = values.get("ClientContactNumber", "").strip() values["ClientContactNumber"] = meeting_num values["ClientContactName"] = meeting_name # blank if no on-site name (option b) onsite_promoted = True if sub_name or sub_num: flags.append("Client Contact set to the on-site contact per rule; " "submitter contact (not on WO): %s %s" % (sub_name, sub_num)) # 5) Select template + fill (reuse the proven, merged-cell-safe fill). template_path, which, tnote = eng.select_template(remote, templates_dir) if template_path is None: # Template genuinely missing -> deterministic fallback rather than refuse. return _deterministic_fallback( request_file, docx_path, output_dir, templates_dir, eng, wf, "FORM %s template not found near %s" % (which, templates_dir), model, keyfile=keyfile) out_name = wf._safe_stem(request_file) + " -- WO.docx" output_path = os.path.join(output_dir, out_name) n = 2 while os.path.exists(output_path): output_path = os.path.join(output_dir, wf._safe_stem(request_file) + " (%d) -- WO.docx" % n) n += 1 # A JOB NUMBER, at last. Taken here -- AFTER the empty-form guard, so a blank # form never burns one -- and written INTO the document, not just the name. # The WO's own "JOB NUMBER:" field is what wo_filename reads, so putting it # in the document is all that is needed: the filename picks it up by itself. _job, _job_note = _take_job_number() if _job: values["JobNumber"] = str(_job) filled, blanked = eng.fill_template(template_path, values, output_path) # A client instruction about WHO should interpret goes into Section 1's # NAME cell -- in front of the staffer at assignment time, and overwritten # the moment they type the assigned interpreter's name. Done BEFORE the # file is renamed, like the job number, so nothing downstream re-reads a # stale copy. _terp_note = str(data.get("_interpreter_note", "") or "").strip() if _terp_note: if _write_interpreter_note_into_docx(output_path, _terp_note): flags.append("Client instruction about the interpreter was written " "into Section 1 NAME (it is replaced when the assigned " "interpreter's name is typed there): " + _terp_note) review.append({"code": "client request", "note": "The client asked for a specific interpreter " "or interpreter type: " + _terp_note}) else: flags.append("Client instruction about the interpreter could NOT be " "written into the WO; it is only here: " + _terp_note) review.append({"code": "client request", "note": "NOT on the Work Order -- " + _terp_note}) # Give the WO the filename staff would have typed. Never fatal -- see _name_wo. # On a map miss, prefer the abbreviation the MODEL suggested in this same # extraction call (free -- no extra API round-trip). If it gave none, the # filename module falls back to a network-free acronym. Map always wins. _ai_org_abbrev = str(data.get("ClientOrgAbbrev", "") or "").strip() _ai_inv = (lambda o: _ai_org_abbrev) if _ai_org_abbrev else None output_path, _fn_notes = _name_wo(output_path, abbrev_inventor=_ai_inv) flags.append("Extraction: AI (%s)." % model) if _job_note: flags.append(_job_note) flags.extend(_fn_notes) flags.append(tnote) if (meeting_name or meeting_num) and not onsite_promoted: flags.append("Meeting contact (not on WO): %s %s" % (meeting_name, meeting_num)) for field, reason in eng.plausibility_flags(values): flags.append("REVIEW %s: %s" % (field, reason)) # Lenient here: mentioning Zoom/Remote is enough to stop demanding a # physical street address (your rule). This CANNOT misinform anyone -- it # only decides whether a flag is raised. missing = eng.missing_value_flags(values, rule_map=None, is_remote=remote_lenient) for m in missing: flags.append("MISSING: %s" % m) # Non-request guard: a genuine interpreter request always carries a date, # a start time, a location (physical OR Zoom), and the word "request" # somewhere in its text. When ALL FOUR are absent this is not a request at # all (e.g. a forwarded text message or a notice that merely mentions an # interpreting job) -- discard it as a refusal rather than emit a hollow # partial WO. The word "request" is the strongest of the four signals. _no_date = not str(values.get("CalendarDate", "")).strip() _no_time = not str(values.get("Time", "")).strip() _no_loc = not any(str(values.get(k, "")).strip() for k in ("LocationName", "JobSiteFullAddress", "ZoomLink")) # The word "request" must appear as an actual word -- not merely inside an # email address such as request@univtrans.com, which routinely shows up in # the forwarded headers of Zoom/Meet-info replies. Require that "request" # is NOT immediately followed by "@". (Still matches request / requested / # requesting / requester -- all genuine request-language.) _no_request_word = not re.search(r"\brequest(?!@)", request_text or "", re.I) if _no_date and _no_time and _no_loc and _no_request_word: reason = ("discarded as non-request: no date, no start time, no location " "(physical or Zoom), and the word 'request' never appears") _log("NON-REQUEST GUARD -> refused: " + reason) return wf._emit({ "status": "refused", "engine_status": "rejected_no_valid_request", "form": None, "wo_path": None, "fields": {}, "flags": [reason], "reason": reason, "request_file": os.path.basename(request_file), "engine_version": AI_PATH_VERSION, "model": model}, 20) # 5b) DETERMINISTIC SAFETY NETS (wo_validator): recompute the arithmetic # and check every value is traceable to the request. Advisory only -- # these add flags, never downgrade a WO -- and wrapped so a validator # fault can never take the pipeline down with it. try: import wo_validator as _wv _v_flags = _wv.validate(values, request_text, data.get("EndTime", ""), data.get("HoursReserved", "")) if _v_flags: flags.extend(_v_flags) _log("VALIDATOR raised %d flag(s)" % len(_v_flags)) except Exception as _exc: _log("VALIDATOR skipped (%s)" % _exc) # 6) Same safety nets as the deterministic path: a WO with no usable # Calendar Date must go out flagged "partial", never clean. norm = "generated" cd = str(values.get("CalendarDate", "")).strip() date_flagged = any(re.search(r"MISSING:\s*Calendar Date|REVIEW\s+CalendarDate", str(f)) for f in flags) if not cd or date_flagged or missing: norm = "partial" out = { "status": norm, "engine_status": "ai_ok", "form": "AI", "template": os.path.basename(template_path), "wo_path": output_path if os.path.isfile(output_path) else None, "fields": values, "flags": flags, # Lenient: if the request so much as MENTIONS Zoom/Remote, makeWOP1 # carries any additional attachments (a separate Zoom-instructions # message) onto the WO email. Over-attaching is harmless; orphaning # the Zoom details is not. "is_remote": bool(remote_lenient), "remote_template": bool(remote_template), "reason": "", "request_file": os.path.basename(request_file), "engine_version": AI_PATH_VERSION, "model": model, } code = {"generated": 0, "partial": 10, "refused": 20}.get(norm, 1) _log("AI RESULT: status=%s template=%s wo=%s" % (norm, os.path.basename(template_path), os.path.basename(output_path) if os.path.isfile(output_path) else "(none)")) return wf._emit(out, code) finally: shutil.rmtree(tmp, ignore_errors=True) if __name__ == "__main__": try: sys.exit(main(sys.argv[1:])) except Exception: last = traceback.format_exc().strip().splitlines()[-1] try: import wo_from_one_request as wf wf._emit({"status": "error", "reason": "unhandled: " + last}, 1) except Exception: sys.stdout.write(json.dumps({"status": "error", "reason": "unhandled: " + last}) + "\n") sys.exit(1)