Compare commits

..

9 Commits

Author SHA1 Message Date
MangoPig 0f1692cc56 Image processing 2026-02-22 01:26:00 +00:00
MangoPig 991d5a0eae Nil fix 2026-02-22 00:20:00 +00:00
MangoPig ffd463df94 Ollama no curl 2026-02-22 00:02:30 +00:00
MangoPig 1f47e70d5c Fix 2026-02-21 23:58:53 +00:00
MangoPig 6fc84b20b6 Go orchestrator 2026-02-21 23:52:55 +00:00
MangoPig 6dce685906 Test Excel endpoint 2026-02-21 23:25:13 +00:00
MangoPig 50ed6e48f5 Fixed starting command 2026-02-21 23:13:27 +00:00
MangoPig 3b95d65e2d Increase Health Check Start Period 2026-02-21 23:10:24 +00:00
MangoPig 3b52c2ba02 Download separately 2026-02-21 23:07:54 +00:00
18 changed files with 2072 additions and 33 deletions
+1
View File
@@ -14,6 +14,7 @@ __pycache__/
# Sample outputs
Tools/Samples/outputs/
*output/
# Environment
.env
+67 -8
View File
@@ -49,7 +49,15 @@ Rationale: De-risk unknowns (Docling OCR quality, Ollama CPU inference speed) be
**Tested Formats:**
- PDF (Filing Receipt) — extracted text, tables, CJK content ✅
- PNG (Email screenshot) — extracted subject, dates, recipient ✅ (CJK trademark garbled, may need `ocr_lang`)
- PNG (Email screenshot) — extracted subject, dates, recipient ✅
- PNG with `ocr_lang=ch_tra` — Traditional Chinese extracted correctly (官藥坊) ✅
**Portable Chinese OCR:**
- Auto-downloads `ch_sim` (Simplified) and `ch_tra` (Traditional) models on first run
- Models persist in `docling_models` volume
- First run takes ~1-2 min extra for model download; subsequent runs instant
- Use `ocr_lang=ch_tra` for HK/Taiwan, `ocr_lang=ch_sim` for Mainland China
### 1b. Ollama (DONE)
@@ -68,16 +76,67 @@ Rationale: De-risk unknowns (Docling OCR quality, Ollama CPU inference speed) be
- Response: `{ "response": "<json>" }` (strip `<think>...</think>` 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
+12
View File
@@ -0,0 +1,12 @@
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
+382
View File
@@ -0,0 +1,382 @@
"""
Generic Excel API Service
A FastAPI service for reading and writing Excel files (.xlsx).
All file paths are relative to /data/ mount.
Endpoints:
- GET /health - Health check
- GET /excel/sheets - List all sheet names
- 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, 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 and writing xlsx files",
version="0.3.0",
)
DATA_ROOT = Path(os.getenv("DATA_ROOT", "/data"))
# 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}")
if not full_path.suffix.lower() == ".xlsx":
raise HTTPException(status_code=400, detail="Only .xlsx files supported")
try:
return load_workbook(full_path, read_only=True, data_only=True)
except Exception as e:
raise HTTPException(
status_code=500, detail=f"Failed to open workbook: {str(e)}"
)
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:
raise HTTPException(
status_code=404,
detail=f"Sheet '{sheet_name}' not found. Available: {wb.sheetnames}",
)
return wb[sheet_name]
def cell_to_str(cell_value) -> str:
"""Convert cell value to string, empty string for None."""
if cell_value is None:
return ""
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.3.0"}
@app.get("/excel/sheets")
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_readonly(file_path)
sheets = wb.sheetnames
wb.close()
return {"file_path": file_path, "sheets": sheets, "count": len(sheets)}
@app.get("/excel/schema")
async def get_schema(
file_path: str = Query(..., description="Path to xlsx file relative to /data/"),
sheet_name: str = Query(..., description="Sheet name to read"),
header_row: int = Query(1, description="Row number containing headers (1-indexed)"),
):
"""Get column headers from the specified sheet."""
wb = get_workbook_readonly(file_path)
ws = get_sheet(wb, sheet_name)
headers = get_headers(ws, header_row)
wb.close()
return {
"file_path": file_path,
"sheet_name": sheet_name,
"header_row": header_row,
"columns": headers,
"column_count": len(headers),
}
@app.get("/excel/read")
async def read_all(
file_path: str = Query(..., description="Path to xlsx file relative to /data/"),
sheet_name: str = Query(..., description="Sheet name to read"),
header_row: int = Query(1, description="Row number containing headers (1-indexed)"),
start_row: Optional[int] = Query(
None,
description="Start reading from this row (1-indexed, defaults to header_row + 1)",
),
limit: Optional[int] = Query(None, description="Max rows to return"),
):
"""Read all rows from the specified sheet."""
wb = get_workbook_readonly(file_path)
ws = get_sheet(wb, sheet_name)
headers = get_headers(ws, header_row)
data_start = start_row if start_row else header_row + 1
rows = []
row_count = 0
for row_num, row in enumerate(ws.iter_rows(min_row=data_start), start=data_start):
if limit and row_count >= limit:
break
values = [cell_to_str(cell.value) for cell in row]
if all(v == "" for v in values):
continue
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]
rows.append(row_data)
row_count += 1
wb.close()
return {
"file_path": file_path,
"sheet_name": sheet_name,
"columns": headers,
"rows": rows,
"row_count": len(rows),
}
@app.get("/excel/row/{row_num}")
async def get_row(
row_num: int,
file_path: str = Query(..., description="Path to xlsx file relative to /data/"),
sheet_name: str = Query(..., description="Sheet name to read"),
header_row: int = Query(1, description="Row number containing headers (1-indexed)"),
):
"""Get a specific row by row number (1-indexed)."""
if row_num < 1:
raise HTTPException(status_code=400, detail="Row number must be >= 1")
wb = get_workbook_readonly(file_path)
ws = get_sheet(wb, sheet_name)
headers = get_headers(ws, header_row)
try:
row = ws[row_num]
except Exception:
wb.close()
raise HTTPException(status_code=404, detail=f"Row {row_num} not found")
values = [cell_to_str(cell.value) for cell in row]
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 {"file_path": file_path, "sheet_name": sheet_name, "row": row_data}
@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",
}
+4
View File
@@ -0,0 +1,4 @@
fastapi==0.115.0
uvicorn==0.30.6
openpyxl==3.1.5
pydantic==2.9.0
+31
View File
@@ -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"]
+136
View File
@@ -0,0 +1,136 @@
// 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")
writer.WriteField("image_export_mode", "placeholder") // Avoid base64 blobs flooding LLM context
// 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"
}
}
+165
View File
@@ -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 <think> 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 <think>...</think> tags
re := regexp.MustCompile(`(?s)<think>.*?</think>`)
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
}
+216
View File
@@ -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
}
+37
View File
@@ -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
+73
View File
@@ -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
}
+228
View File
@@ -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
}
+11
View File
@@ -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
+10
View File
@@ -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=
+68
View File
@@ -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.")
}
+335
View File
@@ -0,0 +1,335 @@
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": getStringField(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 {
v, ok := m[key]
if !ok || v == nil {
return ""
}
if s, ok := v.(string); ok {
return s
}
return fmt.Sprintf("%v", v)
}
+241
View File
@@ -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)
}
+52 -22
View File
@@ -18,19 +18,24 @@ services:
command:
- |
# Download Chinese OCR models if missing (first run only)
if [ ! -f /opt/app-root/src/.cache/docling/models/EasyOcr/zh_sim_g2.pth ] || [ ! -f /opt/app-root/src/.cache/docling/models/EasyOcr/zh_tra_g2.pth ]; then
echo "Downloading Chinese OCR models (Simplified + Traditional)..."
python3 -c "import easyocr; easyocr.Reader(['en','ch_sim','ch_tra'], gpu=False)"
# Copy downloaded models to correct location
# Note: ch_sim and ch_tra are mutually exclusive, must download separately
if [ ! -f /opt/app-root/src/.cache/docling/models/EasyOcr/zh_sim_g2.pth ]; then
echo "Downloading Simplified Chinese OCR model..."
python3 -c "import easyocr; easyocr.Reader(['en','ch_sim'], gpu=False)"
cp -n /opt/app-root/src/.EasyOCR/model/*.pth /opt/app-root/src/.cache/docling/models/EasyOcr/ 2>/dev/null || true
fi
exec uvicorn docling_serve.app:app --host 0.0.0.0 --port 5001
if [ ! -f /opt/app-root/src/.cache/docling/models/EasyOcr/zh_tra_g2.pth ]; then
echo "Downloading Traditional Chinese OCR model..."
python3 -c "import easyocr; easyocr.Reader(['en','ch_tra'], gpu=False)"
cp -n /opt/app-root/src/.EasyOCR/model/*.pth /opt/app-root/src/.cache/docling/models/EasyOcr/ 2>/dev/null || true
fi
exec docling-serve run --host 0.0.0.0 --port 5001
healthcheck:
test: ["CMD", "curl", "-sf", "http://localhost:5001/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 120s
retries: 5
start_period: 500s
# Ollama
ollama:
@@ -47,26 +52,51 @@ services:
- "11434:11434"
entrypoint: ["/bin/sh", "-c", "ollama serve & sleep 5 && ollama pull qwen3:1.7b && wait"]
healthcheck:
test: ["CMD", "curl", "-sf", "http://localhost:11434/api/tags"]
test: ["CMD", "ollama", "list"]
interval: 30s
timeout: 10s
retries: 5
start_period: 120s
# Tools (Python FastAPI - Excel operations)
# tools:
# build: ./Tools/Service
# container_name: jt-tools
# restart: unless-stopped
# volumes:
# - /data/jingtian/BenjaminTeam:/data:rw
# ports:
# - "8080:8080"
# healthcheck:
# test: ["CMD", "curl", "-sf", "http://localhost:8080/health"]
# interval: 30s
# timeout: 10s
# retries: 3
# Tools (Python FastAPI - Generic Excel API)
tools:
build: ./Tools/Service
container_name: jt-tools
restart: unless-stopped
environment:
DATA_ROOT: /data
volumes:
- /data/jingtian/BenjaminTeam:/data:rw
ports:
- "8000:8000"
healthcheck:
test: ["CMD", "python3", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"]
interval: 30s
timeout: 10s
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: