From 6fc84b20b6746570d9db91898d8b2a828fd41f9a Mon Sep 17 00:00:00 2001 From: MangoPig Date: Sat, 21 Feb 2026 23:52:55 +0000 Subject: [PATCH] Go orchestrator --- Code/TODO.md | 65 +++++- Code/Tools/Service/main.py | 260 +++++++++++++++++++---- Code/Tracker/Dockerfile | 31 +++ Code/Tracker/clients/docling.go | 135 ++++++++++++ Code/Tracker/clients/ollama.go | 165 +++++++++++++++ Code/Tracker/clients/tools.go | 216 +++++++++++++++++++ Code/Tracker/config.yaml.example | 37 ++++ Code/Tracker/config/config.go | 73 +++++++ Code/Tracker/db/sqlite.go | 228 ++++++++++++++++++++ Code/Tracker/go.mod | 11 + Code/Tracker/go.sum | 10 + Code/Tracker/main.go | 68 ++++++ Code/Tracker/pipeline/pipeline.go | 341 ++++++++++++++++++++++++++++++ Code/Tracker/watcher/watcher.go | 241 +++++++++++++++++++++ Code/docker-compose.yaml | 22 ++ 15 files changed, 1857 insertions(+), 46 deletions(-) create mode 100644 Code/Tracker/Dockerfile create mode 100644 Code/Tracker/clients/docling.go create mode 100644 Code/Tracker/clients/ollama.go create mode 100644 Code/Tracker/clients/tools.go create mode 100644 Code/Tracker/config.yaml.example create mode 100644 Code/Tracker/config/config.go create mode 100644 Code/Tracker/db/sqlite.go create mode 100644 Code/Tracker/go.mod create mode 100644 Code/Tracker/go.sum create mode 100644 Code/Tracker/main.go create mode 100644 Code/Tracker/pipeline/pipeline.go create mode 100644 Code/Tracker/watcher/watcher.go diff --git a/Code/TODO.md b/Code/TODO.md index a689643..e8961cb 100644 --- a/Code/TODO.md +++ b/Code/TODO.md @@ -76,16 +76,67 @@ Rationale: De-risk unknowns (Docling OCR quality, Ollama CPU inference speed) be - Response: `{ "response": "" }` (strip `...` tags in post-processing) - Note: Qwen3 includes reasoning by default; `/no_think` leaves empty tags, so strip instead -### 1c. Python Tools Service +### 1c. Python Tools Service (DONE) -- [ ] FastAPI + openpyxl service -- [ ] Endpoints: read tracker, write/update rows, get column schema -- [ ] Test with sample Tracker.xlsx +- [x] FastAPI + openpyxl service +- [x] Generic Excel API (not tracker-specific) +- [x] Test with Invoice_Schedule_2026Q1.xlsx -### 1d. Manual Integration Test +**Tools API Spec:** -- [ ] Chain test: PDF → Docling → Ollama → Tools → Excel updated -- [ ] Document any issues / adjustments needed +- Image: Custom Dockerfile (python:3.12-slim + fastapi + openpyxl) +- Port: `8000` +- Volume: `/data` → `/data/jingtian/BenjaminTeam` + +**Endpoints:** + +| Endpoint | Method | Description | +| -------------------- | ------ | ---------------------------- | +| `/health` | GET | Health check | +| `/excel/sheets` | GET | List sheet names in workbook | +| `/excel/schema` | GET | Get column headers | +| `/excel/read` | GET | Read all rows | +| `/excel/row/{n}` | GET | Get specific row (1-indexed) | + +**Query Params:** + +- `file_path` (required) — path relative to `/data/` +- `sheet_name` (required) — name of the sheet to read +- `header_row` (optional, default=1) — which row contains headers + +**Tested:** + +- Invoice_Schedule_2026Q1.xlsx with `header_row=4` — extracted 6 invoice rows ✅ +- Columns: Client, Matter Ref, TM Number, Description, Amount (HKD), Due Date, Status + +**Future Endpoints (Phase 2+):** + +- `POST /excel/row` — Add new row +- `PUT /excel/row/{n}` — Update specific row +- `DELETE /excel/row/{n}` — Delete row +- `POST /excel/format` — Apply formatting + +### 1d. Manual Integration Test (DONE) + +- [x] Chain test: PDF → Docling → Ollama → JSON extraction +- [x] Chain test: PNG → Docling (Chinese OCR) → Ollama → JSON extraction +- [x] Chain test: DOCX → Docling → Ollama → JSON extraction +- [x] Chain test: XLSX → Tools API → Ollama → JSON summary + +**Tested Files:** + +| File | Type | Docling | Ollama | Key Extractions | +| --------------------------------- | ---- | ------- | ------ | ------------------------------------------------ | +| Filing_Receipt_307800905.pdf | PDF | ✅ | ✅ | TM#307800905, NOVA, Class 9, deadline 30-05-2026 | +| Email_Monee_...20260221.png | PNG | ✅ | ✅ | TM#306527151, 官藥坊, evidence of use | +| Letter_Re_TM306735835.docx | DOCX | ✅ | ✅ | TM#306735835, CALIFORNIA BABY, hearing 17-04-2026| +| Invoice_Schedule_2026Q1.xlsx | XLSX | ✅ | ✅ | HKD 474,000 total, 6 invoices, 3 sent/1 paid/2 draft | + +**Notes:** + +- Qwen3 1.7b correctly extracted Chinese trademark 官藥坊 from PNG +- All deadlines, TM numbers, client names extracted accurately +- Minor: XLSX currency guessed as USD (HKD in source) — fix with better prompting ## Phase 2: Go Orchestrator diff --git a/Code/Tools/Service/main.py b/Code/Tools/Service/main.py index 512dbcd..55ca270 100644 --- a/Code/Tools/Service/main.py +++ b/Code/Tools/Service/main.py @@ -1,7 +1,7 @@ """ Generic Excel API Service -A FastAPI service for reading Excel files (.xlsx). +A FastAPI service for reading and writing Excel files (.xlsx). All file paths are relative to /data/ mount. Endpoints: @@ -10,26 +10,52 @@ Endpoints: - GET /excel/schema - Get column headers - GET /excel/read - Read all rows - GET /excel/row/{row_num} - Get specific row (1-indexed) +- GET /excel/find - Find row by column value +- POST /excel/row - Add new row +- PUT /excel/row/{row_num} - Update row +- POST /excel/create - Create new workbook """ from fastapi import FastAPI, HTTPException, Query from pydantic import BaseModel from pathlib import Path -from typing import Optional -from openpyxl import load_workbook +from typing import Optional, List, Any +from openpyxl import Workbook, load_workbook +from openpyxl.styles import Font, Alignment import os app = FastAPI( title="JingTian Tools Service", - description="Generic Excel API for reading xlsx files", - version="0.2.0", + description="Generic Excel API for reading and writing xlsx files", + version="0.3.0", ) DATA_ROOT = Path(os.getenv("DATA_ROOT", "/data")) -def get_workbook(file_path: str): - """Load workbook from file path relative to DATA_ROOT.""" +# Request models +class AddRowRequest(BaseModel): + file_path: str + sheet_name: str + header_row: int = 1 + row: dict[str, Any] + + +class UpdateRowRequest(BaseModel): + file_path: str + sheet_name: str + header_row: int = 1 + row: dict[str, Any] + + +class CreateWorkbookRequest(BaseModel): + file_path: str + sheet_name: str + columns: List[str] + + +def get_workbook_readonly(file_path: str): + """Load workbook from file path relative to DATA_ROOT (read-only).""" full_path = DATA_ROOT / file_path if not full_path.exists(): raise HTTPException(status_code=404, detail=f"File not found: {file_path}") @@ -43,6 +69,21 @@ def get_workbook(file_path: str): ) +def get_workbook_writable(file_path: str): + """Load workbook from file path relative to DATA_ROOT (writable).""" + full_path = DATA_ROOT / file_path + if not full_path.exists(): + raise HTTPException(status_code=404, detail=f"File not found: {file_path}") + if not full_path.suffix.lower() == ".xlsx": + raise HTTPException(status_code=400, detail="Only .xlsx files supported") + try: + return load_workbook(full_path), full_path + except Exception as e: + raise HTTPException( + status_code=500, detail=f"Failed to open workbook: {str(e)}" + ) + + def get_sheet(wb, sheet_name: str): """Get sheet by name.""" if sheet_name not in wb.sheetnames: @@ -60,9 +101,28 @@ def cell_to_str(cell_value) -> str: return str(cell_value) +def get_headers(ws, header_row: int) -> list[str]: + """Get headers from a worksheet.""" + headers = [] + for cell in ws[header_row]: + headers.append(cell_to_str(cell.value)) + return headers + + +def get_column_index(headers: list[str], column_name: str) -> int: + """Get 1-indexed column index for a header name.""" + try: + return headers.index(column_name) + 1 + except ValueError: + raise HTTPException( + status_code=400, + detail=f"Column '{column_name}' not found. Available: {headers}", + ) + + @app.get("/health") async def health(): - return {"status": "healthy", "service": "tools", "version": "0.2.0"} + return {"status": "healthy", "service": "tools", "version": "0.3.0"} @app.get("/excel/sheets") @@ -70,7 +130,7 @@ async def list_sheets( file_path: str = Query(..., description="Path to xlsx file relative to /data/"), ): """List all sheet names in the workbook.""" - wb = get_workbook(file_path) + wb = get_workbook_readonly(file_path) sheets = wb.sheetnames wb.close() return {"file_path": file_path, "sheets": sheets, "count": len(sheets)} @@ -83,13 +143,9 @@ async def get_schema( header_row: int = Query(1, description="Row number containing headers (1-indexed)"), ): """Get column headers from the specified sheet.""" - wb = get_workbook(file_path) + wb = get_workbook_readonly(file_path) ws = get_sheet(wb, sheet_name) - - headers = [] - for cell in ws[header_row]: - headers.append(cell_to_str(cell.value)) - + headers = get_headers(ws, header_row) wb.close() return { "file_path": file_path, @@ -112,29 +168,21 @@ async def read_all( limit: Optional[int] = Query(None, description="Max rows to return"), ): """Read all rows from the specified sheet.""" - wb = get_workbook(file_path) + wb = get_workbook_readonly(file_path) ws = get_sheet(wb, sheet_name) - - # Get headers - headers = [cell_to_str(cell.value) for cell in ws[header_row]] - - # Determine data start row + headers = get_headers(ws, header_row) data_start = start_row if start_row else header_row + 1 - # Read rows rows = [] row_count = 0 for row_num, row in enumerate(ws.iter_rows(min_row=data_start), start=data_start): - # Stop if limit reached if limit and row_count >= limit: break - # Skip completely empty rows values = [cell_to_str(cell.value) for cell in row] if all(v == "" for v in values): continue - # Build row dict with headers as keys row_data = {"_row_num": row_num} for i, header in enumerate(headers): if i < len(values): @@ -165,13 +213,10 @@ async def get_row( if row_num < 1: raise HTTPException(status_code=400, detail="Row number must be >= 1") - wb = get_workbook(file_path) + wb = get_workbook_readonly(file_path) ws = get_sheet(wb, sheet_name) + headers = get_headers(ws, header_row) - # Get headers - headers = [cell_to_str(cell.value) for cell in ws[header_row]] - - # Get the specific row try: row = ws[row_num] except Exception: @@ -179,8 +224,6 @@ async def get_row( raise HTTPException(status_code=404, detail=f"Row {row_num} not found") values = [cell_to_str(cell.value) for cell in row] - - # Build row dict row_data = {"_row_num": row_num} for i, header in enumerate(headers): if i < len(values): @@ -191,10 +234,149 @@ async def get_row( return {"file_path": file_path, "sheet_name": sheet_name, "row": row_data} -# Future endpoints (Phase 2 & 3) -# POST /excel/row - Add new row -# PUT /excel/row/{row_num} - Update row -# DELETE /excel/row/{row_num} - Delete row -# POST /excel/column - Add column -# POST /excel/format - Apply formatting -# POST /excel/create - Create new workbook +@app.get("/excel/find") +async def find_row( + file_path: str = Query(..., description="Path to xlsx file relative to /data/"), + sheet_name: str = Query(..., description="Sheet name to search"), + header_row: int = Query(1, description="Row number containing headers (1-indexed)"), + column: str = Query(..., description="Column name to search"), + value: str = Query(..., description="Value to find"), +): + """Find a row by column value.""" + wb = get_workbook_readonly(file_path) + ws = get_sheet(wb, sheet_name) + headers = get_headers(ws, header_row) + col_idx = get_column_index(headers, column) + data_start = header_row + 1 + + for row_num, row in enumerate(ws.iter_rows(min_row=data_start), start=data_start): + values = [cell_to_str(cell.value) for cell in row] + if col_idx - 1 < len(values) and values[col_idx - 1] == value: + row_data = {"_row_num": row_num} + for i, header in enumerate(headers): + if i < len(values): + key = header if header else f"_col_{i + 1}" + row_data[key] = values[i] + wb.close() + return { + "found": True, + "row_num": row_num, + "row": row_data, + "message": f"Found at row {row_num}", + } + + wb.close() + raise HTTPException( + status_code=404, + detail=f"No row found with {column}='{value}'", + ) + + +@app.post("/excel/row") +async def add_row(request: AddRowRequest): + """Add a new row to the sheet.""" + wb, full_path = get_workbook_writable(request.file_path) + ws = get_sheet(wb, request.sheet_name) + headers = get_headers(ws, request.header_row) + + # Find next empty row + next_row = ws.max_row + 1 + + # Write values + for col_name, value in request.row.items(): + if col_name.startswith("_"): + continue + try: + col_idx = headers.index(col_name) + 1 + ws.cell(row=next_row, column=col_idx, value=value) + except ValueError: + pass # Skip unknown columns + + wb.save(full_path) + wb.close() + + return { + "success": True, + "row_num": next_row, + "message": f"Added row at {next_row}", + } + + +@app.put("/excel/row/{row_num}") +async def update_row(row_num: int, request: UpdateRowRequest): + """Update an existing row.""" + if row_num < 1: + raise HTTPException(status_code=400, detail="Row number must be >= 1") + + wb, full_path = get_workbook_writable(request.file_path) + ws = get_sheet(wb, request.sheet_name) + headers = get_headers(ws, request.header_row) + + # Check row exists + if row_num > ws.max_row: + wb.close() + raise HTTPException(status_code=404, detail=f"Row {row_num} not found") + + # Update values + for col_name, value in request.row.items(): + if col_name.startswith("_"): + continue + try: + col_idx = headers.index(col_name) + 1 + ws.cell(row=row_num, column=col_idx, value=value) + except ValueError: + pass # Skip unknown columns + + wb.save(full_path) + wb.close() + + return { + "success": True, + "row_num": row_num, + "message": f"Updated row {row_num}", + } + + +@app.post("/excel/create") +async def create_workbook(request: CreateWorkbookRequest): + """Create a new workbook with specified columns.""" + full_path = DATA_ROOT / request.file_path + + # Create parent directories if needed + full_path.parent.mkdir(parents=True, exist_ok=True) + + # Check if file exists + if full_path.exists(): + raise HTTPException( + status_code=409, + detail=f"File already exists: {request.file_path}", + ) + + # Create workbook + wb = Workbook() + ws = wb.active + ws.title = request.sheet_name + + # Write headers with formatting + header_font = Font(bold=True) + for col_idx, col_name in enumerate(request.columns, start=1): + cell = ws.cell(row=1, column=col_idx, value=col_name) + cell.font = header_font + cell.alignment = Alignment(horizontal="center") + + # Auto-adjust column widths + for col_idx, col_name in enumerate(request.columns, start=1): + ws.column_dimensions[ws.cell(row=1, column=col_idx).column_letter].width = max( + len(col_name) + 2, 12 + ) + + wb.save(full_path) + wb.close() + + return { + "success": True, + "file_path": request.file_path, + "sheet_name": request.sheet_name, + "columns": request.columns, + "message": f"Created workbook with {len(request.columns)} columns", + } diff --git a/Code/Tracker/Dockerfile b/Code/Tracker/Dockerfile new file mode 100644 index 0000000..b45ab0c --- /dev/null +++ b/Code/Tracker/Dockerfile @@ -0,0 +1,31 @@ +FROM golang:1.22-alpine AS builder + +# Install build dependencies (for sqlite) +RUN apk add --no-cache gcc musl-dev + +WORKDIR /app + +# Copy go mod files +COPY go.mod go.sum ./ +RUN go mod download + +# Copy source +COPY . . + +# Build with CGO enabled (required for sqlite) +RUN CGO_ENABLED=1 go build -o tracker . + +# Runtime image +FROM alpine:3.19 + +RUN apk add --no-cache ca-certificates tzdata curl + +WORKDIR /app + +COPY --from=builder /app/tracker . + +# Config is mounted at runtime via docker-compose volume +# Default config path: /app/config.yaml + +ENTRYPOINT ["./tracker"] +CMD ["-config", "/app/config.yaml"] diff --git a/Code/Tracker/clients/docling.go b/Code/Tracker/clients/docling.go new file mode 100644 index 0000000..0dc7e70 --- /dev/null +++ b/Code/Tracker/clients/docling.go @@ -0,0 +1,135 @@ +// Path: Code/Tracker/clients/docling.go + +package clients + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "mime/multipart" + "net/http" + "os" + "path/filepath" + "strings" + "time" +) + +type DoclingClient struct { + baseURL string + httpClient *http.Client +} + +type DoclingResponse struct { + Document struct { + MdContent string `json:"md_content"` + } `json:"document"` + Status string `json:"status"` + ProcessingTime float64 `json:"processing_time"` +} + +func NewDoclingClient(baseURL string) *DoclingClient { + return &DoclingClient{ + baseURL: strings.TrimSuffix(baseURL, "/"), + httpClient: &http.Client{ + Timeout: 5 * time.Minute, // OCR can take a while + }, + } +} + +func (c *DoclingClient) Health() error { + resp, err := c.httpClient.Get(c.baseURL + "/health") + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("docling unhealthy: status %d", resp.StatusCode) + } + return nil +} + +func (c *DoclingClient) Convert(filePath string) (string, error) { + // Open file + f, err := os.Open(filePath) + if err != nil { + return "", fmt.Errorf("failed to open file: %w", err) + } + defer f.Close() + + // Create multipart form + var buf bytes.Buffer + writer := multipart.NewWriter(&buf) + + // Add file + part, err := writer.CreateFormFile("files", filepath.Base(filePath)) + if err != nil { + return "", fmt.Errorf("failed to create form file: %w", err) + } + if _, err := io.Copy(part, f); err != nil { + return "", fmt.Errorf("failed to copy file: %w", err) + } + + // Add options + writer.WriteField("to_formats", "md") + writer.WriteField("do_ocr", "true") + + // For images, force OCR + ext := strings.ToLower(filepath.Ext(filePath)) + if ext == ".png" || ext == ".jpg" || ext == ".jpeg" { + writer.WriteField("force_ocr", "true") + writer.WriteField("ocr_lang", "en") + writer.WriteField("ocr_lang", "ch_tra") + } + + writer.Close() + + // Make request + req, err := http.NewRequest("POST", c.baseURL+"/v1/convert/file", &buf) + if err != nil { + return "", fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("Content-Type", writer.FormDataContentType()) + + resp, err := c.httpClient.Do(req) + if err != nil { + return "", fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return "", fmt.Errorf("docling error: status %d, body: %s", resp.StatusCode, string(body)) + } + + // Parse response + var result DoclingResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", fmt.Errorf("failed to decode response: %w", err) + } + + return result.Document.MdContent, nil +} + +func getMimeType(path string) string { + ext := strings.ToLower(filepath.Ext(path)) + switch ext { + case ".pdf": + return "application/pdf" + case ".docx": + return "application/vnd.openxmlformats-officedocument.wordprocessingml.document" + case ".doc": + return "application/msword" + case ".xlsx": + return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + case ".xls": + return "application/vnd.ms-excel" + case ".png": + return "image/png" + case ".jpg", ".jpeg": + return "image/jpeg" + default: + return "application/octet-stream" + } +} diff --git a/Code/Tracker/clients/ollama.go b/Code/Tracker/clients/ollama.go new file mode 100644 index 0000000..c08d491 --- /dev/null +++ b/Code/Tracker/clients/ollama.go @@ -0,0 +1,165 @@ +// Path: Code/Tracker/clients/ollama.go + +package clients + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "regexp" + "strings" + "time" +) + +type OllamaClient struct { + baseURL string + model string + temperature float64 + httpClient *http.Client +} + +type OllamaRequest struct { + Model string `json:"model"` + Prompt string `json:"prompt"` + Stream bool `json:"stream"` + Options OllamaOptions `json:"options"` +} + +type OllamaOptions struct { + Temperature float64 `json:"temperature"` +} + +type OllamaResponse struct { + Response string `json:"response"` + Done bool `json:"done"` +} + +func NewOllamaClient(baseURL, model string, temperature float64) *OllamaClient { + return &OllamaClient{ + baseURL: strings.TrimSuffix(baseURL, "/"), + model: model, + temperature: temperature, + httpClient: &http.Client{ + Timeout: 5 * time.Minute, // LLM inference can take a while on CPU + }, + } +} + +func (c *OllamaClient) Health() error { + resp, err := c.httpClient.Get(c.baseURL + "/api/tags") + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("ollama unhealthy: status %d", resp.StatusCode) + } + return nil +} + +func (c *OllamaClient) Extract(documentContent, documentType string) (map[string]interface{}, error) { + prompt := buildExtractionPrompt(documentContent, documentType) + + reqBody := OllamaRequest{ + Model: c.model, + Prompt: prompt, + Stream: false, + Options: OllamaOptions{ + Temperature: c.temperature, + }, + } + + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return nil, fmt.Errorf("failed to marshal request: %w", err) + } + + resp, err := c.httpClient.Post(c.baseURL+"/api/generate", "application/json", bytes.NewReader(jsonBody)) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("ollama error: status %d, body: %s", resp.StatusCode, string(body)) + } + + var result OllamaResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + // Clean up response - remove tags + response := cleanResponse(result.Response) + + // Parse JSON from response + extracted, err := parseJSON(response) + if err != nil { + return nil, fmt.Errorf("failed to parse JSON from response: %w", err) + } + + return extracted, nil +} + +func buildExtractionPrompt(content, docType string) string { + return fmt.Sprintf(`/no_think +Extract structured information from this document as JSON. Include ALL fields that are present. + +Required fields (include if found): +- document_type: The type of document (e.g., "Filing Receipt", "Legal Letter", "Email", "Invoice", "Memo") +- client_name: Client or applicant name +- matter_reference: Matter/case reference number (e.g., "JT/2026/1234") +- tm_number: Trademark application number +- trademark_name: Name of the trademark +- trademark_class: Trademark class number(s) +- filing_date: Date of filing (format: DD-MM-YYYY) +- response_deadline: Deadline for response (format: DD-MM-YYYY) +- hearing_date: Hearing date if mentioned (format: DD-MM-YYYY) +- amount: Invoice/fee amount (number only) +- currency: Currency code (e.g., "HKD", "USD") +- status: Document or payment status +- sender_name: Name of sender/author +- recipient_name: Name of recipient +- subject: Email/letter subject +- summary: Brief 1-2 sentence summary of the document + +Document type hint: %s + +Document content: +%s + +Respond ONLY with valid JSON. No explanation.`, docType, content) +} + +func cleanResponse(response string) string { + // Remove ... tags + re := regexp.MustCompile(`(?s).*?`) + response = re.ReplaceAllString(response, "") + return strings.TrimSpace(response) +} + +func parseJSON(response string) (map[string]interface{}, error) { + // Try to find JSON in the response + response = strings.TrimSpace(response) + + // Find JSON object boundaries + start := strings.Index(response, "{") + end := strings.LastIndex(response, "}") + + if start == -1 || end == -1 || end < start { + return nil, fmt.Errorf("no JSON object found in response") + } + + jsonStr := response[start : end+1] + + var result map[string]interface{} + if err := json.Unmarshal([]byte(jsonStr), &result); err != nil { + return nil, fmt.Errorf("invalid JSON: %w", err) + } + + return result, nil +} diff --git a/Code/Tracker/clients/tools.go b/Code/Tracker/clients/tools.go new file mode 100644 index 0000000..9ae9e36 --- /dev/null +++ b/Code/Tracker/clients/tools.go @@ -0,0 +1,216 @@ +// Path: Code/Tracker/clients/tools.go + +package clients + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "time" +) + +type ToolsClient struct { + baseURL string + httpClient *http.Client +} + +type ExcelReadResponse struct { + FilePath string `json:"file_path"` + SheetName string `json:"sheet_name"` + HeaderRow int `json:"header_row"` + Columns []string `json:"columns"` + RowCount int `json:"row_count"` + Rows []map[string]interface{} `json:"rows"` +} + +type ExcelFindResponse struct { + Found bool `json:"found"` + RowNum int `json:"row_num"` + Row map[string]interface{} `json:"row"` + Message string `json:"message"` +} + +type ExcelWriteRequest struct { + FilePath string `json:"file_path"` + SheetName string `json:"sheet_name"` + HeaderRow int `json:"header_row"` + Row map[string]interface{} `json:"row"` +} + +type ExcelUpdateRequest struct { + FilePath string `json:"file_path"` + SheetName string `json:"sheet_name"` + HeaderRow int `json:"header_row"` + RowNum int `json:"row_num"` + Row map[string]interface{} `json:"row"` +} + +func NewToolsClient(baseURL string) *ToolsClient { + return &ToolsClient{ + baseURL: strings.TrimSuffix(baseURL, "/"), + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + } +} + +func (c *ToolsClient) Health() error { + resp, err := c.httpClient.Get(c.baseURL + "/health") + if err != nil { + return err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("tools unhealthy: status %d", resp.StatusCode) + } + return nil +} + +func (c *ToolsClient) ReadExcel(filePath, sheetName string, headerRow int) (*ExcelReadResponse, error) { + params := url.Values{} + params.Set("file_path", filePath) + params.Set("sheet_name", sheetName) + params.Set("header_row", strconv.Itoa(headerRow)) + + resp, err := c.httpClient.Get(c.baseURL + "/excel/read?" + params.Encode()) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("tools error: status %d, body: %s", resp.StatusCode, string(body)) + } + + var result ExcelReadResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return &result, nil +} + +func (c *ToolsClient) FindRow(filePath, sheetName string, headerRow int, column, value string) (*ExcelFindResponse, error) { + params := url.Values{} + params.Set("file_path", filePath) + params.Set("sheet_name", sheetName) + params.Set("header_row", strconv.Itoa(headerRow)) + params.Set("column", column) + params.Set("value", value) + + resp, err := c.httpClient.Get(c.baseURL + "/excel/find?" + params.Encode()) + if err != nil { + return nil, fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusNotFound { + return &ExcelFindResponse{Found: false}, nil + } + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("tools error: status %d, body: %s", resp.StatusCode, string(body)) + } + + var result ExcelFindResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("failed to decode response: %w", err) + } + + return &result, nil +} + +func (c *ToolsClient) AddRow(filePath, sheetName string, headerRow int, row map[string]interface{}) error { + reqBody := ExcelWriteRequest{ + FilePath: filePath, + SheetName: sheetName, + HeaderRow: headerRow, + Row: row, + } + + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return fmt.Errorf("failed to marshal request: %w", err) + } + + resp, err := c.httpClient.Post(c.baseURL+"/excel/row", "application/json", bytes.NewReader(jsonBody)) + if err != nil { + return fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("tools error: status %d, body: %s", resp.StatusCode, string(body)) + } + + return nil +} + +func (c *ToolsClient) UpdateRow(filePath, sheetName string, headerRow, rowNum int, row map[string]interface{}) error { + reqBody := ExcelUpdateRequest{ + FilePath: filePath, + SheetName: sheetName, + HeaderRow: headerRow, + RowNum: rowNum, + Row: row, + } + + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return fmt.Errorf("failed to marshal request: %w", err) + } + + req, err := http.NewRequest("PUT", c.baseURL+"/excel/row/"+strconv.Itoa(rowNum), bytes.NewReader(jsonBody)) + if err != nil { + return fmt.Errorf("failed to create request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + return fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("tools error: status %d, body: %s", resp.StatusCode, string(body)) + } + + return nil +} + +func (c *ToolsClient) CreateTracker(filePath, sheetName string, columns []string) error { + reqBody := map[string]interface{}{ + "file_path": filePath, + "sheet_name": sheetName, + "columns": columns, + } + + jsonBody, err := json.Marshal(reqBody) + if err != nil { + return fmt.Errorf("failed to marshal request: %w", err) + } + + resp, err := c.httpClient.Post(c.baseURL+"/excel/create", "application/json", bytes.NewReader(jsonBody)) + if err != nil { + return fmt.Errorf("request failed: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + body, _ := io.ReadAll(resp.Body) + return fmt.Errorf("tools error: status %d, body: %s", resp.StatusCode, string(body)) + } + + return nil +} diff --git a/Code/Tracker/config.yaml.example b/Code/Tracker/config.yaml.example new file mode 100644 index 0000000..39f02ac --- /dev/null +++ b/Code/Tracker/config.yaml.example @@ -0,0 +1,37 @@ +watch: + root: /data + extensions: + - .pdf + - .docx + - .doc + - .png + - .jpg + - .jpeg + exclude: + - _LLM + - .git + - "~$" + stability_seconds: 120 + +services: + # Use Docker container names when running in Docker Compose + # Use localhost URLs when running standalone + docling: http://docling:5001 + ollama: http://ollama:11434 + tools: http://tools:8000 + +ollama: + model: qwen3:1.7b + temperature: 0 + +tracker: + path: Admin/Tracker.xlsx + sheet: Tracker + header_row: 1 + +database: + path: /data/_LLM/state.db + +logging: + level: info + error_file: /data/_LLM/errors.log diff --git a/Code/Tracker/config/config.go b/Code/Tracker/config/config.go new file mode 100644 index 0000000..7cf7bff --- /dev/null +++ b/Code/Tracker/config/config.go @@ -0,0 +1,73 @@ +// Path: Code/Tracker/config/config.go + +package config + +import ( + "os" + + "gopkg.in/yaml.v3" +) + +type Config struct { + Watch WatchConfig `yaml:"watch"` + Services ServicesConfig `yaml:"services"` + Ollama OllamaConfig `yaml:"ollama"` + Tracker TrackerConfig `yaml:"tracker"` + Database DatabaseConfig `yaml:"database"` + Logging LoggingConfig `yaml:"logging"` +} + +type WatchConfig struct { + Root string `yaml:"root"` + Extensions []string `yaml:"extensions"` + Exclude []string `yaml:"exclude"` + StabilitySeconds int `yaml:"stability_seconds"` +} + +type ServicesConfig struct { + Docling string `yaml:"docling"` + Ollama string `yaml:"ollama"` + Tools string `yaml:"tools"` +} + +type OllamaConfig struct { + Model string `yaml:"model"` + Temperature float64 `yaml:"temperature"` +} + +type TrackerConfig struct { + Path string `yaml:"path"` + Sheet string `yaml:"sheet"` + HeaderRow int `yaml:"header_row"` +} + +type DatabaseConfig struct { + Path string `yaml:"path"` +} + +type LoggingConfig struct { + Level string `yaml:"level"` + ErrorFile string `yaml:"error_file"` +} + +func Load(path string) (*Config, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + + var cfg Config + if err := yaml.Unmarshal(data, &cfg); err != nil { + return nil, err + } + + // Set defaults + if cfg.Watch.StabilitySeconds == 0 { + cfg.Watch.StabilitySeconds = 120 + } + if cfg.Tracker.HeaderRow == 0 { + cfg.Tracker.HeaderRow = 1 + } + + return &cfg, nil +} diff --git a/Code/Tracker/db/sqlite.go b/Code/Tracker/db/sqlite.go new file mode 100644 index 0000000..5dc0717 --- /dev/null +++ b/Code/Tracker/db/sqlite.go @@ -0,0 +1,228 @@ +// Path: Code/Tracker/db/sqlite.go + +package db + +import ( + "database/sql" + "encoding/json" + "os" + "path/filepath" + "time" + + _ "github.com/mattn/go-sqlite3" +) + +type DB struct { + conn *sql.DB +} + +type Document struct { + ID int64 + FilePath string + FileHash string + FileSize int64 + FirstSeen time.Time + LastModified time.Time + LastProcessed sql.NullTime + Status string // pending, processing, completed, failed +} + +type Extraction struct { + ID int64 + DocumentID int64 + ExtractionJSON string + ExtractedAt time.Time +} + +type ProcessingLog struct { + ID int64 + DocumentID int64 + Action string // processed, updated, skipped, failed + Details string + CreatedAt time.Time +} + +func Open(dbPath string) (*DB, error) { + // Ensure directory exists + dir := filepath.Dir(dbPath) + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, err + } + + conn, err := sql.Open("sqlite3", dbPath+"?_journal_mode=WAL") + if err != nil { + return nil, err + } + + db := &DB{conn: conn} + if err := db.migrate(); err != nil { + conn.Close() + return nil, err + } + + return db, nil +} + +func (db *DB) Close() error { + return db.conn.Close() +} + +func (db *DB) migrate() error { + schema := ` + CREATE TABLE IF NOT EXISTS documents ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + file_path TEXT UNIQUE NOT NULL, + file_hash TEXT NOT NULL, + file_size INTEGER NOT NULL, + first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_modified TIMESTAMP NOT NULL, + last_processed TIMESTAMP, + status TEXT DEFAULT 'pending' + ); + + CREATE INDEX IF NOT EXISTS idx_documents_path ON documents(file_path); + CREATE INDEX IF NOT EXISTS idx_documents_status ON documents(status); + + CREATE TABLE IF NOT EXISTS extractions ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + document_id INTEGER NOT NULL REFERENCES documents(id), + extraction_json TEXT NOT NULL, + extracted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE INDEX IF NOT EXISTS idx_extractions_doc ON extractions(document_id); + + CREATE TABLE IF NOT EXISTS processing_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + document_id INTEGER NOT NULL REFERENCES documents(id), + action TEXT NOT NULL, + details TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + + CREATE INDEX IF NOT EXISTS idx_log_doc ON processing_log(document_id); + ` + + _, err := db.conn.Exec(schema) + return err +} + +// GetDocument retrieves a document by file path +func (db *DB) GetDocument(filePath string) (*Document, error) { + row := db.conn.QueryRow(` + SELECT id, file_path, file_hash, file_size, first_seen, last_modified, last_processed, status + FROM documents WHERE file_path = ? + `, filePath) + + var doc Document + err := row.Scan(&doc.ID, &doc.FilePath, &doc.FileHash, &doc.FileSize, + &doc.FirstSeen, &doc.LastModified, &doc.LastProcessed, &doc.Status) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, err + } + return &doc, nil +} + +// UpsertDocument creates or updates a document record +func (db *DB) UpsertDocument(filePath, fileHash string, fileSize int64, lastModified time.Time) (*Document, error) { + existing, err := db.GetDocument(filePath) + if err != nil { + return nil, err + } + + if existing == nil { + // Insert new + result, err := db.conn.Exec(` + INSERT INTO documents (file_path, file_hash, file_size, last_modified, status) + VALUES (?, ?, ?, ?, 'pending') + `, filePath, fileHash, fileSize, lastModified) + if err != nil { + return nil, err + } + id, _ := result.LastInsertId() + return db.getDocumentByID(id) + } + + // Update existing + _, err = db.conn.Exec(` + UPDATE documents SET file_hash = ?, file_size = ?, last_modified = ?, status = 'pending' + WHERE id = ? + `, fileHash, fileSize, lastModified, existing.ID) + if err != nil { + return nil, err + } + + return db.getDocumentByID(existing.ID) +} + +func (db *DB) getDocumentByID(id int64) (*Document, error) { + row := db.conn.QueryRow(` + SELECT id, file_path, file_hash, file_size, first_seen, last_modified, last_processed, status + FROM documents WHERE id = ? + `, id) + + var doc Document + err := row.Scan(&doc.ID, &doc.FilePath, &doc.FileHash, &doc.FileSize, + &doc.FirstSeen, &doc.LastModified, &doc.LastProcessed, &doc.Status) + if err != nil { + return nil, err + } + return &doc, nil +} + +// UpdateStatus updates the status of a document +func (db *DB) UpdateStatus(docID int64, status string) error { + _, err := db.conn.Exec(`UPDATE documents SET status = ? WHERE id = ?`, status, docID) + return err +} + +// MarkProcessed marks a document as successfully processed +func (db *DB) MarkProcessed(docID int64) error { + _, err := db.conn.Exec(` + UPDATE documents SET status = 'completed', last_processed = CURRENT_TIMESTAMP + WHERE id = ? + `, docID) + return err +} + +// SaveExtraction saves the extraction result +func (db *DB) SaveExtraction(docID int64, extractionJSON map[string]interface{}) error { + jsonBytes, err := json.Marshal(extractionJSON) + if err != nil { + return err + } + + _, err = db.conn.Exec(` + INSERT INTO extractions (document_id, extraction_json) VALUES (?, ?) + `, docID, string(jsonBytes)) + return err +} + +// LogAction logs a processing action +func (db *DB) LogAction(docID int64, action, details string) error { + _, err := db.conn.Exec(` + INSERT INTO processing_log (document_id, action, details) VALUES (?, ?, ?) + `, docID, action, details) + return err +} + +// NeedsProcessing checks if a document needs processing based on hash +func (db *DB) NeedsProcessing(filePath, currentHash string) (bool, error) { + doc, err := db.GetDocument(filePath) + if err != nil { + return false, err + } + if doc == nil { + return true, nil // New file + } + if doc.FileHash != currentHash { + return true, nil // File changed + } + if doc.Status == "failed" { + return true, nil // Retry failed + } + return false, nil // Already processed with same hash +} diff --git a/Code/Tracker/go.mod b/Code/Tracker/go.mod new file mode 100644 index 0000000..4c8bdc6 --- /dev/null +++ b/Code/Tracker/go.mod @@ -0,0 +1,11 @@ +module github.com/goko/jingtian-tracker + +go 1.22 + +require ( + github.com/fsnotify/fsnotify v1.7.0 + github.com/mattn/go-sqlite3 v1.14.22 + gopkg.in/yaml.v3 v3.0.1 +) + +require golang.org/x/sys v0.4.0 // indirect diff --git a/Code/Tracker/go.sum b/Code/Tracker/go.sum new file mode 100644 index 0000000..8fb5151 --- /dev/null +++ b/Code/Tracker/go.sum @@ -0,0 +1,10 @@ +github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA= +github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM= +github.com/mattn/go-sqlite3 v1.14.22 h1:2gZY6PC6kBnID23Tichd1K+Z0oS6nE/XwU+Vz/5o4kU= +github.com/mattn/go-sqlite3 v1.14.22/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= +golang.org/x/sys v0.4.0 h1:Zr2JFtRQNX3BCZ8YtxRE9hNJYC8J6I1MVbMg6owUp18= +golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/Code/Tracker/main.go b/Code/Tracker/main.go new file mode 100644 index 0000000..48b8ee2 --- /dev/null +++ b/Code/Tracker/main.go @@ -0,0 +1,68 @@ +package main + +import ( + "flag" + "log" + "os" + "os/signal" + "syscall" + + "github.com/goko/jingtian-tracker/config" + "github.com/goko/jingtian-tracker/db" + "github.com/goko/jingtian-tracker/pipeline" + "github.com/goko/jingtian-tracker/watcher" +) + +func main() { + // Parse flags + configPath := flag.String("config", "config.yaml", "Path to config file") + flag.Parse() + + log.SetFlags(log.LstdFlags | log.Lshortfile) + log.Println("JingTian Document Tracker starting...") + + // Load config + cfg, err := config.Load(*configPath) + if err != nil { + log.Fatalf("Failed to load config: %v", err) + } + log.Printf("Config loaded: watching %s", cfg.Watch.Root) + + // Initialize database + database, err := db.Open(cfg.Database.Path) + if err != nil { + log.Fatalf("Failed to open database: %v", err) + } + defer database.Close() + log.Printf("Database initialized: %s", cfg.Database.Path) + + // Initialize watcher + w, err := watcher.New(cfg) + if err != nil { + log.Fatalf("Failed to create watcher: %v", err) + } + + // Initialize pipeline + p, err := pipeline.New(cfg, database, w) + if err != nil { + log.Fatalf("Failed to create pipeline: %v", err) + } + defer p.Close() + + // Handle graceful shutdown + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) + + go func() { + <-sigChan + log.Println("Shutdown signal received, stopping...") + w.Stop() + }() + + // Run pipeline + if err := p.Run(); err != nil { + log.Fatalf("Pipeline error: %v", err) + } + + log.Println("JingTian Document Tracker stopped.") +} diff --git a/Code/Tracker/pipeline/pipeline.go b/Code/Tracker/pipeline/pipeline.go new file mode 100644 index 0000000..2bb636f --- /dev/null +++ b/Code/Tracker/pipeline/pipeline.go @@ -0,0 +1,341 @@ +package pipeline + +import ( + "fmt" + "log" + "os" + "path/filepath" + "strings" + "time" + + "github.com/goko/jingtian-tracker/clients" + "github.com/goko/jingtian-tracker/config" + "github.com/goko/jingtian-tracker/db" + "github.com/goko/jingtian-tracker/watcher" +) + +// TrackerColumns defines the tracker spreadsheet columns +var TrackerColumns = []string{ + "Document Path", + "Document Type", + "Client Name", + "Matter Reference", + "TM Number", + "Trademark Name", + "Trademark Class", + "Filing Date", + "Response Deadline", + "Hearing Date", + "Amount", + "Currency", + "Status", + "First Processed", + "Last Updated", + "Update Source", + "Processing Notes", +} + +type Pipeline struct { + cfg *config.Config + db *db.DB + watcher *watcher.Watcher + docling *clients.DoclingClient + ollama *clients.OllamaClient + tools *clients.ToolsClient + errorLog *os.File +} + +func New(cfg *config.Config, database *db.DB, w *watcher.Watcher) (*Pipeline, error) { + // Open error log file + var errorLog *os.File + if cfg.Logging.ErrorFile != "" { + // Ensure directory exists + dir := filepath.Dir(cfg.Logging.ErrorFile) + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, fmt.Errorf("failed to create log directory: %w", err) + } + + f, err := os.OpenFile(cfg.Logging.ErrorFile, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) + if err != nil { + return nil, fmt.Errorf("failed to open error log: %w", err) + } + errorLog = f + } + + return &Pipeline{ + cfg: cfg, + db: database, + watcher: w, + docling: clients.NewDoclingClient(cfg.Services.Docling), + ollama: clients.NewOllamaClient(cfg.Services.Ollama, cfg.Ollama.Model, cfg.Ollama.Temperature), + tools: clients.NewToolsClient(cfg.Services.Tools), + errorLog: errorLog, + }, nil +} + +func (p *Pipeline) Close() { + if p.errorLog != nil { + p.errorLog.Close() + } +} + +func (p *Pipeline) Run() error { + // Check service health + if err := p.checkServices(); err != nil { + return fmt.Errorf("service health check failed: %w", err) + } + + // Ensure tracker exists + if err := p.ensureTracker(); err != nil { + log.Printf("[pipeline] Warning: could not ensure tracker: %v", err) + // Don't fail - tracker might be created manually + } + + // Start watcher + if err := p.watcher.Start(); err != nil { + return fmt.Errorf("failed to start watcher: %w", err) + } + + log.Println("[pipeline] Started processing loop") + + // Process events + for event := range p.watcher.Events { + p.processFile(event) + } + + return nil +} + +func (p *Pipeline) checkServices() error { + log.Println("[pipeline] Checking service health...") + + if err := p.docling.Health(); err != nil { + return fmt.Errorf("docling: %w", err) + } + log.Println("[pipeline] Docling: OK") + + if err := p.ollama.Health(); err != nil { + return fmt.Errorf("ollama: %w", err) + } + log.Println("[pipeline] Ollama: OK") + + if err := p.tools.Health(); err != nil { + return fmt.Errorf("tools: %w", err) + } + log.Println("[pipeline] Tools: OK") + + return nil +} + +func (p *Pipeline) ensureTracker() error { + // Check if tracker exists by trying to read it + _, err := p.tools.ReadExcel(p.cfg.Tracker.Path, p.cfg.Tracker.Sheet, p.cfg.Tracker.HeaderRow) + if err != nil { + // Try to create it + log.Printf("[pipeline] Creating tracker: %s", p.cfg.Tracker.Path) + if createErr := p.tools.CreateTracker(p.cfg.Tracker.Path, p.cfg.Tracker.Sheet, TrackerColumns); createErr != nil { + return fmt.Errorf("failed to create tracker: %w", createErr) + } + log.Printf("[pipeline] Tracker created successfully") + } + return nil +} + +func (p *Pipeline) processFile(event watcher.FileEvent) { + // Get relative path + relPath, err := filepath.Rel(p.cfg.Watch.Root, event.Path) + if err != nil { + relPath = event.Path + } + + log.Printf("[pipeline] Processing: %s", relPath) + + // Check if processing needed + needsProcessing, err := p.db.NeedsProcessing(relPath, event.Hash) + if err != nil { + p.logError(relPath, fmt.Sprintf("failed to check processing status: %v", err)) + return + } + + if !needsProcessing { + log.Printf("[pipeline] Skipping (already processed with same hash): %s", relPath) + return + } + + // Upsert document record + doc, err := p.db.UpsertDocument(relPath, event.Hash, event.Size, event.ModTime) + if err != nil { + p.logError(relPath, fmt.Sprintf("failed to upsert document: %v", err)) + return + } + + // Update status to processing + p.db.UpdateStatus(doc.ID, "processing") + + // Process based on file type + ext := strings.ToLower(filepath.Ext(event.Path)) + var extracted map[string]interface{} + + switch ext { + case ".xlsx", ".xls": + // For Excel files, we don't OCR, just log that we saw it + log.Printf("[pipeline] Excel file detected, skipping OCR: %s", relPath) + p.db.LogAction(doc.ID, "skipped", "Excel files are not processed through OCR") + p.db.MarkProcessed(doc.ID) + return + + default: + // Send to Docling for OCR/extraction + content, doclingErr := p.docling.Convert(event.Path) + if doclingErr != nil { + p.logError(relPath, fmt.Sprintf("Docling failed: %v", doclingErr)) + p.db.UpdateStatus(doc.ID, "failed") + p.db.LogAction(doc.ID, "failed", fmt.Sprintf("Docling error: %v", doclingErr)) + return + } + + if content == "" { + p.logError(relPath, "Docling returned empty content") + p.db.UpdateStatus(doc.ID, "failed") + p.db.LogAction(doc.ID, "failed", "Empty content from Docling") + return + } + + // Detect document type from extension + docType := detectDocType(ext) + + // Send to Ollama for extraction + extracted, err = p.ollama.Extract(content, docType) + if err != nil { + p.logError(relPath, fmt.Sprintf("Ollama failed: %v", err)) + p.db.UpdateStatus(doc.ID, "failed") + p.db.LogAction(doc.ID, "failed", fmt.Sprintf("Ollama error: %v", err)) + return + } + } + + // Save extraction to database + if err := p.db.SaveExtraction(doc.ID, extracted); err != nil { + p.logError(relPath, fmt.Sprintf("failed to save extraction: %v", err)) + } + + // Write to tracker + if err := p.writeToTracker(relPath, extracted); err != nil { + p.logError(relPath, fmt.Sprintf("failed to write to tracker: %v", err)) + p.db.UpdateStatus(doc.ID, "failed") + p.db.LogAction(doc.ID, "failed", fmt.Sprintf("Tracker write error: %v", err)) + return + } + + // Mark as completed + p.db.MarkProcessed(doc.ID) + p.db.LogAction(doc.ID, "processed", "Successfully processed and added to tracker") + log.Printf("[pipeline] Successfully processed: %s", relPath) +} + +func (p *Pipeline) writeToTracker(docPath string, extracted map[string]interface{}) error { + // Check if row exists for this document + existing, err := p.tools.FindRow( + p.cfg.Tracker.Path, + p.cfg.Tracker.Sheet, + p.cfg.Tracker.HeaderRow, + "Document Path", + docPath, + ) + if err != nil { + // Row not found is OK - we'll add a new one + existing = &clients.ExcelFindResponse{Found: false} + } + + now := time.Now().Format("02-01-2006 15:04:05") + + // Build row data + row := map[string]interface{}{ + "Document Path": docPath, + "Document Type": getStringField(extracted, "document_type"), + "Client Name": getStringField(extracted, "client_name"), + "Matter Reference": getStringField(extracted, "matter_reference"), + "TM Number": getStringField(extracted, "tm_number"), + "Trademark Name": getStringField(extracted, "trademark_name"), + "Trademark Class": getStringField(extracted, "trademark_class"), + "Filing Date": getStringField(extracted, "filing_date"), + "Response Deadline": getStringField(extracted, "response_deadline"), + "Hearing Date": getStringField(extracted, "hearing_date"), + "Amount": getField(extracted, "amount"), + "Currency": getStringField(extracted, "currency"), + "Status": getStringField(extracted, "status"), + "Last Updated": now, + "Update Source": "Pipeline", + "Processing Notes": "", + } + + if existing.Found { + // Check if manually edited + if updateSource, ok := existing.Row["Update Source"].(string); ok && updateSource == "Manual" { + log.Printf("[pipeline] Skipping update (manually edited): %s", docPath) + return nil + } + + // Update existing row + // Preserve First Processed + if fp, ok := existing.Row["First Processed"]; ok { + row["First Processed"] = fp + } + + return p.tools.UpdateRow( + p.cfg.Tracker.Path, + p.cfg.Tracker.Sheet, + p.cfg.Tracker.HeaderRow, + existing.RowNum, + row, + ) + } + + // Add new row + row["First Processed"] = now + return p.tools.AddRow( + p.cfg.Tracker.Path, + p.cfg.Tracker.Sheet, + p.cfg.Tracker.HeaderRow, + row, + ) +} + +func (p *Pipeline) logError(path, msg string) { + errMsg := fmt.Sprintf("[%s] %s: %s", time.Now().Format("2006-01-02 15:04:05"), path, msg) + log.Printf("[pipeline] ERROR: %s", errMsg) + + if p.errorLog != nil { + p.errorLog.WriteString(errMsg + "\n") + } +} + +func detectDocType(ext string) string { + switch ext { + case ".pdf": + return "PDF document" + case ".docx", ".doc": + return "Word document" + case ".png", ".jpg", ".jpeg": + return "Image/screenshot" + default: + return "Document" + } +} + +func getStringField(m map[string]interface{}, key string) string { + if v, ok := m[key]; ok { + if s, ok := v.(string); ok { + return s + } + return fmt.Sprintf("%v", v) + } + return "" +} + +func getField(m map[string]interface{}, key string) interface{} { + if v, ok := m[key]; ok { + return v + } + return "" +} diff --git a/Code/Tracker/watcher/watcher.go b/Code/Tracker/watcher/watcher.go new file mode 100644 index 0000000..34bff47 --- /dev/null +++ b/Code/Tracker/watcher/watcher.go @@ -0,0 +1,241 @@ +package watcher + +import ( + "crypto/sha256" + "encoding/hex" + "io" + "log" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/fsnotify/fsnotify" + "github.com/goko/jingtian-tracker/config" +) + +type FileEvent struct { + Path string + Hash string + Size int64 + ModTime time.Time +} + +type Watcher struct { + cfg *config.Config + watcher *fsnotify.Watcher + pending map[string]time.Time // path -> first seen time + pendingMu sync.Mutex + Events chan FileEvent + done chan struct{} +} + +func New(cfg *config.Config) (*Watcher, error) { + fsWatcher, err := fsnotify.NewWatcher() + if err != nil { + return nil, err + } + + w := &Watcher{ + cfg: cfg, + watcher: fsWatcher, + pending: make(map[string]time.Time), + Events: make(chan FileEvent, 100), + done: make(chan struct{}), + } + + return w, nil +} + +func (w *Watcher) Start() error { + // Add all directories recursively + err := filepath.Walk(w.cfg.Watch.Root, func(path string, info os.FileInfo, err error) error { + if err != nil { + return nil // Skip errors + } + if info.IsDir() { + // Check exclusions + for _, excl := range w.cfg.Watch.Exclude { + if strings.Contains(path, excl) { + return filepath.SkipDir + } + } + return w.watcher.Add(path) + } + return nil + }) + if err != nil { + return err + } + + // Start event loop + go w.eventLoop() + + // Start stability checker + go w.stabilityChecker() + + log.Printf("[watcher] Started watching: %s", w.cfg.Watch.Root) + return nil +} + +func (w *Watcher) Stop() { + close(w.done) + w.watcher.Close() + close(w.Events) +} + +func (w *Watcher) eventLoop() { + for { + select { + case <-w.done: + return + case event, ok := <-w.watcher.Events: + if !ok { + return + } + + // Handle directory creation - add to watch + if event.Op&fsnotify.Create != 0 { + info, err := os.Stat(event.Name) + if err == nil && info.IsDir() { + // Check exclusions + excluded := false + for _, excl := range w.cfg.Watch.Exclude { + if strings.Contains(event.Name, excl) { + excluded = true + break + } + } + if !excluded { + w.watcher.Add(event.Name) + log.Printf("[watcher] Added directory: %s", event.Name) + } + continue + } + } + + // Handle file create/write + if event.Op&(fsnotify.Create|fsnotify.Write) != 0 { + if w.isWatchedFile(event.Name) { + w.pendingMu.Lock() + if _, exists := w.pending[event.Name]; !exists { + w.pending[event.Name] = time.Now() + log.Printf("[watcher] File detected, starting stability check: %s", event.Name) + } else { + // File was modified again, reset timer + w.pending[event.Name] = time.Now() + } + w.pendingMu.Unlock() + } + } + + case err, ok := <-w.watcher.Errors: + if !ok { + return + } + log.Printf("[watcher] Error: %v", err) + } + } +} + +func (w *Watcher) stabilityChecker() { + ticker := time.NewTicker(10 * time.Second) + defer ticker.Stop() + + for { + select { + case <-w.done: + return + case <-ticker.C: + w.checkStability() + } + } +} + +func (w *Watcher) checkStability() { + w.pendingMu.Lock() + defer w.pendingMu.Unlock() + + stabilityDuration := time.Duration(w.cfg.Watch.StabilitySeconds) * time.Second + now := time.Now() + + for path, firstSeen := range w.pending { + // Check if file has been stable long enough + if now.Sub(firstSeen) < stabilityDuration { + continue + } + + // Get current file info + info, err := os.Stat(path) + if err != nil { + // File was deleted + delete(w.pending, path) + continue + } + + // Calculate hash + hash, err := hashFile(path) + if err != nil { + log.Printf("[watcher] Error hashing file %s: %v", path, err) + delete(w.pending, path) + continue + } + + // Emit event + event := FileEvent{ + Path: path, + Hash: hash, + Size: info.Size(), + ModTime: info.ModTime(), + } + + select { + case w.Events <- event: + log.Printf("[watcher] File stable, emitting event: %s", path) + default: + log.Printf("[watcher] Event channel full, dropping: %s", path) + } + + delete(w.pending, path) + } +} + +func (w *Watcher) isWatchedFile(path string) bool { + // Check exclusions + for _, excl := range w.cfg.Watch.Exclude { + if strings.Contains(path, excl) { + return false + } + } + + // Check extension + ext := strings.ToLower(filepath.Ext(path)) + for _, watchExt := range w.cfg.Watch.Extensions { + if ext == strings.ToLower(watchExt) { + return true + } + } + + return false +} + +func hashFile(path string) (string, error) { + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + + h := sha256.New() + if _, err := io.Copy(h, f); err != nil { + return "", err + } + + return hex.EncodeToString(h.Sum(nil)), nil +} + +// HashFile exports the hash function for use by other packages +func HashFile(path string) (string, error) { + return hashFile(path) +} diff --git a/Code/docker-compose.yaml b/Code/docker-compose.yaml index d4f50b5..277f85c 100644 --- a/Code/docker-compose.yaml +++ b/Code/docker-compose.yaml @@ -76,6 +76,28 @@ services: retries: 3 start_period: 10s + # Tracker (Go Orchestrator - Document Processing Pipeline) + tracker: + build: ./Tracker + container_name: jt-tracker + restart: unless-stopped + volumes: + - /data/jingtian/BenjaminTeam:/data:rw + - ./Tracker/config.yaml.example:/app/config.yaml:ro + depends_on: + docling: + condition: service_healthy + ollama: + condition: service_healthy + tools: + condition: service_healthy + healthcheck: + test: ["CMD", "pgrep", "-f", "tracker"] + interval: 30s + timeout: 5s + retries: 3 + start_period: 10s + volumes: docling_models: name: jt-docling-models