78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
# Add project root to sys.path
|
||
|
|
sys.path.append(os.getcwd())
|
||
|
|
|
||
|
|
from lawrisk.utils.env_loader import load_env
|
||
|
|
load_env()
|
||
|
|
|
||
|
|
from lawrisk.services.licensing_repo import filter_permits_advanced
|
||
|
|
import pg8000.dbapi as pg
|
||
|
|
|
||
|
|
def verify():
|
||
|
|
print("Verifying fix...")
|
||
|
|
|
||
|
|
# 1. Fetch '市级' ID
|
||
|
|
conn_params = {
|
||
|
|
"host": os.getenv("LIC_PG_HOST", "172.24.240.1"),
|
||
|
|
"port": int(os.getenv("LIC_PG_PORT", os.getenv("PG_PORT", "5432"))),
|
||
|
|
"user": os.getenv("LIC_PG_USER", os.getenv("PG_USER", "postgres")),
|
||
|
|
"password": os.getenv("LIC_PG_PASSWORD", ""),
|
||
|
|
"database": os.getenv("LIC_PG_DATABASE", "licensing_risks"),
|
||
|
|
}
|
||
|
|
try:
|
||
|
|
conn = pg.connect(**conn_params)
|
||
|
|
cur = conn.cursor()
|
||
|
|
cur.execute("SELECT id FROM regions WHERE name = '市级'")
|
||
|
|
row = cur.fetchone()
|
||
|
|
if not row:
|
||
|
|
print("Region '市级' not found")
|
||
|
|
return
|
||
|
|
region_id = str(row[0])
|
||
|
|
conn.close()
|
||
|
|
|
||
|
|
print(f"Region ID for 市级: {region_id}")
|
||
|
|
|
||
|
|
# 2. Call filter_permits_advanced
|
||
|
|
# Note: filter_permits_advanced returns {permits: [...], pagination: {...}} NOT Dict[str, Any] directly?
|
||
|
|
# Checking signature in licensing_repo.py
|
||
|
|
# It ends with: return {"permits": permits_list, "pagination": {...}} (in my rewrite??)
|
||
|
|
# Wait, I rewrote the SQL but did I verify the return construction?
|
||
|
|
# The function was returning `Dict[str, Any]`.
|
||
|
|
# I need to ensure the return statement was preserved or I added it.
|
||
|
|
# My replace_file_content ONLY Replaced the SQL and Params.
|
||
|
|
# It kept the end of the function?
|
||
|
|
# I replaced from `sql = ...` to `count_params.append(...)`.
|
||
|
|
# The result construction logic `permits_map` and `return` were AFTER.
|
||
|
|
# So it should be fine.
|
||
|
|
|
||
|
|
result = filter_permits_advanced(regions=[region_id], limit=100)
|
||
|
|
|
||
|
|
# In licensing_repo.py, check return.
|
||
|
|
# It returns { "permits": permits_list, "pagination": ... } usually.
|
||
|
|
# Wait, my previous view showed `return permits_list`? No.
|
||
|
|
|
||
|
|
permits = result.get('permits', [])
|
||
|
|
count = len(permits)
|
||
|
|
pagination = result.get('pagination', {})
|
||
|
|
total = pagination.get('total', 0)
|
||
|
|
|
||
|
|
print(f"Filter returned {count} permits.")
|
||
|
|
print(f"Total count according to pagination: {total}")
|
||
|
|
|
||
|
|
# Check for unbound permits
|
||
|
|
unbound = [p for p in permits if not p['themes']]
|
||
|
|
print(f"Unbound permits in result: {len(unbound)}")
|
||
|
|
|
||
|
|
if count == 66:
|
||
|
|
print("SUCCESS: 66 permits returned.")
|
||
|
|
else:
|
||
|
|
print(f"FAILURE: Expected 66, got {count}.")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error: {e}")
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
verify()
|