#!/usr/bin/env python3 """ winston_pipeline.py — Winston Linux quote pipeline (DVI-815). Port of the legacy outlook_auto/main.py + clickup_quotes.py to run on fileshare.icastinc.com (Linux), using app-only Graph (winston_graph.py) and the B1 convert-to-PDF recipe proven in DVI-807. No Windows/Excel COM required. Entry point: run(date='YYYY-MM-DD') Callable from the P2 scheduler and the P4 "Run now" UI button. Pipeline steps per quote task ------------------------------ 1. ClickUp pull — fetch quote tasks matching target date 2. Find + download — locate the .xlsm workbook in OneDrive by quote number 3. openpyxl extract — strip to 'Quote (Automated)', set print area, read H137 4. Graph convert — upload .xlsx to scratch drive, GET ?format=pdf, delete scratch 5. Mail draft — create draft in sales mailbox (idempotent: skip if exists) 6. ClickUp writeback — attach PDF to task + set Quote Amount custom field 7. Routed PDF upload — upload PDF to OneDrive (owner personal path or shared path) Configuration (env vars, with fallback to winston_settings.json) ----------------------------------------------------------------- CLICKUP_API_KEY ClickUp personal API token CLICKUP_LIST_ID ClickUp list ID to query for quote tasks CLICKUP_DATE_FIELD Task date field to filter on (default: date_updated) CLICKUP_QUOTE_AMOUNT_FIELD_ID Custom field ID for Quote Amount (auto-detected if absent) WINSTON_ONEDRIVE_USER UPN of the OneDrive owner holding Takeoffs folder WINSTON_TAKEOFFS_FOLDER OneDrive path to the .xlsm workbooks (see default below) WINSTON_SERVICE_USER UPN for the scratch drive used during xlsx→PDF conversion WINSTON_MAIL_FROM Draft mailbox UPN (default: sales@icastinc.com) AZURE_CLIENT_ID } App-only Graph credentials (already wired in DVI-806) AZURE_CLIENT_SECRET } AZURE_TENANT_ID } """ import html as _html import io import json import logging import os import re import threading from datetime import datetime, timezone from pathlib import Path import openpyxl import requests as _req from togen import winston_graph from togen.clickup_client import ClickUpClient log = logging.getLogger("togen.winston_pipeline") # --------------------------------------------------------------------------- # Cooperative cancellation (DVI-893) # # Python threads can't be forcibly killed, so a long-running Winston run is # stopped cooperatively: the "Stop" endpoint records the run_id here, and the # pipeline loop checks is_cancelled() at safe points (between quote tasks) and # returns early, marking the DiagnosticRun as "cancelled". # --------------------------------------------------------------------------- _cancel_lock = threading.Lock() _cancel_requested: set[str] = set() def request_cancel(run_id: str) -> None: """Flag a run for cooperative cancellation (called from the stop endpoint).""" with _cancel_lock: _cancel_requested.add(run_id) def is_cancelled(run_id: str) -> bool: with _cancel_lock: return run_id in _cancel_requested def _clear_cancel(run_id: str) -> None: with _cancel_lock: _cancel_requested.discard(run_id) # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- _SETTINGS_FILE = Path(__file__).resolve().parent / "winston_settings.json" _USER_SETTINGS_DIR = Path(__file__).resolve().parent / "user_settings" _EMAIL_TEMPLATE_FILE = Path(__file__).resolve().parent / "winston_email_template.json" _QUOTE_SHEET = "Quote (Automated)" _GRAND_TOTAL_CELL = "H137" _PRINT_AREA = "A1:AM138" # OneDrive folder containing the per-quote .xlsm workbooks. # Mirrors the legacy Windows path "Infrastructure Precast Inc_ - Takeoffs". _DEFAULT_TAKEOFFS_FOLDER = "Infrastructure Precast Inc_ - Takeoffs" # --------------------------------------------------------------------------- # Settings helpers (mirrors _load_winston_settings in app.py; no app.py import # since app.py starts background loops at import time) # --------------------------------------------------------------------------- _WINSTON_DEFAULTS: dict = { "draft_mailbox": "sales@icastinc.com", "clickup_api_key": "", "clickup_list_id": "", # Team/workspace ID used when no list_id is set (matches PoC team-wide query). # iCast Inc Workspace = 9014248643 "clickup_team_id": "9014248643", "clickup_date_field": "", "onedrive_user": "", "takeoffs_folder": "", "shared_quotes_location": "", "shared_quote_takeoff_location": "", } def _load_settings() -> dict: settings = dict(_WINSTON_DEFAULTS) if _SETTINGS_FILE.is_file(): try: saved = json.loads(_SETTINGS_FILE.read_text()) if isinstance(saved, dict): settings.update(saved) except (json.JSONDecodeError, OSError): pass return settings def _load_user_docs(email: str) -> dict: """Return the 'documents' sub-dict from a user's personal settings JSON.""" safe = re.sub(r"[^a-zA-Z0-9@._-]", "_", email.lower()) fp = _USER_SETTINGS_DIR / f"{safe}.json" if fp.is_file(): try: return json.loads(fp.read_text()).get("documents", {}) except (json.JSONDecodeError, OSError): pass return {} # --------------------------------------------------------------------------- # Email draft template (DVI-857) # # The subject/body/recipients of Winston draft emails are driven by an # admin-editable template stored in winston_email_template.json and managed # from the Winston → Admin → Email tab. The DEFAULTS below reproduce the exact # subject and body Winston produced before DVI-857, so behavior is unchanged # until an admin edits the template. # # Templates use {token} placeholders. Available tokens (see EMAIL_TOKENS) are # fed from the pipeline source data for each quote. Token VALUES are HTML-escaped # when substituted into the body (values come from ClickUp/OneDrive data), while # the template body itself is authored by a togen admin and treated as trusted. # --------------------------------------------------------------------------- EMAIL_TEMPLATE_DEFAULTS: dict = { "subject": "Quote {quote_number} — iCast Infrastructure", "body_html": ( "

Please find attached the quote for {quote_number}.

" "

Quote Total: {quote_total}

" "

This quote was prepared by iCast Infrastructure. " "Please review and let us know if you have any questions.

" "

Thank you,
iCast Infrastructure Sales Team

" ), "cc": [], "bcc": [], "fallback_to": [], } # Token catalogue surfaced in the Email tab legend: token -> where it comes from. EMAIL_TOKENS: list[dict] = [ {"token": "quote_number", "source": "ClickUp task — the quote number (workbook / PDF name)"}, {"token": "quote_total", "source": "Take-off workbook cell H137, currency-formatted (e.g. $1,234.56)"}, {"token": "recipient_email", "source": "ClickUp task — the quote's To recipient"}, {"token": "owner_email", "source": "ClickUp task — the quote owner / salesperson"}, {"token": "date", "source": "Pipeline run target date (YYYY-MM-DD)"}, {"token": "mailbox", "source": "Configured send-as mailbox (Settings → Draft mailbox)"}, ] # --------------------------------------------------------------------------- # Per-pipeline templates (DVI-859 — Option C) # # The on-disk store keeps the single global template's fields at the top level # (backward compatible with the DVI-857 format) plus two additive keys: # per_pipeline_enabled : bool — feature flag; when false the single global # template is always used (current behavior). # pipelines : dict — { pipeline_id: {sparse field overrides} } # # Per-pipeline overrides are SPARSE: only fields the admin actually customized # are stored. Any field a pipeline doesn't set falls back to the global template # ("fallback to using the single customization"). The feature ships disabled so # behavior is unchanged; turning the flag on "simply enables" per-pipeline # customization without any code change. # --------------------------------------------------------------------------- def _load_email_store() -> dict: """Return the raw on-disk store (global fields + pipelines + flag).""" if _EMAIL_TEMPLATE_FILE.is_file(): try: saved = json.loads(_EMAIL_TEMPLATE_FILE.read_text()) if isinstance(saved, dict): return saved except (json.JSONDecodeError, OSError): pass return {} def _global_template(store: dict) -> dict: """Resolve the global template = DEFAULTS overlaid with saved top-level fields.""" tpl = dict(EMAIL_TEMPLATE_DEFAULTS) for k, default in EMAIL_TEMPLATE_DEFAULTS.items(): if k in store and type(store[k]) is type(default): tpl[k] = store[k] return tpl def per_pipeline_enabled() -> bool: """Whether per-pipeline email templates are turned on.""" return bool(_load_email_store().get("per_pipeline_enabled")) def set_per_pipeline_enabled(enabled: bool) -> None: store = _load_email_store() store["per_pipeline_enabled"] = bool(enabled) _EMAIL_TEMPLATE_FILE.write_text(json.dumps(store, indent=2)) def get_pipeline_override(pipeline_id: str) -> dict: """Return the sparse override dict saved for a pipeline (empty if none).""" pipelines = _load_email_store().get("pipelines") ov = pipelines.get(pipeline_id) if isinstance(pipelines, dict) else None return dict(ov) if isinstance(ov, dict) else {} def pipeline_override_ids() -> list: """Pipeline ids that currently have a saved per-pipeline override.""" pipelines = _load_email_store().get("pipelines") return list(pipelines.keys()) if isinstance(pipelines, dict) else [] def load_email_template(pipeline_id: str | None = None) -> dict: """Resolve the effective email template for an optional pipeline. Returns the global template, overlaid per-field with any saved override for ``pipeline_id`` (Option C). The overlay only applies when the per-pipeline feature is enabled AND a (non-empty) override exists for that field; otherwise the single global template is used (fallback). """ store = _load_email_store() tpl = _global_template(store) if pipeline_id and store.get("per_pipeline_enabled"): overrides = store.get("pipelines") ov = overrides.get(pipeline_id) if isinstance(overrides, dict) else None if isinstance(ov, dict): for k, default in EMAIL_TEMPLATE_DEFAULTS.items(): if k in ov and type(ov[k]) is type(default): tpl[k] = ov[k] return tpl def save_email_template(tpl: dict, pipeline_id: str | None = None) -> None: """Persist template fields. When ``pipeline_id`` is None the fields update the single global template. When set, the fields are stored as a SPARSE override for that pipeline: empty strings / empty lists are dropped so those fields keep inheriting the global template. An override with no non-empty fields is removed entirely. """ store = _load_email_store() if pipeline_id: pipelines = store.get("pipelines") if not isinstance(pipelines, dict): pipelines = {} override: dict = {} for k, default in EMAIL_TEMPLATE_DEFAULTS.items(): if k not in tpl or type(tpl[k]) is not type(default): continue val = tpl[k] if (isinstance(val, str) and not val.strip()) or (isinstance(val, list) and not val): continue # empty → inherit global override[k] = val if override: pipelines[pipeline_id] = override else: pipelines.pop(pipeline_id, None) store["pipelines"] = pipelines else: for k, default in EMAIL_TEMPLATE_DEFAULTS.items(): if k in tpl and type(tpl[k]) is type(default): store[k] = tpl[k] _EMAIL_TEMPLATE_FILE.write_text(json.dumps(store, indent=2)) def _format_total(grand_total: object) -> str: if isinstance(grand_total, (int, float)): return f"${grand_total:,.2f}" return str(grand_total) if grand_total is not None else "—" def build_email_context(task: dict, grand_total: object, date: str, mailbox: str) -> dict: """Map pipeline source data to the substitution tokens for a quote.""" return { "quote_number": task.get("quote_number", ""), "quote_total": _format_total(grand_total), "recipient_email": task.get("recipient_email", ""), "owner_email": task.get("owner_email", ""), "date": date or "", "mailbox": mailbox or "", } def render_template(text: str, context: dict, *, escape: bool = False) -> str: """Substitute {token} placeholders from context. Unknown tokens are left untouched (so stray braces in authored HTML/CSS are preserved). When escape=True, substituted values are HTML-escaped — used for the body, where values come from external data. Subject uses escape=False. """ def _repl(m: "re.Match") -> str: key = m.group(1) if key in context: val = str(context[key]) return _html.escape(val) if escape else val return m.group(0) return re.sub(r"\{(\w+)\}", _repl, text or "") def render_email(template: dict, context: dict) -> tuple[str, str]: """Render (subject, body_html) from a template + context.""" subject = render_template(template.get("subject", ""), context) \ or render_template(EMAIL_TEMPLATE_DEFAULTS["subject"], context) body_html = render_template(template.get("body_html", ""), context, escape=True) \ or render_template(EMAIL_TEMPLATE_DEFAULTS["body_html"], context, escape=True) return subject, body_html def sample_email_context() -> dict: """A representative context for the Email tab 'Preview with sample data'.""" return { "quote_number": "26-2904", "quote_total": "$48,750.00", "recipient_email": "customer@example.com", "owner_email": "jsmith@icastinc.com", "date": datetime.now(timezone.utc).strftime("%Y-%m-%d"), "mailbox": "sales@icastinc.com", } # --------------------------------------------------------------------------- # Step 2: Quote Take-offs workbook lookup (isolated — DVI-843) # --------------------------------------------------------------------------- def _winston_find_workbook( token: str, task: dict, settings: dict, onedrive_user: str, takeoffs_folder: str ) -> tuple["dict | None", str, str]: """Locate the xlsm workbook using shared-first → personal-fallback rule. Search order: 1. Shared Quote Take-offs (shared_quote_takeoff_location; falls back to the legacy takeoffs_folder when the new field is empty). 2. Owner's personal Quote Take-offs (quote_takeoff_location from their User Settings Documents tab) — only tried when the owner email is known and the personal path is configured. Returns (item, user_upn, folder_path) on success, or (None, '', '') if the workbook was not found in either location. Existing behavior is preserved when shared_quote_takeoff_location is empty (falls back to takeoffs_folder). """ qn = task["quote_number"] owner_email = task.get("owner_email", "").strip() # 1. Shared Quote Take-offs (primary) — SharePoint site drive (DVI-947). # The shared library lives in SharePoint, not the pipeline account's # OneDrive (togen@ has no mysite), so address the site drive directly. shared_takeoff = settings.get("shared_quote_takeoff_location", "").strip() primary_folder = shared_takeoff or takeoffs_folder # legacy fallback if primary_folder: item = winston_graph.site_find_quote_workbook(token, primary_folder, qn) if item: return item, winston_graph.SITE_DRIVE_UPN, primary_folder # 2. Personal Quote Take-offs (fallback) if owner_email: docs = _load_user_docs(owner_email) personal_takeoff = docs.get("quote_takeoff_location", "").strip() if personal_takeoff: item = winston_graph.find_quote_workbook(token, owner_email, personal_takeoff, qn) if item: return item, owner_email, personal_takeoff return None, "", "" # --------------------------------------------------------------------------- # Step 3: openpyxl extract (B1 recipe — DVI-807) # --------------------------------------------------------------------------- def sample_workbook_structure( wb, *, sample_rows: int = 20, sample_cols: int = 12, cell_cap: int = 200 ) -> dict: """Summarize a loaded workbook for the run-diagnostics inspector (DVI-948). For every sheet: name, max_row×max_col dimensions, and a top-left cell grid (sample_rows × sample_cols of cached values) so the dialog can show which sheet holds the quote and where the grand total sits. Small enough to store inline in the step detail and as a JSON artifact. """ sheets: list[dict] = [] for name in wb.sheetnames: ws = wb[name] try: max_row = int(ws.max_row or 0) max_col = int(ws.max_column or 0) except Exception: max_row = max_col = 0 grid: list[list[str]] = [] for r in range(1, min(sample_rows, max_row) + 1): row_vals: list[str] = [] for c in range(1, min(sample_cols, max_col) + 1): v = ws.cell(row=r, column=c).value s = "" if v is None else str(v) if len(s) > cell_cap: s = s[:cell_cap] + "…" row_vals.append(s) grid.append(row_vals) sheets.append({ "name": name, "max_row": max_row, "max_col": max_col, "sample": grid, }) return { "sheets": sheets, "sheet_names": list(wb.sheetnames), "target_sheet": _QUOTE_SHEET, "grand_total_cell": _GRAND_TOTAL_CELL, "matched": _QUOTE_SHEET in wb.sheetnames, } def _extract_from_workbook(wb) -> tuple[bytes, object]: """Strip a loaded workbook to the quote sheet, apply print area, read H137.""" if _QUOTE_SHEET not in wb.sheetnames: raise ValueError( f"Sheet '{_QUOTE_SHEET}' not found in workbook. " f"Available sheets: {wb.sheetnames}" ) # Drop every sheet except the quote sheet (removes the 15.9 MB Structure Input # sheet that causes Graph to return 406 UnsupportedMediaType). for name in [s for s in wb.sheetnames if s != _QUOTE_SHEET]: del wb[name] wb.active = wb[_QUOTE_SHEET] ws = wb.active ws.print_area = _PRINT_AREA # prevents trailing blank page on PDF grand_total = ws[_GRAND_TOTAL_CELL].value buf = io.BytesIO() wb.save(buf) xlsx_bytes = buf.getvalue() log.info( "_extract_from_workbook: H137=%r xlsx=%d KB", grand_total, len(xlsx_bytes) // 1024 ) return xlsx_bytes, grand_total def _extract_quote_sheet(xlsm_bytes: bytes) -> tuple[bytes, object]: """Strip the xlsm to the 'Quote (Automated)' sheet, apply print area, read H137. Returns (xlsx_bytes, grand_total). Raises ValueError if the required sheet is missing. """ wb = openpyxl.load_workbook(io.BytesIO(xlsm_bytes), data_only=True, keep_vba=False) return _extract_from_workbook(wb) # --------------------------------------------------------------------------- # Step 5: Mail draft dedupe # --------------------------------------------------------------------------- def _draft_exists(token: str, mailbox: str, quote_number: str) -> bool: """Return True if the sales mailbox Drafts folder already has a message whose subject contains quote_number (Graph OData contains() filter). """ url = f"{winston_graph._GRAPH_BASE}/users/{mailbox}/mailFolders/drafts/messages" # contains() on subject is supported for mail messages in Graph OData safe_qn = quote_number.replace("'", "''") params = { "$filter": f"contains(subject,'{safe_qn}')", "$select": "id,subject", "$top": "1", } try: r = _req.get( url, headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"}, params=params, timeout=30, ) if r.status_code == 200: if r.json().get("value"): log.info("Draft already exists for %s — skipping (dedupe)", quote_number) return True else: log.warning( "_draft_exists: Graph returned %s for %s: %s", r.status_code, quote_number, r.text[:200], ) except Exception: log.exception("_draft_exists error for quote %s", quote_number) return False # --------------------------------------------------------------------------- # Step 7: Routing (isolated per spec — swap rule here without touching pipeline) # --------------------------------------------------------------------------- def _winston_route_destination(quote: dict, settings: dict) -> tuple[str, str]: """Return (user_upn, folder_path) for the completed PDF upload (Completed Quotes). Rule (board decision 2 — DVI-805, updated DVI-843): If the quote owner has a personal Completed Quotes path (quotes_location) in their User Settings Documents tab → route there (their OneDrive). Else → shared Completed Quotes path (shared_quotes_location) in Togen Admin. The routing *criteria* (owner-based: personal overrides shared) is isolated here so it can be swapped without touching the rest of the pipeline. """ owner_email = quote.get("owner_email", "").strip() if owner_email: docs = _load_user_docs(owner_email) personal_path = docs.get("quotes_location", "").strip() if personal_path: return owner_email, personal_path # Shared fallback: the configured shared Completed Quotes location lives in # the SharePoint site library, not a personal OneDrive (DVI-947). Signal the # site drive with the SITE_DRIVE_UPN sentinel. shared_path = settings.get("shared_quotes_location", "").strip() return winston_graph.SITE_DRIVE_UPN, shared_path # --------------------------------------------------------------------------- # Per-run summary report (DVI-976) # # After every run (scheduled or manual, including dry runs and early-exit # paths) email the configured Report Recipients an aggregate summary. On a # dry run the report is clearly banner-flagged so recipients know no external # writes were performed. # --------------------------------------------------------------------------- _REPORT_METRICS = [ ("quotes_found", "Quotes found"), ("drafts_created", "Drafts created"), ("drafts_skipped", "Drafts skipped (already existed)"), ("drafts_failed", "Drafts failed"), ("clickup_amounts_set", "ClickUp amounts set"), ("clickup_amounts_failed", "ClickUp amounts failed"), ("clickup_attached", "ClickUp PDFs attached"), ("clickup_failed", "ClickUp attach failed"), ("pdf_routed", "PDFs routed to OneDrive"), ("pdf_route_failed", "PDF routing failed"), ("writes_skipped_dry_run", "Writes skipped (dry run)"), ] def _build_report_html(summary: dict, date: str, trigger: str) -> str: """Render the per-run summary as branded HTML for the report email.""" dry_run = bool(summary.get("dry_run")) esc = _html.escape banner = "" if dry_run: banner = ( '
' 'DRY RUN — no actual actions were taken. No Outlook drafts, ClickUp ' 'writebacks, or file uploads were performed. The counts below reflect ' 'what would have happened.
' ) notes = [] if summary.get("config_errors"): notes.append("Configuration errors: " + esc("; ".join(summary["config_errors"]))) if summary.get("error"): notes.append("Run error: " + esc(str(summary["error"]))) if summary.get("cancelled"): notes.append("Run was cancelled before completing.") notes_html = "" if notes: notes_html = ( '
' + "
".join(notes) + "
" ) metric_rows = "".join( f'{esc(label)}' f'{int(summary.get(key, 0))}' for key, label in _REPORT_METRICS ) quotes = summary.get("quotes") or [] quote_rows = "" for q in quotes: qn = esc(str(q.get("quote_number", q.get("quote", "—")))) recipient = esc(str(q.get("recipient", "") or "—")) draft = esc(str(q.get("draft_status", "—"))) route = esc(str(q.get("pdf_route", "—"))) err = q.get("error") err_html = (f'
{esc(str(err))}
' if err else "") quote_rows += ( f'{qn}{err_html}' f'{recipient}' f'{draft}' f'{route}' ) if not quote_rows: quote_rows = ('No quotes processed.') return f"""\

Winston Pipeline Run Report

Date: {esc(date)}  |  Trigger: {esc(str(trigger))}  |  Run ID: {esc(str(summary.get('run_id', '')))}

{banner} {notes_html}

Summary

{metric_rows}

Quotes

{quote_rows}
Quote Recipient Draft PDF route

Automated report from Winston. Recipients are configured under Bots → Winston → Admin → Settings → Report Recipients.

""" def _send_run_report(summary: dict, date: str, trigger: str) -> bool: """Email the run summary to the configured Report Recipients (DVI-976). Sends regardless of dry-run mode. Returns True if a message was accepted, False if there were no recipients or the send failed. """ settings = _load_settings() to_list = [e for e in (settings.get("report_to") or []) if e] cc_list = [e for e in (settings.get("report_cc") or []) if e] if not to_list and not cc_list: log.info("Winston report: no Report Recipients configured — skipping send") return False token = winston_graph.acquire_graph_token() if not token: log.error("Winston report: could not acquire Graph token — report not sent") return False dry_run = bool(summary.get("dry_run")) subject = ( f"Winston pipeline report — {date}" + (" [DRY RUN]" if dry_run else "") ) html_body = _build_report_html(summary, date, trigger) mailbox = (settings.get("draft_mailbox") or winston_graph.WINSTON_MAIL_FROM).strip() sent = winston_graph.mail_send( token, subject, html_body, to_list, mailbox=mailbox, cc_addresses=cc_list, ) if sent: log.info("Winston report sent to=%d cc=%d dry_run=%s run_id=%s", len(to_list), len(cc_list), dry_run, summary.get("run_id")) else: log.error("Winston report send failed run_id=%s", summary.get("run_id")) return sent # --------------------------------------------------------------------------- # Main entry point # --------------------------------------------------------------------------- def run(date: str, trigger: str = "manual", run_id: str | None = None, pipeline_id: str | None = None, dry_run: bool = False) -> dict: """Run the pipeline and email the per-run summary report (DVI-976). Thin wrapper around :func:`_run_impl` that always attempts to send the summary report to the configured Report Recipients afterwards — including on dry runs and early-exit paths (config errors, no quotes, token failure). Report delivery never affects the returned summary or raises. """ summary = _run_impl(date, trigger, run_id=run_id, pipeline_id=pipeline_id, dry_run=dry_run) try: _send_run_report(summary, date, trigger) except Exception: log.exception("Winston summary report send failed (run_id=%s)", summary.get("run_id")) return summary def _run_impl(date: str, trigger: str = "manual", run_id: str | None = None, pipeline_id: str | None = None, dry_run: bool = False) -> dict: """Process all quote tasks for the given date (YYYY-MM-DD). Returns a summary dict with aggregate counts and per-quote results. Designed to be called by the scheduler and the "Run now" UI endpoint. trigger: source of the run ("manual", "user", "scheduler:"). run_id: pre-generated UUID to use for the DiagnosticRun (so callers can return it to the UI before the background thread starts). pipeline_id: Date Logic pipeline id driving this run. Selects the per-pipeline email template override when one exists (Option C); falls back to the global template otherwise. dry_run: when True (DVI-908) the pipeline validates the full flow — ClickUp pull, workbook download, xlsx extract, xlsx→PDF convert, and diagnostics artifacts — but performs NO external writes: no Outlook draft, no ClickUp writeback, no Completed-Quotes/OneDrive upload. The generated PDF/xlsx are still saved as diagnostics artifacts (a temporary location) so the result is reviewable in Diagnostics without touching production systems. """ from togen.bot_diagnostics import DiagnosticRun log.info("Winston pipeline starting — date=%s trigger=%s dry_run=%s", date, trigger, dry_run) settings = _load_settings() # Resolve configuration (env vars take precedence over settings file) clickup_api_key = ( os.environ.get("CLICKUP_API_KEY") or settings.get("clickup_api_key", "") ).strip() clickup_list_id = ( settings.get("clickup_list_id", "") or os.environ.get("CLICKUP_LIST_ID", "") ).strip() clickup_team_id = ( settings.get("clickup_team_id", "") or os.environ.get("CLICKUP_TEAM_ID", "") ).strip() # date_field default: date_updated works for both list and team queries _default_date_field = "date_updated" date_field = ( settings.get("clickup_date_field", "") or os.environ.get("CLICKUP_DATE_FIELD", _default_date_field) ) global_quote_amount_field_id = os.environ.get("CLICKUP_QUOTE_AMOUNT_FIELD_ID", "").strip() takeoffs_folder = ( settings.get("takeoffs_folder", "") or os.environ.get("WINSTON_TAKEOFFS_FOLDER", _DEFAULT_TAKEOFFS_FOLDER) ) mailbox = (settings.get("draft_mailbox") or winston_graph.WINSTON_MAIL_FROM).strip() onedrive_user = settings.get("onedrive_user", "") or winston_graph.WINSTON_ONEDRIVE_USER # Email draft template (DVI-857/859): admin-editable subject/body/recipients. # Resolves the per-pipeline override when set (Option C), else the global one. email_template = load_email_template(pipeline_id) diag = DiagnosticRun(bot="winston", trigger=trigger, target_date=date, run_id=run_id) summary: dict = { "run_id": diag.id, "date": date, "dry_run": bool(dry_run), "writes_skipped_dry_run": 0, "quotes_found": 0, "drafts_created": 0, "drafts_skipped": 0, "drafts_failed": 0, "clickup_amounts_set": 0, "clickup_amounts_skipped": 0, "clickup_amounts_failed": 0, "clickup_attached": 0, "clickup_skipped": 0, "clickup_failed": 0, "pdf_routed": 0, "pdf_route_failed": 0, "quotes": [], } with diag: # Guard: required config errors = [] if not clickup_api_key: errors.append("missing_clickup_api_key (set CLICKUP_API_KEY or clickup_api_key in winston_settings.json)") if not clickup_list_id and not clickup_team_id: errors.append("missing_clickup_source: set clickup_list_id or clickup_team_id in winston_settings.json") if not onedrive_user: errors.append("missing_onedrive_user (set WINSTON_ONEDRIVE_USER env var)") if errors: for e in errors: log.error("Winston pipeline config error: %s", e) summary["config_errors"] = errors diag.fail("; ".join(errors)) return summary # ------------------------------------------------------------------ # Step 1: ClickUp pull # ------------------------------------------------------------------ tasks: list = [] with diag.step("clickup_pull") as s1: clickup = ClickUpClient(clickup_api_key) if clickup_list_id: tasks = clickup.get_quote_tasks(clickup_list_id, date, date_field=date_field) else: # PoC-compatible: team-wide query by due_date (no list scope) tasks = clickup.get_quote_tasks_by_team(clickup_team_id, date, date_field=date_field) summary["quotes_found"] = len(tasks) log.info("ClickUp pull: %d task(s) for %s", len(tasks), date) # DVI-972: surface the retrieved ClickUp data in the Diagnostics # pane — a query header + per-task table (rendered client-side) plus # a downloadable clickup_tasks.json artifact for deep inspection. s1.set_detail({ "clickup_query": { "scope": "list" if clickup_list_id else "team", "list_id": clickup_list_id or None, "team_id": None if clickup_list_id else clickup_team_id, "target_date": date, "date_field": date_field, "count": len(tasks), }, "tasks": tasks, }) try: diag.add_artifact( "clickup_tasks.json", json.dumps(tasks, default=str, ensure_ascii=False).encode("utf-8"), "json", step=s1, ) except Exception: log.exception("Failed to persist clickup_tasks.json artifact") if not tasks: log.info("No quote tasks found for %s — done", date) diag.set_summary(summary) return summary # Acquire Graph token once for the whole batch token = winston_graph.acquire_graph_token() if not token: log.error("Failed to acquire Graph token — aborting") summary["error"] = "graph_token_failed" diag.fail("graph_token_failed") return summary for task in tasks: if is_cancelled(diag.id): log.warning("Winston run %s cancelled by user — stopping", diag.id) summary["cancelled"] = True diag.cancel() break qn = task["quote_number"] task_id = task["clickup_task_id"] result: dict = {"quote_number": qn, "task_id": task_id} # ------------------------------------------------------------------ # Step 2: Find + download xlsm from OneDrive # ------------------------------------------------------------------ xlsm_bytes: bytes | None = None xlsm_name = f"{qn}.xlsm" pdf_filename = re.sub(r"\.xlsm$", ".pdf", xlsm_name, flags=re.IGNORECASE) with diag.step(f"find_download:{qn}") as s2: try: item, _src_upn, _src_folder = _winston_find_workbook( token, task, settings, onedrive_user, takeoffs_folder ) if not item: log.warning( "Workbook not found for quote %s in any configured " "Quote Take-offs location", qn ) result["error"] = "workbook_not_found" s2.warn() else: xlsm_name = item.get("name", xlsm_name) pdf_filename = re.sub(r"\.xlsm$", ".pdf", xlsm_name, flags=re.IGNORECASE) # DVI-948: record the found file's metadata so the run # diagnostics dialog can show a File card (name/path/size/ # modified) + an Open-online link. _parent = item.get("parentReference") or {} _ppath = _parent.get("path") or "" _full_path = (f"{_ppath}/{xlsm_name}" if _ppath else f"{_src_folder}/{xlsm_name}") s2.set_detail({ "file_name": xlsm_name, "source": _src_upn, "folder": _src_folder, "full_path": _full_path, "size_kb": (round((item.get("size") or 0) / 1024, 1) if item.get("size") is not None else None), "modified": item.get("lastModifiedDateTime"), "web_url": item.get("webUrl"), }) dl_url = item.get("@microsoft.graph.downloadUrl") if dl_url: dl_resp = _req.get(dl_url, timeout=120) xlsm_bytes = dl_resp.content if dl_resp.status_code == 200 else None elif _src_upn == winston_graph.SITE_DRIVE_UPN: xlsm_bytes = winston_graph.site_drive_read_file( token, f"{_src_folder}/{xlsm_name}" ) else: xlsm_bytes = winston_graph.user_drive_read_file( token, _src_upn, f"{_src_folder}/{xlsm_name}" ) if xlsm_bytes: log.info("Downloaded %s (%d KB)", xlsm_name, len(xlsm_bytes) // 1024) else: log.error("Download failed for %s", xlsm_name) result["error"] = "download_failed" s2.warn() except Exception as exc: log.exception("Download exception for quote %s", qn) result["error"] = f"download_exception: {exc}" s2.warn() if not xlsm_bytes: summary["quotes"].append(result) continue # ------------------------------------------------------------------ # Step 3: openpyxl extract (B1 recipe) # ------------------------------------------------------------------ xlsx_bytes: bytes | None = None grand_total: object = None with diag.step(f"openpyxl_extract:{qn}") as s3: try: # DVI-948: load once, capture a workbook-structure snapshot for # the diagnostics inspector (always), then run the B1 extract. wb = openpyxl.load_workbook( io.BytesIO(xlsm_bytes), data_only=True, keep_vba=False ) structure = sample_workbook_structure(wb) try: diag.add_artifact( "workbook_structure.json", json.dumps(structure, default=str, ensure_ascii=False).encode("utf-8"), "json", step=s3, ) except Exception: log.warning("could not store workbook_structure.json for %s", qn) s3.set_detail({ "target_sheet": _QUOTE_SHEET, "grand_total_cell": _GRAND_TOTAL_CELL, "matched": structure["matched"], "workbook_structure": structure, }) xlsx_bytes, grand_total = _extract_from_workbook(wb) result["grand_total"] = grand_total result["xlsx_kb"] = len(xlsx_bytes) // 1024 diag.add_artifact(f"{qn}.xlsx", xlsx_bytes, "xlsx", step=s3) except Exception as exc: log.error("openpyxl extract failed for %s: %s", qn, exc) result["error"] = f"openpyxl_failed: {exc}" s3.warn() # DVI-948: on failure keep the source .xlsm so it can be # downloaded and inspected from the diagnostics dialog. try: diag.add_artifact(xlsm_name, xlsm_bytes, "xlsm", step=s3) except Exception: log.warning("could not store source xlsm artifact for %s", qn) if xlsx_bytes is None: summary["quotes"].append(result) continue # ------------------------------------------------------------------ # Step 4: Graph xlsx → PDF (B1 recipe) # ------------------------------------------------------------------ pdf_bytes: bytes | None = None with diag.step(f"graph_convert:{qn}") as s4: try: # Stable scratch name prevents collisions with concurrent runs scratch_name = re.sub(r"[^a-zA-Z0-9_-]", "_", qn) + ".xlsx" pdf_bytes = winston_graph.xlsx_to_pdf( token, xlsx_bytes, scratch_filename=scratch_name ) if pdf_bytes: result["pdf_kb"] = len(pdf_bytes) // 1024 diag.add_artifact(pdf_filename, pdf_bytes, "pdf", step=s4) else: log.error("Graph PDF conversion failed for %s", qn) result["error"] = "pdf_conversion_failed" s4.warn() except Exception as exc: log.exception("Graph convert exception for quote %s", qn) result["error"] = f"pdf_conversion_exception: {exc}" s4.warn() if not pdf_bytes: summary["quotes"].append(result) continue # ------------------------------------------------------------------ # Step 5: Outlook draft (idempotent — skip if already in Drafts) # ------------------------------------------------------------------ with diag.step(f"outlook_draft:{qn}") as s5: try: recipient = task.get("recipient_email", "") if dry_run: # Render what would be sent (for Diagnostics visibility) # but create no draft. ctx = build_email_context(task, grand_total, date, mailbox) subject, _body = render_email(email_template, ctx) result["draft_status"] = "dry_run (no draft created)" result["recipient"] = recipient or "(no recipient)" summary["writes_skipped_dry_run"] += 1 log.info( "[DRY RUN] Would create draft quote=%s to=%s mailbox=%s subject=%r", qn, recipient or "(none)", mailbox, subject, ) elif _draft_exists(token, mailbox, qn): result["draft_status"] = "skipped" summary["drafts_skipped"] += 1 else: # Recipients: prefer the quote's ClickUp recipient; if none, # fall back to the template's configured fallback list. if recipient: to_list = [recipient] else: to_list = [a for a in email_template.get("fallback_to", []) if a] ctx = build_email_context(task, grand_total, date, mailbox) subject, body_html = render_email(email_template, ctx) draft = winston_graph.mail_create_draft( token, subject=subject, body_html=body_html, to_addresses=to_list, mailbox=mailbox, cc_addresses=[a for a in email_template.get("cc", []) if a], bcc_addresses=[a for a in email_template.get("bcc", []) if a], attachments=[ winston_graph.encode_attachment(pdf_filename, pdf_bytes) ], ) if draft: result["draft_status"] = "created" result["draft_mailbox"] = mailbox result["recipient"] = recipient or "(no recipient)" summary["drafts_created"] += 1 log.info( "Draft created quote=%s to=%s mailbox=%s", qn, recipient or "(none)", mailbox, ) else: result["draft_status"] = "failed" summary["drafts_failed"] += 1 s5.warn() except Exception as exc: log.exception("Mail draft exception for quote %s", qn) result["draft_status"] = "failed" summary["drafts_failed"] += 1 s5.warn() # ------------------------------------------------------------------ # Step 6: ClickUp write-back + Step 7: Routed PDF upload # ------------------------------------------------------------------ with diag.step(f"clickup_writeback_route:{qn}") as s67: try: field_id = global_quote_amount_field_id or task.get("quote_amount_field_id", "") if dry_run: # Validate routing/targets and log intent, but write nothing. result["clickup_amount"] = ( f"dry_run (would set {_format_total(grand_total)})" if field_id and grand_total is not None else "dry_run (skipped)" ) result["clickup_attach"] = f"dry_run (would attach {pdf_filename})" dest_upn, dest_folder = _winston_route_destination(task, settings) if dest_folder: dest_path = f"{dest_folder.rstrip('/')}/{pdf_filename}" result["pdf_route"] = f"dry_run (would upload to {dest_upn}:{dest_path})" else: result["pdf_route"] = "dry_run (no destination configured)" summary["writes_skipped_dry_run"] += 1 log.info( "[DRY RUN] Would write back quote=%s: clickup_amount=%r attach=%r route=%r", qn, result["clickup_amount"], result["clickup_attach"], result["pdf_route"], ) summary["quotes"].append(result) continue # 6a: Quote Amount custom field if field_id and grand_total is not None: ok = clickup.set_custom_field(task_id, field_id, grand_total) if ok: result["clickup_amount"] = "set" summary["clickup_amounts_set"] += 1 else: result["clickup_amount"] = "failed" summary["clickup_amounts_failed"] += 1 else: reason = "no_field_id" if not field_id else "no_grand_total" result["clickup_amount"] = f"skipped ({reason})" summary["clickup_amounts_skipped"] += 1 # 6b: PDF attachment (skip if already attached) if clickup.attachment_exists(task_id, pdf_filename): result["clickup_attach"] = "skipped (already attached)" summary["clickup_skipped"] += 1 else: ok = clickup.attach_file(task_id, pdf_filename, pdf_bytes) if ok: result["clickup_attach"] = "attached" summary["clickup_attached"] += 1 else: result["clickup_attach"] = "failed" summary["clickup_failed"] += 1 # 7: Routed PDF upload to OneDrive dest_upn, dest_folder = _winston_route_destination(task, settings) if dest_folder: dest_path = f"{dest_folder.rstrip('/')}/{pdf_filename}" if dest_upn == winston_graph.SITE_DRIVE_UPN: uploaded = winston_graph.site_drive_upload_file( token, dest_path, pdf_bytes, content_type="application/pdf", ) else: uploaded = winston_graph.user_drive_upload_file( token, dest_upn, dest_path, pdf_bytes, content_type="application/pdf", ) if uploaded: result["pdf_route"] = f"{dest_upn}:{dest_path}" summary["pdf_routed"] += 1 else: result["pdf_route"] = "upload_failed" summary["pdf_route_failed"] += 1 else: result["pdf_route"] = "skipped (no destination configured)" log.warning( "No PDF route destination for %s — " "set shared_quotes_location in Togen Admin → Documents → Quotes", qn, ) except Exception as exc: log.exception("ClickUp/route exception for quote %s", qn) result.setdefault("error", f"writeback_exception: {exc}") s67.warn() summary["quotes"].append(result) # Mark overall run status based on soft failures if summary.get("drafts_failed") or summary.get("pdf_route_failed"): diag.warn() diag.set_summary({ "dry_run": summary["dry_run"], "writes_skipped_dry_run": summary["writes_skipped_dry_run"], "quotes_found": summary["quotes_found"], "drafts_created": summary["drafts_created"], "drafts_skipped": summary["drafts_skipped"], "drafts_failed": summary["drafts_failed"], "clickup_amounts_set": summary["clickup_amounts_set"], "clickup_amounts_failed": summary["clickup_amounts_failed"], "clickup_attached": summary["clickup_attached"], "clickup_failed": summary["clickup_failed"], "pdf_routed": summary["pdf_routed"], "pdf_route_failed": summary["pdf_route_failed"], "quotes": summary["quotes"], }) log.info( "Winston pipeline complete — date=%s found=%d drafts=+%d/~%d/!%d" " attached=+%d/~%d/!%d routed=%d", date, summary["quotes_found"], summary["drafts_created"], summary["drafts_skipped"], summary["drafts_failed"], summary["clickup_attached"], summary["clickup_skipped"], summary["clickup_failed"], summary["pdf_routed"], ) return summary # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- if __name__ == "__main__": import argparse import json as _json import sys logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s %(message)s") ap = argparse.ArgumentParser(description="Winston quote pipeline (DVI-815)") ap.add_argument("--date", required=True, help="Target date YYYY-MM-DD") args = ap.parse_args() result = run(args.date) print(_json.dumps(result, indent=2, default=str)) has_error = bool(result.get("error") or result.get("config_errors")) sys.exit(1 if has_error else 0)