FULL WORKING V 1.0

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
2025-12-22 16:15:50 -07:00
parent 12407b74e4
commit f81dcccbb6
28 changed files with 5136 additions and 12 deletions
+89 -1
View File
@@ -1,11 +1,13 @@
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import FileResponse, PlainTextResponse
from fastapi.responses import FileResponse, PlainTextResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import zipfile
import os
import shutil
from pathlib import Path
from qc_validator import validate_shapefiles
from verofy_uploader import upload_to_verofy
app = FastAPI()
@@ -21,6 +23,12 @@ app.add_middleware(
TEMP_DIR = Path("../temp")
TEMP_DIR.mkdir(exist_ok=True)
class VerofyMapRequest(BaseModel):
mapId: int
verofyEmail: str
verofyPassword: str
@app.post("/upload")
async def upload_shapefile(file: UploadFile = File(...)):
"""Handle shapefile ZIP upload and QC validation"""
@@ -42,6 +50,19 @@ async def upload_shapefile(file: UploadFile = File(...)):
try:
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
zip_ref.extractall(TEMP_DIR)
# Check if files are in a subdirectory and flatten if needed
shp_files = list(TEMP_DIR.glob("*.shp"))
if len(shp_files) == 0:
# Look for shapefiles in subdirectories
subdirs = [d for d in TEMP_DIR.iterdir() if d.is_dir()]
if len(subdirs) == 1:
# Move all files from subdirectory to temp root
subdir = subdirs[0]
for item in subdir.iterdir():
shutil.move(str(item), str(TEMP_DIR / item.name))
# Remove empty subdirectory
subdir.rmdir()
except Exception as e:
return PlainTextResponse(f"Error extracting ZIP file: {str(e)}", status_code=400)
@@ -65,6 +86,73 @@ async def upload_shapefile(file: UploadFile = File(...)):
filename="QC_report.txt"
)
@app.post("/push-to-verofy")
async def push_to_verofy(request: VerofyMapRequest):
"""Push shapefiles from temp folder to Verofy"""
# Use credentials from the request
verofy_email = request.verofyEmail
verofy_password = request.verofyPassword
if not verofy_email or not verofy_password:
return JSONResponse(
status_code=400,
content={
"success": False,
"error": "Verofy credentials are required. Please provide your email and password."
}
)
# Check if temp directory has shapefiles
shapefile_count = len(list(TEMP_DIR.glob("*.shp")))
if shapefile_count == 0:
return JSONResponse(
status_code=400,
content={
"success": False,
"error": "No shapefiles found in temp directory. Please upload files first."
}
)
# Upload to Verofy
try:
result = upload_to_verofy(
temp_dir=str(TEMP_DIR),
map_id=request.mapId,
email=verofy_email,
password=verofy_password,
limit=None # Upload ALL records (no limit)
)
if result["success"]:
return JSONResponse(
status_code=200,
content={
"success": True,
"message": "Successfully uploaded to Verofy",
"uploaded": result["uploaded"],
"errors": result.get("errors", [])
}
)
else:
return JSONResponse(
status_code=500,
content={
"success": False,
"error": "Upload to Verofy failed",
"details": result.get("errors", [])
}
)
except Exception as e:
return JSONResponse(
status_code=500,
content={
"success": False,
"error": f"Error uploading to Verofy: {str(e)}"
}
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
+2
View File
@@ -2,3 +2,5 @@ fastapi==0.104.1
uvicorn==0.24.0
python-multipart==0.0.6
pyshp==2.3.1
geopandas>=0.14.0
requests>=2.31.0
+84
View File
@@ -0,0 +1,84 @@
"""
Test upload script for Access Points
Uploads first 10 records to test the field name fix (isLocked -> locked)
"""
import os
import sys
from pathlib import Path
from verofy_uploader import VerofyUploader
# Get credentials from environment
EMAIL = os.getenv("VEROFY_USER")
PASSWORD = os.getenv("VEROFY_PASS")
if not EMAIL or not PASSWORD:
print("❌ Missing environment variables: Please set VEROFY_USER and VEROFY_PASS")
sys.exit(1)
# Configuration
MAP_ID = 16950
TEMP_DIR = Path("../temp")
LIMIT = 10 # Upload only first 10 records
print("=" * 60)
print("VEROFY API ACCESS POINTS TEST")
print("=" * 60)
print(f"Map ID: {MAP_ID}")
print(f"Temp Directory: {TEMP_DIR}")
print(f"Limit: {LIMIT} records")
print(f"Email: {EMAIL}")
print(f"Fix Applied: isLocked → locked")
print("=" * 60)
print()
# Initialize uploader
uploader = VerofyUploader(EMAIL, PASSWORD)
# Authenticate
if not uploader.authenticate():
print("❌ Authentication failed")
sys.exit(1)
results = {
"success": True,
"uploaded": {},
"errors": []
}
# Test access points
print("📤 Testing access_points.shp...")
shapefile_path = TEMP_DIR / "access_points.shp"
if shapefile_path.exists():
try:
count, errors = uploader._upload_access_points(shapefile_path, MAP_ID, LIMIT)
results["uploaded"]["access_points.shp"] = count
if errors:
results["errors"].extend(errors)
print(f"✅ Uploaded {count} access points")
except Exception as e:
error_msg = f"Error uploading access_points.shp: {str(e)}"
print(f"{error_msg}")
results["errors"].append(error_msg)
results["success"] = False
else:
print("⚠️ access_points.shp not found")
# Display results
print()
print("=" * 60)
print("ACCESS POINTS TEST RESULTS")
print("=" * 60)
print(f"Success: {results['success']}")
print()
print("Uploaded counts:")
for shapefile, count in results.get('uploaded', {}).items():
print(f" {shapefile}: {count} records")
if results.get('errors'):
print()
print("Errors encountered:")
for error in results['errors']:
print(f" - {error}")
print("=" * 60)
+93
View File
@@ -0,0 +1,93 @@
"""
Test upload script for Info Layers
Uploads first 10 records from each layer to test the fix
"""
import os
import sys
from pathlib import Path
from verofy_uploader import VerofyUploader
# Get credentials from environment
EMAIL = os.getenv("VEROFY_USER")
PASSWORD = os.getenv("VEROFY_PASS")
if not EMAIL or not PASSWORD:
print("❌ Missing environment variables: Please set VEROFY_USER and VEROFY_PASS")
sys.exit(1)
# Configuration
MAP_ID = 16950
TEMP_DIR = Path("../temp")
LIMIT = 10 # Upload only first 10 records
print("=" * 60)
print("VEROFY API INFO LAYERS TEST")
print("=" * 60)
print(f"Map ID: {MAP_ID}")
print(f"Temp Directory: {TEMP_DIR}")
print(f"Limit: {LIMIT} records per layer")
print(f"Email: {EMAIL}")
print(f"Fixes Applied:")
print(f" - Removed JSON-encoding of data field")
print(f" - Send data as plain array")
print("=" * 60)
print()
# Initialize uploader
uploader = VerofyUploader(EMAIL, PASSWORD)
# Authenticate
if not uploader.authenticate():
print("❌ Authentication failed")
sys.exit(1)
results = {
"success": True,
"uploaded": {},
"errors": []
}
# Test each info layer
info_layers = [
("cabinet_boundaries.shp", uploader._upload_cabinet_boundaries),
("cables.shp", uploader._upload_cables),
("parcels.shp", uploader._upload_parcels)
]
for shapefile_name, upload_method in info_layers:
print(f"📤 Testing {shapefile_name}...")
shapefile_path = TEMP_DIR / shapefile_name
if shapefile_path.exists():
try:
count, errors = upload_method(shapefile_path, MAP_ID, LIMIT)
results["uploaded"][shapefile_name] = count
if errors:
results["errors"].extend(errors)
print(f"✅ Uploaded {count} {shapefile_name.replace('.shp', '')} records")
except Exception as e:
error_msg = f"Error uploading {shapefile_name}: {str(e)}"
print(f"{error_msg}")
results["errors"].append(error_msg)
results["success"] = False
else:
print(f"⚠️ {shapefile_name} not found")
# Display results
print()
print("=" * 60)
print("INFO LAYERS TEST RESULTS")
print("=" * 60)
print(f"Success: {results['success']}")
print()
print("Uploaded counts:")
for shapefile, count in results.get('uploaded', {}).items():
print(f" {shapefile}: {count} records")
if results.get('errors'):
print()
print("Errors encountered:")
for error in results['errors']:
print(f" - {error}")
print("=" * 60)
+84
View File
@@ -0,0 +1,84 @@
"""
Test upload script for Network Elements
Uploads first 10 records to test the reverse-engineered fix
"""
import os
import sys
from pathlib import Path
from verofy_uploader import VerofyUploader
# Get credentials from environment
EMAIL = os.getenv("VEROFY_USER")
PASSWORD = os.getenv("VEROFY_PASS")
if not EMAIL or not PASSWORD:
print("❌ Missing environment variables: Please set VEROFY_USER and VEROFY_PASS")
sys.exit(1)
# Configuration
MAP_ID = 16950
TEMP_DIR = Path("../temp")
LIMIT = 10 # Upload only first 10 records
print("=" * 60)
print("VEROFY API NETWORK ELEMENTS TEST")
print("=" * 60)
print(f"Map ID: {MAP_ID}")
print(f"Temp Directory: {TEMP_DIR}")
print(f"Limit: {LIMIT} records")
print(f"Email: {EMAIL}")
print(f"Fix Applied: Added 'custom': 0 field + debug output")
print("=" * 60)
print()
# Initialize uploader
uploader = VerofyUploader(EMAIL, PASSWORD)
# Authenticate
if not uploader.authenticate():
print("❌ Authentication failed")
sys.exit(1)
results = {
"success": True,
"uploaded": {},
"errors": []
}
# Test network elements
print("📤 Testing network_elements.shp...")
shapefile_path = TEMP_DIR / "network_elements.shp"
if shapefile_path.exists():
try:
count, errors = uploader._upload_network_elements(shapefile_path, MAP_ID, LIMIT)
results["uploaded"]["network_elements.shp"] = count
if errors:
results["errors"].extend(errors)
print(f"✅ Uploaded {count} network elements")
except Exception as e:
error_msg = f"Error uploading network_elements.shp: {str(e)}"
print(f"{error_msg}")
results["errors"].append(error_msg)
results["success"] = False
else:
print("⚠️ network_elements.shp not found")
# Display results
print()
print("=" * 60)
print("NETWORK ELEMENTS TEST RESULTS")
print("=" * 60)
print(f"Success: {results['success']}")
print()
print("Uploaded counts:")
for shapefile, count in results.get('uploaded', {}).items():
print(f" {shapefile}: {count} records")
if results.get('errors'):
print()
print("Errors encountered:")
for error in results['errors']:
print(f" - {error}")
print("=" * 60)
+91
View File
@@ -0,0 +1,91 @@
"""
Test upload script for Permits
Uploads first 10 records to test the reverse-engineered fix
"""
import os
import sys
from pathlib import Path
from verofy_uploader import VerofyUploader
# Get credentials from environment
EMAIL = os.getenv("VEROFY_USER")
PASSWORD = os.getenv("VEROFY_PASS")
if not EMAIL or not PASSWORD:
print("❌ Missing environment variables: Please set VEROFY_USER and VEROFY_PASS")
sys.exit(1)
# Configuration
MAP_ID = 16950
TEMP_DIR = Path("../temp")
LIMIT = 10 # Upload only first 10 records
print("=" * 60)
print("VEROFY API PERMITS TEST")
print("=" * 60)
print(f"Map ID: {MAP_ID}")
print(f"Temp Directory: {TEMP_DIR}")
print(f"Limit: {LIMIT} records")
print(f"Email: {EMAIL}")
print(f"Fixes Applied:")
print(f" - Added mappermitstatusId: 1")
print(f" - Added mappermitentitytypeId: 6")
print(f" - Added mappermitulrtypeId: 3")
print(f" - Added mappermitentitymeetId: 1")
print(f" - Added mappermitrequirementsId: 1")
print(f" - Changed group1 → permitgroup")
print(f" - Wrapped poly in double array [[...]]")
print("=" * 60)
print()
# Initialize uploader
uploader = VerofyUploader(EMAIL, PASSWORD)
# Authenticate
if not uploader.authenticate():
print("❌ Authentication failed")
sys.exit(1)
results = {
"success": True,
"uploaded": {},
"errors": []
}
# Test permits
print("📤 Testing permits.shp...")
shapefile_path = TEMP_DIR / "permits.shp"
if shapefile_path.exists():
try:
count, errors = uploader._upload_permits(shapefile_path, MAP_ID, LIMIT)
results["uploaded"]["permits.shp"] = count
if errors:
results["errors"].extend(errors)
print(f"✅ Uploaded {count} permits")
except Exception as e:
error_msg = f"Error uploading permits.shp: {str(e)}"
print(f"{error_msg}")
results["errors"].append(error_msg)
results["success"] = False
else:
print("⚠️ permits.shp not found")
# Display results
print()
print("=" * 60)
print("PERMITS TEST RESULTS")
print("=" * 60)
print(f"Success: {results['success']}")
print()
print("Uploaded counts:")
for shapefile, count in results.get('uploaded', {}).items():
print(f" {shapefile}: {count} records")
if results.get('errors'):
print()
print("Errors encountered:")
for error in results['errors']:
print(f" - {error}")
print("=" * 60)
+87
View File
@@ -0,0 +1,87 @@
"""
Test upload script for Splicing
Uploads first 10 records to test the reverse-engineered fix
"""
import os
import sys
from pathlib import Path
from verofy_uploader import VerofyUploader
# Get credentials from environment
EMAIL = os.getenv("VEROFY_USER")
PASSWORD = os.getenv("VEROFY_PASS")
if not EMAIL or not PASSWORD:
print("❌ Missing environment variables: Please set VEROFY_USER and VEROFY_PASS")
sys.exit(1)
# Configuration
MAP_ID = 16950
TEMP_DIR = Path("../temp")
LIMIT = 10 # Upload only first 10 records
print("=" * 60)
print("VEROFY API SPLICING TEST")
print("=" * 60)
print(f"Map ID: {MAP_ID}")
print(f"Temp Directory: {TEMP_DIR}")
print(f"Limit: {LIMIT} records")
print(f"Email: {EMAIL}")
print(f"Fixes Applied:")
print(f" - Changed 'name' field to 'aka' field")
print(f" - Fixed endpoint: /map-splice/create")
print(f" - Added debug output")
print("=" * 60)
print()
# Initialize uploader
uploader = VerofyUploader(EMAIL, PASSWORD)
# Authenticate
if not uploader.authenticate():
print("❌ Authentication failed")
sys.exit(1)
results = {
"success": True,
"uploaded": {},
"errors": []
}
# Test splicing
print("📤 Testing splicing.shp...")
shapefile_path = TEMP_DIR / "splicing.shp"
if shapefile_path.exists():
try:
count, errors = uploader._upload_splicing(shapefile_path, MAP_ID, LIMIT)
results["uploaded"]["splicing.shp"] = count
if errors:
results["errors"].extend(errors)
print(f"✅ Uploaded {count} splicing points")
except Exception as e:
error_msg = f"Error uploading splicing.shp: {str(e)}"
print(f"{error_msg}")
results["errors"].append(error_msg)
results["success"] = False
else:
print("⚠️ splicing.shp not found")
# Display results
print()
print("=" * 60)
print("SPLICING TEST RESULTS")
print("=" * 60)
print(f"Success: {results['success']}")
print()
print("Uploaded counts:")
for shapefile, count in results.get('uploaded', {}).items():
print(f" {shapefile}: {count} records")
if results.get('errors'):
print()
print("Errors encountered:")
for error in results['errors']:
print(f" - {error}")
print("=" * 60)
+62
View File
@@ -0,0 +1,62 @@
"""
Test upload script for Verofy API
Uploads first 10 records of each shapefile layer to test the mapping
"""
import os
import sys
from pathlib import Path
from verofy_uploader import upload_to_verofy
# Get credentials from environment
EMAIL = os.getenv("VEROFY_USER")
PASSWORD = os.getenv("VEROFY_PASS")
if not EMAIL or not PASSWORD:
print("❌ Missing environment variables: Please set VEROFY_USER and VEROFY_PASS")
sys.exit(1)
# Configuration
MAP_ID = 16950
TEMP_DIR = "../temp"
LIMIT = 10 # Upload only first 10 records per layer
print("=" * 60)
print("VEROFY API UPLOAD TEST")
print("=" * 60)
print(f"Map ID: {MAP_ID}")
print(f"Temp Directory: {TEMP_DIR}")
print(f"Limit per layer: {LIMIT} records")
print(f"Email: {EMAIL}")
print("=" * 60)
print()
# Run the upload
result = upload_to_verofy(
temp_dir=TEMP_DIR,
map_id=MAP_ID,
email=EMAIL,
password=PASSWORD,
limit=LIMIT
)
# Display results
print()
print("=" * 60)
print("UPLOAD RESULTS")
print("=" * 60)
print(f"Success: {result['success']}")
print()
print("Uploaded counts by layer:")
for shapefile, count in result.get('uploaded', {}).items():
print(f" {shapefile}: {count} records")
if result.get('errors'):
print()
print("Errors encountered:")
for error in result['errors'][:20]: # Show first 20 errors
print(f" - {error}")
if len(result['errors']) > 20:
print(f" ... and {len(result['errors']) - 20} more errors")
print("=" * 60)
File diff suppressed because it is too large Load Diff