kissflow integration test

fredrik0x · fredrik0x · commit f7344b59b23a · 2025-06-27T09:20:32.000+02:00
diff --git a/.env.example b/.env.example
@@ -5,3 +5,6 @@ AWS_REGION='us-east-1'
 SES_FROM_EMAIL=''
 TURNSTILE_SITE_KEY=''
 TURNSTILE_SECRET_KEY=''
+KISSFLOW_API_KEY=''
+KISSFLOW_ACCOUNT_ID=''
+KISSFLOW_PROCESS_ID=''
diff --git a/README.md b/README.md
@@ -18,14 +18,28 @@ Docker Compose.
 
 ### Third Party Services
 
-* Sendgrid
-* Google reCAPTCHA
+* AWS SES (for email delivery)
+* Cloudflare Turnstile (for bot protection)
+* Kissflow API (optional - for KYC submission tracking)
 
 
 ## New setup
 
 Make a fork of the repository. Set environment variables in `.env` file, using the provided example. Customise the templates and code. Update public keys in [static/js/public-keys.js](static/js/public-keys.js). Deploy to your web server or K8s cluster.
 
+### Kissflow Integration (Optional)
+
+The application now supports automatic integration with Kissflow for KYC submission tracking. When enabled, legal submissions with a Grant ID will automatically update the corresponding AOG (Approval of Grants) item in Kissflow with the submission identifier.
+
+To enable Kissflow integration:
+1. Add the following to your `.env` file:
+   ```
+   KISSFLOW_API_KEY=your_api_key
+   KISSFLOW_ACCOUNT_ID=your_account_id
+   KISSFLOW_PROCESS_ID=your_aog_process_id
+   ```
+2. Ensure your Kissflow API has permissions to read and update AOG items
+3. Test the integration using `python test_kissflow_integration.py`
 
 ## Security
 
@@ -39,4 +53,4 @@ A server operator should follow best practises for security when setting up and
 docker compose up
 ```
 
-The server will be listening on 4200 port.
+The server will be listening on 4200 port.
diff --git a/server.py b/server.py
@@ -4,6 +4,7 @@
 from random import Random
 import requests
 import base64
+import json
 
 from flask import Flask, render_template, request, jsonify
 from flask_limiter import Limiter
@@ -184,6 +185,108 @@ def get_forwarded_address():
     # Otherwise use the default function
     return get_remote_address()
 
+def find_aog_item_by_grant_id(grant_id):
+    """
+    Finds an AOG (Approval of Grants) item in Kissflow by Grant ID.
+    Returns the item ID if found, None otherwise.
+    """
+    try:
+        api_key = os.getenv('KISSFLOW_API_KEY')
+        account_id = os.getenv('KISSFLOW_ACCOUNT_ID')
+        process_id = os.getenv('KISSFLOW_PROCESS_ID')
+        
+        if not all([api_key, account_id, process_id]):
+            logging.error("Missing Kissflow configuration")
+            return None
+        
+        # Kissflow API endpoint to search for items
+        url = f"https://api.kissflow.com/api/v1/accounts/{account_id}/processes/{process_id}/items"
+        
+        headers = {
+            'Authorization': f'Bearer {api_key}',
+            'Content-Type': 'application/json'
+        }
+        
+        # Search for items with matching Grant ID in Request_number field
+        params = {
+            'filter': f'Request_number eq "{grant_id}"',
+            'limit': 1
+        }
+        
+        response = requests.get(url, headers=headers, params=params)
+        
+        if response.status_code == 200:
+            data = response.json()
+            items = data.get('data', [])
+            if items:
+                return items[0].get('_id')
+        else:
+            logging.error(f"Kissflow API error: {response.status_code} - {response.text}")
+            
+    except Exception as e:
+        logging.error(f"Error finding AOG item: {str(e)}")
+    
+    return None
+
+def update_aog_kyc_comments(item_id, legal_identifier):
+    """
+    Updates the KYC_Comments field in a Kissflow AOG item with the legal identifier.
+    """
+    try:
+        api_key = os.getenv('KISSFLOW_API_KEY')
+        account_id = os.getenv('KISSFLOW_ACCOUNT_ID')
+        process_id = os.getenv('KISSFLOW_PROCESS_ID')
+        
+        if not all([api_key, account_id, process_id]):
+            logging.error("Missing Kissflow configuration")
+            return False
+        
+        # Kissflow API endpoint to update an item
+        url = f"https://api.kissflow.com/api/v1/accounts/{account_id}/processes/{process_id}/items/{item_id}"
+        
+        headers = {
+            'Authorization': f'Bearer {api_key}',
+            'Content-Type': 'application/json'
+        }
+        
+        # Update the KYC_Comments field
+        payload = {
+            'KYC_Comments': legal_identifier
+        }
+        
+        response = requests.patch(url, headers=headers, json=payload)
+        
+        if response.status_code in [200, 204]:
+            logging.info(f"Successfully updated AOG item {item_id} with legal identifier {legal_identifier}")
+            return True
+        else:
+            logging.error(f"Kissflow API error: {response.status_code} - {response.text}")
+            
+    except Exception as e:
+        logging.error(f"Error updating AOG item: {str(e)}")
+    
+    return False
+
+def send_identifier_to_kissflow(grant_id, legal_identifier):
+    """
+    Sends the legal identifier to the Kissflow AOG item based on Grant ID.
+    """
+    if not grant_id:
+        logging.warning("No Grant ID provided, skipping Kissflow update")
+        return False
+    
+    # Find the AOG item by Grant ID
+    item_id = find_aog_item_by_grant_id(grant_id)
+    
+    if not item_id:
+        logging.warning(f"No AOG item found for Grant ID: {grant_id}")
+        return False
+    
+    # Update the KYC_Comments field
+    success = update_aog_kyc_comments(item_id, legal_identifier)
+    
+    return success
+
 # Validate required environment variables
 required_env_vars = ['TURNSTILE_SITE_KEY', 'TURNSTILE_SECRET_KEY', 'AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY', 'AWS_REGION', 'SES_FROM_EMAIL']
 validate_env_vars(required_env_vars)
@@ -264,6 +367,16 @@ def submit():
 
         send_email(message)
 
+        # If this is a legal submission with a Grant ID (reference), send to Kissflow
+        if recipient == 'legal' and reference:
+            kissflow_success = send_identifier_to_kissflow(reference, identifier)
+            if kissflow_success:
+                logging.info(f"Successfully sent identifier {identifier} to Kissflow for Grant ID {reference}")
+            else:
+                logging.warning(f"Failed to send identifier {identifier} to Kissflow for Grant ID {reference}")
+                # Note: We don't fail the submission if Kissflow update fails
+                # The email has already been sent successfully
+
         notice = f'Thank you! The relevant team was notified of your submission. Please record the identifier and refer to it in correspondence: {identifier}'
 
         return jsonify({'status': 'success', 'message': notice})
diff --git a/test_kissflow_integration.py b/test_kissflow_integration.py
@@ -0,0 +1,182 @@
+"""
+Test script for Kissflow API integration
+This script helps verify that the Kissflow API integration is working correctly
+"""
+
+import os
+import sys
+from datetime import datetime
+from random import Random
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# Add the current directory to the Python path
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+# Import the functions we want to test
+from server import (
+    get_identifier,
+    find_aog_item_by_grant_id,
+    update_aog_kyc_comments,
+    send_identifier_to_kissflow
+)
+
+
+def test_environment_variables():
+    """Test that all required Kissflow environment variables are set"""
+    print("Testing environment variables...")
+    
+    required_vars = ['KISSFLOW_API_KEY', 'KISSFLOW_ACCOUNT_ID', 'KISSFLOW_PROCESS_ID']
+    missing_vars = []
+    
+    for var in required_vars:
+        value = os.getenv(var)
+        if value:
+            # Mask the value for security
+            masked_value = value[:4] + '...' + value[-4:] if len(value) > 8 else '***'
+            print(f"✓ {var}: {masked_value}")
+        else:
+            print(f"✗ {var}: NOT SET")
+            missing_vars.append(var)
+    
+    if missing_vars:
+        print(f"\nERROR: Missing environment variables: {', '.join(missing_vars)}")
+        print("Please set these variables in your .env file")
+        return False
+    
+    print("\nAll environment variables are set!")
+    return True
+
+
+def test_identifier_generation():
+    """Test the identifier generation function"""
+    print("\nTesting identifier generation...")
+    
+    # Test with default parameters
+    identifier1 = get_identifier('legal')
+    print(f"Generated identifier: {identifier1}")
+    
+    # Test with custom parameters
+    custom_date = datetime(2025, 1, 15, 14, 30, 45)
+    identifier2 = get_identifier('legal', now=custom_date, randint=1234)
+    expected = 'legal:2025:01:15:14:30:45:1234'
+    
+    if identifier2 == expected:
+        print(f"✓ Custom identifier matches expected: {identifier2}")
+    else:
+        print(f"✗ Custom identifier mismatch!")
+        print(f"  Expected: {expected}")
+        print(f"  Got: {identifier2}")
+    
+    return True
+
+
+def test_find_aog_item(grant_id):
+    """Test finding an AOG item by Grant ID"""
+    print(f"\nTesting AOG item lookup for Grant ID: {grant_id}")
+    
+    try:
+        item_id = find_aog_item_by_grant_id(grant_id)
+        
+        if item_id:
+            print(f"✓ Found AOG item with ID: {item_id}")
+            return item_id
+        else:
+            print(f"✗ No AOG item found for Grant ID: {grant_id}")
+            return None
+            
+    except Exception as e:
+        print(f"✗ Error finding AOG item: {str(e)}")
+        return None
+
+
+def test_update_kyc_comments(item_id, identifier):
+    """Test updating the KYC_Comments field"""
+    print(f"\nTesting KYC_Comments update...")
+    print(f"Item ID: {item_id}")
+    print(f"Identifier: {identifier}")
+    
+    try:
+        success = update_aog_kyc_comments(item_id, identifier)
+        
+        if success:
+            print("✓ Successfully updated KYC_Comments field")
+            return True
+        else:
+            print("✗ Failed to update KYC_Comments field")
+            return False
+            
+    except Exception as e:
+        print(f"✗ Error updating KYC_Comments: {str(e)}")
+        return False
+
+
+def test_full_integration(grant_id):
+    """Test the full integration flow"""
+    print(f"\nTesting full integration flow for Grant ID: {grant_id}")
+    
+    # Generate a test identifier
+    identifier = get_identifier('legal')
+    print(f"Generated identifier: {identifier}")
+    
+    try:
+        success = send_identifier_to_kissflow(grant_id, identifier)
+        
+        if success:
+            print("✓ Full integration test passed!")
+            print(f"  - Grant ID: {grant_id}")
+            print(f"  - Identifier: {identifier}")
+            print("  - Successfully updated in Kissflow")
+            return True
+        else:
+            print("✗ Full integration test failed")
+            return False
+            
+    except Exception as e:
+        print(f"✗ Error in full integration: {str(e)}")
+        return False
+
+
+def main():
+    """Main test function"""
+    print("=" * 60)
+    print("Kissflow Integration Test Suite")
+    print("=" * 60)
+    
+    # Test 1: Environment variables
+    if not test_environment_variables():
+        print("\nCannot proceed without proper environment variables.")
+        return
+    
+    # Test 2: Identifier generation
+    test_identifier_generation()
+    
+    # Get Grant ID for testing
+    print("\n" + "-" * 60)
+    grant_id = input("Enter a Grant ID to test (or press Enter to skip API tests): ").strip()
+    
+    if not grant_id:
+        print("Skipping API tests.")
+        return
+    
+    # Test 3: Find AOG item
+    item_id = test_find_aog_item(grant_id)
+    
+    if item_id:
+        # Test 4: Update KYC Comments
+        test_identifier = f"TEST:{get_identifier('legal')}"
+        test_update_kyc_comments(item_id, test_identifier)
+    
+    # Test 5: Full integration
+    print("\n" + "-" * 60)
+    test_full_integration(grant_id)
+    
+    print("\n" + "=" * 60)
+    print("Test suite completed!")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    main()
