feat: field amendments MVP — extraction, pricing, signing, PDF to Drive

.dockerignore

@@ -0,0 +1,6 @@
+.venv
+.git
+tests
+.pytest_cache
+.ruff_cache
+__pycache__

.github/workflows/ci.yml

@@ -0,0 +1,17 @@
+name: ci
+on:
+  push:
+    branches: [main]
+  pull_request:
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+    - uses: actions/checkout@v4
+    - uses: actions/setup-python@v5
+      with: {python-version: '3.11'}
+    - run: sudo apt-get update && sudo apt-get install -y libpango-1.0-0 libpangoft2-1.0-0 libgdk-pixbuf-2.0-0
+    - run: pip install -r requirements-dev.txt
+    - run: ruff check .
+    - run: python -m pytest -v

.github/workflows/image.yml

@@ -0,0 +1,33 @@
+name: image
+on:
+  push:
+    branches: [main]
+permissions:
+  contents: read
+  packages: write
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v4
+    - uses: docker/setup-buildx-action@v3
+    - uses: docker/login-action@v3
+      with:
+        registry: ghcr.io
+        username: ${{ github.actor }}
+        password: ${{ secrets.GITHUB_TOKEN }}
+    - uses: docker/metadata-action@v5
+      id: meta
+      with:
+        images: ghcr.io/siliconsaga/skipta
+        tags: |
+          type=sha,format=long
+          type=raw,value=latest,enable={{is_default_branch}}
+    - uses: docker/build-push-action@v6
+      with:
+        context: .
+        push: true
+        tags: ${{ steps.meta.outputs.tags }}
+        labels: ${{ steps.meta.outputs.labels }}
+        cache-from: type=gha
+        cache-to: type=gha,mode=max

Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.11-slim
+
+# WeasyPrint runtime libs + a real font for PDF output
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libpango-1.0-0 libpangoft2-1.0-0 libgdk-pixbuf-2.0-0 fonts-dejavu-core \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app/ app/
+EXPOSE 8000
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

app/amendments.py

@@ -0,0 +1,59 @@
+"""The Amendments tab is the state machine: one row per amendment, draft → signed."""
+import re
+from dataclasses import astuple, dataclass
+
+from app.google_clients import read_values
+
+TAB = "Amendments"
+DATA_RANGE = f"{TAB}!A2:J"
+
+
+@dataclass
+class AmendmentRecord:
+    amendment_id: str
+    created_at: str
+    customer_name: str
+    voice_text: str
+    extracted_json: str
+    line_items_json: str
+    total: float
+    status: str  # "draft" | "signed"
+    pdf_drive_url: str
+    signed_at: str
+
+    def to_row(self) -> list:
+        row = list(astuple(self))
+        row[6] = f"{self.total:.2f}"
+        return row
+
+    @classmethod
+    def from_row(cls, row: list) -> "AmendmentRecord":
+        padded = list(row) + [""] * (10 - len(row))
+        padded[6] = float(padded[6] or 0)
+        return cls(*padded)
+
+
+def make_amendment_id(customer_name: str, now) -> str:
+    cleaned = re.sub(r"['’]", "", customer_name.lower())  # straight + curly apostrophes: O'Brien → obrien, not o-brien
+    slug = re.sub(r"[^a-z0-9]+", "-", cleaned).strip("-")
+    return f"amend_{slug}_{now.strftime('%Y%m%d%H%M%S')}"
+
+
+def append_amendment(sheets, spreadsheet_id: str, record: AmendmentRecord) -> None:
+    sheets.spreadsheets().values().append(
+        spreadsheetId=spreadsheet_id, range=DATA_RANGE, valueInputOption="RAW", body={"values": [record.to_row()]}
+    ).execute()
+
+
+def find_amendment(sheets, spreadsheet_id: str, amendment_id: str):
+    for index, row in enumerate(read_values(sheets, spreadsheet_id, DATA_RANGE)):
+        if row and row[0] == amendment_id:
+            return index + 2, AmendmentRecord.from_row(row)  # +2: 1-based rows below the header
+    return None
+
+
+def mark_signed(sheets, spreadsheet_id: str, row: int, pdf_url: str, signed_at: str) -> None:
+    sheets.spreadsheets().values().update(
+        spreadsheetId=spreadsheet_id, range=f"{TAB}!H{row}:J{row}", valueInputOption="RAW",
+        body={"values": [["signed", pdf_url, signed_at]]},
+    ).execute()

app/config.py

@@ -0,0 +1,34 @@
+"""Environment-driven settings. A .env at the component root is honored for local dev."""
+import os
+from dataclasses import dataclass, field
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+DEFAULT_MODELS = "gemini-2.5-flash,gemini-2.5-flash-lite,gemini-2.0-flash-001"
+
+
+@dataclass(frozen=True)
+class Settings:
+    project_id: str
+    region: str
+    spreadsheet_id: str
+    drive_folder_id: str
+    base_url: str
+    model_names: list[str] = field(default_factory=list)
+    max_output_tokens: int = 1024
+    rate_limit_per_minute: int = 10
+
+    @classmethod
+    def from_env(cls) -> "Settings":
+        return cls(
+            project_id=os.getenv("GCP_PROJECT_ID", ""),
+            region=os.getenv("GCP_REGION", "us-east1"),
+            spreadsheet_id=os.getenv("SKIPTA_SPREADSHEET_ID", ""),
+            drive_folder_id=os.getenv("SKIPTA_DRIVE_FOLDER_ID", ""),
+            base_url=os.getenv("SKIPTA_BASE_URL", "http://localhost:8000"),
+            model_names=[m.strip() for m in os.getenv("SKIPTA_MODEL_NAMES", DEFAULT_MODELS).split(",") if m.strip()],
+            max_output_tokens=int(os.getenv("MAX_OUTPUT_TOKENS", "1024")),
+            rate_limit_per_minute=int(os.getenv("RATE_LIMIT_PER_MINUTE", "10")),
+        )

app/drive.py

@@ -0,0 +1,45 @@
+"""Drive integration: locate (or create) the customer's subfolder under the shared Skipta folder, upload signed PDFs idempotently."""
+import io
+
+from googleapiclient.http import MediaIoBaseUpload
+
+FOLDER_MIME = "application/vnd.google-apps.folder"
+
+
+def _escape(value: str) -> str:
+    return value.replace("\\", "\\\\").replace("'", "\\'")
+
+
+def find_customer_folder(drive, root_folder_id: str, customer_name: str):
+    query = (
+        f"'{_escape(root_folder_id)}' in parents and mimeType = '{FOLDER_MIME}' "
+        f"and name contains '{_escape(customer_name)}' and trashed = false"
+    )
+    result = drive.files().list(q=query, fields="files(id, name)", pageSize=5).execute()
+    files = result.get("files", [])
+    return files[0]["id"] if files else None
+
+
+def ensure_customer_folder(drive, root_folder_id: str, customer_name: str) -> str:
+    existing = find_customer_folder(drive, root_folder_id, customer_name)
+    if existing:
+        return existing
+    created = drive.files().create(
+        body={"name": customer_name, "mimeType": FOLDER_MIME, "parents": [root_folder_id]}, fields="id"
+    ).execute()
+    return created["id"]
+
+
+def find_file_in_folder(drive, folder_id: str, filename: str):
+    query = f"'{_escape(folder_id)}' in parents and name = '{_escape(filename)}' and trashed = false"
+    result = drive.files().list(q=query, fields="files(id, webViewLink)", pageSize=1).execute()
+    files = result.get("files", [])
+    return files[0].get("webViewLink") if files else None
+
+
+def upload_pdf(drive, folder_id: str, filename: str, pdf_bytes: bytes) -> str:
+    media = MediaIoBaseUpload(io.BytesIO(pdf_bytes), mimetype="application/pdf")
+    created = drive.files().create(
+        body={"name": filename, "parents": [folder_id]}, media_body=media, fields="webViewLink"
+    ).execute()
+    return created["webViewLink"]

app/extraction.py

@@ -0,0 +1,80 @@
+"""Gemini structured-output extraction of the amendment payload. Anti-hallucination is downstream and deterministic (pricing match) — this layer only shapes the request."""
+import logging
+
+from pydantic import BaseModel, Field, ValidationError
+
+logger = logging.getLogger("skipta.extraction")
+
+
+class BreakerRequirement(BaseModel):
+    amps: int = Field(..., description="Amperage of the breaker, e.g. 20, 30, 50")
+    poles: int = Field(..., description="Number of poles, usually 1 or 2")
+    quantity: int = Field(..., ge=1, description="Quantity requested")
+
+
+class PanelRequirement(BaseModel):
+    max_amperage: int = Field(..., description="Maximum amperage capacity of the panel, e.g. 100, 200")
+
+
+class AmendmentPayload(BaseModel):
+    customer_name: str = Field(..., min_length=1)
+    panel: PanelRequirement | None = None
+    breakers: list[BreakerRequirement] = Field(default_factory=list)
+
+
+class ExtractionError(Exception):
+    """Every configured model failed to produce a schema-valid payload."""
+
+
+# Vertex structured-output schema (OpenAPI subset — hand-written; pydantic's json_schema
+# emits $defs, which Gemini's response_schema does not accept).
+AMENDMENT_SCHEMA = {
+    "type": "OBJECT",
+    "properties": {
+        "customer_name": {"type": "STRING", "description": "Surname or identifier of the customer"},
+        "panel": {
+            "type": "OBJECT",
+            "nullable": True,
+            "properties": {"max_amperage": {"type": "INTEGER"}},
+            "required": ["max_amperage"],
+        },
+        "breakers": {
+            "type": "ARRAY",
+            "items": {
+                "type": "OBJECT",
+                "properties": {
+                    "amps": {"type": "INTEGER"},
+                    "poles": {"type": "INTEGER"},
+                    "quantity": {"type": "INTEGER"},
+                },
+                "required": ["amps", "poles", "quantity"],
+            },
+        },
+    },
+    "required": ["customer_name"],
+}
+
+PROMPT = (
+    "You are extracting a field change-order for a residential electrical job from a technician's dictated note. "
+    "Extract ONLY parts the note explicitly mentions — never invent parts, quantities, or a customer name.\n\n"
+    "Note:\n{voice_text}"
+)
+
+
+def extract_amendment(voice_text, *, model_factory, model_names, max_output_tokens):
+    from vertexai.generative_models import GenerationConfig
+
+    config = GenerationConfig(
+        response_mime_type="application/json",
+        response_schema=AMENDMENT_SCHEMA,
+        max_output_tokens=max_output_tokens,
+    )
+    for name in model_names:
+        try:
+            response = model_factory(name).generate_content(PROMPT.format(voice_text=voice_text), generation_config=config)
+            return AmendmentPayload.model_validate_json(response.text)
+        except (ValidationError, ValueError) as exc:
+            logger.warning("model %s returned schema-invalid output: %s", name, exc)
+        except Exception as exc:  # API errors: quota, permission, model-not-found
+            logger.warning("model %s failed: %s", name, exc)
+    raise ExtractionError(f"all models failed for extraction: {model_names}")

app/google_clients.py

@@ -0,0 +1,38 @@
+"""All Google client construction lives here; everything downstream takes injected clients."""
+import google.auth
+from googleapiclient.discovery import build
+
+SCOPES = [
+    "https://www.googleapis.com/auth/cloud-platform",
+    "https://www.googleapis.com/auth/drive",
+    "https://www.googleapis.com/auth/spreadsheets",
+]
+
+
+def get_credentials():
+    creds, _ = google.auth.default(scopes=SCOPES)
+    return creds
+
+
+def build_sheets(creds):
+    return build("sheets", "v4", credentials=creds, cache_discovery=False)
+
+
+def build_drive(creds):
+    return build("drive", "v3", credentials=creds, cache_discovery=False)
+
+
+def make_model_factory(project_id: str, region: str):
+    def factory(model_name: str):
+        import vertexai
+        from vertexai.generative_models import GenerativeModel
+
+        vertexai.init(project=project_id, location=region)
+        return GenerativeModel(model_name)
+
+    return factory
+
+
+def read_values(sheets, spreadsheet_id: str, a1_range: str):
+    result = sheets.spreadsheets().values().get(spreadsheetId=spreadsheet_id, range=a1_range).execute()
+    return result.get("values", [])